Make win glob/grep traversal version-independent

Author: beacoder

.github/workflows/ci.yml

@@ -85,7 +85,7 @@ jobs:
       - name: Run tests with coverage
         run: coverage run -m unittest discover -s tests -v
 
-      # `coverage report` enforces `fail_under` from pyproject.toml (90%),
+      # `coverage report` enforces `fail_under` from pyproject.toml (95%),
       # so an insufficiently-tested change fails this job.
       - name: Coverage report (enforces fail_under)
         run: coverage report

python_agent_harness/tools/filesystem.py

@@ -32,11 +32,13 @@
 from __future__ import annotations
 
 import os
+import re
 import shutil  # noqa: F401  (mock target for tests)
 import subprocess  # noqa: F401  (mock target for tests)
 import tempfile
 import threading
 import time
+from collections.abc import Callable, Iterator
 from pathlib import Path
 from typing import TypeGuard
 
@@ -132,6 +134,87 @@ def _natnump(n: object) -> TypeGuard[int]:
     return isinstance(n, int) and not isinstance(n, bool) and n >= 0
 
 
+def _glob_to_regex(pattern: str) -> str:
+    """Translate a pathlib-style glob pattern into a :mod:`re` pattern.
+
+    Follows :mod:`fnmatch` conventions with two deliberate deviations:
+    ``*`` does not cross path separators (unlike :func:`fnmatch.fnmatch`)
+    and ``**`` matches any number of directories (including none).  A
+    pattern without any glob metacharacters is escaped, so plain names
+    match exactly.
+    """
+    i, n = 0, len(pattern)
+    out: list[str] = []
+    while i < n:
+        c = pattern[i]
+        if c == "*":
+            if pattern[i : i + 2] == "**":
+                i += 2
+                if pattern[i : i + 1] == "/":
+                    i += 1
+                out.append("(?:.*/)?")
+                continue
+            out.append("[^/]*")
+            i += 1
+        elif c == "?":
+            out.append("[^/]")
+            i += 1
+        elif c == "[":
+            end = _find_glob_class_end(pattern, i)
+            if end is None:
+                out.append(re.escape(c))
+                i += 1
+            else:
+                body = pattern[i + 1 : end]
+                if body.startswith("!"):
+                    body = "^" + body[1:]
+                out.append("[" + body.replace("\\", "\\\\") + "]")
+                i = end + 1
+        else:
+            out.append(re.escape(c))
+            i += 1
+    return "".join(out)
+
+
+def _find_glob_class_end(pattern: str, start: int) -> int | None:
+    """Index of the ``]`` closing the char class opened at ``start``.
+
+    Mirrors glob semantics: a ``]`` immediately after ``[`` (or ``[!``)
+    is a literal bracket, and backslash escapes are honoured.
+    """
+    i = start + 1
+    if i < len(pattern) and pattern[i] == "!":
+        i += 1
+    if i < len(pattern) and pattern[i] == "]":
+        i += 1
+    while i < len(pattern):
+        if pattern[i] == "\\" and i + 1 < len(pattern):
+            i += 2
+        elif pattern[i] == "]":
+            return i
+        else:
+            i += 1
+    return None
+
+
+def _walk_files(root: Path, onerror: Callable[[OSError], object] | None = None) -> Iterator[Path]:
+    """Depth-first file iterator that never raises on traversal errors.
+
+    Wraps :func:`os.walk` with ``onerror`` so unreadable directories are
+    reported instead of silently swallowed, and wraps the whole iteration
+    in a try/except so a scan that dies mid-flight still yields the
+    entries collected so far (mirroring the 3.13+ ``pathlib`` scan
+    behaviour on every supported version).
+    """
+    try:
+        for dirpath, _dirnames, filenames in os.walk(root, onerror=onerror, followlinks=False):
+            for name in filenames:
+                yield Path(dirpath) / name
+    except OSError as e:
+        if onerror is not None:
+            onerror(e)
+
+
 def _git_root(path: str) -> str | None:
     d = Path(path).resolve()
     for parent in [d, *d.parents]:

python_agent_harness/tools/glob_win.py

@@ -4,8 +4,8 @@
 in :class:`GlobTool` (which shells out to ``tree``) and the macOS
 variant :class:`GlobMac` (which shells out to ``find``) both fail on
 a stock Windows install.  ``GlobWindows`` replaces that fallback with
-a pure-Python :meth:`pathlib.Path.rglob` approach: Python handles
-directory traversal, pattern matching, and mtime sorting.
+a pure-Python :func:`os.walk` approach: Python handles directory
+traversal, pattern matching, and mtime sorting.
 
 This is slower than the C-based ``tree``/``find`` on large directory
 trees, but produces identical results and has no external dependencies
@@ -19,47 +19,60 @@
 from __future__ import annotations
 
 import os
+import re
 from pathlib import Path
 
 from .base import ToolContext
-from .filesystem import _natnump, _spool
+from .filesystem import _glob_to_regex, _natnump, _spool, _walk_files
 from .glob import GlobTool
 
 
 class GlobWindows(GlobTool):
