Preserve file bytes/line endings in edits; harden plan-mode diff guard

Editing tools no longer corrupt files that are not plain LF/UTF-8:
- Edit/Insert/diffapply/Write use errors="surrogateescape" + newline="",
  so invalid UTF-8 bytes (Latin-1, GBK, ...) and the file's own line
  endings survive the read/write round trip; model-supplied text (LF)
  is translated to the file's CRLF convention instead of rewriting it
- diffapply adopts the file's ending (CRLF/CR/LF) for added lines, so a
  diff no longer leaves mixed endings behind
- Write no longer fails on a non-UTF-8 existing file (UnicodeDecodeError
  escaped the OSError handler and aborted the write)
- recorded diffs are sanitized for display (lone surrogates -> U+FFFD),
  so the TUI can print diffs of non-UTF-8 files

Plan mode: an Edit diff is applied only when every section — old and new
path — resolves to the plan file.  The macOS/Windows Python applier
honors absolute `+++` targets, which bypassed the path-only guard;
unparseable diffs now fail closed.  Session._plan_diff_verdict reuses
the shared _patch_cwd/diff_targets resolution so the guard approves
exactly what the backends would write.

Tests: regression coverage for all of the above, plus suite hygiene —
SESSION_DIR redirects are restored and sandbox temp dirs removed, so a
run no longer leaks /tmp dirs or resurrects deleted ones.

Author: beacoder

python_agent_harness/diffrender.py

@@ -3,18 +3,39 @@
 ``unified_diff`` builds a standard unified diff between two file
 contents (used by Edit/Write to record what actually changed).
 ``render_diff`` turns that text into a red/green ``rich`` renderable
