fix(tools): refuse patch/write_file when content contains a truncation placeholder

tests/tools/test_file_tools.py

@@ -999,3 +999,136 @@ def test_notify_other_tool_call_clears_not_found(self):
         assert _check_not_found_cache("read", "/tmp/never-exists-notify", tid) is None, (
             "notify_other_tool_call must clear cached misses"
         )
+
+
+class TestTruncationPlaceholderGuard:
+    """Regression tests for #83714 — the ``patch``/``write_file`` tools must
+    refuse to persist a truncation-placeholder marker (e.g. ``...[truncated]``)
+    instead of writing it literally into the target file.
+    """
+
+    def test_find_truncation_placeholder_detects_known_markers(self):
+        from tools.file_tools import _find_truncation_placeholder
+
+        assert _find_truncation_placeholder("hello ...[truncated]") == "...[truncated]"
+        assert _find_truncation_placeholder("hello … [truncated] world") == "… [truncated]"
+        assert _find_truncation_placeholder("plain [truncated] marker") == "[truncated]"
+        assert _find_truncation_placeholder("// ... unchanged ...\ndef f(): pass")
+        assert _find_truncation_placeholder("normal content, nothing odd here") is None
+        assert _find_truncation_placeholder("") is None
+        assert _find_truncation_placeholder(None) is None
+
+    @patch("tools.file_tools._get_file_ops")
+    def test_write_file_rejects_truncation_placeholder(self, mock_get):
+        from tools.file_tools import write_file_tool
+
+        content = "line one\nline two\n...[truncated]\nline four\n"
+        result = json.loads(write_file_tool("/tmp/corrupt.py", content))
+
+        assert "error" in result
+        assert "truncation placeholder" in result["error"].lower()
+        mock_get.assert_not_called()
+
+    @patch("tools.file_tools._get_file_ops")
+    def test_patch_replace_rejects_truncation_placeholder_in_new_string(self, mock_get):
+        from tools.file_tools import patch_tool
+
+        result = json.loads(patch_tool(
+            mode="replace",
+            path="/tmp/f.py",
+            old_string="def foo():\n    pass\n",
+            new_string="def foo():\n    do_thing()\n...[truncated]\n",
+        ))
+
+        assert "error" in result
+        assert "truncation placeholder" in result["error"].lower()
+        mock_get.return_value.patch_replace.assert_not_called()
+
+    @patch("tools.file_tools._get_file_ops")
+    def test_patch_replace_allows_marker_already_present_in_old_string(self, mock_get):
+        """A legitimate edit to text that already mentions the marker (e.g.
+        this very test file, or docs about the bug) must not be blocked —
+        only NEW occurrences introduced by new_string are suspicious."""
+        mock_ops = MagicMock()
+        result_obj = MagicMock()
+        result_obj.to_dict.return_value = {"status": "ok"}
+        mock_ops.patch_replace.return_value = result_obj
+        mock_get.return_value = mock_ops
+
+        from tools.file_tools import patch_tool
+        marker_text = "marker = '...[truncated]'\n"
+        result = json.loads(patch_tool(
+            mode="replace",
+            path="/tmp/f.py",
+            old_string=marker_text,
+            new_string="marker = '...[truncated]'  # renamed\n",
+        ))
+        assert result["status"] == "ok"
+        mock_ops.patch_replace.assert_called_once()
+
+    @patch("tools.file_tools._get_file_ops")
+    def test_patch_v4a_rejects_truncation_placeholder_in_added_line(self, mock_get):
+        from tools.file_tools import patch_tool
+
+        patch_text = (
+            "*** Begin Patch\n"
+            "*** Update File: /tmp/f.py\n"
+            "@@ def foo():\n"
+            "-    pass\n"
+            "+    do_thing()\n"
+            "+    ...[truncated]\n"
+            "*** End Patch\n"
+        )
+        result = json.loads(patch_tool(mode="patch", patch=patch_text))
+
+        assert "error" in result
+        assert "truncation placeholder" in result["error"].lower()
+        mock_get.return_value.patch_v4a.assert_not_called()
+
+    @patch("tools.file_tools._get_file_ops")
+    def test_patch_v4a_allows_marker_on_removed_or_context_line(self, mock_get):
+        """The marker must only be checked against ADDED content — a patch
+        that removes a line containing it, or merely has it as unchanged
+        context, is not the model truncating its own output."""
+        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"
+            "*** Update File: /tmp/f.py\n"
+            "@@ def foo():\n"
+            " marker = '...[truncated]'\n"
+            "-    old_call()\n"
+            "+    new_call()\n"
+            "*** End Patch\n"
+        )
+        result = json.loads(patch_tool(mode="patch", patch=patch_text))
+        assert result["status"] == "ok"
+        mock_ops.patch_v4a.assert_called_once()
+
+    @patch("tools.file_tools._get_file_ops")
+    def test_patch_v4a_add_file_rejects_truncation_placeholder(self, mock_get):
+        from tools.file_tools import patch_tool
+
+        patch_text = (
+            "*** Begin Patch\n"
+            "*** Add File: /tmp/new_module.py\n"
+            "+def foo():\n"
+            "+    ...[truncated]\n"
+            "*** End Patch\n"
+        )
+        result = json.loads(patch_tool(mode="patch", patch=patch_text))
+
+        assert "error" in result
+        assert "truncation placeholder" in result["error"].lower()
+        mock_get.return_value.patch_v4a.assert_not_called()
+
+    def test_extract_v4a_added_content_falls_back_to_raw_text_on_parse_failure(self):
+        from tools.file_tools import _extract_v4a_added_content
+
+        garbage = "not a real v4a patch at all"
+        assert _extract_v4a_added_content(garbage) == garbage

