From 806435548bb8fe080fa626afa092ba85e4b8d797 Mon Sep 17 00:00:00 2001
From: Rickard Robin <rickard@kumobits.com>
Date: Mon, 27 Jul 2026 01:59:21 +0000
Subject: [PATCH 1/3] fix(agent): content-transition file mutation verifier
 clears stale footers

Extract TurnFileMutationVerifier with pre-dispatch baselines, local
observation, and fail-closed reconciliation so a later legitimate file
change can suppress a failed patch/write footer without proving payload match.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 agent/file_mutation_verifier.py               | 583 ++++++++++++++++++
 agent/tool_dispatch_helpers.py                |  42 +-
 agent/tool_executor.py                        | 181 +++++-
 agent/turn_context.py                         |   6 +
 agent/turn_finalizer.py                       |  45 +-
 run_agent.py                                  | 106 ++--
 .../test_file_mutation_verifier_redesign.py   | 314 ++++++++++
 .../run_agent/test_file_mutation_verifier.py  |  11 +-
 .../run_agent/test_tool_batch_segmentation.py |   9 +-
 9 files changed, 1177 insertions(+), 120 deletions(-)
 create mode 100644 agent/file_mutation_verifier.py
 create mode 100644 tests/agent/test_file_mutation_verifier_redesign.py

