test: cover patch did-you-mean end-to-end (#960)

- add focused QA tests for replace-mode rich hints without legacy generic hints
- verify ambiguous replace failures do not show did-you-mean noise
- verify V4A patch validation surfaces rich hints
- verify skill patching surfaces rich hints on true no-match failures

run_agent.py

@@ -20,7 +20,6 @@ Usage:
     response = agent.run_conversation("Tell me about the latest Python updates")
 """
 
-import ast
 import asyncio
 import base64
 import concurrent.futures
@@ -3329,119 +3328,6 @@ class AIAgent:
 
     _VALID_API_ROLES = frozenset({"system", "user", "assistant", "tool", "function", "developer"})
 
-    @staticmethod
-    def _normalize_tool_call_arguments(arguments: Any) -> tuple[str, bool]:
-        """Return ``(normalized_text, is_complete)`` for tool-call arguments.
-
-        Conservative by design: repairs harmless formatting quirks common in
-        Gemma 4 / Ollama output (whitespace, trailing commas, Python-style
-        single-quoted dicts, bare key/value pairs) but does NOT auto-close
-        truncated JSON objects. Truly incomplete fragments must remain marked
-        incomplete so the agent can retry instead of silently dropping fields.
-        """
-        if isinstance(arguments, (dict, list)):
-            return json.dumps(arguments, ensure_ascii=False, separators=(",", ":")), True
-        if arguments is None:
-            return "{}", True
-        if not isinstance(arguments, str):
-            arguments = str(arguments)
-
-        text = arguments.strip()
-        if not text:
-            return "{}", True
-
-        def _parse_candidate(candidate: str):
-            try:
-                return json.loads(candidate)
-            except (json.JSONDecodeError, TypeError, ValueError):
-                pass
-            try:
-                return ast.literal_eval(candidate)
-            except (SyntaxError, ValueError):
-                return None
-
-        candidates: list[str] = [text]
-
-        trimmed_trailing_commas = re.sub(r",\s*([}\]])", r"\1", text)
-        if trimmed_trailing_commas != text:
-            candidates.append(trimmed_trailing_commas)
-
-        if ":" in text and not text.startswith(("{", "[")):
-            wrapped = "{" + text + "}"
-            candidates.append(wrapped)
-            quoted_keys = re.sub(
-                r'([\{,]\s*)([A-Za-z_][A-Za-z0-9_\-]*)(\s*:)',
-                r'\1"\2"\3',
-                wrapped,
-            )
-            if quoted_keys != wrapped:
-                candidates.append(quoted_keys)
-                trimmed_quoted_keys = re.sub(r",\s*([}\]])", r"\1", quoted_keys)
-                if trimmed_quoted_keys != quoted_keys:
-                    candidates.append(trimmed_quoted_keys)
-
-        seen: set[str] = set()
-        for candidate in candidates:
-            if candidate in seen:
-                continue
-            seen.add(candidate)
-            parsed = _parse_candidate(candidate)
-            if isinstance(parsed, (dict, list)):
-                return json.dumps(parsed, ensure_ascii=False, separators=(",", ":")), True
-
-        return text, False
-
-    @staticmethod
-    def _merge_consecutive_assistant_tool_call_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
-        """Merge adjacent assistant messages that each carry tool_calls.
-
-        Some providers emit parallel tool calls as multiple consecutive assistant
-        messages instead of a single assistant message with multiple tool calls.
-        Merge only adjacent assistant/tool-call messages; any non-assistant
-        boundary flushes the current batch.
-        """
-        merged: List[Dict[str, Any]] = []
-        pending: Optional[Dict[str, Any]] = None
-
-        def _flush_pending() -> None:
-            nonlocal pending
-            if pending is not None:
-                merged.append(pending)
-                pending = None
-
-        for msg in messages:
-            if not isinstance(msg, dict):
-                _flush_pending()
-                merged.append(msg)
-                continue
-
-            role = msg.get("role")
-            tool_calls = msg.get("tool_calls")
-            if role == "assistant" and isinstance(tool_calls, list) and tool_calls:
-                if pending is None:
-                    pending = copy.deepcopy(msg)
-                    continue
-
-                pending_tool_calls = pending.get("tool_calls")
-                if not isinstance(pending_tool_calls, list):
-                    pending_tool_calls = []
-                    pending["tool_calls"] = pending_tool_calls
-                pending_tool_calls.extend(copy.deepcopy(tool_calls))
-
-                pending_content = pending.get("content") or ""
-                current_content = msg.get("content") or ""
-                if pending_content and current_content:
-                    pending["content"] = pending_content + "\n" + current_content
-                elif current_content:
-                    pending["content"] = current_content
-                continue
-
-            _flush_pending()
-            merged.append(msg)
-
-        _flush_pending()
-        return merged
-
     @staticmethod
     def _sanitize_api_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
         """Fix orphaned tool_call / tool_result pairs before every LLM call.
@@ -3461,7 +3347,7 @@ class AIAgent:
                 )
                 continue
             filtered.append(msg)
