perf(tools): negative-result cache for read_file + search misses

When read_file or search hits a non-existent path, ShellFileOperations
spawns a subprocess to stat the path and another to walk the parent
directory for "did you mean..." suggestions. A typo'd path retried 13
times (observed in the wild) costs 26 subprocess invocations + 13 ls
walks for a result we already know.

Add a per-task negative-result cache keyed by (op, resolved_path) with
a 60s TTL and a hard cap of 500 entries. On hit, return the cached
error JSON immediately and skip the subprocess + suggestion walk.

The cache is namespaced by operation ("read" vs "search") because the
two callers return different error JSON shapes ("File not found:" vs
"Path not found:"). Eviction:

  * TTL (60s) — short, so a path that appears later isn't masked.
  * write_file / patch on the same path — _invalidate_dedup_for_path
    now also drops the negative-cache entry so a freshly-written file
    is read from disk on the next call instead of returning a stale
    "not found" stub.

Tests in tests/tools/test_file_tools.py cover:

  * read cache hit skips the subprocess on retry
  * cache is per-task (no cross-task pollution)
  * successful reads do not poison the cache
  * search cache hit skips the subprocess on retry
  * read and search caches are namespaced (different error shapes)
  * write_file invalidates the read negative cache
  * TTL expiry evicts stale entries

tests/tools/test_file_tools.py

@@ -716,3 +716,207 @@ class TestDedupInvalidationTaskResolution:
         assert correct not in remaining, remaining
 
         ft._read_tracker.pop(task_id, None)
+
+
+# ---------------------------------------------------------------------------
+# Negative-result cache tests
+#
+# Without this cache, a typo'd path retried 13 times (observed in the wild)
+# spawned 13 wc -c subprocesses + 13 ls walks for the "did you mean..." hint.
+# The cache returns the same error JSON immediately and skips both shells.
+# ---------------------------------------------------------------------------
+
+class TestNotFoundCache:
+    @patch("tools.file_tools._get_file_ops")
+    def test_read_caches_file_not_found_and_skips_subprocess_on_retry(self, mock_get):
+        mock_ops = MagicMock()
+        result_obj = MagicMock()
+        result_obj.content = None
+        # Shape returned by ShellFileOperations._suggest_similar_files
+        result_obj.to_dict.return_value = {
+            "error": "File not found: /tmp/does-not-exist-neg-1.txt",
+            "similar_files": [],
+        }
+        mock_ops.read_file.return_value = result_obj
+        mock_get.return_value = mock_ops
+
+        from tools.file_tools import read_file_tool, _read_tracker
+        # Use a unique task_id so we don't collide with other tests.
+        tid = "neg-cache-read-1"
+        _read_tracker.pop(tid, None)
+
+        # First call: subprocess runs, error returned, cache populated.
+        first = json.loads(read_file_tool("/tmp/does-not-exist-neg-1.txt", task_id=tid))
+        assert "File not found" in first["error"]
+        assert mock_ops.read_file.call_count == 1
+
+        # Second call: same path → cache hit → no new subprocess call.
+        second = json.loads(read_file_tool("/tmp/does-not-exist-neg-1.txt", task_id=tid))
+        assert "File not found" in second["error"]
+        assert mock_ops.read_file.call_count == 1, (
+            "Negative cache hit must skip the subprocess on retry"
+        )
+
+    @patch("tools.file_tools._get_file_ops")
+    def test_read_cache_isolated_per_task(self, mock_get):
+        mock_ops = MagicMock()
+        result_obj = MagicMock()
+        result_obj.to_dict.return_value = {
+            "error": "File not found: /tmp/does-not-exist-neg-2.txt",
+            "similar_files": [],
+        }
+        mock_ops.read_file.return_value = result_obj
+        mock_get.return_value = mock_ops
+
+        from tools.file_tools import read_file_tool, _read_tracker
+        for tid in ("neg-cache-iso-A", "neg-cache-iso-B"):
+            _read_tracker.pop(tid, None)
+
+        read_file_tool("/tmp/does-not-exist-neg-2.txt", task_id="neg-cache-iso-A")
+        read_file_tool("/tmp/does-not-exist-neg-2.txt", task_id="neg-cache-iso-B")
+        # Each task gets its own miss; B doesn't reuse A's cache entry.
+        assert mock_ops.read_file.call_count == 2
+
+    @patch("tools.file_tools._get_file_ops")
+    def test_read_cache_populated_only_for_not_found(self, mock_get):
+        # A successful read must NOT populate the negative cache.
+        mock_ops = MagicMock()
+        result_obj = MagicMock()
+        result_obj.content = "x"
+        result_obj.to_dict.return_value = {"content": "x", "total_lines": 1}
+        mock_ops.read_file.return_value = result_obj
+        mock_get.return_value = mock_ops
+
+        from tools.file_tools import read_file_tool, _read_tracker
+        tid = "neg-cache-success-only"
+        _read_tracker.pop(tid, None)
+
+        read_file_tool("/tmp/exists-or-mocked.txt", task_id=tid)
+        nf = _read_tracker[tid].get("not_found", {})
+        assert all(k[0] != "read" or "exists-or-mocked" not in k[1] for k in nf), (
+            "Successful reads must not poison the negative cache"
+        )
+
+    @patch("tools.file_tools._get_file_ops")
+    def test_search_caches_path_not_found_and_skips_subprocess_on_retry(self, mock_get):
+        mock_ops = MagicMock()
+        result_obj = MagicMock()
+        result_obj.matches = []
+        result_obj.to_dict.return_value = {
+            "error": "Path not found: /tmp/does-not-exist-search-3",
+            "total_count": 0,
+        }
+        mock_ops.search.return_value = result_obj
+        mock_get.return_value = mock_ops
+
+        from tools.file_tools import search_tool, _read_tracker
+        tid = "neg-cache-search-3"
+        _read_tracker.pop(tid, None)
+
+        first = json.loads(search_tool("foo", path="/tmp/does-not-exist-search-3", task_id=tid))
+        assert "Path not found" in first["error"]
+        assert mock_ops.search.call_count == 1
+
+        second = json.loads(search_tool("foo", path="/tmp/does-not-exist-search-3", task_id=tid))
+        assert "Path not found" in second["error"]
+        assert mock_ops.search.call_count == 1, (
+            "Search negative cache hit must skip the subprocess on retry"
+        )
+
+    @patch("tools.file_tools._get_file_ops")
+    def test_read_and_search_caches_are_namespaced(self, mock_get):
+        # A read that misses must NOT serve a subsequent search call's miss
+        # (different error JSON shapes).
+        mock_ops = MagicMock()
+
+        read_obj = MagicMock()
+        read_obj.to_dict.return_value = {
+            "error": "File not found: /tmp/does-not-exist-namespace-4",
+        }
+        mock_ops.read_file.return_value = read_obj
+
+        search_obj = MagicMock()
+        search_obj.matches = []
+        search_obj.to_dict.return_value = {
+            "error": "Path not found: /tmp/does-not-exist-namespace-4",
+            "total_count": 0,
+        }
+        mock_ops.search.return_value = search_obj
+
+        mock_get.return_value = mock_ops
+
+        from tools.file_tools import read_file_tool, search_tool, _read_tracker
+        tid = "neg-cache-namespace-4"
+        _read_tracker.pop(tid, None)
+
+        read_file_tool("/tmp/does-not-exist-namespace-4", task_id=tid)
+        search_tool("foo", path="/tmp/does-not-exist-namespace-4", task_id=tid)
+        # Both ops must hit their own caller (namespacing prevents read's
+        # error JSON from being returned to search).
+        assert mock_ops.read_file.call_count == 1
+        assert mock_ops.search.call_count == 1
+
+    @patch("tools.file_tools._get_file_ops")
+    def test_write_invalidates_read_negative_cache(self, mock_get):
+        # After write_file on a path, a subsequent read must hit disk,
+        # not return the cached "not found" stub.
+        mock_ops = MagicMock()
+
+        not_found_obj = MagicMock()
+        not_found_obj.to_dict.return_value = {
+            "error": "File not found: /tmp/will-be-created-neg-5.txt",
+        }
+        present_obj = MagicMock()
+        present_obj.content = "after write"
+        present_obj.to_dict.return_value = {"content": "after write", "total_lines": 1}
+
+        # First read → not found; second read (after write) → present.
+        mock_ops.read_file.side_effect = [not_found_obj, present_obj]
+        write_result_obj = MagicMock()
+        write_result_obj.to_dict.return_value = {"status": "ok"}
+        mock_ops.write_file.return_value = write_result_obj
+        mock_get.return_value = mock_ops
+
+        from tools.file_tools import read_file_tool, write_file_tool, _read_tracker
+        tid = "neg-cache-write-invalidate-5"
+        _read_tracker.pop(tid, None)
+
+        first = json.loads(read_file_tool("/tmp/will-be-created-neg-5.txt", task_id=tid))
+        assert "File not found" in first["error"]
+
+        write_file_tool("/tmp/will-be-created-neg-5.txt", "after write", task_id=tid)
+
+        second = json.loads(read_file_tool("/tmp/will-be-created-neg-5.txt", task_id=tid))
+        assert second.get("content") == "after write", (
+            "write_file must invalidate the negative cache so the next read "
+            "hits the now-existing file instead of returning a stale stub"
+        )
+        assert mock_ops.read_file.call_count == 2
+
+    def test_not_found_ttl_expires(self):
+        # A cache entry older than _NOT_FOUND_TTL_SECONDS must be discarded.
+        from tools.file_tools import (
+            _check_not_found_cache,
+            _record_not_found,
+            _read_tracker,
+            _NOT_FOUND_TTL_SECONDS,
+        )
+        import tools.file_tools as ft
+
+        tid = "neg-cache-ttl-6"
+        _read_tracker.pop(tid, None)
+        _record_not_found("read", "/tmp/ttl-test", tid, '{"error":"x"}')
+        # Fresh entry: cache hit.
+        assert _check_not_found_cache("read", "/tmp/ttl-test", tid) is not None
+
+        # Backdate the entry past the TTL.
+        with ft._read_tracker_lock:
+            entry = _read_tracker[tid]["not_found"][("read", "/tmp/ttl-test")]
+            ft._read_tracker[tid]["not_found"][("read", "/tmp/ttl-test")] = (
+                entry[0] - _NOT_FOUND_TTL_SECONDS - 1.0,
+                entry[1],
+            )
+        # Stale entry: cache miss, also evicted.
+        assert _check_not_found_cache("read", "/tmp/ttl-test", tid) is None
+        with ft._read_tracker_lock:
+            assert ("read", "/tmp/ttl-test") not in _read_tracker[tid].get("not_found", {})