tools/file_tools.py

@@ -7,6 +7,7 @@
 import logging
 import os
 import posixpath
+import re
 import sys
 import threading
 from pathlib import Path, PurePosixPath
@@ -1332,6 +1333,81 @@ def _is_internal_file_tool_content(content: str) -> bool:
     )
 
 
+# #83714 — models occasionally emit a truncation-placeholder marker in a
+# ``write_file``/``patch`` string argument instead of the full content it
+# was supposed to replace (observed with deepseek-v4-pro via the deepseek
+# provider). Hermes itself uses this exact marker to signal truncated tool
+# *output* elsewhere (todo_tool.py, file_operations.py's per-line cap,
+# mcp_tool.py) — the leading theory is the model is imitating a pattern it
+# has seen in its own context rather than Hermes truncating the argument
+# in transit (no length-based slicing of ``content``/``new_string`` exists
+# anywhere between JSON-parsing the tool call and the write/patch below).
+# Whatever the mechanism, writing the marker literally corrupts the file,
+# so it's caught here and refused before any bytes touch disk.
+_TRUNCATION_PLACEHOLDER_PATTERNS = [
+    re.compile(r'\.\.\.\s*\[truncated\]', re.IGNORECASE),
+    re.compile(r'…\s*\[truncated\]', re.IGNORECASE),  # "… [truncated]"
+    re.compile(r'\[truncated\]', re.IGNORECASE),
+    re.compile(r'//\s*\.\.\.\s*(?:rest\s+)?unchanged\s*\.\.\.', re.IGNORECASE),
+    re.compile(r'/\*\s*\.\.\.\s*(?:rest\s+)?unchanged\s*\.\.\.\s*\*/', re.IGNORECASE),
+]
+
+
+def _find_truncation_placeholder(text: str | None) -> str | None:
+    """Return the matched placeholder substring if ``text`` contains a
+    truncation-placeholder marker, else None.
+
+    These markers are legitimate in Hermes's own tool *output* (truncated
+    reads, search results) but never legitimate in the content a model is
+    asking to *write* — a real file that happens to contain the literal
+    string ``[truncated]`` is vanishingly unlikely, and the cost of a false
+    positive (a clear error asking the model to retry) is far lower than
+    the cost of a false negative (silent file corruption).
+    """
+    if not text:
+        return None
+    for pattern in _TRUNCATION_PLACEHOLDER_PATTERNS:
+        match = pattern.search(text)
+        if match:
+            return match.group(0)
+    return None
+
+
+def _extract_v4a_added_content(patch_content: str) -> str:
+    """Return only the text a V4A patch is ADDING: '+' hunk lines and full
+    Add-File bodies. Falls back to the raw patch text if it fails to parse
+    so detection degrades gracefully instead of silently skipping — a
+    malformed patch is exactly the kind of thing worth double-checking.
+    """
+    try:
+        from tools.patch_parser import parse_v4a_patch, OperationType
+        operations, parse_error = parse_v4a_patch(patch_content)
+        if parse_error or not operations:
+            return patch_content
+        chunks: list[str] = []
+        for op in operations:
+            if op.operation == OperationType.ADD and op.content:
+                chunks.append(op.content)
+            for hunk in op.hunks:
+                for line in hunk.lines:
+                    if line.prefix == '+':
+                        chunks.append(line.content)
+        return "\n".join(chunks)
+    except Exception:
+        return patch_content
+
+
+def _truncation_placeholder_error(where: str, marker: str) -> str:
+    return (
+        f"Refusing to write {where}: it contains what looks like a "
+        f"truncation placeholder ({marker!r}) instead of the actual "
+        "content. This usually means the content was abbreviated rather "
+        "than written out in full. Emit the complete text with no "
+        "'...[truncated]'-style markers, then retry. The file was NOT "
+        "modified."
+    )
+
+
 def _get_file_ops(task_id: str = "default") -> ShellFileOperations:
     """Get or create ShellFileOperations for a terminal environment.
 
@@ -2127,6 +2203,9 @@ def write_file_tool(path: str, content: str, task_id: str = "default",
             "Strip read_file line-number prefixes or reconstruct the intended "
             "file contents before writing."
         )
+    _placeholder = _find_truncation_placeholder(content)
+    if _placeholder:
+        return tool_error(_truncation_placeholder_error(f"{path!r}", _placeholder))
     try:
         # Resolve once for the registry lock + stale check.  Failures here
         # fall back to the legacy path — write proceeds, per-task staleness
@@ -2304,6 +2383,15 @@ def _reject_v4a_traversal(v4a_path: str) -> str | None:
                     return tool_error("path required")
                 if old_string is None or new_string is None:
                     return tool_error("old_string and new_string required")
+                # #83714 — only flag a marker that's new to new_string. If
+                # old_string already contains the same literal text (e.g. a
+                # legitimate edit to a docstring that mentions "[truncated]"),
+                # this is not the model abbreviating its own output.
+                _placeholder = _find_truncation_placeholder(new_string)
+                if _placeholder and _placeholder.lower() not in (old_string or "").lower():
+                    return tool_error(
+                        _truncation_placeholder_error("new_string", _placeholder)
+                    )
                 # Pass the resolved ABSOLUTE path to the shell layer so it
                 # operates on the exact file the tool layer resolved — the
                 # shell's own cwd may differ (worktree-cwd bug), and a relative
@@ -2314,6 +2402,17 @@ def _reject_v4a_traversal(v4a_path: str) -> str | None:
             elif mode == "patch":
                 if not patch:
                     return tool_error("patch content required")
+                # #83714 — scan only the content the model is actually ADDING
+                # (V4A '+' hunk lines and full Add-File bodies), not the whole
+                # patch text. Scanning the whole patch would false-positive on
+                # a legitimate edit that removes, or merely anchors context
+                # on, a pre-existing placeholder-like literal.
+                _added = _extract_v4a_added_content(patch)
+                _placeholder = _find_truncation_placeholder(_added)
+                if _placeholder:
+                    return tool_error(
+                        _truncation_placeholder_error("patch content", _placeholder)
+                    )
                 # Rewrite V4A headers to the resolved absolute paths so the
                 # shell layer patches the exact files the tool layer resolved
                 # (locked/reported). Without this a relative header re-resolves