diff --git a/agent/file_mutation_verifier.py b/agent/file_mutation_verifier.py
new file mode 100644
index 0000000000000..eb7e4f10c3ec2
--- /dev/null
+++ b/agent/file_mutation_verifier.py
@@ -0,0 +1,583 @@
+"""Per-turn file-mutation verifier — content-transition ledger and observation.
+
+Tracks dispatched ``write_file`` / ``patch`` outcomes against stable local
+filesystem snapshots.  A safely observed later content change may clear a
+stale unresolved entry without proving payload equality (CONTENT-TRANSITION
+contract).  When observation is unavailable the ledger fails closed.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import logging
+import os
+import re
+import stat
+import threading
+import time
+from dataclasses import dataclass
+from enum import Enum
+from pathlib import Path
+from typing import Any, Callable, Dict, List, Optional, Set, Tuple
+
+from agent.tool_result_classification import (
+    FILE_MUTATING_TOOL_NAMES,
+    file_mutation_result_landed,
+)
+from agent.tool_dispatch_helpers import (
+    _extract_error_preview,
+    _extract_file_mutation_targets,
+    _extract_landed_file_mutation_paths,
+)
+
+logger = logging.getLogger(__name__)
+
+# Turn-wide bounds (fail closed via overflow summary when exceeded).
+MAX_LEDGER_ENTRIES = 64
+MAX_SNAPSHOT_ATTEMPTS = 128
+MAX_SNAPSHOT_BYTES = 32 * 1024 * 1024
+MAX_SINGLE_FILE_BYTES = 8 * 1024 * 1024
+OBSERVATION_TIMEOUT_S = 2.0
+
+_GENERIC_FOOTER_FALLBACK = (
+    "⚠️ File-mutation verifier: one or more file edits may not have landed "
+    "this turn. Run `git status` or `read_file` to confirm."
+)
+
+_FOOTER_INJECTION_RE = re.compile(
+    r"(?im)^\s*(MEDIA:|FILE:|ATTACHMENT:|\!\[).*$"
+)
+
+
+class DispatchTriState(str, Enum):
+    NOT_DISPATCHED = "not_dispatched"
+    DISPATCHED = "dispatched"
+    DISPATCHED_NO_RESULT = "dispatched_no_result"
+
+
+@dataclass(frozen=True)
+class MutationIdentity:
+    backend_kind: str
+    authority: str
+    task_scope: str
+    path_dialect: str
+    path: str
+
+    def display_path(self) -> str:
+        return self.path
+
+    def key(self) -> Tuple[str, str, str, str, str]:
+        return (
+            self.backend_kind,
+            self.authority,
+            self.task_scope,
+            self.path_dialect,
+            self.path,
+        )
+
+
+@dataclass
+class ContentFingerprint:
+    size: int
+    digest: bytes
+    mtime_ns: int
+    inode: int
+
+
+@dataclass
+class LedgerEntry:
+    identity: MutationIdentity
+    tool: str
+    error_preview: str
+    baseline: Optional[ContentFingerprint]
+    unresolved: bool = True
+    overflow: bool = False
+
+
+@dataclass
+class TurnObservationBudget:
+    generation: int = 0
+    snapshot_attempts: int = 0
+    snapshot_bytes: int = 0
+    ledger_count: int = 0
+    overflow: bool = False
+
+
+class TurnFileMutationVerifier:
+    """Turn-scoped mutation ledger with local content-transition observation."""
+
+    def __init__(
+        self,
+        *,
+        resolve_backend: Optional[Callable[[str], Tuple[str, str, str]]] = None,
+        now_fn: Callable[[], float] = time.monotonic,
+    ) -> None:
+        self._resolve_backend = resolve_backend or _default_resolve_backend
+        self._now_fn = now_fn
+        self._budget = TurnObservationBudget()
+        self._ledger: Dict[Tuple[str, str, str, str, str], LedgerEntry] = {}
+        self._lock = threading.Lock()
+        self._active_workers: Set[int] = set()
+        self._pre_dispatch_baselines: Dict[Tuple[str, str], Optional[ContentFingerprint]] = {}
+
+    def reset_turn(self, generation: int = 0) -> None:
+        with self._lock:
+            self._budget = TurnObservationBudget(generation=generation)
+            self._ledger.clear()
+            self._active_workers.clear()
+            self._pre_dispatch_baselines.clear()
+
+    def prepare_mutation_dispatch(
+        self,
+        *,
+        tool_name: str,
+        effective_args: Dict[str, Any],
+        effective_task_id: str,
+        turn_generation: Optional[int] = None,
+    ) -> None:
+        """Capture filesystem baselines immediately before registry dispatch."""
+        if turn_generation is not None and turn_generation != self._budget.generation:
+            return
+        if tool_name not in FILE_MUTATING_TOOL_NAMES:
+            return
+        scope = effective_task_id or "default"
+        for raw_path in _extract_file_mutation_targets(tool_name, effective_args):
+            fp = self._capture_baseline(
+                raw_path,
+                scope,
+                turn_generation=turn_generation,
+            )
+            self._pre_dispatch_baselines[(scope, raw_path)] = fp
+
+    def clear_turn(self) -> None:
+        self.reset_turn(0)
+
+    @property
+    def generation(self) -> int:
+        return self._budget.generation
+
+    def record_tool_outcome(
+        self,
+        *,
+        tool_name: str,
+        effective_args: Dict[str, Any],
+        effective_task_id: str,
+        raw_result: Any,
+        dispatch: DispatchTriState,
+        model_is_error: bool,
+        blocked: bool = False,
+        turn_generation: Optional[int] = None,
+    ) -> None:
+        if turn_generation is not None and turn_generation != self._budget.generation:
+            return
+        if tool_name not in FILE_MUTATING_TOOL_NAMES:
+            return
+        if blocked or dispatch is DispatchTriState.NOT_DISPATCHED:
+            return
+        targets = _extract_file_mutation_targets(tool_name, effective_args)
+        if not targets:
+            return
+
+        if dispatch is DispatchTriState.DISPATCHED_NO_RESULT:
+            preview = "tool dispatch finished without a result"
+            for raw_path in targets:
+                self._upsert_unresolved(
+                    tool_name=tool_name,
+                    raw_path=raw_path,
+                    task_id=effective_task_id,
+                    error_preview=preview,
+                    baseline=None,
+                )
+            return
+
+        raw_landed = file_mutation_result_landed(tool_name, raw_result)
+        if raw_landed:
+            landed_paths = _extract_landed_file_mutation_paths(
+                tool_name, effective_args, raw_result,
+            )
+            scope = effective_task_id or "default"
+            for raw_path in _extract_file_mutation_targets(tool_name, effective_args):
+                self._pre_dispatch_baselines.pop((scope, raw_path), None)
+            for raw_path in landed_paths:
+                ident = self._identity_for_path(raw_path, effective_task_id)
+                if ident is not None:
+                    self._ledger.pop(ident.key(), None)
+            return
+
+        if model_is_error or not raw_landed:
+            preview = _extract_error_preview(raw_result)
+            scope = effective_task_id or "default"
+            for raw_path in targets:
+                baseline = self._pre_dispatch_baselines.pop((scope, raw_path), None)
+                if baseline is None:
+                    baseline = self._capture_baseline(
+                        raw_path,
+                        scope,
+                        turn_generation=turn_generation,
+                    )
+                self._upsert_unresolved(
+                    tool_name=tool_name,
+                    raw_path=raw_path,
+                    task_id=effective_task_id,
+                    error_preview=preview,
+                    baseline=baseline,
+                )
+
+    def observe_after_tool(
+        self,
+        *,
+        tool_name: str,
+        effective_task_id: str,
+        blocked: bool = False,
+    ) -> None:
+        if blocked:
+            return
+        self.reconcile_content_transitions(task_id=effective_task_id)
+
+    def reconcile_content_transitions(self, *, task_id: Optional[str] = None) -> None:
+        if not self._ledger:
+            return
+        groups: Dict[str, List[Tuple[Tuple[str, str, str, str, str], LedgerEntry]]] = {}
+        for key, entry in list(self._ledger.items()):
+            if not entry.unresolved or entry.baseline is None:
+                continue
+            ident = entry.identity
+            scope = ident.task_scope
+            if task_id is not None and scope != (task_id or "default"):
+                continue
+            canon = _canonical_observation_key(ident.path, scope) or ident.path
+            groups.setdefault(canon, []).append((key, entry))
+
+        to_remove: List[Tuple[str, str, str, str, str]] = []
+        for canon, items in groups.items():
+            baseline = items[0][1].baseline
+            assert baseline is not None
+            sample_path = items[0][1].identity.path
+            sample_scope = items[0][1].identity.task_scope
+            current = self._capture_baseline(sample_path, sample_scope)
+            if current is None:
+                continue
+            if current.digest != baseline.digest:
+                to_remove.extend(k for k, _ in items)
+        for key in to_remove:
+            self._ledger.pop(key, None)
+
+    def finalize_failed_dict(self) -> Dict[str, Dict[str, Any]]:
+        self.reconcile_content_transitions()
+
+        failed: Dict[str, Dict[str, Any]] = {}
+        for entry in self._ledger.values():
+            if not entry.unresolved:
+                continue
+            path = entry.identity.display_path()
+            if path in failed:
+                continue
+            failed[path] = {
+                "tool": entry.tool,
+                "error_preview": _sanitize_footer_text(entry.error_preview),
+            }
+        if self._budget.overflow:
+            failed["__overflow__"] = {
+                "tool": "patch",
+                "error_preview": "Additional unresolved file mutations were omitted (turn budget).",
+            }
+        return failed
+
+    def _upsert_unresolved(
+        self,
+        *,
+        tool_name: str,
+        raw_path: str,
+        task_id: str,
+        error_preview: str,
+        baseline: Optional[ContentFingerprint],
+    ) -> None:
+        ident = self._identity_for_path(raw_path, task_id)
+        if ident is None:
+            return
+        key = ident.key()
+        with self._lock:
+            if len(self._ledger) >= MAX_LEDGER_ENTRIES and key not in self._ledger:
+                self._budget.overflow = True
+                return
+            existing = self._ledger.get(key)
+            if existing is not None and existing.unresolved:
+                return
+            self._ledger[key] = LedgerEntry(
+                identity=ident,
+                tool=tool_name,
+                error_preview=error_preview,
+                baseline=baseline,
+                unresolved=True,
+            )
+            self._budget.ledger_count = len(self._ledger)
+
+    def _identity_for_path(
+        self, raw_path: str, task_id: str,
+    ) -> Optional[MutationIdentity]:
+        backend_kind, authority, dialect = self._resolve_backend(task_id or "default")
+        display = str(raw_path)
+        if backend_kind != "local":
+            return MutationIdentity(
+                backend_kind=backend_kind,
+                authority=authority,
+                task_scope=task_id or "default",
+                path_dialect=dialect,
+                path=display,
+            )
+        return MutationIdentity(
+            backend_kind=backend_kind,
+            authority=authority,
+            task_scope=task_id or "default",
+            path_dialect=dialect,
+            path=display,
+        )
+
+    def _capture_baseline(
+        self,
+        raw_path: str,
+        task_id: str,
+        *,
+        turn_generation: Optional[int] = None,
+    ) -> Optional[ContentFingerprint]:
+        if turn_generation is not None and turn_generation != self._budget.generation:
+            return None
+        ident = self._identity_for_path(raw_path, task_id)
+        if ident is None or ident.backend_kind != "local":
+            return None
+        if not _path_allowed_for_observation(ident.path):
+            return None
+        with self._lock:
+            if self._budget.snapshot_attempts >= MAX_SNAPSHOT_ATTEMPTS:
+                self._budget.overflow = True
+                return None
+            self._budget.snapshot_attempts += 1
+        resolved = _resolve_local_path(raw_path, task_id)
+        if resolved is None:
+            return None
+        fp, read_bytes = _stable_local_fingerprint(
+            resolved,
+            deadline=self._now_fn() + OBSERVATION_TIMEOUT_S,
+            active_workers=self._active_workers,
+        )
+        with self._lock:
+            self._budget.snapshot_bytes += read_bytes
+            if self._budget.snapshot_bytes > MAX_SNAPSHOT_BYTES:
+                self._budget.overflow = True
+        return fp
+
+
+def _canonical_observation_key(raw_path: str, task_id: str) -> Optional[str]:
+    resolved = _resolve_local_path(raw_path, task_id)
+    if resolved is None:
+        return None
+    try:
+        from agent.tool_dispatch_helpers import _canonical_path
+
+        return str(_canonical_path(str(resolved)))
+    except Exception:
+        return str(resolved.resolve())
+
+
+def _default_resolve_backend(task_id: str) -> Tuple[str, str, str]:
+    try:
+        from tools.terminal_tool import get_active_env
+
+        env = get_active_env(task_id)
+    except Exception:
+        env = None
+    if env is None:
+        return "local", "host", _path_dialect()
+    cls = type(env).__name__
+    if cls == "LocalEnvironment":
+        return "local", "host", _path_dialect()
+    if "SSH" in cls:
+        return "ssh", cls, "posix"
+    if "Docker" in cls or "Singularity" in cls or "Modal" in cls or "Daytona" in cls:
+        return "container", cls, "posix"
+    return "unknown", cls, "unknown"
+
+
+def _path_dialect() -> str:
+    return "windows" if os.name == "nt" else "posix"
+
+
+def _resolve_local_path(raw_path: str, task_id: str) -> Optional[Path]:
+    try:
+        from tools.file_tools import _resolve_path_for_task
+
+        resolved = _resolve_path_for_task(raw_path, task_id or "default")
+    except Exception:
+        try:
+            resolved = Path(raw_path).expanduser()
+        except Exception:
+            return None
+    if not isinstance(resolved, Path):
+        return None
+    return resolved
+
+
+def _path_allowed_for_observation(path: str) -> bool:
+    if not path or not isinstance(path, str):
+        return False
+    p = path.strip()
+    if not p:
+        return False
+    lower = p.lower()
+    if lower.startswith("\\\\.\\") or lower.startswith("\\\\?\\"):
+        return False
+    if p.startswith("\\\\") and not p.startswith("\\\\?\\"):
+        return False
+    if re.match(r"^[a-zA-Z]:\\", p):
+        rest = p[3:]
+        if rest.startswith("\\") and _is_dos_device_component(rest.lstrip("\\").split("\\")[0]):
+            return False
+    parts = re.split(r"[\\/]", p)
+    for part in parts:
+        if _is_dos_device_component(part):
+            return False
+    if ".." in parts:
+        return False
+    try:
+        from agent.file_safety import get_read_block_error, is_write_denied
+
+        if get_read_block_error(p) or is_write_denied(p):
+            return False
+    except Exception:
+        pass
+    return True
+
+
+def _is_dos_device_component(name: str) -> bool:
+    if not name:
+        return False
+    base = name.split(".")[0].upper()
+    return base in {
+        "CON", "PRN", "AUX", "NUL",
+        *{f"COM{i}" for i in range(1, 10)},
+        *{f"LPT{i}" for i in range(1, 10)},
+    }
+
+
+def _stable_local_fingerprint(
+    path: Path,
+    *,
+    deadline: float,
+    active_workers: Set[int],
+) -> Tuple[Optional[ContentFingerprint], int]:
+    if time.monotonic() > deadline:
+        return None, 0
+    try:
+        st = path.lstat()
+    except OSError:
+        return None, 0
+    if stat.S_ISLNK(st.st_mode):
+        return None, 0
+    if not stat.S_ISREG(st.st_mode):
+        return None, 0
+    if st.st_size > MAX_SINGLE_FILE_BYTES:
+        return None, 0
+    try:
+        st_open = os.stat(path, follow_symlinks=True)
+    except OSError:
+        return None, 0
+    if not stat.S_ISREG(st_open.st_mode):
+        return None, 0
+    if st_open.st_size > MAX_SINGLE_FILE_BYTES:
+        return None, 0
+    read_bytes = 0
+    h = hashlib.sha256()
+    try:
+        with open(path, "rb") as fh:
+            while True:
+                if time.monotonic() > deadline:
+                    return None, read_bytes
+                chunk = fh.read(65536)
+                if not chunk:
+                    break
+                read_bytes += len(chunk)
+                h.update(chunk)
+        st_after = os.stat(path, follow_symlinks=True)
+    except OSError:
+        return None, read_bytes
+    if (
+        st_after.st_size != st_open.st_size
+        or st_after.st_mtime_ns != st_open.st_mtime_ns
+        or st_after.st_ino != st_open.st_ino
+    ):
+        return None, read_bytes
+    return (
+        ContentFingerprint(
+            size=st_after.st_size,
+            digest=h.digest(),
+            mtime_ns=st_after.st_mtime_ns,
+            inode=st_after.st_ino,
+        ),
+        read_bytes,
+    )
+
+
+def _sanitize_footer_text(text: str) -> str:
+    if not text:
+        return text
+    cleaned = _FOOTER_INJECTION_RE.sub("[filtered]", text)
+    cleaned = cleaned.replace("\x00", "")
+    return cleaned
+
+
+def format_failure_footer(
+    failed: Dict[str, Dict[str, Any]],
+    *,
+    format_paths: Callable[[str], str],
+) -> str:
+    if not failed:
+        return ""
+    overflow = failed.pop("__overflow__", None)
+    try:
+        lines = [
+            "⚠️ File-mutation verifier: "
+            f"{len(failed)} file(s) were NOT modified this turn despite any "
+            "wording above that may suggest otherwise. Run `git status` or "
+            "`read_file` to confirm."
+        ]
+        shown = 0
+        for path, info in failed.items():
+            if shown >= 10:
+                break
+            preview = _sanitize_footer_text((info.get("error_preview") or "").strip())
+            tool = info.get("tool") or "patch"
+            if preview:
+                lines.append(f"  • `{path}` — [{tool}] {preview}")
+            else:
+                lines.append(f"  • `{path}` — [{tool}] failed")
+            shown += 1
+        remaining = len(failed) - shown
+        if remaining > 0:
+            lines.append(f"  • … and {remaining} more")
+        if overflow:
+            lines.append("  • … additional failures omitted (turn observation budget)")
+        return format_paths("\n".join(lines))
+    except Exception:
+        logger.debug("file-mutation footer formatting failed", exc_info=True)
+        return _GENERIC_FOOTER_FALLBACK
+
+
+def get_verifier(agent: Any) -> Optional[TurnFileMutationVerifier]:
+    return getattr(agent, "_file_mutation_verifier", None)
+
+
+def ensure_verifier(agent: Any) -> TurnFileMutationVerifier:
+    v = getattr(agent, "_file_mutation_verifier", None)
+    if v is None:
+        v = TurnFileMutationVerifier()
+        agent._file_mutation_verifier = v
+    return v
+
+
+def sync_legacy_failed_state(agent: Any) -> None:
+    v = get_verifier(agent)
+    state = getattr(agent, "_turn_failed_file_mutations", None)
+    if v is None or state is None:
+        return
+    state.clear()
+    state.update(v.finalize_failed_dict())
diff --git a/agent/tool_dispatch_helpers.py b/agent/tool_dispatch_helpers.py
index 07b2c2e65e79c..eaf5949d58eff 100644
--- a/agent/tool_dispatch_helpers.py
+++ b/agent/tool_dispatch_helpers.py
@@ -57,8 +57,10 @@
     "web_search",
 })
 
-# File tools can run concurrently when they target independent paths.
-_PATH_SCOPED_TOOLS = frozenset({"read_file", "write_file", "patch"})
+# Read-only path tools can run concurrently when they target independent paths.
+# Structured mutations (write_file/patch) are always sequential barriers —
+# request/execution middleware may alias paths after planning.
+_PATH_SCOPED_TOOLS = frozenset({"read_file"})
 
 # Patterns that indicate a terminal command may modify/delete files.
 _DESTRUCTIVE_PATTERNS = re.compile(
@@ -152,6 +154,10 @@ def _add_sequential(tc) -> None:
             _add_sequential(tool_call)
             continue
 
+        if tool_name in _FILE_MUTATING_TOOLS:
+            _add_sequential(tool_call)
+            continue
+
         try:
             function_args = json.loads(tool_call.function.arguments)
         except Exception:
@@ -353,26 +359,20 @@ def _extract_file_mutation_targets(tool_name: str, args: Dict[str, Any]) -> List
         body = args.get("patch") or ""
         if not isinstance(body, str) or not body:
             return []
+        try:
+            from tools.patch_parser import OperationType, parse_v4a_patch
+
+            operations, parse_error = parse_v4a_patch(body)
+        except Exception:
+            return []
+        if parse_error:
+            return []
         paths: List[str] = []
-        for _m in re.finditer(
-            r'^\*\*\*\s+(?:Update|Add|Delete)\s+File:\s*(.+)$',
-            body,
-            re.MULTILINE,
-        ):
-            p = _m.group(1).strip()
-            if p:
-                paths.append(p)
-        for _m in re.finditer(
-            r'^\*\*\*\s+Move\s+File:\s*(.+?)\s*->\s*(.+)$',
-            body,
-            re.MULTILINE,
-        ):
-            src = _m.group(1).strip()
-            dst = _m.group(2).strip()
-            if src:
-                paths.append(src)
-            if dst:
-                paths.append(dst)
+        for op in operations:
+            if op.file_path:
+                paths.append(op.file_path)
+            if op.operation == OperationType.MOVE and op.new_path:
+                paths.append(op.new_path)
         return paths
     return []
 
diff --git a/agent/tool_executor.py b/agent/tool_executor.py
index d32fe99c0c55c..22f21a66ba846 100644
--- a/agent/tool_executor.py
+++ b/agent/tool_executor.py
@@ -327,12 +327,15 @@ def _run_agent_tool_execution_middleware(
     effective_task_id: str,
     tool_call_id: str,
     execute,
-) -> tuple[Any, dict]:
+) -> tuple[Any, dict, bool]:
     observed_args = function_args
 
+    registry_dispatched = False
+
     def _execute(next_args: dict) -> Any:
-        nonlocal observed_args
+        nonlocal observed_args, registry_dispatched
         observed_args = next_args if isinstance(next_args, dict) else function_args
+        registry_dispatched = True
         return execute(observed_args)
 
     from hermes_cli.middleware import run_tool_execution_middleware
@@ -348,7 +351,88 @@ def _execute(next_args: dict) -> Any:
         turn_id=getattr(agent, "_current_turn_id", "") or "",
         api_request_id=getattr(agent, "_current_api_request_id", "") or "",
     )
-    return result, observed_args
+    return result, observed_args, registry_dispatched
+
+
+def _file_mutation_turn_generation(agent) -> Optional[int]:
+    from agent.file_mutation_verifier import get_verifier
+
+    verifier = get_verifier(agent)
+    return verifier.generation if verifier is not None else None
+
+
+def _prepare_file_mutation_dispatch(
+    agent,
+    tool_name: str,
+    function_args: dict,
+    effective_task_id: str,
+) -> None:
+    try:
+        from agent.file_mutation_verifier import get_verifier
+
+        verifier = get_verifier(agent)
+        if verifier is None:
+            return
+        verifier.prepare_mutation_dispatch(
+            tool_name=tool_name,
+            effective_args=function_args,
+            effective_task_id=effective_task_id or "default",
+            turn_generation=verifier.generation,
+        )
+    except Exception as exc:
+        logging.debug("file-mutation prepare_dispatch failed: %s", exc)
+
+
+def _finalize_file_mutation_tool(
+    agent,
+    *,
+    tool_name: str,
+    function_args: dict,
+    raw_result: Any,
+    model_result: Any,
+    is_error: bool,
+    blocked: bool,
+    registry_dispatched: bool,
+    result_missing: bool,
+    effective_task_id: str,
+) -> None:
+    try:
+        from agent.file_mutation_verifier import (
+            DispatchTriState,
+            get_verifier,
+            sync_legacy_failed_state,
+        )
+
+        if blocked:
+            dispatch = DispatchTriState.NOT_DISPATCHED
+        elif result_missing:
+            dispatch = DispatchTriState.DISPATCHED_NO_RESULT
+        elif not registry_dispatched:
+            dispatch = DispatchTriState.NOT_DISPATCHED
+        else:
+            dispatch = DispatchTriState.DISPATCHED
+        gen = _file_mutation_turn_generation(agent)
+        agent._record_file_mutation_result(
+            tool_name,
+            function_args,
+            model_result,
+            is_error,
+            raw_result=raw_result,
+            dispatch=dispatch.value,
+            blocked=blocked,
+            effective_task_id=effective_task_id,
+            turn_generation=gen,
+        )
+        verifier = get_verifier(agent)
+        if verifier is not None and not blocked:
+            verifier.observe_after_tool(
+                tool_name=tool_name,
+                effective_task_id=effective_task_id or "default",
+                blocked=False,
+            )
+            sync_legacy_failed_state(agent)
+    except Exception as exc:
+        logging.debug("file-mutation finalize failed: %s", exc)
 
 
 def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effective_task_id: str, api_call_count: int = 0, *, finalize: bool = True) -> None:
@@ -638,6 +722,9 @@ def _run_tool(index, tool_call, function_name, function_args, middleware_trace):
         start = time.time()
         try:
             try:
+                _prepare_file_mutation_dispatch(
+                    agent, function_name, function_args, effective_task_id,
+                )
                 result = agent._invoke_tool(
                     function_name,
                     function_args,
@@ -890,6 +977,20 @@ def _run_tool(index, tool_call, function_name, function_args, middleware_trace):
                 middleware_trace=list(middleware_trace),
             )
             tool_duration = float(timeout_s or 0.0)
+            is_error = True
+            blocked = False
+            _finalize_file_mutation_tool(
+                agent,
+                tool_name=name,
+                function_args=args,
+                raw_result=function_result,
+                model_result=function_result,
+                is_error=is_error,
+                blocked=blocked,
+                registry_dispatched=True,
+                result_missing=True,
+                effective_task_id=effective_task_id,
+            )
         elif r is None:
             # Tool was cancelled (interrupt) or thread didn't return
             if agent._interrupt_requested:
@@ -921,6 +1022,20 @@ def _run_tool(index, tool_call, function_name, function_args, middleware_trace):
                     middleware_trace=list(middleware_trace),
                 )
             tool_duration = 0.0
+            is_error = True
+            blocked = False
+            _finalize_file_mutation_tool(
+                agent,
+                tool_name=name,
+                function_args=args,
+                raw_result=function_result,
+                model_result=function_result,
+                is_error=is_error,
+                blocked=blocked,
+                registry_dispatched=not agent._interrupt_requested,
+                result_missing=not agent._interrupt_requested,
+                effective_task_id=effective_task_id,
+            )
         else:
             function_name, function_args, function_result, tool_duration, is_error, blocked, middleware_trace = r
             progress_function_name = function_name
@@ -928,12 +1043,15 @@ def _run_tool(index, tool_call, function_name, function_args, middleware_trace):
                 effect_disposition = "none"
 
             if not blocked:
+                _raw_mutation_result = function_result
                 function_result = agent._append_guardrail_observation(
                     function_name,
                     function_args,
                     function_result,
                     failed=is_error,
                 )
+            else:
+                _raw_mutation_result = function_result
 
             if is_error:
                 _err_text = _multimodal_text_summary(function_result)
@@ -941,15 +1059,18 @@ def _run_tool(index, tool_call, function_name, function_args, middleware_trace):
                 logger.warning("Tool %s returned error (%.2fs): %s", function_name, tool_duration, result_preview)
 
             # Track file-mutation outcome for the turn-end verifier.
-            # `blocked` calls never actually ran — don't let a guardrail
-            # block count as either a failure or a success.
-            if not blocked:
-                try:
-                    agent._record_file_mutation_result(
-                        function_name, function_args, function_result, is_error,
-                    )
-                except Exception as _ver_err:
-                    logging.debug("file-mutation verifier record failed: %s", _ver_err)
+            _finalize_file_mutation_tool(
+                agent,
+                tool_name=function_name,
+                function_args=function_args,
+                raw_result=_raw_mutation_result,
+                model_result=function_result,
+                is_error=is_error,
+                blocked=blocked,
+                registry_dispatched=True,
+                result_missing=False,
+                effective_task_id=effective_task_id,
+            )
 
             if agent.verbose_logging:
                 logging.debug(f"Tool {function_name} completed in {tool_duration:.2f}s")
@@ -1323,7 +1444,7 @@ def _execute(next_args: dict) -> Any:
                     merge=next_args.get("merge", False),
                     store=agent._todo_store,
                 )
-            function_result, function_args = _run_agent_tool_execution_middleware(
+            function_result, function_args, _registry_dispatched = _run_agent_tool_execution_middleware(
                 agent,
                 function_name=function_name,
                 function_args=function_args,
@@ -1352,7 +1473,7 @@ def _execute(next_args: dict) -> Any:
                     db=session_db,
                     current_session_id=agent.session_id,
                 )
-            function_result, function_args = _run_agent_tool_execution_middleware(
+            function_result, function_args, _registry_dispatched = _run_agent_tool_execution_middleware(
                 agent,
                 function_name=function_name,
                 function_args=function_args,
@@ -1389,7 +1510,7 @@ def _execute(next_args: dict) -> Any:
                         ),
                     )
                 return result
-            function_result, function_args = _run_agent_tool_execution_middleware(
+            function_result, function_args, _registry_dispatched = _run_agent_tool_execution_middleware(
                 agent,
                 function_name=function_name,
                 function_args=function_args,
@@ -1409,7 +1530,7 @@ def _execute(next_args: dict) -> Any:
                     multi_select=next_args.get("multi_select", False),
                     callback=agent.clarify_callback,
                 )
-            function_result, function_args = _run_agent_tool_execution_middleware(
+            function_result, function_args, _registry_dispatched = _run_agent_tool_execution_middleware(
                 agent,
                 function_name=function_name,
                 function_args=function_args,
@@ -1428,7 +1549,7 @@ def _execute(next_args: dict) -> Any:
                     count=next_args.get("count"),
                     callback=getattr(agent, "read_terminal_callback", None),
                 )
-            function_result, function_args = _run_agent_tool_execution_middleware(
+            function_result, function_args, _registry_dispatched = _run_agent_tool_execution_middleware(
                 agent,
                 function_name=function_name,
                 function_args=function_args,
@@ -1460,7 +1581,7 @@ def _execute(next_args: dict) -> Any:
             try:
                 def _execute(next_args: dict) -> Any:
                     return agent._dispatch_delegate_task(next_args)
-                function_result, function_args = _run_agent_tool_execution_middleware(
+                function_result, function_args, _registry_dispatched = _run_agent_tool_execution_middleware(
                     agent,
                     function_name=function_name,
                     function_args=function_args,
@@ -1491,7 +1612,7 @@ def _execute(next_args: dict) -> Any:
             try:
                 def _execute(next_args: dict) -> Any:
                     return agent.context_compressor.handle_tool_call(function_name, next_args, messages=messages)
-                function_result, function_args = _run_agent_tool_execution_middleware(
+                function_result, function_args, _registry_dispatched = _run_agent_tool_execution_middleware(
                     agent,
                     function_name=function_name,
                     function_args=function_args,
@@ -1525,7 +1646,7 @@ def _execute(next_args: dict) -> Any:
             try:
                 def _execute(next_args: dict) -> Any:
                     return agent._memory_manager.handle_tool_call(function_name, next_args)
-                function_result, function_args = _run_agent_tool_execution_middleware(
+                function_result, function_args, _registry_dispatched = _run_agent_tool_execution_middleware(
                     agent,
                     function_name=function_name,
                     function_args=function_args,
@@ -1555,6 +1676,9 @@ def _execute(next_args: dict) -> Any:
                 spinner.start()
             _spinner_result = None
             try:
+                _prepare_file_mutation_dispatch(
+                    agent, function_name, function_args, effective_task_id,
+                )
                 function_result = _ra().handle_function_call(
                     function_name, function_args, effective_task_id,
                     tool_call_id=tool_call.id,
@@ -1597,6 +1721,9 @@ def _execute(next_args: dict) -> Any:
                     agent._vprint(f"  {cute_msg}")
         else:
             try:
+                _prepare_file_mutation_dispatch(
+                    agent, function_name, function_args, effective_task_id,
+                )
                 function_result = _ra().handle_function_call(
                     function_name, function_args, effective_task_id,
                     tool_call_id=tool_call.id,
@@ -1666,6 +1793,7 @@ def _execute(next_args: dict) -> Any:
                 middleware_trace=list(middleware_trace),
             )
         if not _execution_blocked:
+            _raw_mutation_result = function_result
             function_result = agent._append_guardrail_observation(
                 function_name,
                 function_args,
@@ -1686,8 +1814,17 @@ def _execute(next_args: dict) -> Any:
         # turn, not just the parallel ones.
         if not _execution_blocked:
             try:
-                agent._record_file_mutation_result(
-                    function_name, function_args, function_result, _is_error_result,
+                _finalize_file_mutation_tool(
+                    agent,
+                    tool_name=function_name,
+                    function_args=function_args,
+                    raw_result=_raw_mutation_result,
+                    model_result=function_result,
+                    is_error=_is_error_result,
+                    blocked=_execution_blocked,
+                    registry_dispatched=True,
+                    result_missing=False,
+                    effective_task_id=effective_task_id,
                 )
             except Exception as _ver_err:
                 logging.debug("file-mutation verifier record failed: %s", _ver_err)
diff --git a/agent/turn_context.py b/agent/turn_context.py
index ac9555d012e4b..74d257f79605b 100644
--- a/agent/turn_context.py
+++ b/agent/turn_context.py
@@ -1106,6 +1106,12 @@ def build_turn_context(
     # Per-turn file-mutation verifier state.
     agent._turn_failed_file_mutations = {}
     agent._turn_file_mutation_paths = set()
+    from agent.file_mutation_verifier import TurnFileMutationVerifier
+
+    agent._file_mutation_verifier = TurnFileMutationVerifier()
+    agent._file_mutation_verifier.reset_turn(
+        generation=getattr(agent, "_user_turn_count", 0) or 0,
+    )
     agent._verification_stop_nudges = 0
     agent._pre_verify_nudges = 0
 
diff --git a/agent/turn_finalizer.py b/agent/turn_finalizer.py
index 4e2d318b2e2c2..e2b8c96bcd9c7 100644
--- a/agent/turn_finalizer.py
+++ b/agent/turn_finalizer.py
@@ -410,18 +410,32 @@ def finalize_turn(
     # the truth is surfaced on every turn, so over-claiming is
     # structurally impossible past the model.
     #
-    # Gate: only applied when a real text response exists for this
-    # turn and the user didn't interrupt.  Empty/interrupted turns
-    # already have other surface text that shouldn't be augmented.
-    if final_response and not interrupted:
-        try:
-            _failed = getattr(agent, "_turn_failed_file_mutations", None) or {}
-            if _failed and agent._file_mutation_verifier_enabled():
-                footer = agent._format_file_mutation_failure_footer(_failed)
-                if footer:
+    # Gate: append when failures remain.  Interrupted/empty turns still
+    # surface the advisory (fail closed).
+    try:
+        from agent.file_mutation_verifier import sync_legacy_failed_state
+
+        sync_legacy_failed_state(agent)
+        _failed = getattr(agent, "_turn_failed_file_mutations", None) or {}
+        if _failed and agent._file_mutation_verifier_enabled():
+            footer = agent._format_file_mutation_failure_footer(_failed)
+            if footer:
+                if final_response:
                     final_response = final_response.rstrip() + "\n\n" + footer
-        except Exception as _ver_err:
-            logger.debug("file-mutation verifier footer failed: %s", _ver_err)
+                else:
+                    final_response = footer
+    except Exception as _ver_err:
+        logger.debug("file-mutation verifier footer failed: %s", _ver_err)
+        try:
+            if agent._file_mutation_verifier_enabled():
+                from agent.file_mutation_verifier import _GENERIC_FOOTER_FALLBACK
+
+                if final_response:
+                    final_response = final_response.rstrip() + "\n\n" + _GENERIC_FOOTER_FALLBACK
+                else:
+                    final_response = _GENERIC_FOOTER_FALLBACK
+        except Exception:
+            pass
 
     # Turn-completion explainer.
     # When a turn ends abnormally after substantive work — empty content
@@ -612,6 +626,15 @@ def finalize_turn(
     # (the response is still returned either way — #8049).
     if _cleanup_errors:
         result["cleanup_errors"] = _cleanup_errors
+    try:
+        from agent.file_mutation_verifier import get_verifier
+
+        _verifier = get_verifier(agent)
+        if _verifier is not None:
+            _verifier.clear_turn()
+        agent._file_mutation_verifier = None
+    except Exception:
+        pass
     # If a /steer landed after the final assistant turn (no more tool
     # batches to drain into), hand it back to the caller so it can be
     # delivered as the next user turn instead of being silently lost.
diff --git a/run_agent.py b/run_agent.py
index fe8db996955df..7bf4d2f6c1647 100644
--- a/run_agent.py
+++ b/run_agent.py
@@ -3195,42 +3195,58 @@ def _record_file_mutation_result(
         args: Dict[str, Any],
         result: Any,
         is_error: bool,
+        *,
+        raw_result: Any = None,
+        dispatch: Optional[str] = None,
+        blocked: bool = False,
+        effective_task_id: str = "default",
+        turn_generation: Optional[int] = None,
     ) -> None:
-        """Record a ``write_file`` / ``patch`` outcome for the turn-end verifier.
+        """Record a ``write_file`` / ``patch`` outcome for the turn-end verifier."""
+        from agent.file_mutation_verifier import (
+            DispatchTriState,
+            ensure_verifier,
+            sync_legacy_failed_state,
+        )
 
-        On failure, store ``{path: {error_preview, tool}}`` entries.  On
-        success, remove any prior failure entries for the same paths (the
-        model recovered within the turn).  Silently no-ops if the per-turn
-        state dict hasn't been initialised yet (e.g. a tool dispatched
-        outside ``run_conversation``).
-        """
-        if tool_name not in _FILE_MUTATING_TOOLS:
-            return
         state = getattr(self, "_turn_failed_file_mutations", None)
         if state is None:
             return
-        targets = _extract_file_mutation_targets(tool_name, args)
-        if not targets:
-            return
-        landed = file_mutation_result_landed(tool_name, result)
-        if landed:
-            changed = getattr(self, "_turn_file_mutation_paths", None)
-            if changed is not None:
-                changed.update(_extract_landed_file_mutation_paths(tool_name, args, result))
-        if is_error and not landed:
-            preview = _extract_error_preview(result)
-            for path in targets:
-                # Keep the FIRST error we saw for a given path unless we
-                # later see success.  A repeated failure with a different
-                # message shouldn't silently overwrite the original.
-                if path not in state:
-                    state[path] = {
-                        "tool": tool_name,
-                        "error_preview": preview,
-                    }
+
+        verifier = ensure_verifier(self)
+        if dispatch is None:
+            tri = (
+                DispatchTriState.NOT_DISPATCHED
+                if blocked
+                else DispatchTriState.DISPATCHED
+            )
         else:
-            for path in targets:
-                state.pop(path, None)
+            tri = DispatchTriState(dispatch)
+        raw = raw_result if raw_result is not None else result
+        verifier.record_tool_outcome(
+            tool_name=tool_name,
+            effective_args=args,
+            effective_task_id=effective_task_id or "default",
+            raw_result=raw,
+            dispatch=tri,
+            model_is_error=is_error,
+            blocked=blocked,
+            turn_generation=turn_generation,
+        )
+        verifier.observe_after_tool(
+            tool_name=tool_name,
+            effective_task_id=effective_task_id or "default",
+            blocked=blocked,
+        )
+        sync_legacy_failed_state(self)
+        if tool_name in _FILE_MUTATING_TOOLS and not blocked and tri is DispatchTriState.DISPATCHED:
+            landed = file_mutation_result_landed(tool_name, raw)
+            if landed:
+                changed = getattr(self, "_turn_file_mutation_paths", None)
+                if changed is not None:
+                    changed.update(
+                        _extract_landed_file_mutation_paths(tool_name, args, raw)
+                    )
 
     def _file_mutation_verifier_enabled(self) -> bool:
         """Check whether the per-turn file-mutation verifier footer is on.
