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 22:41:12 -07:00
parent 670a8259e3 b15ac86686
commit a3cb4e9fb3
4 changed files with 546 additions and 788 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/tools/test_docker_network_config.py
+5 -1
View File
@@ -30,7 +30,11 @@ def test_sibling_container_config_sites_carry_docker_network():
import tools.code_execution_tool as code_execution_tool
import tools.file_tools as file_tools
for module in (terminal_tool, file_tools, code_execution_tool):
# file_tools no longer builds its own container_config: it goes through
# the shared _create_configured_env, which is what carries docker_network.
assert "_create_configured_env(" in inspect.getsource(file_tools)
for module in (terminal_tool, code_execution_tool):
tree = ast.parse(inspect.getsource(module))
sites = 0
for node in ast.walk(tree):
tools/file_tools.py
+198 -355
View File
@@ -17,6 +17,8 @@ import os
import re
import stat
import threading
import time
from contextlib import ExitStack
from pathlib import Path
from agent.file_safety import get_read_block_error
@@ -94,10 +96,8 @@ logger = logging.getLogger(__name__)
_EXPECTED_WRITE_ERRNOS = {errno.EACCES, errno.EPERM, errno.EROFS}
# ---------------------------------------------------------------------------
# Read-size guard. Model-agnostic, so characters proxy tokens: 100K chars is
# ~25-35K tokens across typical tokenisers. Configurable: file_read_max_chars.
# ---------------------------------------------------------------------------
_DEFAULT_MAX_READ_CHARS = 100_000
_max_read_chars_cached: int | None = None
@@ -105,18 +105,14 @@ _max_read_chars_cached: int | None = None
def _get_max_read_chars() -> int:
"""Return ``file_read_max_chars`` from config.yaml (cached per process; default on missing/invalid)."""
global _max_read_chars_cached
if _max_read_chars_cached is not None:
return _max_read_chars_cached
try:
from hermes_cli.config import load_config
cfg = load_config()
val = cfg.get("file_read_max_chars")
if isinstance(val, (int, float)) and val > 0:
_max_read_chars_cached = int(val)
return _max_read_chars_cached
except Exception:
pass
_max_read_chars_cached = _DEFAULT_MAX_READ_CHARS
if _max_read_chars_cached is None:
try:
from hermes_cli.config import load_config
val = load_config().get("file_read_max_chars")
except Exception:
val = None
valid = isinstance(val, (int, float)) and val > 0
_max_read_chars_cached = int(val) if valid else _DEFAULT_MAX_READ_CHARS
return _max_read_chars_cached
@@ -125,10 +121,10 @@ def _truncate_to_char_budget(content: str, max_chars: int) -> tuple[str, int, bo
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 the
accumulation of many lines under the line-count limit (logs, wide CSV).
If not even the first line fits it is clamped mid-line (Python slicing
never splits a code point) so the read is never empty and the cursor advances.
``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.
"""
if len(content) <= max_chars:
return content, (content.count("\n") + 1 if content else 0), False
@@ -142,10 +138,8 @@ def _truncate_to_char_budget(content: str, max_chars: int) -> tuple[str, int, bo
break
kept.append(line)
running += addition
if not kept:
kept.append(lines[0][:max_chars])
return "\n".join(kept), len(kept), True
@@ -154,14 +148,13 @@ def _apply_char_budget(result_dict: dict, content: str, offset: int, total_lines
continuation hint, and return the trimmed text."""
trimmed, lines_kept, _ = _truncate_to_char_budget(content, max_chars)
next_offset = offset + lines_kept
shown_end = offset + lines_kept - 1
result_dict["content"] = trimmed
result_dict["truncated"] = True
result_dict["truncated_by"] = "bytes"
result_dict["next_offset"] = next_offset
result_dict["hint"] = (
f"Output truncated at the {max_chars:,}-char read budget after "
f"{lines_kept} line(s) (showing lines {offset}-{shown_end} of "
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:
@@ -209,18 +202,14 @@ def _file_ops_uses_host_paths(file_ops) -> bool:
return isinstance(env, LocalEnvironment)
# V4A file headers. ``\s*`` after ``***`` mirrors patch_parser's leniency
# (``***Update File:`` with no space parses and applies, so it must be checked).
_V4A_SINGLE_HEADER_RE = re.compile(r'^(\*\*\*\s*(?:Update|Add|Delete)\s+File:\s*)(.+)$', re.MULTILINE)
_V4A_SINGLE_HEADER_OP_RE = re.compile(r'^\*\*\*\s*(Update|Add|Delete)\s+File:\s*(.+)$', re.MULTILINE)
# V4A file headers: group 1 = header prefix, 2 = op, 3 = path. ``\s*`` after
# ``***`` mirrors patch_parser's leniency (``***Update File:`` applies, so it
# must be checked).
_V4A_SINGLE_HEADER_RE = re.compile(r'^(\*\*\*\s*(Update|Add|Delete)\s+File:\s*)(.+)$', re.MULTILINE)
_V4A_MOVE_HEADER_RE = re.compile(r'^(\*\*\*\s*Move\s+File:\s*)(.+?)\s*->\s*(.+)$', re.MULTILINE)
def _rewrite_v4a_patch_paths_for_host(
patch: str,
path_to_resolved: dict,
file_ops,
) -> str:
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.
``patch_tool`` resolves every header against the task's workspace for
@@ -231,17 +220,12 @@ def _rewrite_v4a_patch_paths_for_host(
if not _file_ops_uses_host_paths(file_ops):
return patch
def _resolved_or_original(raw: str) -> str:
def _res(raw: str) -> str:
raw = raw.strip()
return path_to_resolved.get(raw) or raw
patch = _V4A_SINGLE_HEADER_RE.sub(
lambda m: f"{m.group(1)}{_resolved_or_original(m.group(2))}", patch,
)
return _V4A_MOVE_HEADER_RE.sub(
lambda m: f"{m.group(1)}{_resolved_or_original(m.group(2))} -> {_resolved_or_original(m.group(3))}",
patch,
)
patch = _V4A_SINGLE_HEADER_RE.sub(lambda m: f"{m.group(1)}{_res(m.group(3))}", patch)
return _V4A_MOVE_HEADER_RE.sub(lambda m: f"{m.group(1)}{_res(m.group(2))} -> {_res(m.group(3))}", patch)
def _is_blocked_device_path(path: str) -> bool:
@@ -290,41 +274,38 @@ def _is_blocked_device(filepath: str, base_dir: str | Path | None = None) -> boo
return _is_blocked_device_path(resolved)
def _search_result_read_block_error(path: str, task_id: str = "default") -> str | None:
"""Read-safety error for a search result path, resolved against the task cwd
(search backends may return cwd-relative paths; the process cwd can differ)."""
try:
resolved = _resolve_path_for_task(path, task_id)
except (OSError, ValueError, RuntimeError):
return get_read_block_error(path)
return get_read_block_error(str(resolved))
def _filter_read_blocked_search_results(result, task_id: str = "default") -> int:
"""Remove credential/cache/env paths from a SearchResult in-place; return the omitted count."""
"""Remove credential/cache/env paths from a SearchResult in-place; return the omitted count.
Each path is resolved against the task cwd first (search backends may
return cwd-relative paths; the process cwd can differ).
"""
omitted = 0
def _blocked(path: str) -> bool:
def _allowed(path: str) -> bool:
nonlocal omitted
if _search_result_read_block_error(path, task_id):
try:
target = str(_resolve_path_for_task(path, task_id))
except (OSError, ValueError, RuntimeError):
target = path
if get_read_block_error(target):
omitted += 1
return True
return False
return False
return True
if getattr(result, "matches", None):
result.matches = [m for m in result.matches if not _blocked(m.path)]
result.matches = [m for m in result.matches if _allowed(m.path)]
if getattr(result, "files", None):
result.files = [f for f in result.files if not _blocked(f)]
result.files = [f for f in result.files if _allowed(f)]
if getattr(result, "counts", None):
result.counts = {f: c for f, c in result.counts.items() if not _blocked(f)}
result.counts = {f: c for f, c in result.counts.items() if _allowed(f)}
return omitted
def _is_expected_write_exception(exc: Exception) -> bool:
"""Return True for expected write denials that should not hit error logs."""
if isinstance(exc, PermissionError):
return True
return isinstance(exc, OSError) and exc.errno in _EXPECTED_WRITE_ERRNOS
return isinstance(exc, PermissionError) or (
isinstance(exc, OSError) and exc.errno in _EXPECTED_WRITE_ERRNOS)
# ---------------------------------------------------------------------------
@@ -333,41 +314,21 @@ def _is_expected_write_exception(exc: Exception) -> bool:
_file_ops_lock = threading.Lock()
_file_ops_cache: dict = {}
# Per-backend config key that names the sandbox image (overridable per task).
_ENV_IMAGE_KEYS = {
"docker": "docker_image",
"singularity": "singularity_image",
"modal": "modal_image",
"daytona": "daytona_image",
}
_CONTAINER_CONFIG_DEFAULTS = (
("container_cpu", 1),
("container_memory", 5120),
("container_disk", 51200),
("container_persistent", True),
("vercel_runtime", ""),
("docker_volumes", []),
("docker_mount_cwd_to_workspace", False),
("docker_forward_env", []),
("docker_run_as_host_user", False),
("docker_network", True),
)
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 so a file tool that
runs before any terminal command uses the configured backend (docker,
modal, ...) rather than always defaulting to local.
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.
"""
from tools.terminal_tool import (
_CONTAINER_BACKENDS,
_create_environment,
_create_configured_env,
_get_env_config,
_is_container_backend,
_is_unusable_container_cwd,
_resolve_task_host_cwd,
_select_image,
get_session_cwd,
resolve_task_overrides,
)
@@ -375,10 +336,6 @@ def _create_terminal_env_for_file_ops(raw_task_id: str, task_id: str):
config = _get_env_config()
env_type = config["env_type"]
overrides = resolve_task_overrides(raw_task_id)
image_key = _ENV_IMAGE_KEYS.get(env_type)
image = (overrides.get(image_key) or config[image_key]) if image_key else ""
try:
recorded_cwd = get_session_cwd(raw_task_id)
except Exception:
@@ -398,37 +355,11 @@ def _create_terminal_env_for_file_ops(raw_task_id: str, task_id: str):
)
cwd = config["cwd"]
logger.info("Creating new %s environment for task %s...", env_type, task_id[:8])
container_config = None
if _is_container_backend(env_type):
container_config = {key: config.get(key, default) for key, default in _CONTAINER_CONFIG_DEFAULTS}
ssh_config = None
if env_type == "ssh":
ssh_config = {
"host": config.get("ssh_host", ""),
"user": config.get("ssh_user", ""),
"port": config.get("ssh_port", 22),
"key": config.get("ssh_key", ""),
"persistent": config.get("ssh_persistent", False),
}
local_config = None
if env_type == "local":
local_config = {
"persistent": config.get("local_persistent", False),
}
terminal_env = _create_environment(
env_type=env_type,
image=image,
cwd=cwd,
timeout=config["timeout"],
ssh_config=ssh_config,
container_config=container_config,
local_config=local_config,
task_id=task_id,
terminal_env = _create_configured_env(
config, env_type, image=_select_image(env_type, overrides, config), cwd=cwd,
timeout=config["timeout"], task_id=task_id,
host_cwd=_resolve_task_host_cwd(config, raw_task_id),
local_config={"persistent": config.get("local_persistent", False)} if env_type == "local" else None,
)
return env_type, terminal_env
@@ -443,13 +374,10 @@ def _get_file_ops(task_id: str = "default") -> ShellFileOperations:
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,
_active_environments, _env_lock, _last_activity, _start_cleanup_thread,
_creation_locks, _creation_locks_lock, _resolve_container_task_id,
get_session_cwd, record_session_cwd,
)
import time
raw_task_id = task_id or "default"
task_id = _resolve_container_task_id(raw_task_id)
@@ -463,47 +391,34 @@ def _get_file_ops(task_id: str = "default") -> ShellFileOperations:
if task_id in _active_environments:
_last_activity[task_id] = time.time()
return cached
else:
# 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.
old_cwd = getattr(cached, "cwd", None)
if old_cwd:
try:
from tools.terminal_tool import (
get_session_cwd,
record_session_cwd,
)
if get_session_cwd(raw_task_id) is None:
record_session_cwd(raw_task_id, old_cwd)
except Exception:
pass
with _file_ops_lock:
_file_ops_cache.pop(task_id, None)
# 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.
old_cwd = getattr(cached, "cwd", None)
if old_cwd:
try:
if get_session_cwd(raw_task_id) is None:
record_session_cwd(raw_task_id, old_cwd)
except Exception:
pass
with _file_ops_lock:
_file_ops_cache.pop(task_id, None)
with _creation_locks_lock:
if task_id not in _creation_locks:
_creation_locks[task_id] = threading.Lock()
task_lock = _creation_locks[task_id]
task_lock = _creation_locks.setdefault(task_id, threading.Lock())
with task_lock:
# Double-check: another thread may have created it while we waited.
with _env_lock:
if task_id in _active_environments:
terminal_env = _active_environments.get(task_id)
if terminal_env is not None:
_last_activity[task_id] = time.time()
terminal_env = _active_environments[task_id]
else:
terminal_env = None
if terminal_env is None:
env_type, terminal_env = _create_terminal_env_for_file_ops(raw_task_id, task_id)
with _env_lock:
_active_environments[task_id] = terminal_env
_last_activity[task_id] = time.time()
_start_cleanup_thread()
logger.info("%s environment ready for task %s", env_type, task_id[:8])
@@ -540,16 +455,13 @@ def _special_file_kind(path) -> str | None:
flow to the normal read path's own error handling.
"""
try:
st = os.stat(os.fspath(path)) # follows symlinks, matching a real read
mode = os.stat(os.fspath(path)).st_mode # follows symlinks, matching a real read
except OSError:
return None
mode = st.st_mode
if stat.S_ISREG(mode) or stat.S_ISDIR(mode):
return None
for predicate, label in _SPECIAL_FILE_KINDS:
if predicate(mode):
return label
return "a special (non-regular) file"
return next((label for predicate, label in _SPECIAL_FILE_KINDS if predicate(mode)),
"a special (non-regular) file")
def _read_extracted_document(path: str, _resolved, offset: int, limit: int, task_id: str) -> str | None:
@@ -573,17 +485,11 @@ def _read_extracted_document(path: str, _resolved, offset: int, limit: int, task
return None
file_ops = _get_file_ops(task_id)
try:
binary = file_ops.read_file_bytes(
str(_resolved), max_bytes=MAX_DOCUMENT_BYTES
)
binary = file_ops.read_file_bytes(str(_resolved), max_bytes=MAX_DOCUMENT_BYTES)
if binary.error or binary.base64_content is None:
raise ExtractionError(binary.error or "Document bytes unavailable")
document_bytes = base64.b64decode(
binary.base64_content, validate=True
)
extracted_text = extract_document_bytes(
document_bytes, str(_resolved)
)
document_bytes = base64.b64decode(binary.base64_content, validate=True)
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
@@ -592,13 +498,9 @@ def _read_extracted_document(path: str, _resolved, offset: int, limit: int, task
# through — only a specific ExtractionError carries an actionable reason.
_doc_ext = _resolved.suffix.lower()
_binary_doc = _doc_ext in ANYDOC_EXTENSIONS or (
_doc_ext in EXTRACTABLE_EXTENSIONS and _doc_ext != ".ipynb"
)
if (
_binary_doc
and isinstance(exc, ExtractionError)
and not str(exc).startswith("Unsupported document type")
):
_doc_ext in EXTRACTABLE_EXTENSIONS and _doc_ext != ".ipynb")
if (_binary_doc and isinstance(exc, ExtractionError)
and not str(exc).startswith("Unsupported document type")):
return tool_error(
f"Cannot read '{path}' ({_doc_ext}): document "
f"extraction failed — {exc}. Use terminal utilities "
@@ -665,13 +567,12 @@ def _record_successful_read(task_data: dict, task_id: str, path: str, resolved_s
"""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 — it feeds both
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 after our read) 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).
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).
"""
with _read_tracker_lock:
task_data["dedup_hits"].pop(dedup_key, None)
@@ -693,7 +594,6 @@ def _record_successful_read(task_data: dict, task_id: str, path: str, resolved_s
if not partial:
try:
from tools.skill_manager_tool import mark_background_review_skill_read
mark_background_review_skill_read(Path(resolved_str))
except Exception:
logger.debug("background-review read-mark failed", exc_info=True)
@@ -740,9 +640,8 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str =
# The extension is a claim, so this message names only the extension;
# the content-sniffing path names the actual magic-byte type.
if has_binary_extension(str(_resolved)):
_ext = _resolved.suffix.lower()
return tool_error(
f"Cannot read binary file '{path}' ({_ext}). "
f"Cannot read binary file '{path}' ({_resolved.suffix.lower()}). "
"Use vision_analyze for images, or terminal to inspect binary files."
)
@@ -766,7 +665,6 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str =
with _read_tracker_lock:
task_data = _task_data(task_id)
cached_mtime = task_data.get("dedup", {}).get(dedup_key)
if cached_mtime is not None:
try:
if os.path.getmtime(resolved_str) == cached_mtime:
@@ -774,15 +672,13 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str =
except OSError:
pass # stat failed — fall through to full read
file_ops = _get_file_ops(task_id)
result = file_ops.read_file(path, offset, limit)
result = _get_file_ops(task_id).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 block below
# (consecutive-loop detection, dedup bookkeeping) exactly as before the
# cache existed — serving from the cache is the optimization, recording
# must stay side-effect-identical.
# (consecutive-loop detection, dedup bookkeeping) — serving from the
# cache is the optimization, recording must stay side-effect-identical.
_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))
@@ -798,14 +694,12 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str =
result_dict, result.content or "", offset,
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
if (file_size and file_size > _LARGE_FILE_HINT_BYTES
and limit > 200
and result_dict.get("truncated")):
and limit > 200 and result_dict.get("truncated")):
result_dict.setdefault("_hint", (
f"This file is large ({file_size:,} bytes). "
"Consider reading only the section you need with offset and limit "
@@ -814,7 +708,6 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str =
count = _record_successful_read(task_data, task_id, path, resolved_str, offset, limit,
dedup_key, partial=(offset > 1) or bool(result_dict.get("truncated")))
if count >= 4:
return tool_error(
f"BLOCKED: You have read this exact file region {count} times in a row. "
@@ -823,18 +716,21 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str =
path=path,
already_read=count,
)
elif count >= 3:
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."
)
return json.dumps(result_dict, ensure_ascii=False)
except Exception as e:
return tool_error(str(e))
# ---------------------------------------------------------------------------
# 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."""
try:
@@ -843,6 +739,15 @@ 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.
@@ -852,19 +757,40 @@ def _write_precheck_error(paths: list[str], content_paths: list[str], task_id: s
prompt covers every path of a multi-file patch.
"""
for p in paths:
err = _check_sensitive_path(p, task_id)
err = _check_sensitive_path(p, task_id) or (
None if cross_profile else _check_cross_profile_path(p, task_id))
if err:
return err
if not cross_profile:
err = _check_cross_profile_path(p, task_id)
if err:
return err
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))
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),
)
def _edit_warnings(paths: list[str], path_to_resolved: dict, task_id: str) -> list[str]:
"""One pre-edit warning per path, in priority order: cross-agent registry
(names the sibling subagent) > per-task staleness > workspace divergence
(relative path resolving outside the terminal's cwd — the worktree-cwd bug)."""
warnings: list[str] = []
for p in paths:
r = path_to_resolved.get(p)
w = (file_state.check_stale(task_id, r) if r else None) or _check_file_staleness(p, task_id)
if not w and r:
w = _path_resolution_warning(p, Path(r), task_id)
if w:
warnings.append(w)
return warnings
def _note_edited(task_id: str, paths: list[str], path_to_resolved: dict, session_id: str | None) -> None:
"""Post-success bookkeeping: verification-stale marker, then per path refresh
the read stamp (no false staleness on the next edit) and record the write."""
_mark_verification_stale(task_id, [path_to_resolved.get(p) or p for p in paths], session_id=session_id)
for p in paths:
_update_read_timestamp(p, task_id)
if path_to_resolved.get(p):
file_state.note_write(task_id, path_to_resolved[p])
def write_file_tool(path: str, content: str, task_id: str = "default",
@@ -877,71 +803,44 @@ 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.
sensitive_err = _check_sensitive_path(path, task_id)
if sensitive_err:
return tool_error(sensitive_err)
binary_doc_err = _check_binary_document_write(path, task_id)
if binary_doc_err:
return tool_error(binary_doc_err)
protected_err = _check_protected_instruction_write([path], task_id)
if protected_err:
return tool_error(protected_err)
approval_err = _check_approval_required_write([path], task_id)
if approval_err:
return tool_error(approval_err)
if not cross_profile:
cross_warning = _check_cross_profile_path(path, task_id)
if cross_warning:
return tool_error(cross_warning)
if _is_internal_file_tool_content(content):
return tool_error(
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,
)
if err:
return tool_error(err)
try:
# Resolution failure falls back to the legacy unlocked path (the write
# still proceeds; the per-task staleness check still runs).
_resolved = _resolve_or_none(path, task_id)
if _resolved is None:
stale_warning = _check_file_staleness(path, task_id)
file_ops = _get_file_ops(task_id)
result = file_ops.write_file(path, content)
result_dict = result.to_dict()
if stale_warning:
result_dict["_warning"] = stale_warning
if not result_dict.get("error"):
_mark_verification_stale(task_id, [path], session_id=session_id)
_update_read_timestamp(path, task_id)
return json.dumps(result_dict, ensure_ascii=False)
# Per-path lock serializes read→modify→write across concurrent
# subagents; different paths stay fully parallel.
with file_state.lock_path(_resolved):
# Warning priority: cross-agent (names the sibling subagent) >
# per-task staleness > workspace divergence (relative path resolving
# outside the terminal's cwd — the worktree-cwd bug).
cross_warning = file_state.check_stale(task_id, _resolved)
stale_warning = _check_file_staleness(path, task_id)
cwd_warning = _path_resolution_warning(path, Path(_resolved), task_id)
file_ops = _get_file_ops(task_id)
result = file_ops.write_file(_resolved, content)
result_dict = result.to_dict()
effective_warning = cross_warning or stale_warning or cwd_warning
if effective_warning:
result_dict["_warning"] = effective_warning
# Always report the ABSOLUTE path written so a wrong-cwd mismatch is
# visible in the response instead of silently landing elsewhere.
result_dict["resolved_path"] = _resolved
if not result_dict.get("error"):
result_dict["files_modified"] = [_resolved]
_mark_verification_stale(task_id, [_resolved], session_id=session_id)
# Refresh stamps after the write so consecutive edits by the same
# task don't trigger false staleness warnings.
_update_read_timestamp(path, task_id)
if not result_dict.get("error"):
file_state.note_write(task_id, _resolved)
path_to_resolved = {path: _resolved}
with ExitStack() as _lock:
if _resolved:
# Per-path lock serializes read→modify→write across concurrent
# subagents; different paths stay fully parallel.
_lock.enter_context(file_state.lock_path(_resolved))
warnings = _edit_warnings([path], path_to_resolved, task_id)
result_dict = _get_file_ops(task_id).write_file(_resolved or path, content).to_dict()
if warnings:
result_dict["_warning"] = warnings[0]
if _resolved:
# Always report the ABSOLUTE path written so a wrong-cwd mismatch
# is visible in the response instead of silently landing elsewhere.
result_dict["resolved_path"] = _resolved
if result_dict.get("error"):
_update_read_timestamp(path, task_id)
else:
if _resolved:
result_dict["files_modified"] = [_resolved]
_note_edited(task_id, [path], path_to_resolved, session_id)
return json.dumps(result_dict, ensure_ascii=False)
except Exception as e:
if _is_expected_write_exception(e):
@@ -963,7 +862,12 @@ def _collect_v4a_header_paths(patch: str) -> tuple[list[str], list[str]] | str:
"""
from tools.path_security import has_traversal_component
def _reject_v4a_traversal(v4a_path: str) -> str | None:
headers = [(m.group(3), m.group(2) in ("Update", "Add")) for m in _V4A_SINGLE_HEADER_RE.finditer(patch)]
headers += [(g, False) for m in _V4A_MOVE_HEADER_RE.finditer(patch) for g in (m.group(2), m.group(3))]
paths: list[str] = []
content_paths: list[str] = []
for raw, writes_text in headers:
v4a_path = raw.strip()
if has_traversal_component(v4a_path):
return tool_error(
f"V4A patch header contains '..' traversal: {v4a_path!r}. "
@@ -971,25 +875,9 @@ def _collect_v4a_header_paths(patch: str) -> tuple[list[str], list[str]] | str:
"path in '*** Update File:' / '*** Add File:' / "
"'*** Delete File:' / '*** Move File:' headers."
)
return None
paths: list[str] = []
content_paths: list[str] = []
for _m in _V4A_SINGLE_HEADER_OP_RE.finditer(patch):
_op = _m.group(1)
v4a_path = _m.group(2).strip()
_err = _reject_v4a_traversal(v4a_path)
if _err:
return _err
paths.append(v4a_path)
if _op in ("Update", "Add"):
if writes_text:
content_paths.append(v4a_path)
for _m in _V4A_MOVE_HEADER_RE.finditer(patch):
for v4a_path in (_m.group(2).strip(), _m.group(3).strip()):
_err = _reject_v4a_traversal(v4a_path)
if _err:
return _err
paths.append(v4a_path)
return paths, content_paths
@@ -1017,30 +905,11 @@ def patch_tool(mode: str = "replace", path: str = None, old_string: str = None,
# Lock paths in sorted, deduplicated order so concurrent callers with
# overlapping multi-file patches can't deadlock (every caller locks in
# the same order). An unresolvable path is simply not locked.
_path_to_resolved: dict[str, str] = {
_p: _resolve_or_none(_p, task_id) for _p in _paths_to_check
}
_resolved_paths = sorted({_r for _r in _path_to_resolved.values() if _r})
# ExitStack: one lock per path; degenerates to a single lock (or none
# for an unresolvable path) without special-casing.
from contextlib import ExitStack
_path_to_resolved: dict[str, str] = {_p: _resolve_or_none(_p, task_id) for _p in _paths_to_check}
with ExitStack() as _locks:
for _r in _resolved_paths:
for _r in sorted({_r for _r in _path_to_resolved.values() if _r}):
_locks.enter_context(file_state.lock_path(_r))
# Warning priority per path: cross-agent registry (names the
# sibling) > per-task staleness > workspace divergence.
stale_warnings: list[str] = []
for _p in _paths_to_check:
_r = _path_to_resolved[_p]
_cross = file_state.check_stale(task_id, _r) if _r else None
_sw = _cross or _check_file_staleness(_p, task_id)
if not _sw and _r:
_sw = _path_resolution_warning(_p, Path(_r), task_id)
if _sw:
stale_warnings.append(_sw)
stale_warnings = _edit_warnings(_paths_to_check, _path_to_resolved, task_id)
file_ops = _get_file_ops(task_id)
# Hand the shell layer the RESOLVED targets so both layers agree on
@@ -1055,37 +924,23 @@ def patch_tool(mode: str = "replace", path: str = None, old_string: str = None,
elif mode == "patch":
if not patch:
return tool_error("patch content required")
patch_for_ops = _rewrite_v4a_patch_paths_for_host(
patch, _path_to_resolved, file_ops
)
result = file_ops.patch_v4a(patch_for_ops)
result = file_ops.patch_v4a(_rewrite_v4a_patch_paths_for_host(patch, _path_to_resolved, file_ops))
else:
return tool_error(f"Unknown mode: {mode}")
result_dict = result.to_dict()
if stale_warnings:
result_dict["_warning"] = stale_warnings[0] if len(stale_warnings) == 1 else " | ".join(stale_warnings)
# Report the ABSOLUTE path(s) actually patched so a wrong-cwd
# mismatch is visible instead of silently landing elsewhere.
_resolved_modified = [
_path_to_resolved.get(_p) or _p for _p in _paths_to_check
]
result_dict["_warning"] = " | ".join(stale_warnings)
if not result_dict.get("error"):
# Report the ABSOLUTE path(s) actually patched so a wrong-cwd
# mismatch is visible instead of silently landing elsewhere.
_resolved_modified = [_path_to_resolved.get(_p) or _p for _p in _paths_to_check]
result_dict["files_modified"] = _resolved_modified
if len(_resolved_modified) == 1:
result_dict["resolved_path"] = _resolved_modified[0]
_mark_verification_stale(task_id, _resolved_modified, session_id=session_id)
# Refresh stamps for every patched path (no false staleness on
# the next edit) and clear failure counters so a future miss
# starts a fresh count.
for _p in _paths_to_check:
_update_read_timestamp(_p, task_id)
_r = _path_to_resolved.get(_p)
if _r:
file_state.note_write(task_id, _r)
_reset_patch_failures(task_id, [
_r for _r in (_path_to_resolved.get(_p) for _p in _paths_to_check) if _r
])
_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
@@ -1094,9 +949,7 @@ def patch_tool(mode: str = "replace", path: str = None, old_string: str = None,
if result_dict.get("error") and "Could not find" in str(result_dict["error"]):
failure_count = 0
if mode == "replace" and path:
resolved = _path_to_resolved.get(path) or path
failure_count = _record_patch_failure(task_id, resolved)
failure_count = _record_patch_failure(task_id, _path_to_resolved.get(path) or path)
if failure_count >= 3:
result_dict["_hint"] = (
f"This is failure #{failure_count} patching {path!r}. "
@@ -1127,15 +980,7 @@ def search_tool(pattern: str, target: str = "content", path: str = ".",
# Pagination args are part of the key so paging through truncated
# results doesn't trip the repeated-search guard.
search_key = (
"search",
pattern,
target,
str(path),
file_glob or "",
limit,
offset,
)
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(),
@@ -1152,33 +997,31 @@ def search_tool(pattern: str, target: str = "content", path: str = ".",
)
try:
resolved_path = _resolve_path_for_task(path, task_id)
except (OSError, ValueError, RuntimeError):
resolved_path = None
block_error = get_read_block_error(str(resolved_path) if resolved_path else path)
resolved_search_path = str(_resolve_path_for_task(path, task_id))
except (OSError, ValueError, RuntimeError) as exc:
resolved_search_path = path
# A RuntimeError still surfaces as the tool error unless the raw
# path is itself denylisted (that error wins).
if isinstance(exc, RuntimeError) and not get_read_block_error(path):
raise
block_error = get_read_block_error(resolved_search_path)
if block_error:
return tool_error(block_error)
# A missing search root costs two shells (search + parent listing for
# "Similar paths"); cache the miss so a retry skips both.
try:
resolved_search_path = str(_resolve_path_for_task(path, task_id))
except (OSError, ValueError):
resolved_search_path = path
cached_search_nf = _check_not_found_cache("search", resolved_search_path, task_id)
if cached_search_nf is not None:
return cached_search_nf
file_ops = _get_file_ops(task_id)
result = file_ops.search(
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
)
omitted = _filter_read_blocked_search_results(result, task_id)
if hasattr(result, 'matches'):
for m in result.matches:
if hasattr(m, 'content') and m.content:
m.content = redact_sensitive_text(m.content, file_read=True)
for m in getattr(result, "matches", None) or ():
if getattr(m, "content", None):
m.content = redact_sensitive_text(m.content, file_read=True)
result_dict = result.to_dict(densify=True)
if omitted:
tools/fuzzy_match.py
+343 -98
View File
@@ -1,38 +1,19 @@
#!/usr/bin/env python3
"""Fuzzy find-and-replace for LLM-generated edits.
Tries an ordered chain of increasingly permissive matching strategies (see
:mod:`tools.fuzzy_match_strategies`) so whitespace, indentation, escaping and
Unicode drift in tool-call arguments still land on the intended region.
An ordered chain of increasingly permissive strategies (``STRATEGIES``) lets
whitespace, indentation, escaping and Unicode drift in tool-call arguments
still land on the intended region::
new_content, match_count, strategy, error = fuzzy_find_and_replace(
content, old_string, new_string, replace_all=False)
"""
import re
from difflib import SequenceMatcher
from typing import List, Optional, Tuple
from typing import Callable, Optional
from tools.fuzzy_match_strategies import ( # noqa: F401 — re-exported names
SIMILARITY_STRATEGIES,
STRATEGIES,
UNICODE_MAP,
_build_orig_to_norm_map,
_calculate_line_positions,
_invert_norm_map,
_map_normalized_positions,
_map_positions_norm_to_orig,
_norm_end_to_orig,
_strategy_block_anchor,
_strategy_context_aware,
_strategy_escape_normalized,
_strategy_exact,
_strategy_indentation_flexible,
_strategy_line_trimmed,
_strategy_trimmed_boundary,
_strategy_unicode_normalized,
_strategy_whitespace_normalized,
_unicode_normalize,
)
Span = tuple[int, int]
IDENTICAL_STRINGS_ERROR = (
"No edit was applied because old_string and new_string are identical. "
@@ -40,6 +21,310 @@ IDENTICAL_STRINGS_ERROR = (
"replacement text in new_string."
)
UNICODE_MAP = {
"\u201c": '"', "\u201d": '"', # smart double quotes
"\u2018": "'", "\u2019": "'", # smart single quotes
"\u2014": "--", "\u2013": "-", # em/en dashes
"\u2026": "...", "\u00a0": " ", # ellipsis and non-breaking space
"\u2212": "-", # typographic minus (math/scientific docs)
# Space-separator family (Zs) beyond NBSP: otherwise such files miss every
# precise strategy and fall to the similarity fallback (wrong-region risk).
"\u2000": " ", "\u2001": " ", "\u2002": " ", "\u2003": " ",
"\u2004": " ", "\u2005": " ", "\u2006": " ", "\u2007": " ",
"\u2008": " ", "\u2009": " ", "\u200a": " ", "\u202f": " ",
"\u205f": " ", "\u3000": " ",
}
def _unicode_normalize(text: str) -> str:
"""Map typographic Unicode variants to ASCII equivalents."""
for char, repl in UNICODE_MAP.items():
text = text.replace(char, repl)
return text
# ── Position helpers ─────────────────────────────────────────────────────
def _calculate_line_positions(content_lines: list[str], start_line: int,
end_line: int, content_length: int) -> Span:
"""Character span covering ``content_lines[start_line:end_line]`` (end exclusive)."""
start_pos = sum(len(line) + 1 for line in content_lines[:start_line])
end_pos = sum(len(line) + 1 for line in content_lines[:end_line]) - 1
return start_pos, min(content_length, end_pos)
def _window_spans(content: str, content_lines: list[str], n: int,
accept: Callable[[int], bool]) -> list[Span]:
"""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)
]
def _match_transformed_lines(content: str, pattern: str,
transform: Callable[[list[str]], list[str]]) -> list[Span]:
"""Match ``pattern`` against ``content`` after applying ``transform`` to each line block."""
content_lines = content.split('\n')
pattern_norm = transform(pattern.split('\n'))
n = len(pattern_norm)
return _window_spans(content, content_lines, n,
lambda i: transform(content_lines[i:i + n]) == pattern_norm)
def _strip_boundary(lines: list[str]) -> list[str]:
"""Strip whitespace on the first and last lines only."""
lines = list(lines)
lines[0] = lines[0].strip()
if len(lines) > 1:
lines[-1] = lines[-1].strip()
return lines
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.
"""
result: list[int] = []
norm_pos = 0
for char in original:
result.append(norm_pos)
repl = UNICODE_MAP.get(char)
norm_pos += len(repl) if repl is not None else 1
result.append(norm_pos)
return result
def _invert_norm_map(orig_to_norm: list[int]) -> dict[int, int]:
"""norm_pos -> first original position mapping to it."""
inverted: dict[int, int] = {}
for orig_pos, norm_pos in enumerate(orig_to_norm[:-1]):
inverted.setdefault(norm_pos, orig_pos)
return inverted
def _norm_end_to_orig(orig_to_norm: list[int], orig_start: int, norm_end: int) -> int:
"""Walk from ``orig_start`` until the mapped position reaches ``norm_end``."""
orig_len = len(orig_to_norm) - 1
orig_end = orig_start
while orig_end < orig_len and orig_to_norm[orig_end] < norm_end:
orig_end += 1
return orig_end
def _map_positions_norm_to_orig(orig_to_norm: list[int], norm_matches: list[Span]) -> list[Span]:
"""Convert spans in the normalised string to original-string spans."""
norm_to_orig_start = _invert_norm_map(orig_to_norm)
results: list[Span] = []
for norm_start, norm_end in norm_matches:
if norm_start in norm_to_orig_start:
orig_start = norm_to_orig_start[norm_start]
results.append((orig_start, _norm_end_to_orig(orig_to_norm, orig_start, norm_end)))
return results
def _map_normalized_positions(original: str, normalized: str,
normalized_matches: list[Span]) -> list[Span]:
"""Best-effort span mapping for ``[ \\t]+`` -> ``' '`` whitespace collapsing."""
orig_to_norm = [] # orig_to_norm[i] = position in normalized
orig_idx = norm_idx = 0
while orig_idx < len(original) and norm_idx < len(normalized):
if original[orig_idx] == normalized[norm_idx]:
orig_to_norm.append(norm_idx)
orig_idx += 1
norm_idx += 1
elif original[orig_idx] in ' \t' and normalized[norm_idx] == ' ':
# Collapsed run: advance norm_idx only once the run is consumed.
orig_to_norm.append(norm_idx)
orig_idx += 1
if orig_idx < len(original) and original[orig_idx] not in ' \t':
norm_idx += 1
else:
# Extra whitespace in original, or a mismatch normalization should
# never produce — either way, pin to the current norm_idx.
orig_to_norm.append(norm_idx)
orig_idx += 1
orig_to_norm.extend([len(normalized)] * (len(original) - orig_idx))
norm_to_orig_start = {}
norm_to_orig_end = {}
for orig_pos, norm_pos in enumerate(orig_to_norm):
norm_to_orig_start.setdefault(norm_pos, orig_pos)
norm_to_orig_end[norm_pos] = orig_pos
original_matches = []
for norm_start, norm_end in normalized_matches:
if norm_start in norm_to_orig_start:
orig_start = norm_to_orig_start[norm_start]
else:
orig_start = min(i for i, n in enumerate(orig_to_norm) if n >= norm_start)
if norm_end - 1 in norm_to_orig_end:
orig_end = norm_to_orig_end[norm_end - 1] + 1
else:
orig_end = orig_start + (norm_end - norm_start)
# Absorb trailing collapsed whitespace only when the normalized match
# itself ended in a space; otherwise the first whitespace after the
# match is a word boundary that must survive.
if norm_end < len(normalized) and normalized[norm_end - 1] == ' ':
while orig_end < len(original) and original[orig_end] in ' \t':
orig_end += 1
original_matches.append((orig_start, min(orig_end, len(original))))
return original_matches
# ── Strategies ───────────────────────────────────────────────────────────
# Each takes ``(content, pattern)`` and returns ``(start, end)`` spans in the
# 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
def _strategy_line_trimmed(content: str, pattern: str) -> list[Span]:
"""Strategy 2: strip each line before comparing."""
return _match_transformed_lines(content, pattern, lambda ls: [l.strip() for l in ls])
def _strategy_whitespace_normalized(content: str, pattern: str) -> list[Span]:
"""Strategy 3: collapse runs of spaces/tabs to a single space."""
def normalize(s):
return re.sub(r'[ \t]+', ' ', s)
content_normalized = normalize(content)
matches_in_normalized = _strategy_exact(content_normalized, normalize(pattern))
if not matches_in_normalized:
return []
return _map_normalized_positions(content, content_normalized, matches_in_normalized)
def _strategy_indentation_flexible(content: str, pattern: str) -> list[Span]:
"""Strategy 4: ignore leading indentation entirely."""
return _match_transformed_lines(content, pattern, lambda ls: [l.lstrip() for l in ls])
def _strategy_escape_normalized(content: str, pattern: str) -> list[Span]:
"""Strategy 5: treat literal ``\\n``/``\\t``/``\\r`` in the pattern as control chars."""
pattern_unescaped = pattern.replace('\\n', '\n').replace('\\t', '\t').replace('\\r', '\r')
if pattern_unescaped == pattern:
return []
return _strategy_exact(content, pattern_unescaped)
def _strategy_trimmed_boundary(content: str, pattern: str) -> list[Span]:
"""Strategy 6: strip whitespace on the first and last lines only."""
return _match_transformed_lines(content, pattern, _strip_boundary)
def _strategy_unicode_normalized(content: str, pattern: str) -> list[Span]:
"""Strategy 7: exact/line-trimmed match after Unicode->ASCII normalisation of both sides."""
norm_pattern = _unicode_normalize(pattern)
norm_content = _unicode_normalize(content)
if norm_content == content and norm_pattern == pattern:
return []
norm_matches = (_strategy_exact(norm_content, norm_pattern)
or _strategy_line_trimmed(norm_content, norm_pattern))
if not norm_matches:
return []
return _map_positions_norm_to_orig(_build_orig_to_norm_map(content), norm_matches)
def _strategy_block_anchor(content: str, pattern: str) -> list[Span]:
"""Strategy 8: anchor on first+last lines, similarity-score the middle."""
pattern_lines = _unicode_normalize(pattern).split('\n')
if len(pattern_lines) < 2:
return []
first_line = pattern_lines[0].strip()
last_line = pattern_lines[-1].strip()
n = len(pattern_lines)
# Match on normalized lines; compute offsets from the ORIGINAL lines so
# multi-char expansions (em-dash -> '--') don't shift positions.
norm_content_lines = _unicode_normalize(content).split('\n')
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
}
# 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])
def similar(i: int) -> bool:
if i not in potential_matches:
return False
if n <= 2:
return True
content_middle = '\n'.join(norm_content_lines[i + 1:i + n - 1])
return SequenceMatcher(None, content_middle, pattern_middle).ratio() >= threshold
return _window_spans(content, content.split('\n'), n, similar)
def _strategy_context_aware(content: str, pattern: str) -> list[Span]:
"""Strategy 9 (last resort): anchored per-line similarity, every non-blank line >= 0.80.
The first/last-line anchor pre-filter keeps a miss from being an
O(file x pattern) scan; the all-lines requirement stops one coincidental
line match from replacing an unrelated block.
"""
pattern_lines = pattern.split('\n')
content_lines = content.split('\n')
n = len(pattern_lines)
if n > len(content_lines):
return []
first_pat = pattern_lines[0].strip()
last_pat = pattern_lines[-1].strip()
def _sim(a: str, b: str) -> float:
return 1.0 if a == b else SequenceMatcher(None, a, b).ratio()
def accept(i: int) -> bool:
block_lines = content_lines[i:i + n]
if _sim(first_pat, block_lines[0].strip()) < 0.80:
return False
if _sim(last_pat, block_lines[-1].strip()) < 0.80:
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)
)
return _window_spans(content, content_lines, n, accept)
# Ordered chain: precise strategies first, similarity-based last.
STRATEGIES: list[tuple[str, Callable[[str, str], list[Span]]]] = [
("exact", _strategy_exact),
("line_trimmed", _strategy_line_trimmed),
("whitespace_normalized", _strategy_whitespace_normalized),
("indentation_flexible", _strategy_indentation_flexible),
("escape_normalized", _strategy_escape_normalized),
("trimmed_boundary", _strategy_trimmed_boundary),
("unicode_normalized", _strategy_unicode_normalized),
("block_anchor", _strategy_block_anchor),
("context_aware", _strategy_context_aware),
]
# Matches from these only *approximately* resemble old_string — fine for one
# unique replacement, never safe under replace_all.
SIMILARITY_STRATEGIES = frozenset({"block_anchor", "context_aware"})
# ── Orchestrator ─────────────────────────────────────────────────────────
def is_already_applied(content: str, old_string: str, new_string: str) -> bool:
"""True when the requested edit is already present (re-sent edit -> success-shaped no-op).
@@ -47,21 +332,16 @@ def is_already_applied(content: str, old_string: str, new_string: str) -> bool:
Conservative: new_string must be non-trivial (>= 8 chars stripped) and
appear EXACTLY; when it differs from old_string, old_string must be gone.
"""
if not new_string or len(new_string.strip()) < 8:
if not new_string or len(new_string.strip()) < 8 or new_string not in content:
return False
if new_string not in content:
return False
if old_string == new_string:
return True
return old_string not in content
return old_string == new_string or old_string not in content
def _matched_regions(content: str, matches: List[Tuple[int, int]]) -> str:
def _matched_regions(content: str, matches: list[Span]) -> str:
return "".join(content[start:end] for start, end in matches)
def _format_match_locations(content: str, matches: List[Tuple[int, int]],
cap: int = 5) -> str:
def _format_match_locations(content: str, matches: list[Span], cap: int = 5) -> str:
"""Render up to ``cap`` match positions as 'L<line>: <snippet>' rows."""
rows = []
for start, _end in matches[:cap]:
@@ -81,7 +361,7 @@ def _format_match_locations(content: str, matches: List[Tuple[int, int]],
def fuzzy_find_and_replace(content: str, old_string: str, new_string: str,
replace_all: bool = False) -> Tuple[str, int, Optional[str], Optional[str]]:
replace_all: bool = False) -> tuple[str, int, Optional[str], Optional[str]]:
"""Find and replace via the strategy chain.
Returns ``(new_content, match_count, strategy_name, error)``; on failure
@@ -125,9 +405,7 @@ def fuzzy_find_and_replace(content: str, old_string: str, new_string: str,
effective_new = _maybe_unescape_new_string(new_string, content, matches)
if strategy_name == "unicode_normalized":
effective_new = _preserve_unicode_in_replacement(
content, matches, old_string, effective_new,
)
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,
@@ -139,7 +417,7 @@ def fuzzy_find_and_replace(content: str, old_string: str, new_string: str,
# ── Escape-drift guards ──────────────────────────────────────────────────
def _detect_escape_drift(content: str, matches: List[Tuple[int, int]],
def _detect_escape_drift(content: str, matches: list[Span],
old_string: str, new_string: str) -> Optional[str]:
"""Error string when new_string carries tool-call escape artifacts, else None.
@@ -168,19 +446,9 @@ def _detect_escape_drift(content: str, matches: List[Tuple[int, int]],
return _detect_backslash_doubling(matched_regions, old_string, new_string)
def _backslash_runs(s: str) -> List[int]:
def _backslash_runs(s: str) -> list[int]:
"""Lengths of maximal backslash runs in ``s``, in order."""
runs: List[int] = []
n = 0
for ch in s:
if ch == "\\":
n += 1
elif n:
runs.append(n)
n = 0
if n:
runs.append(n)
return runs
return [len(run) for run in re.findall(r"\\+", s)]
def _detect_backslash_doubling(matched_regions: str, old_string: str,
@@ -195,15 +463,11 @@ def _detect_backslash_doubling(matched_regions: str, old_string: str,
"""
old_runs = _backslash_runs(old_string)
file_runs = _backslash_runs(matched_regions)
if not old_runs or not file_runs or len(old_runs) != len(file_runs):
return None
if old_runs == file_runs:
return None
if any(o != f * 2 for o, f in zip(old_runs, file_runs)):
return None
if not (any(f >= 2 for f in file_runs) or len(file_runs) >= 2):
return None
if _backslash_runs(new_string) == file_runs:
if (not old_runs or not file_runs or len(old_runs) != len(file_runs)
or old_runs == file_runs
or any(o != f * 2 for o, f in zip(old_runs, file_runs))
or not (any(f >= 2 for f in file_runs) or len(file_runs) >= 2)
or _backslash_runs(new_string) == file_runs):
return None
return (
"Escape-drift detected: every backslash run in old_string is exactly "
@@ -216,8 +480,7 @@ def _detect_backslash_doubling(matched_regions: str, old_string: str,
)
def _maybe_unescape_new_string(new_string: str, content: str,
matches: List[Tuple[int, int]]) -> str:
def _maybe_unescape_new_string(new_string: str, content: str, matches: list[Span]) -> str:
"""Convert literal ``\\t``/``\\r`` in new_string to control chars, per sequence,
only when the matched file region already contains the real control char.
@@ -229,12 +492,10 @@ def _maybe_unescape_new_string(new_string: str, content: str,
if "\\t" not in new_string and "\\r" not in new_string:
return new_string
matched_regions = _matched_regions(content, matches)
out = new_string
if "\\t" in out and "\t" in matched_regions:
out = out.replace("\\t", "\t")
if "\\r" in out and "\r" in matched_regions:
out = out.replace("\\r", "\r")
return out
for literal, control in (("\\t", "\t"), ("\\r", "\r")):
if literal in new_string and control in matched_regions:
new_string = new_string.replace(literal, control)
return new_string
# ── Replacement shaping ──────────────────────────────────────────────────
@@ -244,10 +505,7 @@ def _leading_whitespace(line: str) -> str:
def _first_meaningful_line(text: str) -> Optional[str]:
for line in text.split("\n"):
if line.strip():
return line
return None
return next((line for line in text.split("\n") if line.strip()), None)
def _reindent_replacement(file_region: str, old_string: str, new_string: str) -> str:
@@ -269,7 +527,7 @@ def _reindent_replacement(file_region: str, old_string: str, new_string: str) ->
if old_indent == file_indent:
return new_string
out_lines: List[str] = []
out_lines: list[str] = []
for line in new_string.split("\n"):
if not line.strip():
out_lines.append(line)
@@ -280,10 +538,8 @@ def _reindent_replacement(file_region: str, old_string: str, new_string: str) ->
return "\n".join(out_lines)
def _preserve_unicode_in_replacement(
content: str, matches: List[Tuple[int, int]],
old_string: str, new_string: str,
) -> str:
def _preserve_unicode_in_replacement(content: str, matches: list[Span],
old_string: str, new_string: str) -> str:
"""Apply only the old->new edits onto the file's original (Unicode) text.
After a unicode_normalized match, writing the LLM's ASCII new_string
@@ -299,7 +555,7 @@ def _preserve_unicode_in_replacement(
file_orig_to_norm = _build_orig_to_norm_map(file_region)
file_norm_to_orig = _invert_norm_map(file_orig_to_norm)
result_parts: List[str] = []
result_parts: list[str] = []
for tag, i1, i2, j1, j2 in SequenceMatcher(None, norm_old, new_string).get_opcodes():
if tag == "equal":
orig_start = file_norm_to_orig.get(i1, 0)
@@ -310,7 +566,7 @@ def _preserve_unicode_in_replacement(
return "".join(result_parts)
def _apply_replacements(content: str, matches: List[Tuple[int, int]],
def _apply_replacements(content: str, matches: list[Span],
new_string: str, old_string: Optional[str] = None) -> str:
"""Splice ``new_string`` over each span (end-to-start so offsets stay valid).
@@ -330,12 +586,9 @@ def _apply_replacements(content: str, matches: List[Tuple[int, int]],
def _visualize_whitespace(line: str) -> str:
"""Render the leading whitespace run visibly (→ = tab, · = space)."""
i = 0
prefix = []
while i < len(line) and line[i] in (" ", "\t"):
prefix.append("→" if line[i] == "\t" else "·")
i += 1
return "".join(prefix) + line[i:]
stripped = line.lstrip(" \t")
prefix = line[:len(line) - len(stripped)]
return prefix.replace("\t", "→").replace(" ", "·") + stripped
def find_closest_lines(old_string: str, content: str, context_lines: int = 2, max_results: int = 3) -> str:
@@ -347,12 +600,9 @@ def find_closest_lines(old_string: str, content: str, context_lines: int = 2, ma
if not old_lines or not content_lines:
return ""
anchor = old_lines[0].strip()
anchor = old_lines[0].strip() or next((l.strip() for l in old_lines if l.strip()), "")
if not anchor:
candidates = [l.strip() for l in old_lines if l.strip()]
if not candidates:
return ""
anchor = candidates[0]
return ""
scored = []
for i, line in enumerate(content_lines):
@@ -376,8 +626,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 ""
@@ -403,11 +652,7 @@ def format_no_match_hint(error: Optional[str], match_count: int,
Ambiguous-match, escape-drift and identical-strings errors also have
``match_count == 0`` but a hint would mislead there.
"""
if match_count != 0:
return ""
if not error or not error.startswith("Could not find"):
if match_count != 0 or not error or not error.startswith("Could not find"):
return ""
hint = find_closest_lines(old_string, content)
if not hint:
return ""
return "\n\nDid you mean one of these sections?\n" + hint
return "\n\nDid you mean one of these sections?\n" + hint if hint else ""
tools/fuzzy_match_strategies.py
-334
View File
@@ -1,334 +0,0 @@
"""Matching strategies for :mod:`tools.fuzzy_match`.
Each strategy takes ``(content, pattern)`` and returns a list of
``(start, end)`` spans in the ORIGINAL content. ``STRATEGIES`` is the ordered
chain the orchestrator tries; ``SIMILARITY_STRATEGIES`` names the ones whose
matches are approximate and therefore unsafe under ``replace_all``.
"""
import re
from difflib import SequenceMatcher
from typing import Callable, Dict, List, Tuple
UNICODE_MAP = {
"\u201c": '"', "\u201d": '"', # smart double quotes
"\u2018": "'", "\u2019": "'", # smart single quotes
"\u2014": "--", "\u2013": "-", # em/en dashes
"\u2026": "...", "\u00a0": " ", # ellipsis and non-breaking space
"\u2212": "-", # typographic minus (math/scientific docs)
# Space-separator family (Zs) beyond NBSP: files with typographic spacing
# otherwise miss every precise strategy and fall through to the
# similarity fallback, which can pick the wrong region.
"\u2000": " ", "\u2001": " ", "\u2002": " ", "\u2003": " ",
"\u2004": " ", "\u2005": " ", "\u2006": " ", "\u2007": " ",
"\u2008": " ", "\u2009": " ", "\u200a": " ", "\u202f": " ",
"\u205f": " ", "\u3000": " ",
}
Span = Tuple[int, int]
def _unicode_normalize(text: str) -> str:
"""Map typographic Unicode variants to ASCII equivalents."""
for char, repl in UNICODE_MAP.items():
text = text.replace(char, repl)
return text
# ── Position helpers ─────────────────────────────────────────────────────
def _calculate_line_positions(content_lines: List[str], start_line: int,
end_line: int, content_length: int) -> Span:
"""Character span covering ``content_lines[start_line:end_line]`` (end exclusive)."""
start_pos = sum(len(line) + 1 for line in content_lines[:start_line])
end_pos = sum(len(line) + 1 for line in content_lines[:end_line]) - 1
return start_pos, min(content_length, end_pos)
def _match_transformed_lines(content: str, pattern: str,
transform: Callable[[str], str]) -> List[Span]:
"""Match ``pattern`` against ``content`` after applying ``transform`` per line."""
content_lines = content.split('\n')
norm_lines = [transform(line) for line in content_lines]
pattern_norm = '\n'.join(transform(line) for line in pattern.split('\n'))
n = pattern_norm.count('\n') + 1
matches = []
for i in range(len(norm_lines) - n + 1):
if '\n'.join(norm_lines[i:i + n]) == pattern_norm:
matches.append(_calculate_line_positions(content_lines, i, i + n, len(content)))
return matches
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.
"""
result: List[int] = []
norm_pos = 0
for char in original:
result.append(norm_pos)
repl = UNICODE_MAP.get(char)
norm_pos += len(repl) if repl is not None else 1
result.append(norm_pos)
return result
def _invert_norm_map(orig_to_norm: List[int]) -> Dict[int, int]:
"""norm_pos -> first original position mapping to it."""
inverted: Dict[int, int] = {}
for orig_pos, norm_pos in enumerate(orig_to_norm[:-1]):
if norm_pos not in inverted:
inverted[norm_pos] = orig_pos
return inverted
def _norm_end_to_orig(orig_to_norm: List[int], orig_start: int, norm_end: int) -> int:
"""Walk from ``orig_start`` until the mapped position reaches ``norm_end``."""
orig_len = len(orig_to_norm) - 1
orig_end = orig_start
while orig_end < orig_len and orig_to_norm[orig_end] < norm_end:
orig_end += 1
return orig_end
def _map_positions_norm_to_orig(orig_to_norm: List[int],
norm_matches: List[Span]) -> List[Span]:
"""Convert spans in the normalised string to original-string spans."""
norm_to_orig_start = _invert_norm_map(orig_to_norm)
results: List[Span] = []
for norm_start, norm_end in norm_matches:
if norm_start not in norm_to_orig_start:
continue
orig_start = norm_to_orig_start[norm_start]
results.append((orig_start, _norm_end_to_orig(orig_to_norm, orig_start, norm_end)))
return results
def _map_normalized_positions(original: str, normalized: str,
normalized_matches: List[Span]) -> List[Span]:
"""Best-effort span mapping for ``[ \\t]+`` -> ``' '`` whitespace collapsing."""
orig_to_norm = [] # orig_to_norm[i] = position in normalized
orig_idx = norm_idx = 0
while orig_idx < len(original) and norm_idx < len(normalized):
if original[orig_idx] == normalized[norm_idx]:
orig_to_norm.append(norm_idx)
orig_idx += 1
norm_idx += 1
elif original[orig_idx] in ' \t' and normalized[norm_idx] == ' ':
# Collapsed run: advance norm_idx only once the run is consumed.
orig_to_norm.append(norm_idx)
orig_idx += 1
if orig_idx < len(original) and original[orig_idx] not in ' \t':
norm_idx += 1
else:
# Extra whitespace in original, or a mismatch that normalization
# should never produce — either way, pin to the current norm_idx.
orig_to_norm.append(norm_idx)
orig_idx += 1
while orig_idx < len(original):
orig_to_norm.append(len(normalized))
orig_idx += 1
norm_to_orig_start = {}
norm_to_orig_end = {}
for orig_pos, norm_pos in enumerate(orig_to_norm):
if norm_pos not in norm_to_orig_start:
norm_to_orig_start[norm_pos] = orig_pos
norm_to_orig_end[norm_pos] = orig_pos
original_matches = []
for norm_start, norm_end in normalized_matches:
if norm_start in norm_to_orig_start:
orig_start = norm_to_orig_start[norm_start]
else:
orig_start = min(i for i, n in enumerate(orig_to_norm) if n >= norm_start)
if norm_end - 1 in norm_to_orig_end:
orig_end = norm_to_orig_end[norm_end - 1] + 1
else:
orig_end = orig_start + (norm_end - norm_start)
# Absorb trailing collapsed whitespace only when the normalized match
# itself ended in a space; otherwise the first whitespace after the
# match is a word boundary that must survive (#52491).
if norm_end < len(normalized) and normalized[norm_end - 1] == ' ':
while orig_end < len(original) and original[orig_end] in ' \t':
orig_end += 1
original_matches.append((orig_start, min(orig_end, len(original))))
return original_matches
# ── Strategies ───────────────────────────────────────────────────────────
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
def _strategy_line_trimmed(content: str, pattern: str) -> List[Span]:
"""Strategy 2: strip each line before comparing."""
return _match_transformed_lines(content, pattern, str.strip)
def _strategy_whitespace_normalized(content: str, pattern: str) -> List[Span]:
"""Strategy 3: collapse runs of spaces/tabs to a single space."""
def normalize(s):
return re.sub(r'[ \t]+', ' ', s)
content_normalized = normalize(content)
matches_in_normalized = _strategy_exact(content_normalized, normalize(pattern))
if not matches_in_normalized:
return []
return _map_normalized_positions(content, content_normalized, matches_in_normalized)
def _strategy_indentation_flexible(content: str, pattern: str) -> List[Span]:
"""Strategy 4: ignore leading indentation entirely."""
return _match_transformed_lines(content, pattern, str.lstrip)
def _strategy_escape_normalized(content: str, pattern: str) -> List[Span]:
"""Strategy 5: treat literal ``\\n``/``\\t``/``\\r`` in the pattern as control chars."""
pattern_unescaped = pattern.replace('\\n', '\n').replace('\\t', '\t').replace('\\r', '\r')
if pattern_unescaped == pattern:
return []
return _strategy_exact(content, pattern_unescaped)
def _strategy_trimmed_boundary(content: str, pattern: str) -> List[Span]:
"""Strategy 6: strip whitespace on the first and last lines only."""
pattern_lines = pattern.split('\n')
pattern_lines[0] = pattern_lines[0].strip()
if len(pattern_lines) > 1:
pattern_lines[-1] = pattern_lines[-1].strip()
modified_pattern = '\n'.join(pattern_lines)
n = len(pattern_lines)
content_lines = content.split('\n')
matches = []
for i in range(len(content_lines) - n + 1):
check_lines = content_lines[i:i + n]
check_lines[0] = check_lines[0].strip()
if n > 1:
check_lines[-1] = check_lines[-1].strip()
if '\n'.join(check_lines) == modified_pattern:
matches.append(_calculate_line_positions(content_lines, i, i + n, len(content)))
return matches
def _strategy_unicode_normalized(content: str, pattern: str) -> List[Span]:
"""Strategy 7: exact/line-trimmed match after Unicode->ASCII normalisation of both sides."""
norm_pattern = _unicode_normalize(pattern)
norm_content = _unicode_normalize(content)
if norm_content == content and norm_pattern == pattern:
return []
norm_matches = _strategy_exact(norm_content, norm_pattern)
if not norm_matches:
norm_matches = _strategy_line_trimmed(norm_content, norm_pattern)
if not norm_matches:
return []
return _map_positions_norm_to_orig(_build_orig_to_norm_map(content), norm_matches)
def _strategy_block_anchor(content: str, pattern: str) -> List[Span]:
"""Strategy 8: anchor on first+last lines, similarity-score the middle."""
pattern_lines = _unicode_normalize(pattern).split('\n')
if len(pattern_lines) < 2:
return []
first_line = pattern_lines[0].strip()
last_line = pattern_lines[-1].strip()
n = len(pattern_lines)
# Match on normalized lines; compute offsets from the ORIGINAL lines so
# multi-char expansions (em-dash -> '--') don't shift positions.
norm_content_lines = _unicode_normalize(content).split('\n')
orig_content_lines = content.split('\n')
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
]
# 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
matches = []
for i in potential_matches:
if n <= 2:
similarity = 1.0
else:
content_middle = '\n'.join(norm_content_lines[i + 1:i + n - 1])
pattern_middle = '\n'.join(pattern_lines[1:-1])
similarity = SequenceMatcher(None, content_middle, pattern_middle).ratio()
if similarity >= threshold:
matches.append(_calculate_line_positions(orig_content_lines, i, i + n, len(content)))
return matches
def _strategy_context_aware(content: str, pattern: str) -> List[Span]:
"""Strategy 9 (last resort): anchored per-line similarity, every non-blank line >= 0.80.
The first/last-line anchor pre-filter keeps a miss from being an
O(file x pattern) scan; the all-lines requirement stops one coincidental
line match from replacing an unrelated block.
"""
pattern_lines = pattern.split('\n')
content_lines = content.split('\n')
n = len(pattern_lines)
if n > len(content_lines):
return []
first_pat = pattern_lines[0].strip()
last_pat = pattern_lines[-1].strip()
ANCHOR_THRESHOLD = 0.80
def _sim(a: str, b: str) -> float:
if a == b:
return 1.0
return SequenceMatcher(None, a, b).ratio()
matches = []
for i in range(len(content_lines) - n + 1):
block_lines = content_lines[i:i + n]
if _sim(first_pat, block_lines[0].strip()) < ANCHOR_THRESHOLD:
continue
if _sim(last_pat, block_lines[-1].strip()) < ANCHOR_THRESHOLD:
continue
all_match = True
for p_line, c_line in zip(pattern_lines, block_lines):
p_stripped = p_line.strip()
if p_stripped and _sim(p_stripped, c_line.strip()) < 0.80:
all_match = False
break
if all_match:
matches.append(_calculate_line_positions(content_lines, i, i + n, len(content)))
return matches
# Ordered chain: precise strategies first, similarity-based last.
STRATEGIES: List[Tuple[str, Callable[[str, str], List[Span]]]] = [
("exact", _strategy_exact),
("line_trimmed", _strategy_line_trimmed),
("whitespace_normalized", _strategy_whitespace_normalized),
("indentation_flexible", _strategy_indentation_flexible),
("escape_normalized", _strategy_escape_normalized),
("trimmed_boundary", _strategy_trimmed_boundary),
("unicode_normalized", _strategy_unicode_normalized),
("block_anchor", _strategy_block_anchor),
("context_aware", _strategy_context_aware),
]
# Matches from these only *approximately* resemble old_string — fine for one
# unique replacement, never safe under replace_all.
SIMILARITY_STRATEGIES = frozenset({"block_anchor", "context_aware"})