fix: block stale write_file overwrites

Require an explicit full-file baseline before replacing existing host-visible files with write_file, and fail closed when that baseline is stale. This prevents stale conversation context from clobbering manual or external edits.\n\nRefs #65604

tests/tools/test_cross_profile_guard.py

@@ -72,8 +72,9 @@ class TestWriteFileCrossProfileGuard:
         the same OS user owns every profile dir and the terminal tool
         always could write them. Cross-profile writes now succeed; the
         system prompt's profile hint is the only steering."""
-        from tools.file_tools import write_file_tool
+        from tools.file_tools import read_file_tool, write_file_tool
         target = fake_hermes["root"] / "skills" / "shared-skill" / "SKILL.md"
+        assert not json.loads(read_file_tool(str(target))).get("error")
         result_json = write_file_tool(str(target), "cross-profile write, allowed")
         result = json.loads(result_json)
         assert not result.get("error"), f"guard retired; write must succeed: {result}"

tests/tools/test_file_read_guards.py

@@ -827,9 +827,12 @@ class TestWriteInvalidatesDedup(unittest.TestCase):
         # Read with different offsets to populate multiple dedup entries.
         read_file_tool(self._tmpfile, offset=1, limit=100, task_id="off")
         read_file_tool(self._tmpfile, offset=50, limit=100, task_id="off")
+        # The last read was partial; a full read restores the write baseline.
+        read_file_tool(self._tmpfile, offset=1, limit=500, task_id="off")
 
         # Write — should invalidate BOTH dedup entries.
-        write_file_tool(self._tmpfile, "replaced\n", task_id="off")
+        write = json.loads(write_file_tool(self._tmpfile, "replaced\n", task_id="off"))
+        self.assertNotIn("error", write)
 
         # Both reads should return fresh content.
         r1 = json.loads(read_file_tool(self._tmpfile, offset=1, limit=100, task_id="off"))

tests/tools/test_file_staleness.py

@@ -2,8 +2,9 @@
 """
 Tests for file staleness detection in write_file and patch.
 
-When a file is modified externally between the agent's read and write,
-the write should include a warning so the agent can re-read and verify.
+write_file refuses (before any disk mutation) to overwrite an existing file the
+task never read in full or that changed on disk since that read; patch stays
+warning-only for stale reads.
 
 Run with:  python -m pytest tests/tools/test_file_staleness.py -v
 """
@@ -65,8 +66,17 @@ def _make_fake_ops(read_content="hello\n", file_size=6):
     return fake
 
 
+def _modify_externally(path: str, content: str) -> None:
+    """Rewrite *path* so its mtime provably differs from the pre-write stamp."""
+    before = os.path.getmtime(path)
+    with open(path, "w") as f:
+        f.write(content)
+    if os.path.getmtime(path) == before:
+        os.utime(path, (before + 1.0, before + 1.0))
+
+
 # ---------------------------------------------------------------------------
-# Core staleness check
+# write_file: refuse stale / unread overwrites before touching the disk
 # ---------------------------------------------------------------------------
 
 class TestStalenessCheck(unittest.TestCase):
@@ -96,6 +106,62 @@ class TestStalenessCheck(unittest.TestCase):
 
         result = json.loads(write_file_tool(self._tmpfile, "new content", task_id="t1"))
         self.assertNotIn("_warning", result)
+        self.assertNotIn("error", result)
+
+    def test_write_file_refuses_before_mutation_when_modified_externally(self):
+        """read → external edit → write_file: refused, external edit preserved;
+        a full re-read heals the baseline and the next write lands."""
+        self.assertNotIn("error", json.loads(read_file_tool(self._tmpfile, task_id="t1")))
+        _modify_externally(self._tmpfile, "someone else changed this\n")
+
+        refused = json.loads(write_file_tool(self._tmpfile, "new content\n", task_id="t1"))
+        self.assertTrue(refused.get("stale_write_blocked"), refused)
+        self.assertIn("modified since you last read", refused["error"])
+        with open(self._tmpfile) as f:
+            self.assertEqual(f.read(), "someone else changed this\n")
+
+        self.assertNotIn("error", json.loads(read_file_tool(self._tmpfile, task_id="t1")))
+        written = json.loads(write_file_tool(self._tmpfile, "merged\n", task_id="t1"))
+        self.assertNotIn("error", written)
+        with open(self._tmpfile) as f:
+            self.assertEqual(f.read(), "merged\n")
+
+    def test_write_file_requires_full_unredacted_read_of_existing_file(self):
+        """Existing file with no baseline is refused untouched: never read, only
+        patched, read partially, or read redacted (the «redacted:…» sentinel must
+        never be persisted). A net-new file needs no baseline and the task's own
+        write is a baseline for its next write."""
+        refused = json.loads(write_file_tool(self._tmpfile, "x\n", task_id="t2"))
+        self.assertTrue(refused.get("stale_write_blocked"), refused)
+        self.assertIn("has not read it in full", refused["error"])
+
+        patched = json.loads(patch_tool(mode="replace", path=self._tmpfile,
+                                        old_string="original", new_string="patched", task_id="t2"))
+        self.assertNotIn("error", patched)
+        self.assertTrue(json.loads(write_file_tool(self._tmpfile, "x\n", task_id="t2")).get("stale_write_blocked"))
+
+        with open(self._tmpfile, "w") as f:
+            f.write("one\ntwo\nthree\n")
+        self.assertNotIn("error", json.loads(read_file_tool(self._tmpfile, offset=1, limit=1, task_id="t2")))
+        self.assertTrue(json.loads(write_file_tool(self._tmpfile, "x\n", task_id="t2")).get("stale_write_blocked"))
+
+        secret = "ghp_" + "A" * 40
+        with open(self._tmpfile, "w") as f:
+            f.write(f"token={secret}\n")
+        with patch("agent.redact._REDACT_ENABLED", True):
+            read = json.loads(read_file_tool(self._tmpfile, task_id="t2"))
+            self.assertNotIn(secret, read["content"])
+            refused = json.loads(write_file_tool(self._tmpfile, "token=«redacted:ghp_…»\n", task_id="t2"))
+        self.assertTrue(refused.get("stale_write_blocked"), refused)
+        with open(self._tmpfile) as f:
+            self.assertEqual(f.read(), f"token={secret}\n")
+
+        new_path = os.path.join(self._tmpdir, "brand_new.txt")
+        self.assertNotIn("error", json.loads(write_file_tool(new_path, "one\n", task_id="t2")))
+        self.assertNotIn("error", json.loads(write_file_tool(new_path, "two\n", task_id="t2")))
+        with open(new_path) as f:
+            self.assertEqual(f.read(), "two\n")
+        os.unlink(new_path)
 
 
     @patch("tools.file_tools._get_file_ops")
@@ -114,6 +180,7 @@ class TestStalenessCheck(unittest.TestCase):
             f.write("live copy\n")
 
         fake_ops = _make_fake_ops("live copy\n", 10)
+        fake_ops.write_file = MagicMock(side_effect=AssertionError("must not write stale content"))
         mock_ops.return_value = fake_ops
 
         from tools import terminal_tool
@@ -135,8 +202,9 @@ class TestStalenessCheck(unittest.TestCase):
         finally:
             terminal_tool.clear_session_cwd("live_task")
 
-        self.assertIn("_warning", result)
-        self.assertIn("modified since you last read", result["_warning"])
+        self.assertTrue(result.get("stale_write_blocked"), result)
+        self.assertIn("modified since you last read", result["error"])
+        fake_ops.write_file.assert_not_called()
 
 
 # ---------------------------------------------------------------------------

tests/tools/test_file_state_registry.py

@@ -211,20 +211,23 @@ class FileToolsIntegrationTests(unittest.TestCase):
             f.write(content)
         return p
 
-    def test_sibling_agent_write_surfaces_warning_through_handler(self):
+    def test_sibling_agent_write_refuses_stale_overwrite_through_handler(self):
         p = self._write_seed("shared.txt")
         r = json.loads(read_file_tool(path=p, task_id="agentA"))
         self.assertNotIn("error", r)
 
+        self.assertNotIn("error", json.loads(read_file_tool(path=p, task_id="agentB")))
         w_b = json.loads(write_file_tool(path=p, content="B wrote\n", task_id="agentB"))
         self.assertNotIn("error", w_b)
 
         w_a = json.loads(write_file_tool(path=p, content="A stale\n", task_id="agentA"))
-        warn = w_a.get("_warning", "")
-        self.assertTrue(warn, f"expected warning, got: {w_a}")
-        # The cross-agent message names the sibling task_id.
-        self.assertIn("agentB", warn)
-        self.assertIn("sibling", warn.lower())
+        err = w_a.get("error", "")
+        self.assertTrue(w_a.get("stale_write_blocked"), f"expected stale write refusal, got: {w_a}")
+        # The cross-agent message names the sibling task_id; B's write survives.
+        self.assertIn("agentB", err)
+        self.assertIn("sibling", err.lower())
+        with open(p) as f:
+            self.assertEqual(f.read(), "B wrote\n")
 
 
     def test_net_new_file_no_warning(self):

tests/tools/test_line_ending_preservation.py

@@ -119,10 +119,12 @@ class TestWriteFileCRLFPreservation:
         """The agent typically sends bare-LF content; if the file existed
         with CRLF, the write should convert to CRLF rather than silently
         flipping the endings."""
-        from tools.file_tools import _handle_write_file
+        from tools.file_tools import _handle_write_file, read_file_tool
 
         target = tmp_path / "config.bat"
         target.write_bytes(b"@echo off\r\nset X=1\r\n")
+        # write_file refuses to overwrite an existing file the task never read.
+        assert "error" not in json.loads(read_file_tool(str(target), task_id="crlf_write_1"))
 
         result = _handle_write_file(
             {
@@ -143,10 +145,11 @@ class TestWriteFileCRLFPreservation:
 
     def test_overwrite_lf_file_stays_lf(self, hermes_home, tmp_path):
         """Pre-existing LF file should not get spurious CRLFs."""
-        from tools.file_tools import _handle_write_file
+        from tools.file_tools import _handle_write_file, read_file_tool
 
         target = tmp_path / "lf.txt"
         target.write_bytes(b"line1\nline2\n")
+        assert "error" not in json.loads(read_file_tool(str(target), task_id="crlf_write_3"))
 
         result = _handle_write_file(
             {"path": str(target), "content": "X\nY\nZ\n"},

tools/file_state.py

@@ -32,6 +32,12 @@ def _disabled() -> bool:
     return os.environ.get("HERMES_DISABLE_FILE_STATE_GUARD", "").strip() == "1"
 
 
+def guard_disabled() -> bool:
+    """True when the user switched the read-before-write guard off; the file
+    tools then warn instead of refusing stale/unread write_file overwrites."""
+    return _disabled()
+
+
 def _mtime_or_none(resolved: str) -> Optional[float]:
     try:
         return os.path.getmtime(resolved)

tools/file_tools.py

@@ -30,12 +30,12 @@ from tools.file_tools_paths import (
 from tools.file_tools_write_guards import (
     _READ_DEDUP_STATUS_MESSAGE, _check_approval_required_write, _check_binary_document_write,
     _check_cross_profile_path, _check_protected_instruction_write, _check_sensitive_path,
-    _is_internal_file_tool_content)
+    _is_internal_file_tool_content, _stale_overwrite_blocker, _stale_write_refusal)
 from tools.file_tools_read_tracking import (
     _bump_consecutive, _cap_read_tracker_data, _check_file_staleness, _check_not_found_cache,
-    _mark_verification_stale, _patch_failure_lock, _patch_failure_tracker, _read_tracker,
-    _read_tracker_lock, _record_not_found, _record_patch_failure, _reset_patch_failures,
-    _task_data, _update_read_timestamp)
+    _mark_full_write_baseline, _mark_verification_stale, _patch_failure_lock,
+    _patch_failure_tracker, _read_tracker, _read_tracker_lock, _record_not_found,
+    _record_patch_failure, _reset_patch_failures, _task_data, _update_read_timestamp)
 
 logger = logging.getLogger(__name__)
 
@@ -492,11 +492,15 @@ def _dedup_stub_or_block(task_data: dict, dedup_key: tuple, path: str) -> str:
 
 
 def _record_successful_read(task_data: dict, task_id: str, path: str, resolved_str: str,
-                            offset: int, limit: int, dedup_key: tuple, *, partial: bool) -> int:
+                            offset: int, limit: int, dedup_key: tuple, *, partial: bool,
+                            redacted: bool = False) -> int:
     """Bookkeeping after a real (non-stub) read; returns the consecutive-read count.
 
     Per-task tracker under the lock (stub counter, history, consecutive count,
-    mtime for dedup + staleness). Then OUTSIDE our lock (no nested locking): the
+    mtime for dedup + staleness, and — for a full UNREDACTED read — the write_file
+    baseline: a redacted read returned a non-round-trippable ``«redacted:…»``
+    sentinel, so it must not bless an overwrite that would persist the sentinel
+    into a credential file). Then OUTSIDE our lock (no nested locking): the
     cross-agent registry, and the background-review read-mark (a FULL read of a
     skill file counts like skill_view so a follow-up skill_manage(patch) is accepted).
     """
@@ -511,6 +515,8 @@ def _record_successful_read(task_data: dict, task_id: str, path: str, resolved_s
             task_data.setdefault("read_timestamps", {})[resolved_str] = _mtime_now
         except OSError:
             pass
+        if not partial and not redacted:
+            task_data.setdefault("full_write_baselines", set()).add(resolved_str)
         _cap_read_tracker_data(task_data)
 
     try:
@@ -620,8 +626,11 @@ def read_file_tool(path: str, offset: int = 1, limit: int = DEFAULT_READ_LIMIT,
             result.content = _apply_char_budget(
                 result_dict, result.content or "", offset,
                 result_dict.get("total_lines", "unknown"), max_chars)
+        redacted = False
         if result.content:
-            result.content = redact_sensitive_text(result.content, file_read=True)
+            unredacted = result.content
+            result.content = redact_sensitive_text(unredacted, file_read=True)
+            redacted = result.content != unredacted
             result_dict["content"] = result.content
 
         if (file_size and file_size > _LARGE_FILE_HINT_BYTES
@@ -632,7 +641,8 @@ def read_file_tool(path: str, offset: int = 1, limit: int = DEFAULT_READ_LIMIT,
                 "to keep context usage efficient."))
 
         count = _record_successful_read(task_data, task_id, path, resolved_str, offset, limit,
-                                        dedup_key, partial=(offset > 1) or bool(result_dict.get("truncated")))
+                                        dedup_key, partial=(offset > 1) or bool(result_dict.get("truncated")),
+                                        redacted=redacted)
         if count >= 4:
             return tool_error(
                 f"BLOCKED: You have read this exact file region {count} times in a row. "
@@ -783,6 +793,12 @@ def write_file_tool(path: str, content: str, task_id: str = "default",
                 # Per-path lock serializes read→modify→write across concurrent
                 # subagents; different paths stay fully parallel.
                 _lock.enter_context(file_state.lock_path(_resolved))
+            # A whole-file overwrite of content this task never saw, or that
+            # changed since, is refused HERE — before the write — instead of
+            # warning after the clobber (#65604). Nothing below runs.
+            blocker = _stale_overwrite_blocker(path, _resolved, task_id)
+            if blocker:
+                return json.dumps(_stale_write_refusal(path, blocker, _resolved), ensure_ascii=False)
             warnings = _edit_warnings([path], path_to_resolved, task_id)
             rewrite_hint = _whole_file_rewrite_hint(task_id, _resolved, content)
             result_dict = _get_file_ops(task_id).write_file(_resolved or path, content).to_dict()
@@ -799,6 +815,9 @@ def write_file_tool(path: str, content: str, task_id: str = "default",
             else:
                 if _resolved:
                     result_dict["files_modified"] = [_resolved]
+                    # Own write = current whole-file content: consecutive
+                    # same-task writes stay unblocked. patch never does this.
+                    _mark_full_write_baseline(_resolved, task_id)
                 _note_edited(task_id, [path], path_to_resolved, session_id)
         return json.dumps(result_dict, ensure_ascii=False)
     except Exception as e:
@@ -1037,7 +1056,7 @@ READ_FILE_SCHEMA = {
 
 WRITE_FILE_SCHEMA = {
     "name": "write_file",
-    "description": "Write content to a file, completely replacing existing content. Use this instead of echo/cat heredoc in terminal. Creates parent directories automatically. OVERWRITES the entire file — use 'patch' for targeted edits. Auto-runs syntax checks on .py/.json/.yaml/.toml and other linted languages; only NEW errors introduced by this write are surfaced (pre-existing errors are filtered out). The result's verified:true means the on-disk content hash was confirmed — do NOT re-read the file to check the write landed.",
+    "description": "Write content to a file, completely replacing existing content. Use this instead of echo/cat heredoc in terminal. Creates parent directories automatically. OVERWRITES the entire file — use 'patch' for targeted edits. For an EXISTING file, call read_file first: write_file refuses (file untouched) when this task has no current full read/write of the file or the file changed on disk since; on refusal, read_file, merge, then retry. Auto-runs syntax checks on .py/.json/.yaml/.toml and other linted languages; only NEW errors introduced by this write are surfaced (pre-existing errors are filtered out). The result's verified:true means the on-disk content hash was confirmed — do NOT re-read the file to check the write landed.",
     "parameters": {
         "type": "object",
         "properties": {

tools/file_tools_read_tracking.py

@@ -7,7 +7,10 @@ call), ``read_history`` (diagnostics), ``dedup`` (key -> mtime; survives context
 compression), ``dedup_generation_reads`` (keys whose full content was served since
 the last compaction boundary; cleared on compression so one recovery read returns
 full content), ``dedup_hits`` (stub-loop breaker), ``read_timestamps``
-(staleness warnings) and ``not_found`` (short-TTL negative cache). Every
+(staleness warnings), ``full_write_baselines`` (resolved paths whose whole-file
+content this task saw via a full unredacted read_file or wrote via write_file;
+required before write_file may overwrite an existing file — patch never
+qualifies) and ``not_found`` (short-TTL negative cache). Every
 container is hard-capped (``_cap_read_tracker_data``) so long sessions stay small.
 """
 
@@ -35,6 +38,7 @@ _PATCH_FAILURE_PATHS_CAP = 64
 _READ_HISTORY_CAP = 500
 _DEDUP_CAP = 1000
 _READ_TIMESTAMPS_CAP = 1000
+_FULL_WRITE_BASELINES_CAP = 1000
 _NOT_FOUND_CAP = 500
 _NOT_FOUND_TTL_SECONDS = 60.0  # a path that didn't exist may be created soon
 
@@ -46,7 +50,8 @@ def _task_data(task_id: str) -> dict:
         "last_key": None, "consecutive": 0, "read_history": set()})
     for key in ("dedup", "dedup_hits", "read_timestamps"):
         task_data.setdefault(key, {})
-    task_data.setdefault("dedup_generation_reads", set())
+    for key in ("dedup_generation_reads", "full_write_baselines"):
+        task_data.setdefault(key, set())
     return task_data
 
 
@@ -80,6 +85,7 @@ def _cap_read_tracker_data(task_data: dict) -> None:
         ("dedup_hits", _DEDUP_CAP),
         ("dedup_generation_reads", _DEDUP_CAP),
         ("read_timestamps", _READ_TIMESTAMPS_CAP),
+        ("full_write_baselines", _FULL_WRITE_BASELINES_CAP),
         ("not_found", _NOT_FOUND_CAP)):
         container = task_data.get(key)
         if container is not None and len(container) > cap:
@@ -160,6 +166,9 @@ def reset_file_dedup(task_id: str = None):
             if "dedup_hits" in task_data:
                 task_data["dedup_hits"].clear()
             task_data.setdefault("dedup_generation_reads", set()).clear()
+            # The summary may have dropped the exact bytes the baseline vouched
+            # for: a full overwrite needs a fresh read_file after compaction.
+            task_data.setdefault("full_write_baselines", set()).clear()
 
 
 def notify_other_tool_call(task_id: str = "default"):
@@ -219,6 +228,22 @@ def _update_read_timestamp(filepath: str, task_id: str) -> None:
             _cap_read_tracker_data(task_data)
 
 
+def _mark_full_write_baseline(resolved: str, task_id: str) -> None:
+    """Record that *task_id* saw the whole current content of *resolved* (full
+    unredacted read_file, or its own successful write_file), so a later
+    write_file may replace the file. Acquires the lock itself."""
+    with _read_tracker_lock:
+        task_data = _task_data(task_id)
+        task_data["full_write_baselines"].add(str(resolved))
+        _cap_read_tracker_data(task_data)
+
+
+def _has_full_write_baseline(resolved: str, task_id: str) -> bool:
+    with _read_tracker_lock:
+        task_data = _read_tracker.get(task_id) or {}
+        return str(resolved) in task_data.get("full_write_baselines", set())
+
+
 def _check_file_staleness(filepath: str, task_id: str) -> str | None:
     """Warn (don't block) when the file's mtime changed since this task last read it.
     ``None`` when never read, fresh, or unstattable (a deleted file is the write's problem)."""

tools/file_tools_write_guards.py

@@ -6,14 +6,18 @@ Guards, in the order the tools apply them: ``_check_sensitive_path`` (hard
 deny), ``_check_binary_document_write``, ``_check_protected_instruction_write``
 (ALWAYS ask), ``_check_approval_required_write`` (normal gate),
 ``_check_cross_profile_path`` (sandbox-mirror lost-work), ``_is_internal_file_tool_content``.
+``_stale_overwrite_blocker`` (write_file only, under the per-path lock) refuses a
+whole-file overwrite of content this task never saw or that changed since.
 """
 
 import fnmatch
 import os
 from pathlib import Path
 
+from tools import file_state
 from tools.binary_extensions import has_opaque_document_extension, is_pdf_path
 from tools.file_tools_paths import _expand_tilde, _resolve_path_for_task
+from tools.file_tools_read_tracking import _check_file_staleness, _has_full_write_baseline
 
 # Prefixes matched after realpath. macOS: /private/var mirrors /var — block the
 # sensitive subtrees only; a blanket "/private/var/" refuses every temp-file
@@ -441,6 +445,53 @@ _READ_DEDUP_STATUS_MESSAGE = (
     "still current — refer to that instead of re-reading.")
 
 
+def _stale_overwrite_blocker(filepath: str, resolved: str | None, task_id: str) -> str | None:
+    """Reason write_file must NOT replace the existing file, else ``None``.
+
+    Refuses BEFORE any disk mutation (the pre-#65604 warning arrived after the
+    clobber): a sibling/external/partial-read staleness finding, or an existing
+    file with no full-content baseline for this task (never read in full, read
+    redacted, only patched, or evicted by compaction). Net-new files, files this
+    task fully read or wrote, unresolvable paths and the file-state kill switch
+    all let the write proceed.
+    """
+    if file_state.guard_disabled():
+        return None
+    stale = (file_state.check_stale(task_id, resolved) if resolved else None) or _check_file_staleness(filepath, task_id)
+    if stale:
+        return stale
+    if not resolved or _has_full_write_baseline(resolved, task_id):
+        return None
+    try:
+        exists = Path(resolved).exists()
+    except OSError:
+        return None
+    if not exists:
+        return None
+    return (
+        f"{resolved} exists but this task has not read it in full (or only saw a "
+        "redacted/partial view). Read the file before using write_file so a stale "
+        "conversation copy cannot overwrite the current disk content.")
+
+
+def _stale_write_refusal(filepath: str, reason: str, resolved: str | None = None) -> dict:
+    """Model-facing refusal payload for write_file; ``stale_write_blocked`` lets
+    callers tell it apart from I/O errors."""
+    result = {
+        "error": (
+            f"Refusing to overwrite {filepath}: {reason} "
+            "The file was NOT modified. Use read_file to reload the current "
+            "contents, merge the requested change, then call write_file again. "
+            "For small edits, prefer patch so existing unrelated changes are "
+            "preserved."),
+        "stale_write_blocked": True,
+        "path": filepath,
+    }
+    if resolved:
+        result["resolved_path"] = resolved
+    return result
+
+
 def _is_internal_file_status_text(content: str) -> bool:
     """True when content is the read_file dedup status message, verbatim or lightly framed
     (contains the full message and is <=2x its length — a real file quoting it would be longer)."""