fix(tools): stabilize sandbox read_file results

tests/tools/test_code_execution_programmatic_read.py

@@ -0,0 +1,236 @@
+from __future__ import annotations
+
+import json
+import sys
+
+import pytest
+
+from tools import code_execution_tool
+from tools import file_tools
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="UDS not available on Windows")
+def test_execute_code_read_file_returns_stable_raw_shape(tmp_path):
+    target = tmp_path / "sandbox.txt"
+    target.write_text("alpha\n1|literal\nomega\n", encoding="utf-8")
+    code = f"""
+import json
+from hermes_tools import read_file
+
+first = read_file({str(target)!r})
+second = read_file({str(target)!r})
+print(json.dumps([first, second], sort_keys=True))
+"""
+
+    result = json.loads(
+        code_execution_tool.execute_code(
+            code=code,
+            task_id=f"sandbox-{tmp_path.name}",
+            enabled_tools=["read_file"],
+        )
+    )
+    reads = json.loads(result["output"].strip())
+
+    assert result["status"] == "success"
+    assert result["tool_calls_made"] == 2
+    assert reads[0]["success"] is True
+    assert reads[0]["content"] == "alpha\n1|literal\nomega\n"
+    assert reads[1] == reads[0]
+
+
+def test_programmatic_read_returns_raw_content_on_repeated_calls(tmp_path):
+    target = tmp_path / "sample.txt"
+    target.write_text("alpha\n1|literal\nomega\n", encoding="utf-8")
+    first = json.loads(
+        file_tools.read_file_programmatic_tool(str(target), task_id="sandbox-read")
+    )
+    second = json.loads(
+        file_tools.read_file_programmatic_tool(str(target), task_id="sandbox-read")
+    )
+
+    assert first["success"] is True
+    assert first["content"] == "alpha\n1|literal\nomega\n"
+    assert second == first
+
+
+def test_programmatic_read_failure_keeps_stable_content_key(tmp_path):
+    result = json.loads(
+        file_tools.read_file_programmatic_tool(
+            str(tmp_path / "missing.txt"), task_id="sandbox-read"
+        )
+    )
+
+    assert result["success"] is False
+    assert result["content"] == ""
+    assert result["error"]
+
+
+def test_programmatic_read_preserves_explicit_failure(monkeypatch):
+    monkeypatch.setattr(
+        file_tools,
+        "read_file_tool",
+        lambda **_kwargs: '{"success": false, "note": "blocked"}',
+    )
+
+    result = json.loads(file_tools.read_file_programmatic_tool("blocked.pipe"))
+
+    assert result == {
+        "success": False,
+        "note": "blocked",
+        "content": "",
+    }
+
+
+def test_programmatic_read_preserves_pagination_without_display_gutters(tmp_path):
+    target = tmp_path / "pages.txt"
+    target.write_text("one\ntwo\nthree\nfour\n", encoding="utf-8")
+
+    result = json.loads(
+        file_tools.read_file_programmatic_tool(
+            str(target), offset=2, limit=2, task_id="sandbox-page"
+        )
+    )
+
+    assert result["content"] == "two\nthree\n"
+    assert result["total_lines"] == 4
+    assert result["truncated"] is True
+
+
+def test_programmatic_read_does_not_change_chat_dedup_contract(tmp_path):
+    target = tmp_path / "chat.txt"
+    target.write_text("alpha\nbeta\n", encoding="utf-8")
+    task_id = f"chat-{tmp_path.name}"
+
+    programmatic = json.loads(
+        file_tools.read_file_programmatic_tool(str(target), task_id=task_id)
+    )
+    chat_first = json.loads(file_tools.read_file_tool(str(target), task_id=task_id))
+    chat_second = json.loads(file_tools.read_file_tool(str(target), task_id=task_id))
+
+    assert programmatic["content"] == "alpha\nbeta\n"
+    assert chat_first["content"] == "1|alpha\n2|beta"
+    assert chat_second["status"] == "unchanged"
+    assert "content" not in chat_second
+
+
+def test_sandbox_dispatch_uses_standard_dispatcher(monkeypatch):
+    captured = {}
+
+    def fake_handle_function_call(tool_name, tool_args, task_id=None):
+        captured.update(
+            tool_name=tool_name,
+            tool_args=tool_args,
+            task_id=task_id,
+            programmatic=file_tools._programmatic_read.get(),
+        )
+        return '{"success": true, "content": "raw"}'
+
+    monkeypatch.setattr(
+        "model_tools.handle_function_call",
+        fake_handle_function_call,
+    )
+
+    result = code_execution_tool._dispatch_sandbox_tool_call(
+        "read_file",
+        {"path": "notes.md", "offset": 4, "limit": 7},
+        task_id="task-1",
+    )
+
+    assert json.loads(result)["content"] == "raw"
+    assert captured == {
+        "tool_name": "read_file",
+        "tool_args": {"path": "notes.md", "offset": 4, "limit": 7},
+        "task_id": "task-1",
+        "programmatic": True,
+    }
+    assert file_tools._programmatic_read.get() is False
+
+
+def test_programmatic_read_and_chat_raw_paths_are_byte_identical(tmp_path):
+    """Pin the two content construction paths to identical raw output.
+
+    The early structured-document branch builds its page natively (no
+    gutter, no per-line truncation), while the file_ops branch either adds
+    numbers natively or — with line_numbers=False — applies the same clamp
+    without a gutter. Both must produce byte-identical raw content for the
+    same window so the "stable" programmatic contract cannot drift from
+    the chat path.
+    """
+    target = tmp_path / "paths.txt"
+    long_line = "x" * 3000  # exceeds the default 2000-char per-line clamp
+    target.write_text(f"alpha\n{long_line}\ngamma\n", encoding="utf-8")
+
+    programmatic = json.loads(
+        file_tools.read_file_programmatic_tool(str(target), task_id="pin-prog")
+    )
+    chat_raw = json.loads(
+        file_tools.read_file_tool(str(target), task_id="pin-chat", line_numbers=False)
+    )
+
+    assert programmatic["content"] == chat_raw["content"]
+    assert programmatic["content"].splitlines()[1] != long_line  # clamped identically
+
+
+def test_programmatic_early_document_branch_matches_file_ops_path_byte_for_byte(
+    tmp_path,
+):
+    """Pin the early structured-document branch against the file_ops path.
+
+    For an extractable document (``.ipynb``), ``read_file_tool`` builds the
+    page natively in its early branch — no gutter, no per-line clamp. The
+    same window (offset=2, limit=1) read through a real
+    ``ShellFileOperations.read_file(..., line_numbers=False)`` call must
+    produce byte-identical content, so the "stable" programmatic contract
+    cannot drift between the two construction paths.
+    """
+    from tools.environments.local import LocalEnvironment
+    from tools.file_operations import ShellFileOperations
+
+    notebook = {
+        "cells": [
+            {"cell_type": "code", "source": ["alpha\n", "beta\n"], "outputs": []},
+            {"cell_type": "markdown", "source": ["tail line\n"]},
+        ],
+        "metadata": {},
+        "nbformat": 4,
+        "nbformat_minor": 5,
+    }
+    target = tmp_path / "pins.ipynb"
+    target.write_text(json.dumps(notebook), encoding="utf-8")
+
+    # Chat path: early structured-document branch builds natively (no gutter).
+    chat = json.loads(
+        file_tools.read_file_tool(
+            str(target), offset=2, limit=1, task_id="pin-doc-chat", line_numbers=False
+        )
+    )
+    assert chat.get("extracted_document") is True
+
+    # Programmatic path: same early branch, same window, no gutter.
+    programmatic = json.loads(
+        file_tools.read_file_programmatic_tool(
+            str(target), offset=2, limit=1, task_id="pin-doc-prog"
+        )
+    )
+
+    # file_ops path: mirror the extracted text into a plain-text file and
+    # run the REAL backend read on the identical window with
+    # line_numbers=False. Both paths share canonical newline semantics:
+    # a raw page ends with "\n" (sed/cut always newline-terminate; the
+    # early branch appends it to its joined page).
+    from tools.read_extract import extract_document_text
+
+    # Line 2 of the extraction is the first source line ("alpha"); line 1
+    # is the "# ── Code cell N ──" header the extractor adds.
+    extracted_line2 = extract_document_text(str(target)).splitlines()[1]
+    assert extracted_line2 == "alpha"  # sanity: non-empty page content
+    mirror = tmp_path / "mirror.txt"
+    mirror.write_text(f"sentinel\n{extracted_line2}\ntail\n", encoding="utf-8")
+
+    file_ops = ShellFileOperations(LocalEnvironment())
+    raw = file_ops.read_file(str(mirror), offset=2, limit=1, line_numbers=False)
+
+    assert not raw.error
+    expected_page = f"{extracted_line2}\n"
+    # Byte-identical page content across all three construction paths.
+    assert programmatic["content"] == chat["content"] == raw.content == expected_page

