perf(tools): take one control-plane snapshot per grep/glob/ls scan (#6551)

The #6503 merge added _control_plane_path() to _is_denied_tool_path(),
which grep, glob and ls call per enumerated entry. Without a snapshot
argument every call rebuilt _control_plane_snapshot() and stat'ed every
protected state file again. A 5,000-file grep went from 0.75 s to ~7 s,
and around 20k files it hits the grep timeout.

Thread an optional snapshot through _is_denied_tool_path() and
_can_traverse_tool_path(), and build it once per scan, the way
process_resources already does for its launch-boundary walk. Root
resolution and bound-resource resolution still observe fresh state.

src/agent_tools/filesystem_tools.py

@@ -801,15 +801,18 @@ class LsTool:
             return {"error": f"ls: {e}", "exit_code": 1}
 
         def _ls():
+            from src.agent_runtime.resources import _control_plane_snapshot
             if not os.path.isdir(root):
                 return None, f"ls: {root}: not a directory"
             rows = []
+            snapshot = _control_plane_snapshot()
             try:
                 with os.scandir(root) as it:
                     for entry in it:
                         if entry.name.startswith("."):
                             continue
-                        if _is_denied_tool_path(os.path.realpath(entry.path)) or not _visible_bound_resource(entry.path):
+                        if (_is_denied_tool_path(os.path.realpath(entry.path), snapshot=snapshot)
+                                or not _visible_bound_resource(entry.path)):
                             continue
                         try:
                             is_dir = entry.is_dir(follow_symlinks=False)
@@ -864,10 +867,12 @@ class GlobTool:
             return {"error": f"glob: {e}", "exit_code": 1}
 
         def _glob():
+            from src.agent_runtime.resources import _control_plane_snapshot
             base = os.path.abspath(root)
             if not os.path.isdir(base):
                 return None, f"glob: {root}: not a directory"
             rbase = os.path.realpath(base)
+            snapshot = _control_plane_snapshot()
             norm_pat = pattern.replace("\\", "/")
             # Fast path: literal pattern (no wildcards) → direct path lookup.
             if not any(c in norm_pat for c in "*?["):
@@ -884,7 +889,8 @@ class GlobTool:
                 # .ssh/id_rsa, …) falls through to the walk, which skips it —
                 # otherwise glob would surface secret paths that read_file /
                 # grep already refuse to touch.
-                if inside and os.path.exists(cand) and not _is_denied_tool_path(cand) and _visible_bound_resource(cand):
+                if (inside and os.path.exists(cand) and not _is_denied_tool_path(cand, snapshot=snapshot)
+                        and _visible_bound_resource(cand)):
                     return [cand], None
                 # Literal not at exact path — fall through to walk so
                 # e.g. "foo.py" still matches at any depth (like rglob).
@@ -894,7 +900,7 @@ class GlobTool:
             cap = _CODENAV_MAX_HITS * 5
             try:
                 for dp, dns, fns in os.walk(base):
-                    if not _can_traverse_tool_path(os.path.realpath(dp)):
+                    if not _can_traverse_tool_path(os.path.realpath(dp), snapshot=snapshot):
                         dns[:] = []
                         continue
                     # Prune skipped dirs before descending (unlike rglob which
@@ -905,7 +911,7 @@ class GlobTool:
                         d for d in dns
                         if d not in _CODENAV_SKIP_DIRS
                         and d not in _SENSITIVE_BASENAMES
-                        and _can_traverse_tool_path(os.path.realpath(os.path.join(dp, d)))
+                        and _can_traverse_tool_path(os.path.realpath(os.path.join(dp, d)), snapshot=snapshot)
                     ]
                     for name in fns + dns:
                         full = os.path.join(dp, name)
@@ -913,7 +919,8 @@ class GlobTool:
                         if regex.fullmatch(rel) or regex.fullmatch(name):
                             # Skip deny-listed sensitive files (.env, id_rsa,
                             # known_hosts, …) the same way grep does.
-                            if _is_denied_tool_path(os.path.realpath(full)) or not _visible_bound_resource(full):
+                            if (_is_denied_tool_path(os.path.realpath(full), snapshot=snapshot)
+                                    or not _visible_bound_resource(full)):
                                 continue
                             try:
                                 mtime = os.stat(full).st_mtime
@@ -981,6 +988,7 @@ class GrepTool:
             import subprocess
             import threading
 
+            from src.agent_runtime.resources import _control_plane_snapshot
             from src.constants import DATA_DIR
 
             rg = shutil.which("rg")
@@ -992,6 +1000,7 @@ class GrepTool:
             if bound is not None:
                 bound.validate()
             files = []
+            snapshot = _control_plane_snapshot()
 
             def check_deadline():
                 if time.monotonic() >= deadline:
@@ -1002,7 +1011,7 @@ class GrepTool:
                 if os.path.islink(path):
                     return
                 canonical = os.path.realpath(path)