@@ -3302,32 +3318,14 @@ def _format_file_mutation_failure_footer(cls, failed: Dict[str, Dict[str, Any]])
         bare-path media extractor can never auto-attach a protected file
         (e.g. ``~/.hermes/config.yaml``) to a messaging channel (#35584).
         """
+        from agent.file_mutation_verifier import format_failure_footer
+
         if not failed:
             return ""
-        lines = [
-            "⚠️ File-mutation verifier: "
-            f"{len(failed)} file(s) were NOT modified this turn despite any "
-            "wording above that may suggest otherwise. Run `git status` or "
-            "`read_file` to confirm."
-        ]
-        shown = 0
-        for path, info in failed.items():
-            if shown >= 10:
-                break
-            preview = (info.get("error_preview") or "").strip()
-            tool = info.get("tool") or "patch"
-            if preview:
-                lines.append(f"  • `{path}` — [{tool}] {preview}")
-            else:
-                lines.append(f"  • `{path}` — [{tool}] failed")
-            shown += 1
-        remaining = len(failed) - shown
-        if remaining > 0:
-            lines.append(f"  • … and {remaining} more")
-        # Neutralize any path the preview text echoed (the bullet path is
-        # already backticked above; the lookbehind keeps it from being
-        # double-wrapped).
-        return cls._neutralize_footer_paths("\n".join(lines))
+        return format_failure_footer(
+            dict(failed),
+            format_paths=cls._neutralize_footer_paths,
+        )
 
     def _turn_completion_explainer_enabled(self) -> bool:
         """Check whether the end-of-turn completion explainer footer is on.
diff --git a/tests/agent/test_file_mutation_verifier_redesign.py b/tests/agent/test_file_mutation_verifier_redesign.py
new file mode 100644
index 0000000000000..803d23c37bd27
--- /dev/null
+++ b/tests/agent/test_file_mutation_verifier_redesign.py
@@ -0,0 +1,314 @@
+"""Behavioral tests for the content-transition file-mutation verifier redesign."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from agent.file_mutation_verifier import (
+    DispatchTriState,
+    TurnFileMutationVerifier,
+    _path_allowed_for_observation,
+    format_failure_footer,
+)
+from run_agent import AIAgent, _extract_file_mutation_targets
+
+
+def _bare_agent() -> AIAgent:
+    from agent.file_mutation_verifier import TurnFileMutationVerifier
+
+    agent = object.__new__(AIAgent)
+    agent._turn_failed_file_mutations = {}
+    agent._turn_file_mutation_paths = set()
+    agent._file_mutation_verifier = TurnFileMutationVerifier()
+    agent._file_mutation_verifier.reset_turn(1)
+    return agent
+
+
+class TestContentTransitionSuppressesFooter:
+    def test_failed_patch_then_external_edit_clears_footer(self, tmp_path, monkeypatch):
+        monkeypatch.chdir(tmp_path)
+        target = tmp_path / "config.yaml"
+        target.write_text("before: true\n", encoding="utf-8")
+
+        agent = _bare_agent()
+        fail = json.dumps({"success": False, "error": "Write denied (simulated)"})
+        agent._record_file_mutation_result(
+            "patch",
+            {"mode": "replace", "path": "config.yaml", "old_string": "x", "new_string": "y"},
+            fail,
+            is_error=True,
+            raw_result=fail,
+            dispatch=DispatchTriState.DISPATCHED.value,
+            effective_task_id="default",
+        )
+        assert agent._turn_failed_file_mutations
+
+        target.write_text("after: true\n", encoding="utf-8")
+        from agent.file_mutation_verifier import sync_legacy_failed_state
+
+        sync_legacy_failed_state(agent)
+        assert agent._turn_failed_file_mutations == {}
+
+    def test_unchanged_file_keeps_footer(self, tmp_path, monkeypatch):
+        monkeypatch.chdir(tmp_path)
+        target = tmp_path / "stale.py"
+        target.write_text("same\n", encoding="utf-8")
+
+        agent = _bare_agent()
+        fail = json.dumps({"error": "Could not find old_string"})
+        agent._record_file_mutation_result(
+            "patch",
+            {"mode": "replace", "path": "stale.py", "old_string": "nope", "new_string": "y"},
+            fail,
+            is_error=True,
+            raw_result=fail,
+            dispatch=DispatchTriState.DISPATCHED.value,
+        )
+        from agent.file_mutation_verifier import sync_legacy_failed_state
+
+        sync_legacy_failed_state(agent)
+        assert any(k.endswith("stale.py") for k in agent._turn_failed_file_mutations)
+
+
+class TestDispatchTriState:
+    def test_not_dispatched_creates_no_ledger_io(self, tmp_path, monkeypatch):
+        monkeypatch.chdir(tmp_path)
+        agent = _bare_agent()
+        agent._record_file_mutation_result(
+            "write_file",
+            {"path": "new.txt", "content": "x"},
+            json.dumps({"bytes_written": 1}),
+            is_error=False,
+            dispatch=DispatchTriState.NOT_DISPATCHED.value,
+            blocked=True,
+        )
+        assert agent._turn_failed_file_mutations == {}
+
+    def test_middleware_short_circuit_does_not_clear_prior_failure(self, tmp_path, monkeypatch):
+        monkeypatch.chdir(tmp_path)
+        (tmp_path / "a.txt").write_text("x\n", encoding="utf-8")
+        agent = _bare_agent()
+        fail = json.dumps({"error": "first"})
+        agent._record_file_mutation_result(
+            "patch",
+            {"mode": "replace", "path": "a.txt", "old_string": "x", "new_string": "y"},
+            fail,
+            is_error=True,
+            raw_result=fail,
+            dispatch=DispatchTriState.DISPATCHED.value,
+        )
+        agent._record_file_mutation_result(
+            "write_file",
+            {"path": "a.txt", "content": "landed"},
+            json.dumps({"bytes_written": 6}),
+            is_error=False,
+            dispatch=DispatchTriState.NOT_DISPATCHED.value,
+        )
+        from agent.file_mutation_verifier import sync_legacy_failed_state
+
+        sync_legacy_failed_state(agent)
+        assert any(k.endswith("a.txt") for k in agent._turn_failed_file_mutations)
+
+
+class TestV4aParserParity:
+    def test_crlf_and_compact_headers_use_parser(self):
+        body = (
+            "*** Begin Patch\r\n"
+            "***Update File: a.py\r\n"
+            "@@ @@\r\n"
+            "-a\r\n"
+            "+b\r\n"
+            "*** End Patch\r\n"
+        )
+        assert _extract_file_mutation_targets("patch", {"mode": "patch", "patch": body}) == ["a.py"]
+
+    def test_invalid_patch_yields_no_targets(self):
+        body = "*** Update File: outside\n*** Begin Patch\n*** End Patch\n"
+        assert _extract_file_mutation_targets("patch", {"mode": "patch", "patch": body}) == []
+
+
+class TestPathSafety:
+    @pytest.mark.skipif(__import__("sys").platform != "win32", reason="DOS device paths")
+    def test_dos_device_rejected(self):
+        assert not _path_allowed_for_observation(r"\\.\CON")
+
+    def test_unc_rejected(self):
+        assert not _path_allowed_for_observation(r"\\server\share\file.txt")
+
+
+class TestFooterSanitization:
+    def test_media_injection_neutralized(self):
+        failed = {
+            "/tmp/x.md": {
+                "tool": "patch",
+                "error_preview": "MEDIA: /etc/passwd\nfailed",
+            }
+        }
+        out = format_failure_footer(failed, format_paths=lambda s: s)
+        assert "MEDIA:" not in out
+        assert "[filtered]" in out
+
+
+class TestMetadataOnlyChange:
+    def test_mtime_only_does_not_suppress_footer(self, tmp_path, monkeypatch):
+        monkeypatch.chdir(tmp_path)
+        target = tmp_path / "meta.py"
+        target.write_text("content\n", encoding="utf-8")
+        agent = _bare_agent()
+        agent._file_mutation_verifier.prepare_mutation_dispatch(
+            tool_name="patch",
+            effective_args={
+                "mode": "replace",
+                "path": "meta.py",
+                "old_string": "x",
+                "new_string": "y",
+            },
+            effective_task_id="default",
+            turn_generation=1,
+        )
+        fail = json.dumps({"error": "failed"})
+        agent._record_file_mutation_result(
+            "patch",
+            {"mode": "replace", "path": "meta.py", "old_string": "x", "new_string": "y"},
+            fail,
+            is_error=True,
+            raw_result=fail,
+            dispatch=DispatchTriState.DISPATCHED.value,
+            turn_generation=1,
+        )
+        import os
+
+        os.utime(target, None)
+        from agent.file_mutation_verifier import sync_legacy_failed_state
+
+        sync_legacy_failed_state(agent)
+        assert any(k.endswith("meta.py") for k in agent._turn_failed_file_mutations)
+
+
+class TestPathAliasReconciliation:
+    def test_relative_and_absolute_same_file_clears(self, tmp_path, monkeypatch):
+        monkeypatch.chdir(tmp_path)
+        target = tmp_path / "alias.py"
+        target.write_text("start\n", encoding="utf-8")
+        agent = _bare_agent()
+        fail = json.dumps({"error": "nope"})
+        agent._record_file_mutation_result(
+            "patch",
+            {"mode": "replace", "path": "alias.py", "old_string": "x", "new_string": "y"},
+            fail,
+            is_error=True,
+            raw_result=fail,
+            dispatch=DispatchTriState.DISPATCHED.value,
+        )
+        target.write_text("end\n", encoding="utf-8")
+        agent._file_mutation_verifier.reconcile_content_transitions(task_id="default")
+        from agent.file_mutation_verifier import sync_legacy_failed_state
+
+        sync_legacy_failed_state(agent)
+        assert agent._turn_failed_file_mutations == {}
+
+
+class TestTurnGenerationBudget:
+    def test_stale_generation_does_not_record(self, tmp_path, monkeypatch):
+        monkeypatch.chdir(tmp_path)
+        agent = _bare_agent()
+        agent._file_mutation_verifier.reset_turn(2)
+        fail = json.dumps({"error": "late"})
+        agent._record_file_mutation_result(
+            "write_file",
+            {"path": "z.txt", "content": "a"},
+            fail,
+            is_error=True,
+            raw_result=fail,
+            dispatch=DispatchTriState.DISPATCHED.value,
+            turn_generation=1,
+        )
+        assert agent._turn_failed_file_mutations == {}
+
+
+class TestPreDispatchBaseline:
+    def test_unchanged_file_after_prepare_keeps_failure(self, tmp_path, monkeypatch):
+        monkeypatch.chdir(tmp_path)
+        target = tmp_path / "partial.py"
+        target.write_text("baseline\n", encoding="utf-8")
+        verifier = TurnFileMutationVerifier()
+        verifier.reset_turn(1)
+        verifier.prepare_mutation_dispatch(
+            tool_name="write_file",
+            effective_args={"path": "partial.py", "content": "changed\n"},
+            effective_task_id="default",
+            turn_generation=1,
+        )
+        verifier.record_tool_outcome(
+            tool_name="write_file",
+            effective_args={"path": "partial.py", "content": "changed\n"},
+            effective_task_id="default",
+            raw_result=json.dumps({"error": "disk full"}),
+            dispatch=DispatchTriState.DISPATCHED,
+            model_is_error=True,
+            turn_generation=1,
+        )
+        assert "partial.py" in verifier.finalize_failed_dict()
+
+    def test_post_dispatch_disk_change_suppresses_using_pre_baseline(self, tmp_path, monkeypatch):
+        monkeypatch.chdir(tmp_path)
+        target = tmp_path / "partial.py"
+        target.write_text("baseline\n", encoding="utf-8")
+        verifier = TurnFileMutationVerifier()
+        verifier.reset_turn(1)
+        verifier.prepare_mutation_dispatch(
+            tool_name="write_file",
+            effective_args={"path": "partial.py", "content": "changed\n"},
+            effective_task_id="default",
+            turn_generation=1,
+        )
+        target.write_text("mutated-on-disk\n", encoding="utf-8")
+        verifier.record_tool_outcome(
+            tool_name="write_file",
+            effective_args={"path": "partial.py", "content": "changed\n"},
+            effective_task_id="default",
+            raw_result=json.dumps({"error": "disk full"}),
+            dispatch=DispatchTriState.DISPATCHED,
+            model_is_error=True,
+            turn_generation=1,
+        )
+        assert "partial.py" not in verifier.finalize_failed_dict()
+
+
+class TestRecoveredThenFailedAgain:
+    def test_later_failure_after_transition(self, tmp_path, monkeypatch):
+        monkeypatch.chdir(tmp_path)
+        p = tmp_path / "flip.py"
+        p.write_text("v1\n", encoding="utf-8")
+        agent = _bare_agent()
+        fail1 = json.dumps({"error": "first"})
+        agent._record_file_mutation_result(
+            "patch",
+            {"mode": "replace", "path": "flip.py", "old_string": "v1", "new_string": "v2"},
+            fail1,
+            is_error=True,
+            raw_result=fail1,
+            dispatch=DispatchTriState.DISPATCHED.value,
+        )
+        p.write_text("v2\n", encoding="utf-8")
+        from agent.file_mutation_verifier import sync_legacy_failed_state
+
+        sync_legacy_failed_state(agent)
+        assert agent._turn_failed_file_mutations == {}
+
+        fail2 = json.dumps({"error": "second failure"})
+        agent._record_file_mutation_result(
+            "patch",
+            {"mode": "replace", "path": "flip.py", "old_string": "v9", "new_string": "v3"},
+            fail2,
+            is_error=True,
+            raw_result=fail2,
+            dispatch=DispatchTriState.DISPATCHED.value,
+        )
+        sync_legacy_failed_state(agent)
+        assert any(k.endswith("flip.py") for k in agent._turn_failed_file_mutations)
+        key = next(k for k in agent._turn_failed_file_mutations if k.endswith("flip.py"))
+        assert "second failure" in agent._turn_failed_file_mutations[key]["error_preview"]
diff --git a/tests/run_agent/test_file_mutation_verifier.py b/tests/run_agent/test_file_mutation_verifier.py
index 578c846e65818..709068c8fab8f 100644
--- a/tests/run_agent/test_file_mutation_verifier.py
+++ b/tests/run_agent/test_file_mutation_verifier.py
@@ -119,17 +119,14 @@ def test_none_returns_empty(self):
 
 
 def _bare_agent() -> AIAgent:
-    """Skip __init__ and only attach the per-turn state dict.
+    """Skip __init__ and only attach the per-turn state dict."""
+    from agent.file_mutation_verifier import TurnFileMutationVerifier
 
-    AIAgent.__init__ takes ~60 parameters and touches network, auth, and
-    the filesystem.  For these tests we only need the two methods —
-    ``_record_file_mutation_result`` and ``_format_file_mutation_failure_footer``.
-    Using ``object.__new__`` mirrors the gateway-test pattern documented in
-    the agent pitfalls list.
-    """
     agent = object.__new__(AIAgent)
     agent._turn_failed_file_mutations = {}
     agent._turn_file_mutation_paths = set()
+    agent._file_mutation_verifier = TurnFileMutationVerifier()
+    agent._file_mutation_verifier.reset_turn(1)
     return agent
 
 
diff --git a/tests/run_agent/test_tool_batch_segmentation.py b/tests/run_agent/test_tool_batch_segmentation.py
index acd75317c28cf..c10e3c9a2084d 100644
--- a/tests/run_agent/test_tool_batch_segmentation.py
+++ b/tests/run_agent/test_tool_batch_segmentation.py
@@ -146,11 +146,10 @@ def test_overlapping_paths_split_across_segments(self, tmp_path, monkeypatch):
             _tc("web_search", call_id="r2"),
         ]
         segments = _plan_tool_batch_segments(calls)
-        # w2 conflicts with w1 → closes the first run; w2+r2 form the second.
-        assert _kinds(segments) == ["parallel", "parallel"]
+        # write_file is always a sequential barrier regardless of path overlap.
+        assert _kinds(segments) == ["parallel", "sequential"]
         assert [tc.id for tc in segments[0][1]] == ["w1", "r1"]
         assert [tc.id for tc in segments[1][1]] == ["w2", "r2"]
-        # Order and completeness preserved.
         assert _flatten_ids(segments) == ["w1", "r1", "w2", "r2"]
 
     def test_path_scoped_tool_without_path_is_a_barrier(self):
@@ -469,11 +468,11 @@ def test_execution_cwd_used_over_process_cwd(self, tmp_path, monkeypatch):
 
         # With execution_cwd supplied the relative path resolves under exec_cwd.
         path_with_cwd = _extract_parallel_scope_path(
-            "write_file", {"path": "x.txt"}, execution_cwd=exec_cwd
+            "read_file", {"path": "x.txt"}, execution_cwd=exec_cwd
         )
         # The absolute path under exec_cwd must match.
         path_absolute = _extract_parallel_scope_path(
-            "write_file", {"path": str(exec_cwd / "x.txt")}
+            "read_file", {"path": str(exec_cwd / "x.txt")}
         )
 
         assert path_with_cwd is not None

From fb37c8257a7aba547a1f45ceb57ac4afbf06be08 Mon Sep 17 00:00:00 2001
From: Rickard Robin <rickard@kumobits.com>
Date: Mon, 27 Jul 2026 02:06:32 +0000
Subject: [PATCH 2/3] fix(agent): harden file-mutation verifier failsafe paths

Use killable subprocess isolation for bounded local hashing, O_NOFOLLOW
opens with stability checks, registry dispatch tracking via ContextVar,
and tests for middleware short-circuit, observation timeout, and turn-end
footer on interrupted/empty turns.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 agent/file_mutation_verifier.py               | 154 +++++++++++++++--
 agent/tool_executor.py                        | 142 +++++++++------
 model_tools.py                                |  25 +++
 .../test_file_mutation_verifier_redesign.py   | 151 +++++++++++++++-
 ...est_turn_finalizer_file_mutation_footer.py | 162 ++++++++++++++++++
 5 files changed, 574 insertions(+), 60 deletions(-)
 create mode 100644 tests/agent/test_turn_finalizer_file_mutation_footer.py

diff --git a/agent/file_mutation_verifier.py b/agent/file_mutation_verifier.py
index eb7e4f10c3ec2..6f95fd20f976b 100644
--- a/agent/file_mutation_verifier.py
+++ b/agent/file_mutation_verifier.py
@@ -10,6 +10,7 @@
 
 import hashlib
 import logging
+import multiprocessing
 import os
 import re
 import stat
@@ -111,20 +112,22 @@ def __init__(
         *,
         resolve_backend: Optional[Callable[[str], Tuple[str, str, str]]] = None,
         now_fn: Callable[[], float] = time.monotonic,
+        use_subprocess_fingerprint: bool = True,
     ) -> None:
         self._resolve_backend = resolve_backend or _default_resolve_backend
         self._now_fn = now_fn
+        self._use_subprocess_fingerprint = use_subprocess_fingerprint
         self._budget = TurnObservationBudget()
         self._ledger: Dict[Tuple[str, str, str, str, str], LedgerEntry] = {}
         self._lock = threading.Lock()
-        self._active_workers: Set[int] = set()
+        self._active_worker_pids: Set[int] = set()
         self._pre_dispatch_baselines: Dict[Tuple[str, str], Optional[ContentFingerprint]] = {}
 
     def reset_turn(self, generation: int = 0) -> None:
         with self._lock:
             self._budget = TurnObservationBudget(generation=generation)
             self._ledger.clear()
-            self._active_workers.clear()
+            self._active_worker_pids.clear()
             self._pre_dispatch_baselines.clear()
 
     def prepare_mutation_dispatch(
@@ -152,6 +155,11 @@ def prepare_mutation_dispatch(
     def clear_turn(self) -> None:
         self.reset_turn(0)
 
+    @property
+    def active_worker_pids(self) -> Set[int]:
+        with self._lock:
+            return set(self._active_worker_pids)
+
     @property
     def generation(self) -> int:
         return self._budget.generation
@@ -358,7 +366,9 @@ def _capture_baseline(
         fp, read_bytes = _stable_local_fingerprint(
             resolved,
             deadline=self._now_fn() + OBSERVATION_TIMEOUT_S,
-            active_workers=self._active_workers,
+            active_worker_pids=self._active_worker_pids,
+            use_subprocess=self._use_subprocess_fingerprint,
+            pid_lock=self._lock,
         )
         with self._lock:
             self._budget.snapshot_bytes += read_bytes
@@ -463,7 +473,104 @@ def _stable_local_fingerprint(
     path: Path,
     *,
     deadline: float,
-    active_workers: Set[int],
+    active_worker_pids: Set[int],
+    use_subprocess: bool = True,
+    pid_lock: Optional[threading.Lock] = None,
+) -> Tuple[Optional[ContentFingerprint], int]:
+    if time.monotonic() > deadline:
+        return None, 0
+    if use_subprocess:
+        return _stable_local_fingerprint_subprocess(
+            path,
+            deadline=deadline,
+            active_worker_pids=active_worker_pids,
+            pid_lock=pid_lock,
+        )
+    return _stable_local_fingerprint_inprocess(path, deadline=deadline)
+
+
+def _mp_fingerprint_worker(path_str: str, out_q: "multiprocessing.queues.Queue") -> None:
+    try:
+        fp, read_bytes = _stable_local_fingerprint_inprocess(
+            Path(path_str),
+            deadline=time.monotonic() + OBSERVATION_TIMEOUT_S,
+        )
+    except Exception:
+        out_q.put((None, 0))
+        return
+    if fp is None:
+        out_q.put((None, read_bytes))
+        return
+    out_q.put(
+        (
+            (fp.size, fp.digest, fp.mtime_ns, fp.inode),
+            read_bytes,
+        )
+    )
+
+
+def _stable_local_fingerprint_subprocess(
+    path: Path,
+    *,
+    deadline: float,
+    active_worker_pids: Set[int],
+    pid_lock: Optional[threading.Lock],
+) -> Tuple[Optional[ContentFingerprint], int]:
+    remaining = max(0.05, deadline - time.monotonic())
+    ctx = multiprocessing.get_context("spawn")
+    out_q = ctx.Queue(maxsize=1)
+    proc = ctx.Process(
+        target=_mp_fingerprint_worker,
+        args=(str(path), out_q),
+        daemon=True,
+    )
+    payload = None
+    read_bytes = 0
+    proc.start()
+    if pid_lock is not None:
+        with pid_lock:
+            active_worker_pids.add(proc.pid)
+    try:
+        proc.join(timeout=remaining)
+        if proc.is_alive():
+            proc.kill()
+            proc.join(timeout=0.5)
+            return None, 0
+        if not out_q.empty():
+            payload, read_bytes = out_q.get_nowait()
+        else:
+            return None, 0
+    finally:
+        if pid_lock is not None:
+            with pid_lock:
+                active_worker_pids.discard(proc.pid)
+        try:
+            proc.close()
+        except Exception:
+            pass
+        try:
+            out_q.close()
+            out_q.join_thread()
+        except Exception:
+            pass
+    if payload is None:
+        return None, read_bytes
+    size, digest, mtime_ns, inode = payload
+    return (
+        ContentFingerprint(
+            size=size,
+            digest=digest,
+            mtime_ns=mtime_ns,
+            inode=inode,
+        ),
+        read_bytes,
+    )
+
+
+def _stable_local_fingerprint_inprocess(
+    path: Path,
+    *,
+    deadline: float,
 ) -> Tuple[Optional[ContentFingerprint], int]:
     if time.monotonic() > deadline:
         return None, 0
@@ -477,18 +584,35 @@ def _stable_local_fingerprint(
         return None, 0
     if st.st_size > MAX_SINGLE_FILE_BYTES:
         return None, 0
+    flags = os.O_RDONLY
+    if hasattr(os, "O_NOFOLLOW"):
+        flags |= os.O_NOFOLLOW
+    fd: Optional[int] = None
     try:
-        st_open = os.stat(path, follow_symlinks=True)
+        fd = os.open(path, flags)
+        st_open = os.fstat(fd)
     except OSError:
+        if fd is not None:
+            os.close(fd)
         return None, 0
     if not stat.S_ISREG(st_open.st_mode):
+        os.close(fd)
         return None, 0
     if st_open.st_size > MAX_SINGLE_FILE_BYTES:
+        os.close(fd)
+        return None, 0
+    if st.st_ino != st_open.st_ino or getattr(st, "st_dev", None) != getattr(st_open, "st_dev", None):
+        os.close(fd)
         return None, 0
     read_bytes = 0
     h = hashlib.sha256()
+    open_ino = st_open.st_ino
+    open_dev = getattr(st_open, "st_dev", None)
+    open_size = st_open.st_size
+    open_mtime = st_open.st_mtime_ns
     try:
-        with open(path, "rb") as fh:
+        with os.fdopen(fd, "rb") as fh:
+            fd = None
             while True:
                 if time.monotonic() > deadline:
                     return None, read_bytes
@@ -497,13 +621,23 @@ def _stable_local_fingerprint(
                     break
                 read_bytes += len(chunk)
                 h.update(chunk)
-        st_after = os.stat(path, follow_symlinks=True)
+    except OSError:
+        return None, read_bytes
+    finally:
+        if fd is not None:
+            try:
+                os.close(fd)
+            except OSError:
+                pass
+    try:
+        st_after = os.stat(path, follow_symlinks=False)
     except OSError:
         return None, read_bytes
     if (
-        st_after.st_size != st_open.st_size
-        or st_after.st_mtime_ns != st_open.st_mtime_ns
-        or st_after.st_ino != st_open.st_ino
+        st_after.st_size != open_size
+        or st_after.st_mtime_ns != open_mtime
+        or st_after.st_ino != open_ino
+        or getattr(st_after, "st_dev", None) != open_dev
     ):
         return None, read_bytes
     return (
diff --git a/agent/tool_executor.py b/agent/tool_executor.py
index 22f21a66ba846..d3f80e460fb62 100644
--- a/agent/tool_executor.py
+++ b/agent/tool_executor.py
@@ -683,7 +683,7 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe
     results = [None] * num_tools
     for i, (tc, name, args, middleware_trace, block_result, blocked_by_guardrail) in enumerate(parsed_calls):
         if block_result is not None:
-            results[i] = (name, args, block_result, 0.0, True, True, middleware_trace)
+            results[i] = (name, args, block_result, 0.0, True, True, False, middleware_trace)
 
     # Touch activity before launching workers so the gateway knows
     # we're executing tools (not stuck).
@@ -720,21 +720,33 @@ def _run_tool(index, tool_call, function_name, function_args, middleware_trace):
         # ContextVars are propagated by propagate_context_to_thread() at the
         # submit site below (GHSA-qg5c-hvr5-hjgr, #13617).
         start = time.time()
+        _registry_dispatched = False
         try:
             try:
-                _prepare_file_mutation_dispatch(
-                    agent, function_name, function_args, effective_task_id,
-                )
-                result = agent._invoke_tool(
-                    function_name,
-                    function_args,
-                    effective_task_id,
-                    tool_call.id,
-                    messages=messages,
-                    pre_tool_block_checked=True,
-                    skip_tool_request_middleware=True,
-                    tool_request_middleware_trace=list(middleware_trace),
+                from model_tools import (
+                    begin_tool_registry_dispatch_tracking,
+                    end_tool_registry_dispatch_tracking,
+                    tool_registry_was_dispatched,
                 )
+
+                _dispatch_token = begin_tool_registry_dispatch_tracking()
+                try:
+                    _prepare_file_mutation_dispatch(
+                        agent, function_name, function_args, effective_task_id,
+                    )
+                    result = agent._invoke_tool(
+                        function_name,
+                        function_args,
+                        effective_task_id,
+                        tool_call.id,
+                        messages=messages,
+                        pre_tool_block_checked=True,
+                        skip_tool_request_middleware=True,
+                        tool_request_middleware_trace=list(middleware_trace),
+                    )
+                finally:
+                    _registry_dispatched = tool_registry_was_dispatched()
+                    end_tool_registry_dispatch_tracking(_dispatch_token)
             except KeyboardInterrupt:
                 try:
                     agent.interrupt("keyboard interrupt")
@@ -751,7 +763,7 @@ def _run_tool(index, tool_call, function_name, function_args, middleware_trace):
                 )
                 duration = time.time() - start
                 logger.info("tool %s cancelled (%.2fs)", function_name, duration)
-                results[index] = (function_name, function_args, result, duration, True, False, middleware_trace)
+                results[index] = (function_name, function_args, result, duration, True, False, False, middleware_trace)
                 return
             except Exception as tool_error:
                 result = f"Error executing tool '{function_name}': {tool_error}"
@@ -762,7 +774,16 @@ def _run_tool(index, tool_call, function_name, function_args, middleware_trace):
                 logger.info("tool %s failed (%.2fs): %s", function_name, duration, result[:200])
             else:
                 logger.info("tool %s completed (%.2fs, %d chars)", function_name, duration, len(result))
-            results[index] = (function_name, function_args, result, duration, is_error, False, middleware_trace)
+            results[index] = (
+                function_name,
+                function_args,
+                result,
+                duration,
+                is_error,
+                False,
+                _registry_dispatched,
+                middleware_trace,
+            )
         finally:
             # Tear down worker-tid tracking.  Clear any interrupt bit we may
             # have set so the next task scheduled onto this recycled tid
@@ -1037,7 +1058,7 @@ def _run_tool(index, tool_call, function_name, function_args, middleware_trace):
                 effective_task_id=effective_task_id,
             )
         else:
-            function_name, function_args, function_result, tool_duration, is_error, blocked, middleware_trace = r
+            function_name, function_args, function_result, tool_duration, is_error, blocked, registry_dispatched, middleware_trace = r
             progress_function_name = function_name
             if blocked:
                 effect_disposition = "none"
@@ -1067,7 +1088,7 @@ def _run_tool(index, tool_call, function_name, function_args, middleware_trace):
                 model_result=function_result,
                 is_error=is_error,
                 blocked=blocked,
-                registry_dispatched=True,
+                registry_dispatched=registry_dispatched,
                 result_missing=False,
                 effective_task_id=effective_task_id,
             )
@@ -1377,6 +1398,8 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
             except Exception as cb_err:
                 logging.debug(f"Tool start callback error: {cb_err}")
 
+        _registry_dispatched = False
+
         # Checkpoint: snapshot working dir before file-mutating tools
         if not _execution_blocked and function_name in {"write_file", "patch"} and agent._checkpoint_mgr.enabled:
             try:
@@ -1675,23 +1698,35 @@ def _execute(next_args: dict) -> Any:
                 spinner = KawaiiSpinner(f"{face} {emoji} {preview}", spinner_type='dots', print_fn=agent._print_fn)
                 spinner.start()
             _spinner_result = None
+            _registry_dispatched = False
             try:
-                _prepare_file_mutation_dispatch(
-                    agent, function_name, function_args, effective_task_id,
-                )
-                function_result = _ra().handle_function_call(
-                    function_name, function_args, effective_task_id,
-                    tool_call_id=tool_call.id,
-                    session_id=agent.session_id or "",
-                    turn_id=getattr(agent, "_current_turn_id", "") or "",
-                    api_request_id=getattr(agent, "_current_api_request_id", "") or "",
-                    enabled_tools=list(agent.valid_tool_names) if agent.valid_tool_names else None,
-                    skip_pre_tool_call_hook=True,
-                    skip_tool_request_middleware=True,
-                    enabled_toolsets=getattr(agent, "enabled_toolsets", None),
-                    disabled_toolsets=getattr(agent, "disabled_toolsets", None),
-                    tool_request_middleware_trace=list(middleware_trace),
+                from model_tools import (
+                    begin_tool_registry_dispatch_tracking,
+                    end_tool_registry_dispatch_tracking,
+                    tool_registry_was_dispatched,
                 )
+
+                _dispatch_token = begin_tool_registry_dispatch_tracking()
+                try:
+                    _prepare_file_mutation_dispatch(
+                        agent, function_name, function_args, effective_task_id,
+                    )
+                    function_result = _ra().handle_function_call(
+                        function_name, function_args, effective_task_id,
+                        tool_call_id=tool_call.id,
+                        session_id=agent.session_id or "",
+                        turn_id=getattr(agent, "_current_turn_id", "") or "",
+                        api_request_id=getattr(agent, "_current_api_request_id", "") or "",
+                        enabled_tools=list(agent.valid_tool_names) if agent.valid_tool_names else None,
+                        skip_pre_tool_call_hook=True,
+                        skip_tool_request_middleware=True,
+                        enabled_toolsets=getattr(agent, "enabled_toolsets", None),
+                        disabled_toolsets=getattr(agent, "disabled_toolsets", None),
+                        tool_request_middleware_trace=list(middleware_trace),
+                    )
+                finally:
+                    _registry_dispatched = tool_registry_was_dispatched()
+                    end_tool_registry_dispatch_tracking(_dispatch_token)
                 _spinner_result = function_result
             except KeyboardInterrupt:
                 function_result = _emit_cancelled_terminal_post_tool_call(
@@ -1721,22 +1756,33 @@ def _execute(next_args: dict) -> Any:
                     agent._vprint(f"  {cute_msg}")
         else:
             try:
-                _prepare_file_mutation_dispatch(
-                    agent, function_name, function_args, effective_task_id,
-                )
-                function_result = _ra().handle_function_call(
-                    function_name, function_args, effective_task_id,
-                    tool_call_id=tool_call.id,
-                    session_id=agent.session_id or "",
-                    turn_id=getattr(agent, "_current_turn_id", "") or "",
-                    api_request_id=getattr(agent, "_current_api_request_id", "") or "",
-                    enabled_tools=list(agent.valid_tool_names) if agent.valid_tool_names else None,
-                    skip_pre_tool_call_hook=True,
-                    skip_tool_request_middleware=True,
-                    enabled_toolsets=getattr(agent, "enabled_toolsets", None),
-                    disabled_toolsets=getattr(agent, "disabled_toolsets", None),
-                    tool_request_middleware_trace=list(middleware_trace),
+                from model_tools import (
+                    begin_tool_registry_dispatch_tracking,
+                    end_tool_registry_dispatch_tracking,
+                    tool_registry_was_dispatched,
                 )
+
+                _dispatch_token = begin_tool_registry_dispatch_tracking()
+                try:
+                    _prepare_file_mutation_dispatch(
+                        agent, function_name, function_args, effective_task_id,
+                    )
+                    function_result = _ra().handle_function_call(
+                        function_name, function_args, effective_task_id,
+                        tool_call_id=tool_call.id,
+                        session_id=agent.session_id or "",
+                        turn_id=getattr(agent, "_current_turn_id", "") or "",
+                        api_request_id=getattr(agent, "_current_api_request_id", "") or "",
+                        enabled_tools=list(agent.valid_tool_names) if agent.valid_tool_names else None,
+                        skip_pre_tool_call_hook=True,
+                        skip_tool_request_middleware=True,
+                        enabled_toolsets=getattr(agent, "enabled_toolsets", None),
+                        disabled_toolsets=getattr(agent, "disabled_toolsets", None),
+                        tool_request_middleware_trace=list(middleware_trace),
+                    )
+                finally:
+                    _registry_dispatched = tool_registry_was_dispatched()
+                    end_tool_registry_dispatch_tracking(_dispatch_token)
             except KeyboardInterrupt:
                 _emit_cancelled_terminal_post_tool_call(
                     agent,
@@ -1822,7 +1868,7 @@ def _execute(next_args: dict) -> Any:
                     model_result=function_result,
                     is_error=_is_error_result,
                     blocked=_execution_blocked,
-                    registry_dispatched=True,
+                    registry_dispatched=_registry_dispatched,
                     result_missing=False,
                     effective_task_id=effective_task_id,
                 )
diff --git a/model_tools.py b/model_tools.py
index 7b5811e8fbfe4..e8d774923e36f 100644
--- a/model_tools.py
+++ b/model_tools.py
@@ -20,6 +20,7 @@
     check_tool_availability(quiet) -> tuple
 """
 
+import contextvars
 import os
 import json
 import re
@@ -29,6 +30,28 @@
 import time
 from typing import Dict, Any, List, Optional, Tuple
 
+_tool_registry_dispatched: contextvars.ContextVar[bool] = contextvars.ContextVar(
+    "tool_registry_dispatched",
+    default=False,
+)
+
+
+def begin_tool_registry_dispatch_tracking() -> contextvars.Token:
+    """Reset per-tool dispatch tracking (thread/task-local)."""
+    return _tool_registry_dispatched.set(False)
+
+
+def end_tool_registry_dispatch_tracking(token: contextvars.Token) -> None:
+    _tool_registry_dispatched.reset(token)
+
+
+def mark_tool_registry_dispatched() -> None:
+    _tool_registry_dispatched.set(True)
+
+
+def tool_registry_was_dispatched() -> bool:
+    return bool(_tool_registry_dispatched.get())
+
 from tools.registry import discover_builtin_tools, registry
 from toolsets import resolve_toolset, validate_toolset
 
@@ -1327,6 +1350,7 @@ def handle_function_call(
                 # the parent's tool set via the process-global.
                 sandbox_enabled = enabled_tools if enabled_tools is not None else _last_resolved_tool_names
                 def _dispatch(next_args: Dict[str, Any]) -> Any:
+                    mark_tool_registry_dispatched()
                     return registry.dispatch(
                         function_name, next_args,
                         task_id=task_id,
@@ -1335,6 +1359,7 @@ def _dispatch(next_args: Dict[str, Any]) -> Any:
                     )
             else:
                 def _dispatch(next_args: Dict[str, Any]) -> Any:
+                    mark_tool_registry_dispatched()
                     return registry.dispatch(
                         function_name, next_args,
                         task_id=task_id,
diff --git a/tests/agent/test_file_mutation_verifier_redesign.py b/tests/agent/test_file_mutation_verifier_redesign.py
index 803d23c37bd27..44b4ae723c7a0 100644
--- a/tests/agent/test_file_mutation_verifier_redesign.py
+++ b/tests/agent/test_file_mutation_verifier_redesign.py
@@ -12,6 +12,7 @@
     TurnFileMutationVerifier,
     _path_allowed_for_observation,
     format_failure_footer,
+    sync_legacy_failed_state,
 )
 from run_agent import AIAgent, _extract_file_mutation_targets
 
@@ -234,7 +235,7 @@ def test_unchanged_file_after_prepare_keeps_failure(self, tmp_path, monkeypatch)
         monkeypatch.chdir(tmp_path)
         target = tmp_path / "partial.py"
         target.write_text("baseline\n", encoding="utf-8")
-        verifier = TurnFileMutationVerifier()
+        verifier = TurnFileMutationVerifier(use_subprocess_fingerprint=False)
         verifier.reset_turn(1)
         verifier.prepare_mutation_dispatch(
             tool_name="write_file",
@@ -257,7 +258,7 @@ def test_post_dispatch_disk_change_suppresses_using_pre_baseline(self, tmp_path,
         monkeypatch.chdir(tmp_path)
         target = tmp_path / "partial.py"
         target.write_text("baseline\n", encoding="utf-8")
-        verifier = TurnFileMutationVerifier()
+        verifier = TurnFileMutationVerifier(use_subprocess_fingerprint=False)
         verifier.reset_turn(1)
         verifier.prepare_mutation_dispatch(
             tool_name="write_file",
@@ -312,3 +313,149 @@ def test_later_failure_after_transition(self, tmp_path, monkeypatch):
         assert any(k.endswith("flip.py") for k in agent._turn_failed_file_mutations)
         key = next(k for k in agent._turn_failed_file_mutations if k.endswith("flip.py"))
         assert "second failure" in agent._turn_failed_file_mutations[key]["error_preview"]
+
+
+def _slow_fingerprint_worker(path_str, out_q):
+    import time
+
+    time.sleep(5)
+    out_q.put((None, 0))
+
+
+class TestFailsafeObservation:
+    def test_subprocess_timeout_leaves_no_live_workers(self, tmp_path, monkeypatch):
+        import agent.file_mutation_verifier as fmv
+
+        target = tmp_path / "slow.bin"
+        target.write_bytes(b"abc")
+
+        monkeypatch.setattr(fmv, "_mp_fingerprint_worker", _slow_fingerprint_worker)
+        monkeypatch.setattr(fmv, "OBSERVATION_TIMEOUT_S", 0.15)
+
+        verifier = TurnFileMutationVerifier(use_subprocess_fingerprint=True)
+        verifier.reset_turn(1)
+        monkeypatch.chdir(tmp_path)
+        fp = verifier._capture_baseline("slow.bin", "default", turn_generation=1)
+        assert fp is None
+        assert verifier.active_worker_pids == set()
+
+    def test_symlink_is_not_observable(self, tmp_path):
+        import os
+        import time
+
+        from agent.file_mutation_verifier import _stable_local_fingerprint_inprocess
+
+        if os.name == "nt":
+            pytest.skip("symlink lstat guard is POSIX-focused in this test")
+        real = tmp_path / "real.txt"
+        real.write_text("data\n", encoding="utf-8")
+        link = tmp_path / "link.txt"
+        link.symlink_to(real)
+        fp, _ = _stable_local_fingerprint_inprocess(
+            link, deadline=time.monotonic() + 5.0,
+        )
+        assert fp is None
+
+    def test_torn_read_rejected(self, tmp_path, monkeypatch):
+        import time
+
+        from agent.file_mutation_verifier import _stable_local_fingerprint_inprocess
+
+        target = tmp_path / "torn.txt"
+        target.write_text("stable\n", encoding="utf-8")
+        real_stat = target.stat()
+
+        def _unstable_stat(path, *, follow_symlinks=True):
+            if not follow_symlinks:
+                raise OSError("simulated torn read")
+            return real_stat
+
+        monkeypatch.setattr(__import__("os"), "stat", _unstable_stat)
+        fp, _ = _stable_local_fingerprint_inprocess(
+            target, deadline=time.monotonic() + 5.0,
+        )
+        assert fp is None
+
+    def test_posix_dialect_preserves_backslash_identity(self):
+        verifier = TurnFileMutationVerifier(
+            resolve_backend=lambda _tid: ("local", "host", "posix"),
+        )
+        a = verifier._identity_for_path("dir\\file.py", "default")
+        b = verifier._identity_for_path("dir/file.py", "default")
+        assert a is not None and b is not None
+        assert a.path != b.path
+
+    def test_ssh_backend_never_local_clears(self, tmp_path, monkeypatch):
+        monkeypatch.chdir(tmp_path)
+        target = tmp_path / "remote.py"
+        target.write_text("a\n", encoding="utf-8")
+        verifier = TurnFileMutationVerifier(
+            resolve_backend=lambda _tid: ("ssh", "SSHSession", "posix"),
+            use_subprocess_fingerprint=False,
+        )
+        verifier.reset_turn(1)
+        verifier.record_tool_outcome(
+            tool_name="write_file",
+            effective_args={"path": "remote.py", "content": "b\n"},
+            effective_task_id="default",
+            raw_result=json.dumps({"error": "fail"}),
+            dispatch=DispatchTriState.DISPATCHED,
+            model_is_error=True,
+            turn_generation=1,
+        )
+        target.write_text("b\n", encoding="utf-8")
+        verifier.reconcile_content_transitions(task_id="default")
+        assert "remote.py" in verifier.finalize_failed_dict()
+
+
+class TestRegistryDispatchAuthority:
+    def test_handle_function_call_middleware_short_circuit_not_dispatched(
+        self, monkeypatch, tmp_path,
+    ):
+        monkeypatch.chdir(tmp_path)
+
+        def _short_circuit(name, args, execute, **kwargs):
+            return json.dumps({"success": True, "bytes_written": 12})
+
+        monkeypatch.setattr(
+            "hermes_cli.middleware.run_tool_execution_middleware",
+            _short_circuit,
+        )
+        from model_tools import (
+            begin_tool_registry_dispatch_tracking,
+            end_tool_registry_dispatch_tracking,
+            handle_function_call,
+            tool_registry_was_dispatched,
+        )
+
+        token = begin_tool_registry_dispatch_tracking()
+        try:
+            handle_function_call(
+                "write_file",
+                {"path": "never.txt", "content": "x"},
+                task_id="default",
+                skip_pre_tool_call_hook=True,
+            )
+            dispatched = tool_registry_was_dispatched()
+        finally:
+            end_tool_registry_dispatch_tracking(token)
+        assert not dispatched
+
+        agent = _bare_agent()
+        agent._record_file_mutation_result(
+            "patch",
+            {"mode": "replace", "path": "a.txt", "old_string": "x", "new_string": "y"},
+            json.dumps({"error": "first"}),
+            is_error=True,
+            raw_result=json.dumps({"error": "first"}),
+            dispatch=DispatchTriState.DISPATCHED.value,
+        )
+        agent._record_file_mutation_result(
+            "write_file",
+            {"path": "never.txt", "content": "x"},
+            json.dumps({"success": True, "bytes_written": 12}),
+            is_error=False,
+            dispatch=DispatchTriState.NOT_DISPATCHED.value,
+        )
+        sync_legacy_failed_state(agent)
+        assert any(k.endswith("a.txt") for k in agent._turn_failed_file_mutations)
diff --git a/tests/agent/test_turn_finalizer_file_mutation_footer.py b/tests/agent/test_turn_finalizer_file_mutation_footer.py
new file mode 100644
index 0000000000000..9c6bdc7b1ab0f
--- /dev/null
+++ b/tests/agent/test_turn_finalizer_file_mutation_footer.py
@@ -0,0 +1,162 @@
+"""Turn-end file-mutation footer delivery on interrupted/empty turns."""
+
+import json
+
+import pytest
+
+from agent.file_mutation_verifier import TurnFileMutationVerifier, sync_legacy_failed_state
+from agent.turn_finalizer import finalize_turn
+
+
+class _FooterAgent:
+    def __init__(self):
+        self.max_iterations = 90
+        self.iteration_budget = type(
+            "B", (), {"used": 1, "max_total": 90, "remaining": 1},
+        )()
+        self.context_compressor = type("C", (), {"last_prompt_tokens": 0})()
+        self.model = "stub"
+        self.provider = "stub"
+        self.base_url = "http://stub"
+        self.session_id = "s1"
+        self.quiet_mode = True
+        self.platform = "cli"
+        self._interrupt_requested = False
+        self._interrupt_message = None
+        self._tool_guardrail_halt_decision = None
+        self._response_was_previewed = False
+        self._skill_nudge_interval = 0
+        self._iters_since_skill = 0
+        self.request_overrides = {}
+        self._turn_failed_file_mutations = {}
+        self._turn_file_mutation_paths = set()
+        self._file_mutation_verifier = TurnFileMutationVerifier(use_subprocess_fingerprint=False)
+        self._file_mutation_verifier.reset_turn(1)
+        for attr in (
+            "session_input_tokens",
+            "session_output_tokens",
+            "session_cache_read_tokens",
+            "session_cache_write_tokens",
+            "session_reasoning_tokens",
+            "session_prompt_tokens",
+            "session_completion_tokens",
+            "session_total_tokens",
+            "session_estimated_cost_usd",
+        ):
+            setattr(self, attr, 0)
+        self.session_cost_status = "ok"
+        self.session_cost_source = "stub"
+
+    def _file_mutation_verifier_enabled(self):
+        return True
+
+    def _format_file_mutation_failure_footer(self, failed):
+        from run_agent import AIAgent
+
+        return AIAgent._format_file_mutation_failure_footer(failed)
+
+    def _turn_completion_explainer_enabled(self):
+        return False
+
+    def _save_trajectory(self, *a, **k):
+        pass
+
+    def _cleanup_task_resources(self, *a, **k):
+        pass
+
+    def _drop_trailing_empty_response_scaffolding(self, *a, **k):
+        pass
+
+    def _persist_session(self, *a, **k):
+        pass
+
+    def _emit_status(self, *a, **k):
+        pass
+
+    def _safe_print(self, *a, **k):
+        pass
+
+    def _handle_max_iterations(self, messages, n):
+        return ""
+
+    def _drain_pending_steer(self):
+        return None
+
+    def clear_interrupt(self):
+        pass
+
+    def _sync_external_memory_for_turn(self, **k):
+        pass
+
+
+def _finalize(agent, *, final_response="", interrupted=False):
+    return finalize_turn(
+        agent,
+        final_response=final_response,
+        api_call_count=1,
+        interrupted=interrupted,
+        failed=False,
+        messages=[{"role": "user", "content": "hi"}],
+        conversation_history=None,
+        effective_task_id="default",
+        turn_id="t1",
+        user_message="hi",
+        original_user_message="hi",
+        _should_review_memory=False,
+        _turn_exit_reason="unknown",
+    )
+
+
+def test_interrupted_empty_turn_still_gets_footer(tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+    (tmp_path / "miss.py").write_text("same\n", encoding="utf-8")
+    agent = _FooterAgent()
+    fail = json.dumps({"error": "denied"})
+    agent._file_mutation_verifier.record_tool_outcome(
+        tool_name="patch",
+        effective_args={
+            "mode": "replace",
+            "path": "miss.py",
+            "old_string": "x",
+            "new_string": "y",
+        },
+        effective_task_id="default",
+        raw_result=fail,
+        dispatch=__import__(
+            "agent.file_mutation_verifier", fromlist=["DispatchTriState"]
+        ).DispatchTriState.DISPATCHED,
+        model_is_error=True,
+        turn_generation=1,
+    )
+    sync_legacy_failed_state(agent)
+    result = _finalize(agent, final_response="", interrupted=True)
+    assert "File-mutation verifier" in (result["final_response"] or "")
+
+
+def test_formatter_exception_uses_generic_fallback(tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+    (tmp_path / "bad.py").write_text("x\n", encoding="utf-8")
+    agent = _FooterAgent()
+    fail = json.dumps({"error": "nope"})
+    agent._file_mutation_verifier.record_tool_outcome(
+        tool_name="write_file",
+        effective_args={"path": "bad.py", "content": "y\n"},
+        effective_task_id="default",
+        raw_result=fail,
+        dispatch=__import__(
+            "agent.file_mutation_verifier", fromlist=["DispatchTriState"]
+        ).DispatchTriState.DISPATCHED,
+        model_is_error=True,
+        turn_generation=1,
+    )
+    sync_legacy_failed_state(agent)
+
+    def _boom(_failed):
+        raise RuntimeError("format broke")
+
+    agent._format_file_mutation_failure_footer = _boom
+    result = _finalize(agent, final_response="partial answer")
+    text = result["final_response"] or ""
+    assert "partial answer" in text
+    assert "File-mutation verifier" in text
+    assert "git status" in text.lower() or "read_file" in text

From 64b7e60fd64db3f96748f58f1fdf238be6947b57 Mon Sep 17 00:00:00 2001
From: Rickard Robin <rickard@kumobits.com>
Date: Mon, 27 Jul 2026 02:13:52 +0000
Subject: [PATCH 3/3] fix(agent): fail closed when dispatched mutation has no
 parseable path

Record an unknown-target ledger entry so invalid or unparseable patch/write
failures still surface in the turn-end footer instead of being dropped.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 agent/file_mutation_verifier.py               | 24 ++++++++++++++++++-
 .../test_file_mutation_verifier_redesign.py   | 20 ++++++++++++++++
 .../run_agent/test_file_mutation_verifier.py  | 13 ++++++----
 3 files changed, 52 insertions(+), 5 deletions(-)

diff --git a/agent/file_mutation_verifier.py b/agent/file_mutation_verifier.py
index 6f95fd20f976b..d8228fc95b9e0 100644
--- a/agent/file_mutation_verifier.py
+++ b/agent/file_mutation_verifier.py
@@ -45,6 +45,9 @@
     "this turn. Run `git status` or `read_file` to confirm."
 )
 
+# Dispatched failures with no parseable path still fail closed in the footer.
+_UNKNOWN_MUTATION_TARGET = "(unknown file-mutation target)"
+
 _FOOTER_INJECTION_RE = re.compile(
     r"(?im)^\s*(MEDIA:|FILE:|ATTACHMENT:|\!\[).*$"
 )
@@ -184,6 +187,25 @@ def record_tool_outcome(
             return
         targets = _extract_file_mutation_targets(tool_name, effective_args)
         if not targets:
+            if dispatch is DispatchTriState.DISPATCHED_NO_RESULT:
+                self._upsert_unresolved(
+                    tool_name=tool_name,
+                    raw_path=_UNKNOWN_MUTATION_TARGET,
+                    task_id=effective_task_id,
+                    error_preview="tool dispatch finished without a result",
+                    baseline=None,
+                )
+                return
+            if dispatch is DispatchTriState.DISPATCHED:
+                raw_landed = file_mutation_result_landed(tool_name, raw_result)
+                if model_is_error or not raw_landed:
+                    self._upsert_unresolved(
+                        tool_name=tool_name,
+                        raw_path=_UNKNOWN_MUTATION_TARGET,
+                        task_id=effective_task_id,
+                        error_preview=_extract_error_preview(raw_result),
+                        baseline=None,
+                    )
             return
 
         if dispatch is DispatchTriState.DISPATCHED_NO_RESULT:
@@ -431,7 +453,7 @@ def _path_allowed_for_observation(path: str) -> bool:
     if not path or not isinstance(path, str):
         return False
     p = path.strip()
-    if not p:
+    if not p or p == _UNKNOWN_MUTATION_TARGET:
         return False
     lower = p.lower()
     if lower.startswith("\\\\.\\") or lower.startswith("\\\\?\\"):
diff --git a/tests/agent/test_file_mutation_verifier_redesign.py b/tests/agent/test_file_mutation_verifier_redesign.py
index 44b4ae723c7a0..7c191bca555bb 100644
--- a/tests/agent/test_file_mutation_verifier_redesign.py
+++ b/tests/agent/test_file_mutation_verifier_redesign.py
@@ -130,6 +130,26 @@ def test_invalid_patch_yields_no_targets(self):
         body = "*** Update File: outside\n*** Begin Patch\n*** End Patch\n"
         assert _extract_file_mutation_targets("patch", {"mode": "patch", "patch": body}) == []
 
+    def test_dispatched_failure_without_targets_fails_closed(self):
+        from agent.file_mutation_verifier import _UNKNOWN_MUTATION_TARGET
+
+        verifier = TurnFileMutationVerifier(use_subprocess_fingerprint=False)
+        verifier.reset_turn(1)
+        body = "*** Update File: outside\n*** Begin Patch\n*** End Patch\n"
+        fail = json.dumps({"error": "Could not apply patch"})
+        verifier.record_tool_outcome(
+            tool_name="patch",
+            effective_args={"mode": "patch", "patch": body},
+            effective_task_id="default",
+            raw_result=fail,
+            dispatch=DispatchTriState.DISPATCHED,
+            model_is_error=True,
+            turn_generation=1,
+        )
+        failed = verifier.finalize_failed_dict()
+        assert _UNKNOWN_MUTATION_TARGET in failed
+        assert "Could not apply patch" in failed[_UNKNOWN_MUTATION_TARGET]["error_preview"]
+
 
 class TestPathSafety:
     @pytest.mark.skipif(__import__("sys").platform != "win32", reason="DOS device paths")
diff --git a/tests/run_agent/test_file_mutation_verifier.py b/tests/run_agent/test_file_mutation_verifier.py
index 709068c8fab8f..1d92510131bd3 100644
--- a/tests/run_agent/test_file_mutation_verifier.py
+++ b/tests/run_agent/test_file_mutation_verifier.py
@@ -282,15 +282,20 @@ def test_no_state_dict_silent_noop(self):
             json.dumps({"error": "x"}), is_error=True,
         )
 
-    def test_missing_path_arg_recorded_nowhere(self):
+    def test_missing_path_arg_fails_closed(self):
         agent = _bare_agent()
         agent._record_file_mutation_result(
             "patch", {"mode": "replace"},  # no path
             json.dumps({"error": "path required"}), is_error=True,
         )
-        # No path → nothing to key on, state stays empty.  The per-turn
-        # state is about file paths, not individual tool-call IDs.
-        assert agent._turn_failed_file_mutations == {}
+        # No parseable path → fail closed: record an unknown-target entry
+        # so the failure still surfaces in the turn-end footer instead of
+        # being dropped.
+        failed = agent._turn_failed_file_mutations
+        assert list(failed) == ["(unknown file-mutation target)"]
+        entry = failed["(unknown file-mutation target)"]
+        assert entry["tool"] == "patch"
+        assert "path required" in entry["error_preview"]
 
 
 # ---------------------------------------------------------------------------