-suitable for the TUI's tool-output panel.
+suitable for the TUI's tool-output panel.  ``sanitize_for_display``
+makes a recorded diff printable when it carries surrogate-escaped
+bytes from a non-UTF-8 file.
 """
 
 from __future__ import annotations
 
 import difflib
+import re
 
 from rich.console import Group
 from rich.text import Text
 
 MAX_DIFF_LINES = 400  # truncation cap for the rendered (not stored) diff
 
+# Lone surrogates (U+DC80-U+DCFF plus any other surrogate code point):
+# what a file read with errors="surrogateescape" produces for bytes that
+# are not valid UTF-8.
+_SURROGATE_RE = re.compile(r"[\ud800-\udfff]")
+
+
+def sanitize_for_display(text: str) -> str:
+    """Replace lone surrogates in TEXT with U+FFFD.
+
+    Files are read with ``errors="surrogateescape"`` so invalid bytes
+    survive a read/write round trip; those bytes appear as lone
+    surrogates in the string, which cannot be encoded to the terminal
+    (``print`` would raise ``UnicodeEncodeError``).  A recorded diff is
+    display-only, so the surrogates are replaced here — the file on
+    disk is never touched.
+    """
+    return _SURROGATE_RE.sub("\ufffd", text)
+
 
 def unified_diff(
     old_content: str,

python_agent_harness/session.py

@@ -18,6 +18,7 @@
 
 from . import config
 from .client import Client, LLMClient
+from .diffrender import sanitize_for_display
 from .mcp.config import MCPConfig
 from .mcp.manager import MCPManager
 from .models import AgentMode
@@ -327,11 +328,16 @@ def execute_tool(
         return result
 
     def record_diff(self, diff_text: str) -> None:
-        """Attach a unified diff to the tool call currently executing."""
+        """Attach a unified diff to the tool call currently executing.
+
+        Sanitized for display: file content read with
+        ``errors="surrogateescape"`` (non-UTF-8 files) carries lone
+        surrogates that a terminal cannot encode.
+        """
         call_id = getattr(self._active_call, "call_id", None)
         if call_id and diff_text:
             with self._tool_diffs_lock:
-                self._tool_diffs[call_id] = diff_text
+                self._tool_diffs[call_id] = sanitize_for_display(diff_text)
 
     def take_diff(self, call_id: str) -> str | None:
         """Pop and return the diff recorded for CALL_ID, if any."""
@@ -350,23 +356,50 @@ def _plan_blocked(self, name: str, args: dict[str, Any]) -> str | None:
                 "Error: blocked by plan mode (read-only phase); "
                 "Bash is disabled — use Read/Glob/Grep for read-only access"
             )
+        if name == "Edit" and args.get("old_str") is None and args.get("diff") is not False:
+            # diff/patch mode: the `path` argument is NOT enough — the
+            # patch applies to the paths inside the diff itself, and the
+            # macOS/Windows Python applier honors absolute `+++` targets.
+            content = args.get("new_str")
+            if isinstance(content, str) and content.strip():
+                return self._plan_diff_verdict(args)
+            # contentless diff call: Edit rejects a missing/invalid
+            # new_str before any patch engine runs, so the plain path
+            # check below is sufficient
         path = self._tool_path(name, args)
         if path and path != self.plan_mode.plan_file:
-            # Edit diff-mode (no old_str, diff not explicitly False) runs
-            # `patch` which can write to arbitrary files via relative paths
-            # in the diff content — block it even if the target path looks
-            # innocent, because patch follows paths within the diff.
-            if name == "Edit" and args.get("old_str") is None and args.get("diff") is not False:
-                return (
-                    "Error: blocked by plan mode (read-only phase); "
-                    "diff/patch mode cannot target files other than the plan "
-                    "file — use string replacement (old_str/new_str) instead"
-                )
             return (
                 "Error: blocked by plan mode (read-only phase); only the plan file may be modified"
             )
         return None
 
+    def _plan_diff_verdict(self, args: dict[str, Any]) -> str | None:
+        """Allow an Edit diff only when every section targets the plan file.
+
+        Resolution mirrors the Edit backends (same cwd/fallback rules,
+        same helpers as the built-in Python applier), so the sections
+        approved here are exactly the files the backend would write.  A
+        diff that cannot be parsed into file sections is refused (fail
+        closed): an unverifiable patch must never reach a patch engine.
+        """
+        from .tools.diffapply import diff_targets
+        from .tools.edit import _patch_cwd, _strip_diff_fence
+
+        plan_file = self.plan_mode.plan_file
+        raw = str(args.get("path", ""))
+        path = os.path.realpath(os.path.abspath(raw))
+        cwd = _patch_cwd(raw, path)
+        text = str(args.get("new_str"))
+        text = text if text.endswith("\n") else text + "\n"
+        targets = diff_targets(_strip_diff_fence(text), cwd, path if os.path.isfile(path) else None)
+        if plan_file and targets is not None and all(target == plan_file for target in targets):
+            return None
+        return (
+            "Error: blocked by plan mode (read-only phase); "
+            "diff/patch mode cannot target files other than the plan "
+            "file — use string replacement (old_str/new_str) instead"
+        )
+
     def _tool_path(self, name: str, args: dict[str, Any]) -> str | None:
         if name == "Write":
             return os.path.realpath(

python_agent_harness/tools/diffapply.py

@@ -53,9 +53,10 @@ def old_seq(self) -> list[tuple[str, str]]:
 
 
 class _Section:
-    __slots__ = ("new_path", "hunks")
+    __slots__ = ("old_path", "new_path", "hunks")
 
-    def __init__(self, new_path: str, hunks: list[_Hunk]) -> None:
+    def __init__(self, old_path: str, new_path: str, hunks: list[_Hunk]) -> None:
+        self.old_path = old_path
         self.new_path = new_path
         self.hunks = hunks
 
@@ -82,6 +83,8 @@ def _parse(diff_text: str) -> list[_Section]:
             i += 1
             continue
         new_path = lines[i + 1][len("+++") :].strip()
+        # git-style "a/... b/..." header, old path on the `---` line
+        old_path = lines[i][len("---") :].strip()
         i += 2
         hunks: list[_Hunk] = []
         while i < n:
@@ -108,7 +111,7 @@ def _parse(diff_text: str) -> list[_Section]:
             if _section_header_at(lines, i):
                 break
             i += 1  # stray line (e.g. "diff --git", "index ...") — skip
-        sections.append(_Section(new_path, hunks))
+        sections.append(_Section(old_path, new_path, hunks))
     return sections
 
 
@@ -157,8 +160,36 @@ def _match_hunk(hunk: _Hunk, file_lines: list[str]) -> int | None:
     return None
 
 
-def _apply_hunk(hunk: _Hunk, pos: int, file_lines: list[str]) -> None:
-    """Replace the matched region with the hunk's new lines in place."""
+def _file_line_ending(file_lines: list[str]) -> str:
+    """The file's line-ending convention: ``"\\r\\n"``, ``"\\r"`` or ``"\\n"``.
+
+    Decided by the first ``"\\n"``-terminated line, so a classic-Mac
+    ``"\\r"`` terminator (or a stray carriage return, which universal
+    newline reading treats as a terminator) never flips a normal file's
+    convention.  ``"\\r"`` is returned only when the file has no
+    ``"\\n"`` terminator at all.  Files without any terminator (a
+    single unterminated line) default to ``"\\n"``.
+    """
+    saw_cr_only = False
+    for line in file_lines:
+        if line.endswith("\r\n"):
+            return "\r\n"
+        if line.endswith("\n"):
+            return "\n"
+        if line.endswith("\r"):
+            saw_cr_only = True
+    return "\r" if saw_cr_only else "\n"
+
+
+def _apply_hunk(hunk: _Hunk, pos: int, file_lines: list[str], ending: str) -> None:
+    """Replace the matched region with the hunk's new lines in place.
+
+    Context lines keep the file's original bytes; added lines adopt the
+    file's line-ending convention (*ending*) — the diff's own ``+``
+    lines usually carry LF (the model / git), and letting them through
+    verbatim would leave a CRLF file with mixed endings.  A line marked
+    ``\\ No newline at end of file`` gets no terminator at all.
+    """
     old_count = sum(1 for kind, _, _ in hunk.body if kind in (" ", "-"))
     new_block: list[str] = []
     p = pos