-                if not _path_within(canonical, base) or _is_denied_tool_path(canonical):
+                if not _path_within(canonical, base) or _is_denied_tool_path(canonical, snapshot=snapshot):
                     return
                 if bound is not None:
                     try:
@@ -1037,7 +1046,7 @@ class GrepTool:
                     while pending_directories:
                         check_deadline()
                         directory = pending_directories.pop()
-                        if not _can_traverse_tool_path(directory):
+                        if not _can_traverse_tool_path(directory, snapshot=snapshot):
                             continue
                         with os.scandir(directory) as entries:
                             for entry in entries:
@@ -1052,8 +1061,8 @@ class GrepTool:
                                     raise ValueError("grep: directory identity changed during enumeration")
                                 if entry.is_dir(follow_symlinks=False):
                                     if (entry.name not in _CODENAV_SKIP_DIRS
-                                            and _can_traverse_tool_path(canonical)
+                                            and _can_traverse_tool_path(canonical, snapshot=snapshot)
-                                            and (bound is None or _is_denied_tool_path(canonical)
+                                            and (bound is None or _is_denied_tool_path(canonical, snapshot=snapshot)
                                                  or _visible_bound_resource(canonical))):
                                         pending_directories.append(canonical)
                                 else:

src/tool_execution.py

@@ -964,24 +964,28 @@ def _is_hardlinked_regular_file(resolved: str) -> bool:
     return stat.S_ISREG(target.st_mode) and getattr(target, "st_nlink", 1) > 1
 
 
-def _is_denied_tool_path(resolved: str) -> bool:
+def _is_denied_tool_path(resolved: str, *, snapshot=None) -> bool:
-    """Apply every path deny to a canonical traversal result."""
+    """Apply every path deny to a canonical traversal result.
+
+    Directory scans pass one ``_control_plane_snapshot()`` for the whole walk;
+    rebuilding it per entry made grep/ls/glob linear in protected-state stats.
+    """
     from src.agent_runtime.resources import _control_plane_path
     return (
         _is_sensitive_path(resolved)
         or _is_app_state_path(resolved)
         or _is_hardlinked_regular_file(resolved)
-        or _control_plane_path(resolved)
+        or _control_plane_path(resolved, snapshot=snapshot)
     )
 
 
-def _can_traverse_tool_path(resolved: str) -> bool:
+def _can_traverse_tool_path(resolved: str, *, snapshot=None) -> bool:
     """Allow walking a denied state parent only to reach safe carve-outs."""
     if _is_sensitive_path(resolved):
         return False
     if not _is_app_state_path(resolved):
         from src.agent_runtime.resources import _control_plane_path
-        return not _control_plane_path(resolved)
+        return not _control_plane_path(resolved, snapshot=snapshot)
     return any(
         _path_within(readable, resolved)
         for readable in _agent_readable_data_subdirs()

tests/test_agent_state_dir_confinement.py

@@ -176,6 +176,41 @@ def test_native_file_tools_hide_control_plane_hardlink_alias(tmp_path, monkeypat
     assert ":1:LIVE_ADMIN_SESSION" not in grep_result["output"]
 
 
+def test_native_file_tools_take_one_control_plane_snapshot_per_scan(tmp_path, monkeypatch):
+    """The deny check must not rebuild the control-plane snapshot per entry.
+
+    Each rebuild stats every protected state file, so a per-entry rebuild made
+    a 5,000-file grep about seven times slower and timed out near 20k files.
+    """
+    resources = importlib.import_module("src.agent_runtime.resources")
+    data_dir = tmp_path / "data"
+    data_dir.mkdir()
+    workspace = _configure_test_data_tree(monkeypatch, data_dir)["AGENT_WORKSPACE_DIR"]
+    for index in range(3):
+        directory = workspace / f"d{index}"
+        directory.mkdir(parents=True)
+        for item in range(20):
+            (directory / f"f{item}.txt").write_text("NEEDLE\n" if item == 0 else "x\n")
+    snapshots = []
+    original = resources._control_plane_snapshot
+
+    def counting_snapshot():
+        snapshots.append(1)
+        return original()
+
+    monkeypatch.setattr(resources, "_control_plane_snapshot", counting_snapshot)
+    for tool, args, expected in (
+        (LsTool(), {"path": str(workspace / "d0")}, "f19.txt"),
+        (GlobTool(), {"pattern": "**/*.txt", "path": str(workspace)}, "f19.txt"),
+        (GrepTool(), {"pattern": "NEEDLE", "path": str(workspace)}, "f0.txt:1:NEEDLE"),
+    ):
+        snapshots.clear()
+        result = asyncio.run(tool.execute(json.dumps(args), {}))
+        assert expected in result.get("output", ""), result
+        # One fresh check resolves the search root; the walk shares one more.
+        assert len(snapshots) <= 2, (type(tool).__name__, len(snapshots))
+
+
 def test_blocks_app_state_on_a_case_insensitive_filesystem():
     """On default macOS a case-variant path opens the same file, and realpath
     does not canonicalise case there the way it does on Windows.