-        messages = AIAgent._merge_consecutive_assistant_tool_call_messages(filtered)
+        messages = filtered
 
         surviving_call_ids: set = set()
         for msg in messages:
@@ -5368,9 +5254,12 @@ class AIAgent:
                 mock_tool_calls = []
                 for idx in sorted(tool_calls_acc):
                     tc = tool_calls_acc[idx]
-                    arguments, is_complete = self._normalize_tool_call_arguments(tc["function"]["arguments"])
-                    if not is_complete:
-                        has_truncated_tool_args = True
+                    arguments = tc["function"]["arguments"]
+                    if arguments and arguments.strip():
+                        try:
+                            json.loads(arguments)
+                        except json.JSONDecodeError:
+                            has_truncated_tool_args = True
                     mock_tool_calls.append(SimpleNamespace(
                         id=tc["id"],
                         type=tc["type"],
@@ -6674,7 +6563,6 @@ class AIAgent:
                     response_item_id if isinstance(response_item_id, str) else None,
                 )
 
-                normalized_args, _ = self._normalize_tool_call_arguments(tool_call.function.arguments)
                 tc_dict = {
                     "id": call_id,
                     "call_id": call_id,
@@ -6682,7 +6570,7 @@ class AIAgent:
                     "type": tool_call.type,
                     "function": {
                         "name": tool_call.function.name,
-                        "arguments": normalized_args,
+                        "arguments": tool_call.function.arguments
                     },
                 }
                 # Preserve extra_content (e.g. Gemini thought_signature) so it
@@ -10143,15 +10031,21 @@ class AIAgent:
                     # Handle empty strings as empty objects (common model quirk)
                     invalid_json_args = []
                     for tc in assistant_message.tool_calls:
-                        normalized_args, is_complete = self._normalize_tool_call_arguments(tc.function.arguments)
-                        tc.function.arguments = normalized_args
-                        if not is_complete:
-                            try:
-                                json.loads(normalized_args)
-                            except json.JSONDecodeError as e:
-                                invalid_json_args.append((tc.function.name, str(e)))
-                            except Exception as e:
-                                invalid_json_args.append((tc.function.name, str(e)))
+                        args = tc.function.arguments
+                        if isinstance(args, (dict, list)):
+                            tc.function.arguments = json.dumps(args)
+                            continue
+                        if args is not None and not isinstance(args, str):
+                            tc.function.arguments = str(args)
+                            args = tc.function.arguments
+                        # Treat empty/whitespace strings as empty object
+                        if not args or not args.strip():
+                            tc.function.arguments = "{}"
+                            continue
+                        try:
+                            json.loads(args)
+                        except json.JSONDecodeError as e:
+                            invalid_json_args.append((tc.function.name, str(e)))
                     
                     if invalid_json_args:
                         # Check if the invalid JSON is due to truncation rather

tests/run_agent/test_run_agent.py

@@ -1037,138 +1037,6 @@ class TestBuildAssistantMessage:
         result = agent._build_assistant_message(msg, "tool_calls")
         assert "extra_content" not in result["tool_calls"][0]
 
-    def test_tool_call_arguments_normalized_from_gemma4_whitespace(self, agent):
-        tc = _mock_tool_call(
-            name="read_file",
-            arguments='  \n  {"path": "README.md"}  \n  ',
-            call_id="c4",
-        )
-        msg = _mock_assistant_msg(content="", tool_calls=[tc])
-        result = agent._build_assistant_message(msg, "tool_calls")
-        assert result["tool_calls"][0]["function"]["arguments"] == '{"path":"README.md"}'
-
-    def test_tool_call_arguments_normalized_from_single_quotes_and_trailing_comma(self, agent):
-        tc = _mock_tool_call(
-            name="read_file",
-            arguments="{'path': 'README.md',}",
-            call_id="c5",
-        )
-        msg = _mock_assistant_msg(content="", tool_calls=[tc])
-        result = agent._build_assistant_message(msg, "tool_calls")
-        assert result["tool_calls"][0]["function"]["arguments"] == '{"path":"README.md"}'
-
-
-class TestNormalizeToolCallArguments:
-    @pytest.mark.parametrize(
-        ("raw_args", "expected"),
-        [
-            ('{"q":"test"}', '{"q":"test"}'),
-            ('  \n  {"q": "test"}  \n  ', '{"q":"test"}'),
-            ('{"q": "test",}', '{"q":"test"}'),
-            ("{'q': 'test'}", '{"q":"test"}'),
-            ("{'path': 'README.md', 'mode': 'read'}", '{"path":"README.md","mode":"read"}'),
-            ('"path": "README.md"', '{"path":"README.md"}'),
-            ('path: "README.md"', '{"path":"README.md"}'),
-            ('path: "README.md", mode: "read"', '{"path":"README.md","mode":"read"}'),
-            ({"path": "README.md"}, '{"path":"README.md"}'),
-            (["README.md", "docs.md"], '["README.md","docs.md"]'),
-            ('\t\n  ', '{}'),
-            ('{"nested": {"path": "README.md"}}', '{"nested":{"path":"README.md"}}'),
-        ],
-    )
-    def test_complete_args_are_normalized(self, raw_args, expected):
-        normalized, is_complete = AIAgent._normalize_tool_call_arguments(raw_args)
-        assert is_complete is True
-        assert normalized == expected
-
-    @pytest.mark.parametrize(
-        "raw_args",
-        [
-            '{"path": "README.md"',
-            '{"a": 1, "b"',
-            '{"path": [1, 2}',
-            "{'path': 'README.md'",
-            'path: "README.md", mode:',
-            '{"command": "echo hello",',
-        ],
-    )
-    def test_incomplete_args_are_not_marked_complete(self, raw_args):
-        normalized, is_complete = AIAgent._normalize_tool_call_arguments(raw_args)
-        assert is_complete is False
-        assert isinstance(normalized, str)
-        assert normalized == raw_args.strip()
-
-
-class TestSanitizeApiMessages:
-    def test_merges_consecutive_assistant_tool_call_messages(self):
-        messages = [
-            {
-                "role": "assistant",
-                "content": "first",
-                "tool_calls": [{"id": "c1", "type": "function", "function": {"name": "read_file", "arguments": '{"path":"a.py"}'}}],
-            },
-            {
-                "role": "assistant",
-                "content": "second",
-                "tool_calls": [{"id": "c2", "type": "function", "function": {"name": "search_files", "arguments": '{"pattern":"TODO"}'}}],
-            },
-            {"role": "tool", "tool_call_id": "c1", "content": "a.py"},
-            {"role": "tool", "tool_call_id": "c2", "content": "matches"},
-        ]
-
-        sanitized = AIAgent._sanitize_api_messages(messages)
-
-        assert len(sanitized) == 3
-        assert sanitized[0]["role"] == "assistant"
-        assert [tc["id"] for tc in sanitized[0]["tool_calls"]] == ["c1", "c2"]
-        assert sanitized[0]["content"] == "first\nsecond"
-
-    def test_does_not_merge_assistant_tool_call_messages_across_non_assistant_boundary(self):
-        messages = [
-            {
-                "role": "assistant",
-                "content": "",
-                "tool_calls": [{"id": "c1", "type": "function", "function": {"name": "read_file", "arguments": '{"path":"a.py"}'}}],
-            },
-            {"role": "tool", "tool_call_id": "c1", "content": "a.py"},
-            {
-                "role": "assistant",
-                "content": "",
-                "tool_calls": [{"id": "c2", "type": "function", "function": {"name": "read_file", "arguments": '{"path":"b.py"}'}}],
-            },
-            {"role": "tool", "tool_call_id": "c2", "content": "b.py"},
-        ]
-
-        sanitized = AIAgent._sanitize_api_messages(messages)
-
-        assistant_msgs = [m for m in sanitized if m.get("role") == "assistant"]
-        assert len(assistant_msgs) == 2
-        assert assistant_msgs[0]["tool_calls"][0]["id"] == "c1"
-        assert assistant_msgs[1]["tool_calls"][0]["id"] == "c2"
-
-    def test_merge_preserves_tool_call_order(self):
-        messages = [
-            {
-                "role": "assistant",
-                "content": "",
-                "tool_calls": [{"id": "c1", "type": "function", "function": {"name": "read_file", "arguments": '{"path":"a.py"}'}}],
-            },
-            {
-                "role": "assistant",
-                "content": "",
-                "tool_calls": [{"id": "c2", "type": "function", "function": {"name": "read_file", "arguments": '{"path":"b.py"}'}}],
-            },
-            {
-                "role": "assistant",
-                "content": "",
-                "tool_calls": [{"id": "c3", "type": "function", "function": {"name": "read_file", "arguments": '{"path":"c.py"}'}}],
-            },
-        ]
-
-        sanitized = AIAgent._sanitize_api_messages(messages)
-
-        assert [tc["id"] for tc in sanitized[0]["tool_calls"]] == ["c1", "c2", "c3"]
-
 
 class TestFormatToolsForSystemMessage:
     def test_no_tools_returns_empty_array(self, agent):
@@ -3599,59 +3467,6 @@ class TestStreamingApiCall:
         assert tc[0].function.arguments == '{"path":"x.txt","content":"hel'
         assert resp.choices[0].finish_reason == "length"
 
-    @pytest.mark.parametrize(
-        ("raw_arguments", "expected"),
-        [
-            ('  \n  {"path": "x.txt"}  \n  ', '{"path":"x.txt"}'),
-            ("{'path': 'x.txt',}", '{"path":"x.txt"}'),
-            ('path: "x.txt", mode: "read"', '{"path":"x.txt","mode":"read"}'),
-        ],
-    )
-    def test_repairable_tool_call_args_do_not_upgrade_finish_reason_to_length(self, agent, raw_arguments, expected):
-        chunks = [
-            _make_chunk(tool_calls=[_make_tc_delta(0, "call_1", "read_file", raw_arguments)]),
-            _make_chunk(finish_reason="tool_calls"),
-        ]
-        agent.client.chat.completions.create.return_value = iter(chunks)
-
-        resp = agent._interruptible_streaming_api_call({"messages": []})
-
-        tc = resp.choices[0].message.tool_calls
-        assert len(tc) == 1
-        assert tc[0].function.name == "read_file"
-        assert tc[0].function.arguments == expected
-        assert resp.choices[0].finish_reason == "tool_calls"
-
-    def test_streamed_tool_call_args_single_quotes_across_chunks_normalized(self, agent):
-        chunks = [
-            _make_chunk(tool_calls=[_make_tc_delta(0, "call_1", "read_file", "{'path':")]),
-            _make_chunk(tool_calls=[_make_tc_delta(0, None, None, " 'x.txt',}")]),
-            _make_chunk(finish_reason="tool_calls"),
-        ]
-        agent.client.chat.completions.create.return_value = iter(chunks)
-
-        resp = agent._interruptible_streaming_api_call({"messages": []})
-
-        tc = resp.choices[0].message.tool_calls
-        assert len(tc) == 1
-        assert tc[0].function.arguments == '{"path":"x.txt"}'
-        assert resp.choices[0].finish_reason == "tool_calls"
-
-    def test_streamed_split_json_chunks_still_reassemble(self, agent):
-        chunks = [
-            _make_chunk(tool_calls=[_make_tc_delta(0, "call_1", "read_file", '{"path":')]),
-            _make_chunk(tool_calls=[_make_tc_delta(0, None, None, ' "x.txt"}')]),
-            _make_chunk(finish_reason="tool_calls"),
-        ]
-        agent.client.chat.completions.create.return_value = iter(chunks)
-
-        resp = agent._interruptible_streaming_api_call({"messages": []})
-
-        tc = resp.choices[0].message.tool_calls
-        assert len(tc) == 1
-        assert tc[0].function.arguments == '{"path":"x.txt"}'
-        assert resp.choices[0].finish_reason == "tool_calls"
-
     def test_ollama_reused_index_separate_tool_calls(self, agent):
         """Ollama sends every tool call at index 0 with different ids.
 

tests/tools/test_fuzzy_match.py

@@ -148,3 +148,184 @@ class TestStrategyNameSurfaced:
         assert count == 0
         assert strategy is None
         assert err is not None
+
+
+class TestEscapeDriftGuard:
+    """Tests for the escape-drift guard that catches bash/JSON serialization
+    artifacts where an apostrophe gets prefixed with a spurious backslash
+    in tool-call transport.
+    """
+
+    def test_drift_blocked_apostrophe(self):
+        """File has ', old_string and new_string both have \\' — classic
+        tool-call drift. Guard must block with a helpful error instead of
+        writing \\' literals into source code."""
+        content = "x = \"hello there\"\n"
+        # Simulate transport-corrupted old_string and new_string where an
+        # apostrophe-like context got prefixed with a backslash. The content
+        # itself has no apostrophe, but both strings do — matching via
+        # whitespace/anchor strategies would otherwise succeed.
+        old_string = "x = \"hello there\" # don\\'t edit\n"
+        new_string = "x = \"hi there\" # don\\'t edit\n"
+        # This particular pair won't match anything, so it exits via
+        # no-match path. Build a case where a non-exact strategy DOES match.
+        content = "line\n    x = 1\nline"
+        old_string = "line\n  x = \\'a\\'\nline"
+        new_string = "line\n  x = \\'b\\'\nline"
+        new, count, strategy, err = fuzzy_find_and_replace(content, old_string, new_string)
+        assert count == 0
+        assert err is not None and "Escape-drift" in err
+        assert "backslash" in err.lower()
+        assert new == content  # file untouched
+
+    def test_drift_blocked_double_quote(self):
+        """Same idea but with \\" drift instead of \\'."""
+        content = 'line\n    x = 1\nline'
+        old_string = 'line\n  x = \\"a\\"\nline'
+        new_string = 'line\n  x = \\"b\\"\nline'
+        new, count, strategy, err = fuzzy_find_and_replace(content, old_string, new_string)
+        assert count == 0
+        assert err is not None and "Escape-drift" in err
+
+    def test_drift_allowed_when_file_genuinely_has_backslash_escapes(self):
+        """If the file already contains \\' (e.g. inside an existing escaped
+        string), the model is legitimately preserving it. Guard must NOT
+        fire."""
+        content = "line\n  x = \\'a\\'\nline"
+        old_string = "line\n  x = \\'a\\'\nline"
+        new_string = "line\n  x = \\'b\\'\nline"
+        new, count, strategy, err = fuzzy_find_and_replace(content, old_string, new_string)
+        assert err is None
+        assert count == 1
+        assert "\\'b\\'" in new
+
+    def test_drift_allowed_on_exact_match(self):
+        """Exact matches bypass the drift guard entirely — if the file
+        really contains the exact bytes old_string specified, it's not
+        drift."""
+        content = "hello \\'world\\'"
+        new, count, strategy, err = fuzzy_find_and_replace(
+            content, "hello \\'world\\'", "hello \\'there\\'"
+        )
+        assert err is None
+        assert count == 1
+        assert strategy == "exact"
+
+    def test_drift_allowed_when_adding_escaped_strings(self):
+        """Model is adding new content with \\' that wasn't in the original.
+        old_string has no \\', so guard doesn't fire."""
+        content = "line1\nline2\nline3"
+        old_string = "line1\nline2\nline3"
+        new_string = "line1\nprint(\\'added\\')\nline2\nline3"
+        new, count, strategy, err = fuzzy_find_and_replace(content, old_string, new_string)
+        assert err is None
+        assert count == 1
+        assert "\\'added\\'" in new
+
+    def test_no_drift_check_when_new_string_lacks_suspect_chars(self):
+        """Fast-path: if new_string has no \\' or \\", guard must not
+        fire even on fuzzy match."""
+        content = "def foo():\n    pass"  # extra space ignored by line_trimmed
+        old_string = "def foo():\n  pass"
+        new_string = "def bar():\n  return 1"
+        new, count, strategy, err = fuzzy_find_and_replace(content, old_string, new_string)
+        assert err is None
+        assert count == 1
+
+
+class TestFindClosestLines:
+    def setup_method(self):
+        from tools.fuzzy_match import find_closest_lines
+        self.find_closest_lines = find_closest_lines
+
+    def test_finds_similar_line(self):
+        content = "def foo():\n    pass\ndef bar():\n    return 1\n"
+        result = self.find_closest_lines("def baz():", content)
+        assert "def foo" in result or "def bar" in result
+
+    def test_returns_empty_for_no_match(self):
+        content = "completely different content here"
+        result = self.find_closest_lines("xyzzy_no_match_possible_!!!", content)
+        assert result == ""
+
+    def test_returns_empty_for_empty_inputs(self):
+        assert self.find_closest_lines("", "some content") == ""
+        assert self.find_closest_lines("old string", "") == ""
+
+    def test_includes_context_lines(self):
+        content = "line1\nline2\ndef target():\n    pass\nline5\n"
+        result = self.find_closest_lines("def target():", content)
+        assert "target" in result
+
+    def test_includes_line_numbers(self):
+        content = "line1\nline2\ndef foo():\n    pass\n"
+        result = self.find_closest_lines("def foo():", content)
+        # Should include line numbers in format "N| content"
+        assert "|" in result
+
+
+class TestFormatNoMatchHint:
+    """Gating tests for format_no_match_hint — the shared helper that decides
+    whether a 'Did you mean?' snippet should be appended to an error.
+    """
+
+    def setup_method(self):
+        from tools.fuzzy_match import format_no_match_hint
+        self.fmt = format_no_match_hint
+
+    def test_fires_on_could_not_find_with_match(self):
+        """Classic no-match: similar content exists → hint fires."""
+        content = "def foo():\n    pass\ndef bar():\n    pass\n"
+        result = self.fmt(
+            "Could not find a match for old_string in the file",
+            0, "def baz():", content,
+        )
+        assert "Did you mean" in result
+        assert "foo" in result or "bar" in result
+
+    def test_silent_on_ambiguous_match_error(self):
+        """'Found N matches' is not a missing-match failure — no hint."""
+        content = "aaa bbb aaa\n"
+        result = self.fmt(
+            "Found 2 matches for old_string. Provide more context to make it unique, or use replace_all=True.",
+            0, "aaa", content,
+        )
+        assert result == ""
+
+    def test_silent_on_escape_drift_error(self):
+        """Escape-drift errors are intentional blocks — hint would mislead."""
+        content = "x = 1\n"
+        result = self.fmt(
+            "Escape-drift detected: old_string and new_string contain the literal sequence '\\\\''...",
+            0, "x = \\'1\\'", content,
+        )
+        assert result == ""
+
+    def test_silent_on_identical_strings(self):
+        """old_string == new_string — hint irrelevant."""
+        result = self.fmt(
+            "old_string and new_string are identical",
+            0, "foo", "foo bar\n",
+        )
+        assert result == ""
+
+    def test_silent_when_match_count_nonzero(self):
+        """If match succeeded, we shouldn't be in the error path — defense in depth."""
+        result = self.fmt(
+            "Could not find a match for old_string in the file",
+            1, "foo", "foo bar\n",
+        )
+        assert result == ""
+
+    def test_silent_on_none_error(self):
+        """No error at all — no hint."""
+        result = self.fmt(None, 0, "foo", "bar\n")
+        assert result == ""
+
+    def test_silent_when_no_similar_content(self):
+        """Even for a valid no-match error, skip hint when nothing similar exists."""
+        result = self.fmt(
+            "Could not find a match for old_string in the file",
+            0, "totally_unique_xyzzy_qux", "abc\nxyz\n",
+        )
+        assert result == ""

tests/tools/test_patch_did_you_mean.py

@@ -0,0 +1,114 @@
+import json
+import os
+import textwrap
+from pathlib import Path
+
+import tools.skill_manager_tool as skill_manager_tool
+from tools.file_tools import patch_tool
+from tools.skill_manager_tool import _create_skill, _patch_skill
+
+
+def _disable_patch_tool_guards(monkeypatch):
+    monkeypatch.setattr("tools.file_tools._check_sensitive_path", lambda _path: None)
+    monkeypatch.setattr("tools.file_tools._check_file_staleness", lambda _path, _task_id: None)
+    monkeypatch.setattr("tools.file_tools._log_and_check_conflict", lambda _path, _task_id, _action: None)
+
+
+def test_patch_tool_replace_no_match_shows_rich_hint_without_legacy_hint(tmp_path, monkeypatch):
+    _disable_patch_tool_guards(monkeypatch)
+    sample = tmp_path / "sample.py"
+    sample.write_text("def foo():\n    return 1\n\ndef bar():\n    return 2\n", encoding="utf-8")
+
+    raw = patch_tool(
+        mode="replace",
+        path=str(sample),
+        old_string="def barycentric():",
+        new_string="def barycentric_new():",
+        task_id="qa960-replace-rich-hint",
+    )
+
+    result = json.loads(raw)
+    assert result["success"] is False
+    assert "Could not find a match" in result["error"]
+    assert "Did you mean one of these sections?" in result["error"]
+    assert "def bar():" in result["error"] or "def foo():" in result["error"]
+    assert "[Hint:" not in raw
+
+
+def test_patch_tool_replace_ambiguous_error_does_not_show_did_you_mean(tmp_path, monkeypatch):
+    _disable_patch_tool_guards(monkeypatch)
+    sample = tmp_path / "sample.py"
+    sample.write_text("aaa\nbbb\naaa\n", encoding="utf-8")
+
+    raw = patch_tool(
+        mode="replace",
+        path=str(sample),
+        old_string="aaa",
+        new_string="ccc",
+        task_id="qa960-replace-ambiguous",
+    )
+
+    result = json.loads(raw)
+    assert result["success"] is False
+    assert "Found 2 matches" in result["error"]
+    assert "Did you mean one of these sections?" not in result["error"]
+    assert "[Hint:" not in raw
+
+
+def test_patch_tool_v4a_no_match_shows_rich_hint(tmp_path, monkeypatch):
+    _disable_patch_tool_guards(monkeypatch)
+    sample = tmp_path / "sample.py"
+    sample.write_text("def foo():\n    return 1\n", encoding="utf-8")
+
+    patch = textwrap.dedent(
+        f"""\
+        *** Begin Patch
+        *** Update File: {sample}
+        @@
+        -def barycentric():
+        +def barycentric_new():
+        *** End Patch
+        """
+    )
+
+    raw = patch_tool(mode="patch", patch=patch, task_id="qa960-v4a-rich-hint")
+    result = json.loads(raw)
+    assert result["success"] is False
+    assert "Patch validation failed" in result["error"]
+    assert "Did you mean one of these sections?" in result["error"]
+    assert "def foo():" in result["error"]
+
+
+def test_skill_patch_no_match_shows_rich_hint(tmp_path, monkeypatch):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    skills_dir = tmp_path / "skills"
+    skills_dir.mkdir(parents=True, exist_ok=True)
+    monkeypatch.setattr(skill_manager_tool, "SKILLS_DIR", skills_dir)
+    monkeypatch.setattr(skill_manager_tool, "_security_scan_skill", lambda _skill_dir: None)
+
+    _create_skill(
+        "qa-skill",
+        textwrap.dedent(
+            """\
+            ---
+            name: qa-skill
+            description: test
+            ---
+
+            Step 1: Do the thing.
+            Step 2: Verify the thing.
+            """
+        ),
+    )
+
+    result = _patch_skill(
+        "qa-skill",
+        "Step 1: Do the production rollout.",
+        "Step 1: Updated.",
+    )
+
+    assert result["success"] is False
+    assert "Could not find a match" in result["error"]
+    assert "Did you mean one of these sections?" in result["error"]
+    assert "Step 1: Do the thing." in result["error"]
+    assert "file_preview" in result

tools/file_operations.py

@@ -757,12 +757,14 @@ class ShellFileOperations(FileOperations):
             content, old_string, new_string, replace_all
         )
         
