diff --git a/tests/tools/test_docker_network_config.py b/tests/tools/test_docker_network_config.py index ee0eef9ead..d2c779f53a 100644 --- a/tests/tools/test_docker_network_config.py +++ b/tests/tools/test_docker_network_config.py @@ -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): diff --git a/tools/file_tools.py b/tools/file_tools.py index a76ec2e8af..7c24dae8a4 100644 --- a/tools/file_tools.py +++ b/tools/file_tools.py @@ -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: diff --git a/tools/fuzzy_match.py b/tools/fuzzy_match.py index a991c00f6b..e0c4e98f07 100644 --- a/tools/fuzzy_match.py +++ b/tools/fuzzy_match.py @@ -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: ' 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 "" diff --git a/tools/fuzzy_match_strategies.py b/tools/fuzzy_match_strategies.py deleted file mode 100644 index eb3e3464bd..0000000000 --- a/tools/fuzzy_match_strategies.py +++ /dev/null @@ -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"})