tests/tools/test_file_read_guards.py

@@ -50,7 +50,7 @@ def to_dict(self):
 
 def _make_fake_ops(content="hello\n", total_lines=1, file_size=6):
     fake = MagicMock()
-    fake.read_file = lambda path, offset=1, limit=500: _FakeReadResult(
+    fake.read_file = lambda path, offset=1, limit=500, **kwargs: _FakeReadResult(
         content=content, total_lines=total_lines, file_size=file_size,
     )
     return fake
@@ -694,8 +694,8 @@ def test_large_truncated_file_gets_hint(self, mock_ops):
         fake = _make_fake_ops(content=content, total_lines=10000, file_size=600_000)
         # Make to_dict return truncated=True
         orig_read = fake.read_file
-        def patched_read(path, offset=1, limit=500):
-            r = orig_read(path, offset, limit)
+        def patched_read(path, offset=1, limit=500, **kwargs):
+            r = orig_read(path, offset, limit, **kwargs)
             orig_to_dict = r.to_dict
             def new_to_dict():
                 d = orig_to_dict()
@@ -787,7 +787,7 @@ def test_write_invalidates_dedup_same_second(self, mock_ops):
         stub because the mtime comparison saw no change.
         """
         fake = MagicMock()
-        fake.read_file = lambda path, offset=1, limit=500: _FakeReadResult(
+        fake.read_file = lambda path, offset=1, limit=500, **kwargs: _FakeReadResult(
             content="original content\n", total_lines=1, file_size=18,
         )
         fake.write_file = lambda path, content: MagicMock(
@@ -804,7 +804,7 @@ def test_write_invalidates_dedup_same_second(self, mock_ops):
         write_file_tool(self._tmpfile, "new content\n", task_id="wr")
 
         # 3. Read again — should get full content, NOT dedup stub.
-        fake.read_file = lambda path, offset=1, limit=500: _FakeReadResult(
+        fake.read_file = lambda path, offset=1, limit=500, **kwargs: _FakeReadResult(
             content="new content\n", total_lines=1, file_size=13,
         )
         r2 = json.loads(read_file_tool(self._tmpfile, task_id="wr"))
@@ -816,7 +816,7 @@ def test_write_invalidates_dedup_same_second(self, mock_ops):
     def test_write_invalidates_all_offsets(self, mock_ops):
         """A write invalidates dedup entries for ALL offset/limit combos."""
         fake = MagicMock()
-        fake.read_file = lambda path, offset=1, limit=500: _FakeReadResult(
+        fake.read_file = lambda path, offset=1, limit=500, **kwargs: _FakeReadResult(
             content="line1\nline2\nline3\n", total_lines=3, file_size=20,
         )
         fake.write_file = lambda path, content: MagicMock(

tests/tools/test_file_staleness.py

@@ -58,7 +58,7 @@ def to_dict(self):
 
 def _make_fake_ops(read_content="hello\n", file_size=6):
     fake = MagicMock()
-    fake.read_file = lambda path, offset=1, limit=500: _FakeReadResult(
+    fake.read_file = lambda path, offset=1, limit=500, **kwargs: _FakeReadResult(
         content=read_content, total_lines=1, file_size=file_size,
     )
     fake.write_file = lambda path, content: _FakeWriteResult()

tests/tools/test_file_tools.py

@@ -6,6 +6,7 @@
 
 import json
 import logging
+from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -70,7 +71,7 @@ def test_writes_content(self, mock_get):
         from tools.file_tools import write_file_tool
         result = json.loads(write_file_tool("/tmp/out.txt", "hello world!\n"))
         assert result["status"] == "ok"
-        mock_ops.write_file.assert_called_once_with("/tmp/out.txt", "hello world!\n")
+        mock_ops.write_file.assert_called_once_with(str(Path("/tmp/out.txt").resolve()), "hello world!\n")
 
     @patch("tools.file_tools._get_file_ops")
     def test_permission_error_returns_error_json_without_error_log(self, mock_get, caplog):
@@ -161,7 +162,7 @@ def test_replace_mode_calls_patch_replace(self, mock_get):
             old_string="foo", new_string="bar"
         ))
         assert result["status"] == "ok"
-        mock_ops.patch_replace.assert_called_once_with("/tmp/f.py", "foo", "bar", False)
+        mock_ops.patch_replace.assert_called_once_with(str(Path("/tmp/f.py").resolve()), "foo", "bar", False)
 
 
     @patch("tools.file_tools._get_file_ops")

tests/tools/test_programmatic_read_raw_repeat.py

@@ -0,0 +1,24 @@
+"""Regression for #93749: programmatic reads must stay raw on repeat."""
+
+from __future__ import annotations
+
+import json
+
+from tools import file_tools
+
+
+def test_programmatic_read_returns_raw_content_on_repeated_calls(tmp_path):
+    target = tmp_path / "sample.txt"
+    raw = "alpha\n1|literal\nomega\n"
+    target.write_text(raw, encoding="utf-8")
+    programmatic = getattr(file_tools, "read_file_programmatic_tool", None)
+    if programmatic is not None:
+        first = json.loads(programmatic(str(target), task_id="sandbox-read"))
+        second = json.loads(programmatic(str(target), task_id="sandbox-read"))
+    else:
+        first = json.loads(file_tools.read_file_tool(str(target), task_id="sandbox-read"))
+        second = json.loads(file_tools.read_file_tool(str(target), task_id="sandbox-read"))
+
+    assert first["success"] is True
+    assert first["content"] == raw
+    assert second == first

tests/tools/test_read_loop_detection.py

@@ -34,7 +34,7 @@ def to_dict(self):
         return {"content": self.content, "total_lines": self._total_lines}
 
 
-def _fake_read_file(path, offset=1, limit=500):
+def _fake_read_file(path, offset=1, limit=500, **kwargs):
     return _FakeReadResult(content=f"content of {path}", total_lines=10)
 
 

tools/code_execution_rpc.py

@@ -57,7 +57,12 @@ def _handle_rpc_request(request: dict, *, allowed_tools: frozenset, tool_call_co
     # Silence handler status prints so they don't leak into the CLI spinner.
     try:
         with thread_scoped_silence():
-            result = dispatch(tool_name, tool_args)
+            if tool_name == "read_file":
+                from tools.file_tools import programmatic_read_context
+                with programmatic_read_context():
+                    result = dispatch(tool_name, tool_args)
+            else:
+                result = dispatch(tool_name, tool_args)
     except Exception as exc:
         logger.error("Tool call failed in %s: %s", where, exc, exc_info=True)
         result = tool_error(str(exc))