-        if error:
-            return PatchResult(error=error)
-        
-        if match_count == 0:
-            return PatchResult(error=f"Could not find match for old_string in {path}")
-        
+        if error or match_count == 0:
+            err_msg = error or f"Could not find match for old_string in {path}"
+            try:
+                from tools.fuzzy_match import format_no_match_hint
+                err_msg += format_no_match_hint(err_msg, match_count, old_string, content)
+            except Exception:
+                pass
+            return PatchResult(error=err_msg)
         # Write back
         write_result = self.write_file(path, new_content)
         if write_result.error:

tools/file_tools.py

@@ -8,6 +8,7 @@ import os
 import threading
 import time
 from pathlib import Path
+from typing import Any, Dict, Optional
 from tools.binary_extensions import has_binary_extension
 from tools.file_operations import ShellFileOperations
 from agent.redact import redact_sensitive_text
@@ -690,8 +691,11 @@ def patch_tool(mode: str = "replace", path: str = None, old_string: str = None,
         result_json = json.dumps(result_dict, ensure_ascii=False)
         # Hint when old_string not found — saves iterations where the agent
         # retries with stale content instead of re-reading the file.
+        # Suppressed when patch_replace already attached a rich "Did you mean?"
+        # snippet (which is strictly more useful than the generic hint).
         if result_dict.get("error") and "Could not find" in str(result_dict["error"]):
-            result_json += "\n\n[Hint: old_string not found. Use read_file to verify the current content, or search_files to locate the text.]"
+            if "Did you mean one of these sections?" not in str(result_dict["error"]):
+                result_json += "\n\n[Hint: old_string not found. Use read_file to verify the current content, or search_files to locate the text.]"
         return result_json
     except Exception as e:
         return tool_error(str(e))

tools/fuzzy_match.py

@@ -93,6 +93,21 @@ def fuzzy_find_and_replace(content: str, old_string: str, new_string: str,
                     f"Provide more context to make it unique, or use replace_all=True."
                 )
 
+            # Escape-drift guard: when the matched strategy is NOT `exact`,
+            # we matched via some form of normalization. If new_string
+            # contains shell/JSON-style escape sequences (\\' or \\") that
+            # would be written literally into the file but the matched
+            # region of the file has no such sequences, this is almost
+            # certainly tool-call serialization drift — the model typed
+            # an apostrophe/quote and the transport added a stray
+            # backslash. Writing new_string as-is would corrupt the file.
+            # Block with a helpful error so the model re-reads and retries
+            # instead of the caller silently persisting garbage (or not).
+            if strategy_name != "exact":
+                drift_err = _detect_escape_drift(content, matches, old_string, new_string)
+                if drift_err:
+                    return content, 0, None, drift_err
+
             # Perform replacement
             new_content = _apply_replacements(content, matches, new_string)
             return new_content, len(matches), strategy_name, None
@@ -101,6 +116,46 @@ def fuzzy_find_and_replace(content: str, old_string: str, new_string: str,
     return content, 0, None, "Could not find a match for old_string in the file"
 
 
+def _detect_escape_drift(content: str, matches: List[Tuple[int, int]],
+                         old_string: str, new_string: str) -> Optional[str]:
+    """Detect tool-call escape-drift artifacts in new_string.
+
+    Looks for ``\\'`` or ``\\"`` sequences that are present in both
+    old_string and new_string (i.e. the model copy-pasted them as "context"
+    it intended to preserve) but don't exist in the matched region of the
+    file. That pattern indicates the transport layer inserted spurious
+    shell-style escapes around apostrophes or quotes — writing new_string
+    verbatim would literally insert ``\\'`` into source code.
+
+    Returns an error string if drift is detected, None otherwise.
+    """
+    # Cheap pre-check: bail out unless new_string actually contains a
+    # suspect escape sequence. This keeps the guard free for all the
+    # common, correct cases.
+    if "\\'" not in new_string and '\\"' not in new_string:
+        return None
+
+    # Aggregate matched regions of the file — that's what new_string will
+    # replace. If the suspect escapes are present there already, the
+    # model is genuinely preserving them (valid for some languages /
+    # escaped strings); accept the patch.
+    matched_regions = "".join(content[start:end] for start, end in matches)
+
+    for suspect in ("\\'", '\\"'):
+        if suspect in new_string and suspect in old_string and suspect not in matched_regions:
+            plain = suspect[1]  # "'" or '"'
+            return (
+                f"Escape-drift detected: old_string and new_string contain "
+                f"the literal sequence {suspect!r} but the matched region of "
+                f"the file does not. This is almost always a tool-call "
+                f"serialization artifact where an apostrophe or quote got "
+                f"prefixed with a spurious backslash. Re-read the file with "
+                f"read_file and pass old_string/new_string without "
+                f"backslash-escaping {plain!r} characters."
+            )
+    return None
+
+
 def _apply_replacements(content: str, matches: List[Tuple[int, int]], new_string: str) -> str:
     """
     Apply replacements at the given positions.
@@ -564,3 +619,86 @@ def _map_normalized_positions(original: str, normalized: str,
         original_matches.append((orig_start, min(orig_end, len(original))))
     
     return original_matches
+
+
+def find_closest_lines(old_string: str, content: str, context_lines: int = 2, max_results: int = 3) -> str:
+    """Find lines in content most similar to old_string for "did you mean?" feedback.
+
+    Returns a formatted string showing the closest matching lines with context,
+    or empty string if no useful match is found.
+    """
+    if not old_string or not content:
+        return ""
+
+    old_lines = old_string.splitlines()
+    content_lines = content.splitlines()
+
+    if not old_lines or not content_lines:
+        return ""
+
+    # Use first line of old_string as anchor for search
+    anchor = old_lines[0].strip()
+    if not anchor:
+        # Try second line if first is blank
+        candidates = [l.strip() for l in old_lines if l.strip()]
+        if not candidates:
+            return ""
+        anchor = candidates[0]
+
+    # Score each line in content by similarity to anchor
+    scored = []
+    for i, line in enumerate(content_lines):
+        stripped = line.strip()
+        if not stripped:
+            continue
+        ratio = SequenceMatcher(None, anchor, stripped).ratio()
+        if ratio > 0.3:
+            scored.append((ratio, i))
+
+    if not scored:
+        return ""
+
+    # Take top matches
+    scored.sort(key=lambda x: -x[0])
+    top = scored[:max_results]
+
+    parts = []
+    seen_ranges = set()
+    for _, line_idx in top:
+        start = max(0, line_idx - context_lines)
+        end = min(len(content_lines), line_idx + len(old_lines) + context_lines)
+        key = (start, end)
+        if key in seen_ranges:
+            continue
+        seen_ranges.add(key)
+        snippet = "\n".join(
+            f"{start + j + 1:4d}| {content_lines[start + j]}"
+            for j in range(end - start)
+        )
+        parts.append(snippet)
+
+    if not parts:
+        return ""
+
+    return "\n---\n".join(parts)
+
+
+def format_no_match_hint(error: Optional[str], match_count: int,
+                         old_string: str, content: str) -> str:
+    """Return a '\\n\\nDid you mean...' snippet for plain no-match errors.
+
+    Gated so the hint only fires for actual "old_string not found" failures.
+    Ambiguous-match ("Found N matches"), escape-drift, and identical-strings
+    errors all have ``match_count == 0`` but a "did you mean?" snippet would
+    be misleading — those failed for unrelated reasons.
+
+    Returns an empty string when there's nothing useful to append.
+    """
+    if match_count != 0:
+        return ""
+    if not error or not error.startswith("Could not find"):
+        return ""
+    hint = find_closest_lines(old_string, content)
+    if not hint:
+        return ""
+    return "\n\nDid you mean one of these sections?\n" + hint

tools/patch_parser.py

@@ -290,10 +290,16 @@ def _validate_operations(
                 )
                 if count == 0:
                     label = f"'{hunk.context_hint}'" if hunk.context_hint else "(no hint)"
-                    errors.append(
+                    msg = (
                         f"{op.file_path}: hunk {label} not found"
                         + (f" — {match_error}" if match_error else "")
                     )
+                    try:
+                        from tools.fuzzy_match import format_no_match_hint
+                        msg += format_no_match_hint(match_error, count, search_pattern, simulated)
+                    except Exception:
+                        pass
+                    errors.append(msg)
                 else:
                     # Advance simulation so subsequent hunks validate correctly.
                     # Reuse the result from the call above — no second fuzzy run.
@@ -537,7 +543,13 @@ def _apply_update(op: PatchOperation, file_ops: Any) -> Tuple[bool, str]:
                             error = None
                 
                 if error:
-                    return False, f"Could not apply hunk: {error}"
+                    err_msg = f"Could not apply hunk: {error}"
+                    try:
+                        from tools.fuzzy_match import format_no_match_hint
+                        err_msg += format_no_match_hint(error, 0, search_pattern, new_content)
+                    except Exception:
+                        pass
+                    return False, err_msg
         else:
             # Addition-only hunk (no context or removed lines).
             # Insert at the location indicated by the context hint, or at end of file.

tools/skill_manager_tool.py

@@ -575,9 +575,15 @@ def _patch_skill(
     if match_error:
         # Show a short preview of the file so the model can self-correct
         preview = content[:500] + ("..." if len(content) > 500 else "")
+        err_msg = match_error
+        try:
+            from tools.fuzzy_match import format_no_match_hint
+            err_msg += format_no_match_hint(match_error, match_count, old_string, content)
+        except Exception:
+            pass
         return {
             "success": False,
-            "error": match_error,
+            "error": err_msg,
             "file_preview": preview,
         }