-    """Glob with a pure-Python ``pathlib.rglob`` non-git fallback for Windows."""
-
-    def _rglob_fallback(self, pattern: str, base: str, depth: object) -> str:
-        """Use ``pathlib.Path.rglob`` for traversal + matching, Python for mtime sort.
-
-        Walks the directory tree with :meth:`pathlib.Path.rglob`, filters
-        hidden directories (``.git``, etc.), and sorts results by
-        modification time (newest first), matching the ``tree --sort=mtime``
-        order of the Linux fallback.  Symlinks are followed by default
-        via ``Path.rglob``.
+    """Glob with a fault-tolerant pure-Python non-git fallback for Windows."""
+
+    def _walk_fallback(self, pattern: str, base: str, depth: object) -> str:
+        """Walk the tree with :func:`os.walk`, match, and sort by mtime.
+
+        Matches files against a pathlib-style glob translated via
+        :func:`_glob_to_regex`, filters hidden directories (``.git``,
+        etc.), and sorts results by modification time (newest first),
+        matching the ``tree --sort=mtime`` order of the Linux fallback.
+        Unlike ``Path.rglob`` (whose OSError suppression only exists on
+        3.13+), this traversal tolerates races and unreadable directories
+        identically on every supported Python version.
         """
         root = Path(base)
         max_depth = depth if _natnump(depth) else None
+        # pathlib rglob semantics: a pattern without a directory part is
+        # matched against the basename at any depth; a pattern with one is
+        # matched against the path relative to the root.
+        if "/" in pattern or os.sep in pattern:
+            rx = re.compile(_glob_to_regex(pattern.replace(os.sep, "/")))
+        else:
+            rx = re.compile(r"(?:.*/)?" + _glob_to_regex(pattern))
+        errors: list[str] = []
+
+        def onerror(e: OSError) -> None:
+            errors.append(str(e))
 
         matches: list[tuple[float, str]] = []
-        try:
-            for p in root.rglob(pattern):
-                if not p.is_file():
-                    continue
-                # skip hidden directories (.git, etc.)
-                if any(part.startswith(".") for part in p.relative_to(root).parts[:-1]):
-                    continue
-                if max_depth is not None:
-                    rel_depth = len(p.relative_to(root).parts)
-                    if rel_depth > max_depth:
-                        continue
-                try:
-                    mtime = p.stat().st_mtime
-                except OSError:
-                    mtime = 0.0
-                matches.append((mtime, str(p)))
-        except OSError as e:
-            return f"Error: {e}"
+        for p in _walk_files(root, onerror=onerror):
+            rel = p.relative_to(root)
+            rel_parts = rel.parts
+            if any(part.startswith(".") for part in rel_parts[:-1]):
+                continue
+            if max_depth is not None and len(rel_parts) > max_depth:
+                continue
+            if not rx.fullmatch(rel.as_posix()):
+                continue
+            try:
+                mtime = p.stat().st_mtime
+            except OSError:
+                mtime = 0.0
+            matches.append((mtime, str(p)))
+
+        if not matches and errors:
+            return f"Error: {errors[0]}"
 
         matches.sort(key=lambda t: t[0], reverse=True)
         out = "\n".join(path for _, path in matches)
@@ -88,4 +101,4 @@ def run(self, args: dict, ctx: ToolContext) -> str:
         if git_root:
             return super().run(args, ctx)
 
-        return self._rglob_fallback(pattern, base, depth)
+        return self._walk_fallback(pattern, base, depth)

python_agent_harness/tools/grep_win.py

@@ -8,10 +8,10 @@
 
 ``GrepWindows`` extends the fallback chain with a pure-Python
 ``re``-based search: it walks the directory tree with
-:meth:`pathlib.Path.rglob`, applies the regex to each file, and
-collects matches with line numbers and optional context lines.  This
-ensures the Grep tool always works on a stock Windows install with
-only Python and Git installed.
+:func:`os.walk` (via :func:`_walk_files`), applies the regex to each
+file, and collects matches with line numbers and optional context
+lines.  This ensures the Grep tool always works on a stock Windows
+install with only Python and Git installed.
 
 Only the fallback strategy differs; the git path (``git grep -P``),
 the tool name, and the result format are inherited so callers, the
@@ -26,7 +26,7 @@
 import subprocess
 from pathlib import Path
 
-from .filesystem import _spool
+from .filesystem import _spool, _walk_files
 from .grep import Grep, _grep_out
 
 
@@ -98,17 +98,11 @@ def _python_grep(self, regex: str, path: str, glob: str | None, context: int | N
             files = [Path(path)]
         else:
             root = Path(path)
-            collected: list[Path] = []
-            try:
-                for p in root.rglob("*"):
-                    if not p.is_file():
-                        continue
-                    if any(part.startswith(".") for part in p.relative_to(root).parts[:-1]):
-                        continue
-                    collected.append(p)
-            except OSError:
-                pass
-            files = collected
+            files = [
+                p
+                for p in _walk_files(root)
+                if not any(part.startswith(".") for part in p.relative_to(root).parts[:-1])
+            ]
 
         for file_path in files:
             if match_count >= max_matches: