"""Tests for the file tools module (schema, handler wiring, error paths). Tests verify tool schemas, handler dispatch, validation logic, and error handling without requiring a running terminal environment. """ import json import logging from unittest.mock import MagicMock, patch import pytest from tools.file_tools import ( read_file_tool, ) class TestReadFileHandler: @patch("tools.file_tools._get_file_ops") def test_dispatch_without_limit_uses_schema_default(self, mock_get): """The model omits ``limit`` on most reads; the dispatch handler must fall back to the SAME default the schema advertises (drifted to 500 vs 2000).""" from tools.file_tools import READ_FILE_SCHEMA, _handle_read_file 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 _handle_read_file({"path": "/tmp/test.txt"}, task_id="t-default") schema_default = READ_FILE_SCHEMA["parameters"]["properties"]["limit"]["default"] assert mock_ops.read_file.call_args.args[2] == schema_default @patch("tools.file_tools._get_file_ops") def test_exception_returns_error_json(self, mock_get): mock_get.side_effect = RuntimeError("terminal not available") from tools.file_tools import read_file_tool result = json.loads(read_file_tool("/tmp/test.txt")) assert "error" in result assert "terminal not available" in result["error"] class TestWriteFileHandler: @patch("tools.file_tools._get_file_ops") def test_permission_error_returns_error_json_without_error_log(self, mock_get, caplog): mock_get.side_effect = PermissionError("read-only filesystem") from tools.file_tools import write_file_tool with caplog.at_level(logging.DEBUG, logger="tools.file_tools"): result = json.loads(write_file_tool("/tmp/out.txt", "data")) assert "error" in result assert "read-only" in result["error"] assert not any(r.levelno >= logging.ERROR for r in caplog.records) @patch("tools.file_tools._get_file_ops") def test_rejects_read_file_line_numbered_content(self, mock_get): """#19798 — do not persist read_file's LINE_NUM|CONTENT display format.""" from tools.file_tools import write_file_tool content = " 1|setting: new_value\n 2|other: thing\n" result = json.loads(write_file_tool("/tmp/config.yaml", content)) assert "error" in result assert "line-number" in result["error"].lower() mock_get.assert_not_called() def test_missing_content_key_returns_error(self): """#19096 — handler must reject tool calls where 'content' key is absent.""" from tools.file_tools import _handle_write_file result = json.loads(_handle_write_file({"path": "/tmp/oops.md"})) assert "error" in result assert "content" in result["error"] def test_missing_path_key_returns_error(self): """#19096 — handler must reject tool calls where 'path' key is absent.""" from tools.file_tools import _handle_write_file result = json.loads(_handle_write_file({"content": "hello"})) assert "error" in result def test_explicit_empty_content_is_allowed(self): """#19096 — explicit empty string content (file truncation) must still work.""" from tools.file_tools import _handle_write_file with patch("tools.file_tools._get_file_ops") as mock_get: mock_ops = MagicMock() result_obj = MagicMock() result_obj.to_dict.return_value = {"status": "ok", "path": "/tmp/empty.txt", "bytes": 0} mock_ops.write_file.return_value = result_obj mock_get.return_value = mock_ops result = json.loads(_handle_write_file({"path": "/tmp/empty.txt", "content": ""})) assert result["status"] == "ok" def test_non_string_content_returns_error(self): """#19096 — content must be a string, not a dict or list.""" from tools.file_tools import _handle_write_file result = json.loads(_handle_write_file({"path": "/tmp/x.txt", "content": {"nested": "dict"}})) assert "error" in result assert "string" in result["error"].lower() or "content" in result["error"].lower() class TestPatchHandler: @patch("tools.file_tools._get_file_ops") def test_unknown_mode_errors(self, mock_get): from tools.file_tools import patch_tool result = json.loads(patch_tool(mode="invalid_mode")) assert "error" in result @patch("tools.file_tools._get_file_ops") def test_patch_v4a_rejects_traversal_in_update_header(self, mock_get): """V4A '*** Update File:' headers come from patch content, which can carry prompt-injection-controlled paths (skill content, web extract). ``..`` traversal in the header must be rejected before the patch is applied, even though the explicit ``path=`` arg is allowed to use ``..`` for legitimate cross-worktree edits.""" from tools.file_tools import patch_tool result = json.loads(patch_tool( mode="patch", patch=( "*** Begin Patch\n" "*** Update File: ../../../etc/shadow\n" "@@ -1,3 +1,3 @@\n" "-old\n" "+new\n" "*** End Patch\n" ), )) assert "error" in result assert "traversal" in result["error"].lower() # patch_v4a must not be invoked when the header is rejected mock_get.return_value.patch_v4a.assert_not_called() @patch("tools.file_tools._get_file_ops") def test_patch_v4a_rejects_traversal_in_add_header(self, mock_get): from tools.file_tools import patch_tool result = json.loads(patch_tool( mode="patch", patch=( "*** Begin Patch\n" "*** Add File: ../../../tmp/dropped.py\n" "+print('pwned')\n" "*** End Patch\n" ), )) assert "error" in result assert "traversal" in result["error"].lower() class TestPatchSensitivePathExtraction: """Regression tests for patch_tool sensitive-path extraction. The sensitive path check relies on a regex that parses V4A patch headers. These tests cover: 1. ``*** Move File:`` operations (previously missed — the regex only matched Update/Add/Delete, so Move could target /etc/* without hitting the check). 2. ``***Keyword File:`` with no space after ``***`` (previously missed — the regex required ``\\s+`` even though patch_parser accepts ``\\s*``). 3. ``..`` traversal in Move headers (the Move endpoints run through the same traversal rejection as the other V4A headers). """ @patch("tools.file_tools._get_file_ops") def test_patch_move_to_sensitive_dst_blocked(self, mock_get): from tools.file_tools import patch_tool patch_text = ( "*** Begin Patch\n" "*** Move File: /tmp/work.txt -> /etc/crontab\n" "*** End Patch\n" ) result = json.loads(patch_tool(mode="patch", patch=patch_text)) assert "error" in result assert "sensitive" in result["error"].lower() mock_get.assert_not_called() @patch("tools.file_tools._get_file_ops") def test_patch_update_no_space_after_asterisks_blocked(self, mock_get): """``***Update File:`` (no space after asterisks) must also be caught. patch_parser.py accepts this form (``\\s*`` in its regex), so the sensitive path check must be at least as lenient or the check is bypassed. """ from tools.file_tools import patch_tool patch_text = ( "*** Begin Patch\n" "***Update File: /etc/resolv.conf\n" "@@ @@\n" "-old\n" "+new\n" "*** End Patch\n" ) result = json.loads(patch_tool(mode="patch", patch=patch_text)) assert "error" in result assert "sensitive" in result["error"].lower() mock_get.assert_not_called() @patch("tools.file_tools._get_file_ops") def test_patch_move_safe_paths_not_blocked(self, mock_get): """Safe Move operations should still reach the file_ops dispatch.""" mock_ops = MagicMock() result_obj = MagicMock() result_obj.to_dict.return_value = {"status": "ok"} mock_ops.patch_v4a.return_value = result_obj mock_get.return_value = mock_ops from tools.file_tools import patch_tool patch_text = ( "*** Begin Patch\n" "*** Move File: /tmp/a.txt -> /tmp/b.txt\n" "*** End Patch\n" ) result = json.loads(patch_tool(mode="patch", patch=patch_text)) assert "error" not in result mock_ops.patch_v4a.assert_called_once() class TestSearchHandler: @patch("tools.file_tools._get_file_ops") def test_search_exception_returns_error(self, mock_get): mock_get.side_effect = RuntimeError("no terminal") from tools.file_tools import search_tool result = json.loads(search_tool(pattern="x")) assert "error" in result # --------------------------------------------------------------------------- # Windows MSYS path resolution (salvage of #50488 / #46995) # --------------------------------------------------------------------------- class TestWindowsMsysPathResolution: """File tools must translate Git Bash drive paths before Path resolution.""" @pytest.mark.platforms("windows") def test_absolute_msys_path_normalized_before_windows_resolve(self, monkeypatch): """Windows-only: ``_resolve_path_for_task`` hands the translated path to ``ntpath``/``Path``, and only a real Windows ``Path`` renders ``C:\\Users\\...`` — faking ``sys.platform`` left PosixPath in place.""" import tools.file_tools_paths as file_tools monkeypatch.setattr(file_tools, "_uses_container_paths", lambda task_id="default": False) resolved = file_tools._resolve_path_for_task("/c/Users/Mark/project/app.py") assert str(resolved) == r"C:\Users\Mark\project\app.py" @pytest.mark.platforms("windows") def test_container_paths_skip_msys_translation(self, monkeypatch): """WSL/docker Linux paths must not be rewritten as Windows drives. Windows-only: the translation this guards against only happens when the host really is Windows, so the negative is only meaningful there. """ import tools.file_tools_paths as file_tools monkeypatch.setattr(file_tools, "_uses_container_paths", lambda task_id="default": True) monkeypatch.setattr( file_tools, "_authoritative_workspace_root", lambda task_id="default": "/home/don/project", ) resolved = file_tools._resolve_path_for_task("/home/don/.env") assert str(resolved) == "/home/don/.env" # --------------------------------------------------------------------------- # Tool result hint tests (#722) # --------------------------------------------------------------------------- class TestPatchHints: """Patch tool should hint when old_string is not found.""" @patch("tools.file_tools._get_file_ops") def test_no_match_includes_hint(self, mock_get): mock_ops = MagicMock() result_obj = MagicMock() result_obj.to_dict.return_value = { "error": "Could not find match for old_string in foo.py" } mock_ops.patch_replace.return_value = result_obj mock_get.return_value = mock_ops from tools.file_tools import patch_tool raw = patch_tool(mode="replace", path="foo.py", old_string="x", new_string="y") # patch_tool surfaces the hint as a structured "_hint" field on the # JSON error payload (not an inline "[Hint: ..." tail). assert json.loads(raw).get("_hint") @patch("tools.file_tools._get_file_ops") def test_success_no_hint(self, mock_get): mock_ops = MagicMock() result_obj = MagicMock() result_obj.to_dict.return_value = {"success": True, "diff": "--- a\n+++ b"} mock_ops.patch_replace.return_value = result_obj mock_get.return_value = mock_ops from tools.file_tools import patch_tool raw = patch_tool(mode="replace", path="foo.py", old_string="x", new_string="y") assert "_hint" not in raw class TestSearchHints: """Search tool should hint when results are truncated.""" def setup_method(self): """Clear read/search tracker between tests to avoid cross-test state.""" from tools.file_tools_read_tracking import _read_tracker _read_tracker.clear() @patch("tools.file_tools._get_file_ops") def test_truncated_hint_with_nonzero_offset(self, mock_get): mock_ops = MagicMock() result_obj = MagicMock() result_obj.to_dict.return_value = { "total_count": 150, "matches": [{"path": "a.py", "line": 1, "content": "x"}] * 50, "truncated": True, } mock_ops.search.return_value = result_obj mock_get.return_value = mock_ops from tools.file_tools import search_tool raw = search_tool(pattern="foo", offset=50, limit=50) parsed = json.loads(raw) assert "offset=100" in parsed["_hint"] # --------------------------------------------------------------------------- # PATCH_SCHEMA shape tests (issue #15524) # --------------------------------------------------------------------------- class TestSensitivePathCheck: """Verify that _check_sensitive_path blocks writes to protected locations.""" def test_hermes_config_blocked_for_write_file(self, tmp_path, monkeypatch): fake_config = tmp_path / "config.yaml" monkeypatch.setattr("tools.file_tools_write_guards._hermes_config_resolved", str(fake_config)) monkeypatch.setattr("tools.file_tools_write_guards._hermes_config_resolved_loaded", True) from tools.file_tools import write_file_tool result = json.loads(write_file_tool(str(fake_config), "approvals:\n mode: off\n")) assert "error" in result assert "Hermes config" in result["error"] def test_system_path_still_blocked(self, monkeypatch): monkeypatch.setattr("tools.file_tools_write_guards._hermes_config_resolved", "/some/other/path") monkeypatch.setattr("tools.file_tools_write_guards._hermes_config_resolved_loaded", True) from tools.file_tools import write_file_tool result = json.loads(write_file_tool("/etc/passwd", "evil")) assert "error" in result assert "sensitive system path" in result["error"] def test_macos_private_var_carveouts(self): """macOS temp dirs under /private/var must not be blanket-blocked, while the genuinely-sensitive /private/var subtrees still are.""" from tools.file_tools_write_guards import _check_sensitive_path # $TMPDIR / /tmp / /var/folders realpath into these on macOS. assert _check_sensitive_path("/private/var/folders/xy/T/tmp.txt") is None assert _check_sensitive_path("/private/var/tmp/build.log") is None # Sensitive subtrees remain blocked. assert _check_sensitive_path("/private/var/db/secret") is not None assert _check_sensitive_path("/private/var/root/x") is not None # /etc (and its macOS /private/etc mirror) stay blocked. assert _check_sensitive_path("/private/etc/hosts") is not None @patch("tools.file_tools._get_file_ops") def test_normal_file_not_blocked(self, mock_get, monkeypatch): monkeypatch.setattr("tools.file_tools_write_guards._hermes_config_resolved", "/home/user/.hermes/config.yaml") monkeypatch.setattr("tools.file_tools_write_guards._hermes_config_resolved_loaded", True) mock_ops = MagicMock() result_obj = MagicMock() result_obj.to_dict.return_value = {"status": "ok", "path": "/tmp/other.txt", "bytes": 5} mock_ops.write_file.return_value = result_obj mock_get.return_value = mock_ops from tools.file_tools import write_file_tool result = json.loads(write_file_tool("/tmp/other.txt", "hello")) assert result["status"] == "ok" class TestPatchSchemaShape: """The BASE schema is replace-only (V4A layers on for OpenAI-family mains via _patch_schema_overrides — see test_patch_v4a_gate.py). The kimi-k2.x per-mode-description concern now applies to the V4A LAYER, whose composed variant still documents both modes' requirements.""" def test_v4a_layer_keeps_per_mode_documentation(self): """When the V4A layer IS rendered (OpenAI-family), the strict-model guidance survives: per-mode requirements in description text, no anyOf/oneOf (breaks Anthropic/Fireworks/Kimi sanitizers).""" from unittest.mock import patch as _p import tools.file_tools as ft with _p("agent.auxiliary_client._read_main_provider", return_value="openai"), \ _p("agent.auxiliary_client._read_main_model", return_value="gpt-5.2"): o = ft._patch_schema_overrides() params = o["parameters"] assert params["required"] == ["mode"] assert {"mode", "patch", "path", "old_string", "new_string"} <= set(params["properties"]) assert "anyOf" not in params and "oneOf" not in params # --------------------------------------------------------------------------- # Session-cwd persistence across env recreation (#26211: silent file creation # failure in long conversations). The durable anchor is the per-session cwd # record in terminal_tool; env cleanup cannot lose it because it never lived # on the env. # --------------------------------------------------------------------------- class TestSessionCwdSurvivesEnvRecreation: """ When the terminal environment is cleaned up and re-created during a long conversation, the session's cwd record preserves the working directory so subsequent file writes with relative paths land in the right directory. Regression guard for issue #26211. """ @patch("tools.terminal_tool._active_environments", new_callable=dict) @patch("tools.file_tools._file_ops_cache", new_callable=dict) @patch("tools.terminal_tool._get_env_config") @patch("tools.terminal_tool_backends._create_environment") def test_recorded_cwd_used_for_recreated_env( self, mock_create_env, mock_config, mock_cache, mock_active ): import tools.terminal_tool as tt from tools.file_tools import _get_file_ops mock_env = MagicMock() mock_env.cwd = "/Users/user/project" mock_create_env.return_value = mock_env mock_config.return_value = { "env_type": "local", "cwd": "/default/path", "timeout": 30, } task_id = "default" # The session's record holds the directory (written by the last # completed terminal command before the env was cleaned up). tt.record_session_cwd(task_id, "/Users/user/project") try: _get_file_ops(task_id) create_call = mock_create_env.call_args assert create_call is not None, "_create_environment was not called" kwargs = create_call.kwargs if create_call.kwargs else {} cwd_passed = kwargs.get("cwd", None) if cwd_passed is None: args = create_call.args if create_call.args else [] if len(args) >= 3: cwd_passed = args[2] assert cwd_passed == "/Users/user/project", \ f"Expected cwd='/Users/user/project', got {cwd_passed!r}" finally: tt.clear_session_cwd(task_id) @patch("tools.terminal_tool._active_environments", new_callable=dict) @patch("tools.file_tools._file_ops_cache", new_callable=dict) @patch("tools.terminal_tool._get_env_config") @patch("tools.terminal_tool_backends._create_environment") def test_stale_cache_cwd_rescued_into_record_on_cleanup_detection( self, mock_create_env, mock_config, mock_cache, mock_active ): """If the env died but the file-ops cache entry survived, its cwd is rescued into the session record before the cache entry is dropped — the recreated env starts where the user left off.""" import tools.terminal_tool as tt from tools.file_tools import _get_file_ops task_id = "default" tt.clear_session_cwd(task_id) # Stale cache entry: env was cleaned up, cache still holds the old cwd. cached = MagicMock() cached.env = None cached.cwd = "/Users/user/project" mock_cache[task_id] = cached mock_env = MagicMock() mock_env.cwd = "/Users/user/project" mock_create_env.return_value = mock_env mock_config.return_value = { "env_type": "local", "cwd": "/config/default/path", "timeout": 30, } try: _get_file_ops(task_id) create_call = mock_create_env.call_args assert create_call is not None, "_create_environment was not called" kwargs = create_call.kwargs if create_call.kwargs else {} cwd_passed = kwargs.get("cwd", None) if cwd_passed is None: args = create_call.args if create_call.args else [] if len(args) >= 3: cwd_passed = args[2] # Rebuilt env restored the rescued cwd, NOT the config default. assert cwd_passed == "/Users/user/project", \ f"Expected restored cwd='/Users/user/project', got {cwd_passed!r}" finally: tt.clear_session_cwd(task_id) class TestSilentFileMisplacementE2E: """Real-IO regression for #26211. Exercises the actual write_file_tool path against a temp filesystem: an agent cd's into a project, the cleanup thread kills the env, and a later relative-path write must land in the project dir (not the config default). Mocks miss this because resolution (_resolve_path_for_task) runs BEFORE _get_file_ops rebuilds the env — only the durable session-cwd record makes the resolved path correct. """ def test_relative_write_after_env_cleanup_lands_in_user_cwd(self, tmp_path, monkeypatch): import tools.terminal_tool as tt import tools.file_tools as ft project = tmp_path / "project" config_default = tmp_path / "config_default" project.mkdir() config_default.mkdir() monkeypatch.delenv("TERMINAL_CWD", raising=False) _orig = tt._get_env_config monkeypatch.setattr( tt, "_get_env_config", lambda: {**_orig(), "env_type": "local", "cwd": str(config_default)}, ) task_id = "default" tt.clear_session_cwd(task_id) # 1) Env alive; agent has cd'd into the project (the completed command # recorded the session cwd — simulate that write here). fo = ft._get_file_ops(task_id) fo.env.cwd = str(project) tt.record_session_cwd(task_id, str(project)) ft.write_file_tool("alive.txt", "1\n", task_id) assert (project / "alive.txt").exists() # 2) Cleanup thread kills the env AND clears the file_ops cache. with tt._env_lock: tt._active_environments.pop(task_id, None) tt._last_activity.pop(task_id, None) with ft._file_ops_lock: ft._file_ops_cache.pop(task_id, None) # 3) The next relative write must still land in the project dir. res = json.loads(ft.write_file_tool("report.txt", "hello\n", task_id)) assert res.get("resolved_path") == str(project / "report.txt"), res assert (project / "report.txt").exists(), "file should be in the user's cwd" assert not (config_default / "report.txt").exists(), \ "file silently misplaced into config default (the #26211 bug)" tt.clear_session_cwd(task_id) class TestDedupInvalidationTaskResolution: """Real-IO regression: dedup eviction must resolve paths per-task. ``_invalidate_dedup_for_path`` looked up the read-tracker under the correct task_id but resolved the path with the DEFAULT task, so for any task whose workspace cwd differs from the process cwd (every ``-w``/Desktop/ACP session using relative paths) the computed key never matched the cached key and the stale-read entry was never evicted. A read after a write could then be served the OLD content stub. """ def test_invalidate_evicts_the_task_resolved_key(self, tmp_path, monkeypatch): import tools.terminal_tool as tt import tools.file_tools as ft workspace = tmp_path / "workspace" proc = tmp_path / "proc" workspace.mkdir() proc.mkdir() monkeypatch.delenv("TERMINAL_CWD", raising=False) monkeypatch.chdir(proc) # process cwd != task workspace task_id = "acp-dedup" monkeypatch.setattr(tt, "_task_env_overrides", {task_id: {"cwd": str(workspace)}}) (workspace / "data.txt").write_text("v1\n", encoding="utf-8") # The task resolves the relative path into the workspace; the default # task (the old buggy resolution) would resolve into proc. from tools.file_tools_paths import _resolve_path_for_task from tools.file_tools_read_tracking import _read_tracker correct = str(_resolve_path_for_task("data.txt", task_id)) buggy = str(_resolve_path_for_task("data.txt")) assert correct != buggy, "test precondition: cwds must diverge" # Populate the dedup cache via a real read. ft.read_file_tool("data.txt", task_id=task_id) keys = [k[0] for k in _read_tracker.get(task_id, {}).get("dedup", {})] assert correct in keys, keys # Invalidate as write_file_tool does; the entry must be gone. from tools.file_tools_read_tracking import _invalidate_dedup_for_path _invalidate_dedup_for_path("data.txt", task_id) remaining = [k[0] for k in _read_tracker.get(task_id, {}).get("dedup", {})] assert correct not in remaining, remaining _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 from tools.file_tools_read_tracking import _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 from tools.file_tools_read_tracking import _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_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 from tools.file_tools_read_tracking import _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 from tools.file_tools_read_tracking import _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 from tools.file_tools_read_tracking import _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_read_tracking import ( _NOT_FOUND_TTL_SECONDS, _check_not_found_cache, _read_tracker, _read_tracker_lock, _record_not_found) 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 _read_tracker_lock: entry = _read_tracker[tid]["not_found"][("read", "/tmp/ttl-test")] _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 _read_tracker_lock: assert ("read", "/tmp/ttl-test") not in _read_tracker[tid].get("not_found", {}) def test_out_of_band_creation_defeats_cached_miss(self, tmp_path): """CRITICAL staleness contract: a file created AFTER a cached miss — by a terminal command or any external process, NOT write_file_tool — must be served for real on the next read. The agent pattern 'check for file → create it → read it' breaks otherwise.""" from tools.file_tools_read_tracking import _check_not_found_cache, _record_not_found, _read_tracker tid = "neg-cache-oob-read" _read_tracker.pop(tid, None) target = tmp_path / "created-later.txt" _record_not_found("read", str(target), tid, '{"error":"File not found: x"}') assert _check_not_found_cache("read", str(target), tid) is not None # Out-of-band creation: plain filesystem write, no tool hook fires. target.write_text("real content\n", encoding="utf-8") # The cached miss must NOT be served once the path exists… assert _check_not_found_cache("read", str(target), tid) is None, ( "stale 'File not found' served after the file was created " "out-of-band — the existence guard regressed" ) # …and the entry is evicted, not just skipped. with __import__("tools.file_tools", fromlist=["x"])._read_tracker_lock: assert ("read", str(target)) not in _read_tracker[tid].get("not_found", {}) def test_out_of_band_creation_defeats_cached_search_miss(self, tmp_path): """Same contract for search roots: creating a file under a previously-missing directory must defeat the cached 'Path not found'.""" from tools.file_tools_read_tracking import _check_not_found_cache, _record_not_found, _read_tracker tid = "neg-cache-oob-search" _read_tracker.pop(tid, None) missing_dir = tmp_path / "later-dir" _record_not_found("search", str(missing_dir), tid, '{"error":"Path not found: x"}') assert _check_not_found_cache("search", str(missing_dir), tid) is not None missing_dir.mkdir() (missing_dir / "x.txt").write_text("hi\n", encoding="utf-8") assert _check_not_found_cache("search", str(missing_dir), tid) is None, ( "stale 'Path not found' served after the directory was created" ) def test_notify_other_tool_call_clears_not_found(self): """Belt-and-suspenders: any non-read tool (terminal etc.) invalidates the task's negative cache via the dispatcher's notify hook.""" from tools.file_tools_read_tracking import _check_not_found_cache, _record_not_found, _read_tracker from tools.file_tools_read_tracking import notify_other_tool_call tid = "neg-cache-notify" _read_tracker.pop(tid, None) _record_not_found("read", "/tmp/never-exists-notify", tid, '{"error":"x"}') assert _check_not_found_cache("read", "/tmp/never-exists-notify", tid) is not None notify_other_tool_call(tid) assert _check_not_found_cache("read", "/tmp/never-exists-notify", tid) is None, ( "notify_other_tool_call must clear cached misses" ) class TestSSHConfigWriteGate: """~/.ssh/config can run commands (ProxyCommand / Match exec), so a write routes through the approval gate and fails closed with nobody to approve. #93201: the gate call once raised TypeError (missing required kwarg) instead of returning an approval decision — drive the real gate end to end.""" @pytest.fixture() def ssh_config(self, tmp_path, monkeypatch): monkeypatch.setenv("HOME", str(tmp_path)) return tmp_path / ".ssh" / "config" def test_no_human_present_blocks_and_writes_nothing(self, ssh_config): from tools.file_tools import write_file_tool result = json.loads(write_file_tool(str(ssh_config), "Host x\n ProxyCommand evil\n")) assert "BLOCKED" in result["error"] assert not ssh_config.exists() def test_single_query_session_denies_with_the_q_mode_message(self, ssh_config, monkeypatch): monkeypatch.setenv("HERMES_SINGLE_QUERY_SESSION", "1") monkeypatch.setenv("HERMES_INTERACTIVE", "1") # `hermes chat -q` exports it too from tools.file_tools import write_file_tool result = json.loads(write_file_tool(str(ssh_config), "Host x\n ProxyCommand evil\n")) assert "BLOCKED" in result["error"] assert "single-query" in result["error"] assert not ssh_config.exists() class TestSecretFileReadRedaction: """#110567: read_file / search_files must classify the RESOLVED path and run the assignment passes for a secret-bearing file, instead of returning an opaque prefix-less credential in cleartext. Same classifier the terminal side uses (``_is_secret_file_arg``), so the two surfaces cannot drift.""" SYNTH = "3JcQ1UqZ8mNp4Rt6vWx2Yb9Ad0Ef7Gh5Ij2kS" # 40-char opaque, no vendor prefix class _Match: def __init__(self, path, content): self.path = path self.content = content class _SearchResult: def __init__(self, matches): self.matches = matches self.files = [] self.counts = {} def to_dict(self, densify=False): return { "total_count": len(self.matches), "matches": [{"path": m.path, "content": m.content} for m in self.matches], } @pytest.fixture def hermes_home(self, tmp_path, monkeypatch): """A Hermes home with no ``.hermes`` segment, like ``%LOCALAPPDATA%\\hermes``.""" import agent.file_safety as file_safety home = tmp_path / "hermes" home.mkdir() monkeypatch.setattr(file_safety, "_hermes_home_path", lambda: home) monkeypatch.setattr(file_safety, "_hermes_root_path", lambda: home) return home @staticmethod def _read_ops(body): ops = MagicMock() result_obj = MagicMock() result_obj.content = body result_obj.to_dict.return_value = {"content": body, "total_lines": body.count("\n")} ops.read_file.return_value = result_obj return ops @patch("tools.file_tools._get_file_ops") def test_read_file_of_hermes_config_masks_opaque_token(self, mock_get, hermes_home): # read_file renders line-numbered content ("5| ADS_API_TOKEN: …"); the gutter is # part of the text the redactor sees, so the fixture must carry it (a gutter-free # fixture would pass even though the real read leaks). body = (f"1|mcp_servers:\n2| nasa_ads:\n3| env:\n" f"4| ADS_API_TOKEN: {self.SYNTH}\n5|MAX_TOKENS: 100\n") mock_get.return_value = self._read_ops(body) from tools.file_tools import read_file_tool out = json.loads(read_file_tool(str(hermes_home / "config.yaml"), task_id="secret-read")) assert self.SYNTH not in out["content"] assert "«redacted" in out["content"] assert "5|MAX_TOKENS: 100" in out["content"] # non-secret scalar and rendered gutter survive # A project's own config.yaml is NOT secret-bearing: source dumps are never mangled. mock_get.return_value = self._read_ops(f"4| ADS_API_TOKEN: {self.SYNTH}\n") out = json.loads(read_file_tool(str(hermes_home.parent / "proj-config.yaml"), task_id="plain-read")) assert self.SYNTH in out["content"] @patch("tools.file_tools._get_file_ops") def test_search_in_hermes_home_masks_opaque_token(self, mock_get, hermes_home): config = hermes_home / "config.yaml" ops = MagicMock() ops.search.return_value = self._SearchResult( [self._Match(str(config), f" ADS_API_TOKEN: {self.SYNTH}")]) mock_get.return_value = ops from tools.file_tools import search_tool raw = search_tool(pattern="ADS_API_TOKEN", path=str(hermes_home), task_id="secret-search") assert self.SYNTH not in raw assert "«redacted" in raw class TestConflictMarkerFlag: def test_read_flags_balanced_conflict_blocks_only(self, tmp_path): conflicted = tmp_path / "c.py" conflicted.write_text("x=1\n<<<<<<< HEAD\ny=2\n=======\ny=3\n>>>>>>> feature\nz=4\n", encoding="utf-8") result = json.loads(read_file_tool(str(conflicted))) assert result["conflict_blocks"] == 1 assert "merge-conflict" in result["_hint"] prose = tmp_path / "p.py" prose.write_text("print('<<<<<<< not a conflict')\n", encoding="utf-8") assert "conflict_blocks" not in json.loads(read_file_tool(str(prose)))