@@ -169,12 +200,8 @@ def _apply_hunk(hunk: _Hunk, pos: int, file_lines: list[str]) -> None:
         elif kind == "-":
             p += 1
         else:  # "+"
-            if no_newline:
-                new_block.append(content.rstrip("\n"))
-            elif content.endswith("\n"):
-                new_block.append(content)
-            else:
-                new_block.append(content + "\n")
+            body = content.rstrip("\r\n")
+            new_block.append(body if no_newline else body + ending)
     file_lines[pos : pos + old_count] = new_block
 
 
@@ -199,8 +226,12 @@ def _apply_section(section: _Section, cwd: str, fallback_path: str | None) -> tu
     target = _resolve_target(section.new_path, cwd, fallback_path)
     if not os.path.isfile(target):
         return False, f"target file does not exist: {target}"
+    # surrogateescape + newline="": invalid UTF-8 bytes and the file's own
+    # line endings survive untouched for every line the diff does not
+    # change (errors="replace" used to persist U+FFFD for such bytes, and
+    # universal newlines rewrote the whole file's endings).
     try:
-        with open(target, encoding="utf-8", errors="replace") as f:
+        with open(target, encoding="utf-8", errors="surrogateescape", newline="") as f:
             file_lines = f.readlines()
     except OSError as e:
         return False, f"cannot read {target}: {e}"
@@ -216,10 +247,11 @@ def _apply_section(section: _Section, cwd: str, fallback_path: str | None) -> tu
             )
         plan.append((hunk, pos))
     new_lines = list(file_lines)
+    ending = _file_line_ending(file_lines)
     for hunk, pos in reversed(plan):  # bottom-up: earlier positions stay valid
-        _apply_hunk(hunk, pos, new_lines)
+        _apply_hunk(hunk, pos, new_lines, ending)
     try:
-        with open(target, "w", encoding="utf-8") as f:
+        with open(target, "w", encoding="utf-8", errors="surrogateescape", newline="") as f:
             f.writelines(new_lines)
     except OSError as e:
         return False, f"cannot write {target}: {e}"
