fix(file-tools): abort write/patch when AI truncation placeholders detected

tests/tools/test_file_tools.py

@@ -871,3 +871,203 @@ def test_relative_write_after_env_cleanup_lands_in_user_cwd(self, tmp_path, monk
             "file silently misplaced into config default (the #26211 bug)"
 
         tt.clear_session_cwd(task_id)
+
+
+class TestTruncationSignatureGuard:
+    """Regression tests for AI truncation placeholder detection (issue #20805).
+
+    When an AI model omits file sections it considers unchanged, it often
+    emits placeholder strings like '// ... unchanged ...' or
+    '/* ... full function ... */'. Writing these to disk produces corrupt
+    files that cannot be compiled or run. The guard must block write, replace,
+    and V4A patch paths -- but the V4A check must only inspect content the
+    patch actually ADDS, not removed or context lines (a legitimate edit
+    that deletes, or merely anchors on, a pre-existing placeholder-like
+    literal must not be blocked).
+    """
+
+    def test_check_truncation_signatures_detects_placeholder(self):
+        from tools.file_tools import _check_truncation_signatures
+        result = _check_truncation_signatures("def foo():\n    // ... unchanged ...\n    pass")
+        assert result is not None
+        assert "truncation placeholder" in result.lower()
+
+    def test_check_truncation_signatures_allows_preexisting(self):
+        from tools.file_tools import _check_truncation_signatures
+        original = "def foo():\n    // ... unchanged ...\n    pass"
+        new_content = original + "\ndef bar(): pass"
+        result = _check_truncation_signatures(new_content, original)
+        assert result is None, "Pre-existing placeholder must not block the write"
+
+    def test_check_truncation_signatures_clean_content_passes(self):
+        from tools.file_tools import _check_truncation_signatures
+        result = _check_truncation_signatures("def foo():\n    return 42\n")
+        assert result is None
+
+    def test_write_file_blocks_on_truncation_placeholder(self, tmp_path, monkeypatch):
+        import tools.file_tools as ft
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        task_file = tmp_path / "blocked.py"
+        content_with_placeholder = (
+            "def foo():\n"
+            "    # ... rest of file ...\n"
+            "    pass\n"
+        )
+        result = json.loads(ft.write_file_tool(str(task_file), content_with_placeholder))
+        assert result.get("error"), (
+            "write_file_tool must return an error for content with "
+            "AI truncation placeholders, got: " + str(result)
+        )
+        assert "truncation" in str(result.get("error", "")).lower()
+
+    def test_write_file_allows_clean_content(self, tmp_path, monkeypatch):
+        import tools.file_tools as ft
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        task_file = tmp_path / "clean.py"
+        result = json.loads(ft.write_file_tool(str(task_file), "def foo():\n    return 42\n"))
+        assert not result.get("error"), f"Clean write must succeed: {result}"
+
+    # -- replace mode ---------------------------------------------------
+
+    @patch("tools.file_tools._get_file_ops")
+    def test_replace_mode_blocks_placeholder_in_new_string(self, mock_get):
+        from tools.file_tools import patch_tool
+        mock_get.return_value = MagicMock()
+
+        result = json.loads(patch_tool(
+            mode="replace", path="/tmp/f.py",
+            old_string="def foo(): pass",
+            new_string="def foo():\n    // ... unchanged ...\n    pass",
+        ))
+        assert result.get("error")
+        assert "truncation" in str(result.get("error", "")).lower()
+        mock_get.return_value.patch_replace.assert_not_called()
+
+    @patch("tools.file_tools._get_file_ops")
+    def test_replace_mode_allows_preexisting_placeholder_in_old_string(self, mock_get):
+        """A placeholder already present in old_string (being replaced away,
+        or retained in new_string unchanged) must not block the edit."""
+        from tools.file_tools import patch_tool
+        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
+
+        old = "def foo():\n    // ... unchanged ...\n    pass"
+        new = "def foo():\n    // ... unchanged ...\n    return 1"
+        result = json.loads(patch_tool(mode="replace", path="/tmp/f.py", old_string=old, new_string=new))
+        assert not result.get("error"), result
+        mock_ops.patch_replace.assert_called_once()
+
+    # -- V4A patch mode ---------------------------------------------------
+
+    @patch("tools.file_tools._get_file_ops")
+    def test_patch_v4a_blocks_placeholder_on_added_line(self, mock_get):
+        from tools.file_tools import patch_tool
+        mock_get.return_value = MagicMock()
+
+        patch_text = (
+            "*** Begin Patch\n"
+            "*** Update File: foo.py\n"
+            "@@ def foo():\n"
+            "-    return 1\n"
+            "+    // ... unchanged ...\n"
+            "*** End Patch\n"
+        )
+        result = json.loads(patch_tool(mode="patch", patch=patch_text))
+        assert result.get("error"), result
+        assert "truncation" in str(result.get("error", "")).lower()
+        mock_get.return_value.patch_v4a.assert_not_called()
+
+    @patch("tools.file_tools._get_file_ops")
+    def test_patch_v4a_allows_placeholder_on_removed_line(self, mock_get):
+        """Regression: deleting a pre-existing placeholder-like literal (it
+        appears only on a '-' line, never introduced by this patch) must not
+        be blocked. Scanning the raw serialized patch text -- as the
+        original PR's unconditional check did -- would false-positive here."""
+        from tools.file_tools import patch_tool
+        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
+
+        patch_text = (
+            "*** Begin Patch\n"
+            "*** Update File: foo.py\n"
+            "@@ def foo():\n"
+            "-    // ... unchanged ...\n"
+            "+    return 1\n"
+            "*** End Patch\n"
+        )
+        result = json.loads(patch_tool(mode="patch", patch=patch_text))
+        assert not result.get("error"), result
+        mock_ops.patch_v4a.assert_called_once()
+
+    @patch("tools.file_tools._get_file_ops")
+    def test_patch_v4a_allows_placeholder_on_context_line(self, mock_get):
+        """A placeholder-like literal that only appears as unchanged context
+        (anchoring the hunk) must not be blocked either."""
+        from tools.file_tools import patch_tool
+        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
+
+        patch_text = (
+            "*** Begin Patch\n"
+            "*** Update File: foo.py\n"
+            "@@ def foo():\n"
+            "     // ... unchanged ...\n"
+            "-    return 1\n"
+            "+    return 2\n"
+            "*** End Patch\n"
+        )
+        result = json.loads(patch_tool(mode="patch", patch=patch_text))
+        assert not result.get("error"), result
+        mock_ops.patch_v4a.assert_called_once()
+
+    @patch("tools.file_tools._get_file_ops")
+    def test_patch_v4a_blocks_placeholder_in_add_file_content(self, mock_get):
+        """A brand-new file (Add File) containing a placeholder must still
+        be blocked -- the whole file content is 'added'."""
+        from tools.file_tools import patch_tool
+        mock_get.return_value = MagicMock()
+
+        patch_text = (
+            "*** Begin Patch\n"
+            "*** Add File: new_module.py\n"
+            "+def foo():\n"
+            "+    // ... unchanged ...\n"
+            "*** End Patch\n"
+        )
+        result = json.loads(patch_tool(mode="patch", patch=patch_text))
+        assert result.get("error"), result
+        mock_get.return_value.patch_v4a.assert_not_called()
+
+    def test_extract_v4a_added_content_only_includes_plus_lines(self):
+        from tools.file_tools import _extract_v4a_added_content
+
+        patch_text = (
+            "*** Begin Patch\n"
+            "*** Update File: foo.py\n"
+            "@@ def foo():\n"
+            "     context line\n"
+            "-    removed line\n"
+            "+    added line\n"
+            "*** End Patch\n"
+        )
+        added = _extract_v4a_added_content(patch_text)
+        assert "added line" in added
+        assert "removed line" not in added
+        assert "context line" not in added
+
+    def test_extract_v4a_added_content_falls_back_on_parse_failure(self):
+        """Malformed/unparseable patches fall back to raw-text scanning so a
+        real truncation placeholder in a broken patch isn't silently missed."""
+        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

@@ -1569,6 +1569,82 @@ def _mark_verification_stale(
         logger.debug("verification stale marker failed", exc_info=True)
 
 
+# AI truncation signatures: placeholder strings that indicate the model
+# produced incomplete/abbreviated content rather than real file content.
+_TRUNCATION_SIGNATURES: tuple[str, ...] = (
+    "/* ... full function ... */",
+    "/* ... unchanged ... */",
+    "// ... unchanged ...",
+    "// ... rest of file ...",
+    "# ... rest of file ...",
+    "# ... unchanged ...",
+    "/* ... rest unchanged ... */",
+    "... (rest of file unchanged)",
+    "... (unchanged)",
+    "<!-- ... unchanged ... -->",
+    "# ... (rest of the function remains the same)",
+    "// ... (rest of the function remains the same)",
+)
+
+
+def _check_truncation_signatures(content: str, original: str | None = None) -> str | None:
+    """Return an error message if content contains AI truncation placeholders.
+
+    Only flags signatures absent from the original file so legitimate comments
+    are not blocked. When original is None the check is unconditional.
+    Must be called through the backend-aware file_ops pipeline so docker/modal/
+    SSH environments see the same file as the eventual write.
+    """
+    content_lower = content.lower()
+    for sig in _TRUNCATION_SIGNATURES:
+        sig_lower = sig.lower()
+        if sig_lower in content_lower:
+            if original is None or sig_lower not in original.lower():
+                return (
+                    f"Refusing to write: content contains AI truncation placeholder "
+                    f"{sig!r}. This indicates the content is incomplete — "
+                    "re-read the file fully and reconstruct the complete content "
+                    "before writing."
+                )
+    return None
+
+
+def _extract_v4a_added_content(patch_content: str) -> str:
+    """Return only the content a V4A patch actually *adds*.
+
+    V4A hunks represent added ('+'), removed ('-'), and unchanged context
+    (' ') lines distinctly. Scanning the whole serialized patch for
+    truncation placeholders would false-positive on a legitimate edit that
+    removes, or merely anchors context on, a pre-existing placeholder-like
+    literal (e.g. deleting an old "// ... unchanged ..." stub). Restricting
+    the check to '+' hunk lines and full Add-file content means only text
+    the model is actually introducing gets checked.
+    """
+    try:
+        from tools.patch_parser import OperationType, parse_v4a_patch
+
+        operations, error = parse_v4a_patch(patch_content)
+    except Exception:
+        operations, error = [], "parse failed"
+
+    if error or not operations:
+        # Fall back to scanning the raw patch text if parsing fails here --
+        # the real V4A application below will surface the actual parse
+        # error; better a possible false positive on malformed input than
+        # silently skipping the truncation check entirely.
+        return patch_content
+
+    added_parts: list[str] = []
+    for op in operations:
+        if op.operation == OperationType.ADD and op.content:
+            added_parts.append(op.content)
+        for hunk in op.hunks:
+            for hunk_line in hunk.lines:
+                if hunk_line.prefix == "+":
+                    added_parts.append(hunk_line.content)
+    return "\n".join(added_parts)
+
+
 def write_file_tool(path: str, content: str, task_id: str = "default",
                     cross_profile: bool = False,
                     session_id: str | None = None) -> str:
@@ -1593,6 +1669,12 @@ 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."
         )
+    # Reject AI truncation placeholders unconditionally on write: there is no
+    # original to compare against at this point, and a write_file that contains
+    # "// ... unchanged ..." is never correct.
+    _trunc_err = _check_truncation_signatures(content)
+    if _trunc_err:
+        return tool_error(_trunc_err)
     try:
         # Resolve once for the registry lock + stale check.  Failures here
         # fall back to the legacy path — write proceeds, per-task staleness
@@ -1771,10 +1853,25 @@ def _reject_v4a_traversal(v4a_path: str) -> str | None:
                 # path would let the two layers disagree about which file is
                 # being edited.
                 _replace_target = _path_to_resolved.get(path) or path
+                # Guard against AI truncation placeholders in new_string.
+                # Compare against old_string so a pre-existing placeholder
+                # in the region being replaced is not re-flagged.
+                _trunc_err = _check_truncation_signatures(new_string, old_string)
+                if _trunc_err:
+                    return tool_error(_trunc_err)
                 result = file_ops.patch_replace(_replace_target, old_string, new_string, replace_all)
             elif mode == "patch":
                 if not patch:
                     return tool_error("patch content required")
+                # V4A "+/* ... full function ... */" patterns are a common
+                # truncation placeholder in patch payloads. Only the content
+                # the patch actually ADDS is checked -- a legitimate edit
+                # that removes, or anchors context on, a pre-existing
+                # placeholder-like literal on a '-' or ' ' hunk line must not
+                # be blocked.
+                _trunc_err = _check_truncation_signatures(_extract_v4a_added_content(patch))
+                if _trunc_err:
+                    return tool_error(_trunc_err)
                 result = file_ops.patch_v4a(patch)
             else:
                 return tool_error(f"Unknown mode: {mode}")