tools/file_tools.py

@@ -872,6 +872,8 @@ def _reset_patch_failures(task_id: str, resolved_paths: list) -> None:
 _READ_HISTORY_CAP = 500       # set; used only by get_read_files_summary
 _DEDUP_CAP = 1000             # dict; skip-identical-reread guard
 _READ_TIMESTAMPS_CAP = 1000   # dict; external-edit detection for write/patch
+_NOT_FOUND_CAP = 500          # dict; per-task negative-result cache for missing paths
+_NOT_FOUND_TTL_SECONDS = 60.0 # short TTL — a path that didn't exist may be created soon
 _READ_DEDUP_STATUS_MESSAGE = (
     "File unchanged since last read. The content from "
     "the earlier read_file result in this conversation is "
@@ -929,6 +931,60 @@ def _cap_read_tracker_data(task_data: dict) -> None:
             except (StopIteration, KeyError):
                 break
 
+    nf = task_data.get("not_found")
+    if nf is not None and len(nf) > _NOT_FOUND_CAP:
+        excess = len(nf) - _NOT_FOUND_CAP
+        for _ in range(excess):
+            try:
+                nf.pop(next(iter(nf)))
+            except (StopIteration, KeyError):
+                break
+
+
+def _check_not_found_cache(op: str, resolved_str: str, task_id: str) -> str | None:
+    """Return cached not-found JSON for *(op, resolved_str)* if still fresh.
+
+    Skips the expensive subprocess + suggestion walk when the model retries
+    the same missing path. Observed in agent.log: a single typo'd path was
+    retried 13 times — each retry forked a shell to walk the parent directory
+    and score similar names.
+
+    *op* is "read" or "search" — kept separate because the two callers return
+    different error JSON shapes ("File not found:" vs "Path not found:").
+
+    Eviction: TTL or write_file/patch on the path (see invalidate_for_path).
+    """
+    import time
+    with _read_tracker_lock:
+        task_data = _read_tracker.get(task_id)
+        if not task_data:
+            return None
+        nf = task_data.get("not_found")
+        if not nf:
+            return None
+        entry = nf.get((op, resolved_str))
+        if entry is None:
+            return None
+        ts, cached_json = entry
+        if time.monotonic() - ts > _NOT_FOUND_TTL_SECONDS:
+            nf.pop((op, resolved_str), None)
+            return None
+        return cached_json
+
+
+def _record_not_found(op: str, resolved_str: str, task_id: str, error_json: str) -> None:
+    """Cache a not-found error so the next *op* call for *resolved_str* skips I/O."""
+    import time
+    with _read_tracker_lock:
+        task_data = _read_tracker.setdefault(task_id, {
+            "last_key": None, "consecutive": 0,
+            "read_history": set(), "dedup": {},
+            "dedup_hits": {}, "read_timestamps": {},
+        })
+        nf = task_data.setdefault("not_found", {})
+        nf[(op, resolved_str)] = (time.monotonic(), error_json)
+        _cap_read_tracker_data(task_data)
+
 
 def _is_internal_file_status_text(content: str) -> bool:
     """Return True when content looks like an internal file-tool status, not real file bytes.
@@ -1282,6 +1338,15 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 500, task_id: str =
         if block_error:
             return tool_error(block_error)
 
+        # ── Negative-result cache ─────────────────────────────────────
+        # If we already discovered this path doesn't exist (within TTL),
+        # return the cached error without spawning the subprocess +
+        # similar-files walk. Cleared by write_file/patch on the same path.
+        resolved_str_for_neg = str(_resolve_path_for_task(path, task_id))
+        cached_not_found = _check_not_found_cache("read", resolved_str_for_neg, task_id)
+        if cached_not_found is not None:
+            return cached_not_found
+
         # ── Dedup check ───────────────────────────────────────────────
         # If we already read this exact (path, offset, limit) and the
         # file hasn't been modified since, return a lightweight stub
@@ -1345,6 +1410,15 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 500, task_id: str =
         result = file_ops.read_file(path, offset, limit)
         result_dict = result.to_dict()
 
+        # ── Populate negative-result cache on not-found ───────────────
+        # _suggest_similar_files returns ReadResult(error="File not found: ..").
+        # Cache the JSON we'd return so a retry skips the parent-dir walk.
+        _err = result_dict.get("error") or ""
+        if isinstance(_err, str) and _err.startswith("File not found:"):
+            _not_found_json = json.dumps(result_dict, ensure_ascii=False)
+            _record_not_found("read", resolved_str_for_neg, task_id, _not_found_json)
+            return _not_found_json
+
         # ── Character-count guard ─────────────────────────────────────
         # We're model-agnostic so we can't count tokens; characters are
         # the best proxy we have.  If the read produced an unreasonable
@@ -1544,12 +1618,18 @@ def _invalidate_dedup_for_path(filepath: str, task_id: str) -> None:
         if task_data is None:
             return
         dedup = task_data.get("dedup")
-        if not dedup:
-            return
-        # Collect keys to remove (can't mutate dict during iteration).
-        stale_keys = [k for k in dedup if k[0] == resolved]
-        for k in stale_keys:
-            del dedup[k]
+        if dedup:
+            # Collect keys to remove (can't mutate dict during iteration).
+            stale_keys = [k for k in dedup if k[0] == resolved]
+            for k in stale_keys:
+                del dedup[k]
+        # Also evict from the negative-result cache: a write_file that
+        # creates the path means subsequent reads (or searches under it)
+        # must hit disk.
+        nf = task_data.get("not_found")
+        if nf:
+            nf.pop(("read", resolved), None)
+            nf.pop(("search", resolved), None)
 
 
 def _update_read_timestamp(filepath: str, task_id: str) -> None:
@@ -1974,6 +2054,19 @@ def search_tool(pattern: str, target: str = "content", path: str = ".",
         if block_error:
             return tool_error(block_error)
 
+        # ── Negative-result cache ─────────────────────────────────────
+        # Search returns "Path not found: <path>" when the search root
+        # doesn't exist. The error path also lists the parent directory
+        # (file_operations.py:1402) — expensive to repeat. Cache so the
+        # next call to a known-missing root skips both shells.
+        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(
             pattern=pattern, path=path, target=target, file_glob=file_glob,
@@ -1992,6 +2085,13 @@ def search_tool(pattern: str, target: str = "content", path: str = ".",
                 "token, cache, or secret-bearing environment files."
             )
 
+        # Populate negative cache when search root was missing.
+        _search_err = result_dict.get("error") or ""
+        if isinstance(_search_err, str) and _search_err.startswith("Path not found:"):
+            _search_nf_json = json.dumps(result_dict, ensure_ascii=False)
+            _record_not_found("search", resolved_search_path, task_id, _search_nf_json)
+            return _search_nf_json
+
         if count >= 3:
             result_dict["_warning"] = (
                 f"You have run this exact search {count} times consecutively. "