@@ -244,3 +276,27 @@ def apply_unified_diff(
         if not ok:
             return False, msg
     return True, f"applied {len(sections)} file section(s)"
+
+
+def diff_targets(diff_text: str, cwd: str, fallback_path: str | None = None) -> list[str] | None:
+    """Paths a patch could plausibly touch for this diff.
+
+    Includes every section's ``---`` (old) AND ``+++`` (new) path,
+    resolved exactly like the applier resolves the new path.  The
+    built-in applier only ever writes the new path, but external patch
+    engines (GNU patch) may consult the old path too, so a caller
+    gating writes must treat both as in scope.  Returns ``None`` when
+    the text yields no file section — the caller decides whether an
+    unverifiable diff is acceptable (the plan-mode guard treats it as a
+    refusal).
+    """
+    sections = _parse(diff_text)
+    if not sections:
+        return None
+    targets: list[str] = []
+    for section in sections:
+        for path in (section.old_path, section.new_path):
+            resolved = os.path.realpath(_resolve_target(path, cwd, fallback_path))
+            if resolved not in targets:
+                targets.append(resolved)
+    return targets

python_agent_harness/tools/edit.py

@@ -18,6 +18,41 @@
 from ..diffrender import unified_diff
 from .base import Tool, ToolContext
 
+# Any line ending, as a single compiled pattern (CRLF first so a CRLF is
+# never rewritten as two endings).
+_NEWLINE_RE = re.compile(r"\r\n|\r|\n")
+
+
+def _uses_crlf(text: str) -> bool:
+    """Whether the FIRST line ending in TEXT is CRLF.
+
+    Files are read with ``newline=""`` so their endings are preserved
+    byte-for-byte; this detector decides how model-provided text (which
+    is almost always LF) must be translated to match the file's own
+    convention instead of silently rewriting it.
+    """
+    idx = text.find("\n")
+    return idx > 0 and text[idx - 1] == "\r"
+
+
+def _to_crlf(text: str) -> str:
+    """Rewrite every line ending in TEXT to CRLF."""
+    return _NEWLINE_RE.sub("\r\n", text)
+
+
+def _patch_cwd(raw: str, path: str) -> str:
+    """The working directory a diff applies in (Emacs `file-name-directory`).
+
+    A trailing-slash directory path -> that directory itself, so a
+    multi-file diff applies to files within it; otherwise the file's
+    parent.  Shared by `Edit.run` and the plan-mode diff guard
+    (``Session._plan_diff_verdict``) so both always agree on where a
+    patch's section paths resolve.
+    """
+    if raw.endswith("/") or raw.endswith(os.sep):
+        return path or "/"
+    return os.path.dirname(path) or "/"
+
 
 class Edit(Tool):
     name = "Edit"
@@ -73,13 +108,11 @@ def run(self, args: dict, ctx: ToolContext) -> str:
         # gptel: string mode when `diff` is false OR `old_str` is provided.
         if diffp is False or old is not None:
             return self._string_replace(path, old, new_str, ctx)
-        # Diff mode runs `patch` in Emacs `file-name-directory' of the path:
-        # a trailing-slash directory path -> that directory itself, so a
-        # multi-file diff applies to files within it; otherwise the parent.
-        if raw.endswith("/") or raw.endswith(os.sep):
-            cwd = path or "/"
-        else:
-            cwd = os.path.dirname(path) or "/"
+        # Diff mode runs `patch` in Emacs `file-name-directory' of the path
+        # (see _patch_cwd): a trailing-slash directory path -> that
+        # directory itself, so a multi-file diff applies to files within
+        # it; otherwise the parent.
+        cwd = _patch_cwd(raw, path)
         return self._apply_patch(path, cwd, new_str, ctx)
 
     def _string_replace(self, path: str, old: str | None, new_str: str, ctx: ToolContext) -> str:
@@ -89,11 +122,23 @@ def _string_replace(self, path: str, old: str | None, new_str: str, ctx: ToolCon
             )
         if old is None:
             return "Error: old_str is required for non-diff edits"
+        if not isinstance(old, str) or not isinstance(new_str, str):
+            return "Error: old_str and new_str must be strings"
+        # surrogateescape: invalid UTF-8 bytes (Latin-1, GBK, ...) survive
+        # the read/rewrite round trip — errors="replace" would silently
+        # persist U+FFFD for every such byte.  newline="": the file's own
+        # line endings are preserved instead of being rewritten to LF.
         try:
-            with open(path, encoding="utf-8", errors="replace") as f:
+            with open(path, encoding="utf-8", errors="surrogateescape", newline="") as f:
                 content = f.read()
         except OSError as e:
             return f"Error: cannot read {path}: {e}"
+        if _uses_crlf(content):
+            # the file uses CRLF; the model's old_str/new_str are nearly
+            # always LF, so translate them to the file's convention —
+            # the match still works and the edit never mixes endings
+            old = _to_crlf(old)
+            new_str = _to_crlf(new_str)
         count = content.count(old)
         if count == 0:
             return f'Error: Could not find old_str "{old[:20]}" in file {path}'
@@ -104,7 +149,7 @@ def _string_replace(self, path: str, old: str | None, new_str: str, ctx: ToolCon
             )
         new = content.replace(old, new_str, 1)
         try:
-            with open(path, "w", encoding="utf-8") as f:
+            with open(path, "w", encoding="utf-8", errors="surrogateescape", newline="") as f:
                 f.write(new)
         except OSError as e:
             return f"Error: {e}"