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

cli.py

@@ -589,7 +589,6 @@ from tools.terminal_tool import set_sudo_password_callback, set_approval_callbac
 from tools.skills_tool import set_secret_capture_callback
 from hermes_cli.callbacks import prompt_for_secret
 from tools.browser_tool import _emergency_cleanup_all_sessions as _cleanup_all_browsers
-from utils import repair_and_load_json
 
 # Guard to prevent cleanup from running multiple times on exit
 _cleanup_done = False
@@ -3570,11 +3569,7 @@ class HermesCLI:
                 result_json = _asyncio.run(
                     vision_analyze_tool(image_url=str(img_path), user_prompt=analysis_prompt)
                 )
-                result = repair_and_load_json(
-                    result_json,
-                    default={},
-                    context="cli_image_analysis",
-                ) if isinstance(result_json, str) else {}
+                result = _json.loads(result_json)
                 if result.get("success"):
                     description = result.get("analysis", "")
                     enriched_parts.append(
@@ -4965,14 +4960,7 @@ class HermesCLI:
         from tools.cronjob_tools import cronjob as cronjob_tool
 
         def _cron_api(**kwargs):
-            result = repair_and_load_json(
-                cronjob_tool(**kwargs),
-                default=None,
-                context="cli_cron_command",
-            )
-            if isinstance(result, dict):
-                return result
-            return {"success": False, "error": "Invalid JSON from cronjob tool"}
+            return json.loads(cronjob_tool(**kwargs))
 
         def _normalize_skills(values):
             normalized = []

tests/cli/test_cli_json_repair.py

@@ -1,62 +0,0 @@
-import sys
-import types
-from unittest.mock import patch
-
-
-def _stub_auxiliary_client():
-    stub = types.ModuleType("agent.auxiliary_client")
-    stub.call_llm = lambda *args, **kwargs: None
-    stub.resolve_provider_client = lambda *args, **kwargs: (None, None)
-    stub.get_text_auxiliary_client = lambda *args, **kwargs: (None, None)
-    stub.async_call_llm = lambda *args, **kwargs: None
-    stub.extract_content_or_reasoning = lambda *args, **kwargs: ""
-    stub._OR_HEADERS = {}
-    stub._get_task_timeout = lambda *args, **kwargs: 30
-    sys.modules["agent.auxiliary_client"] = stub
-
-
-def _stub_vision_tools(vision_analyze_tool):
-    stub = types.ModuleType("tools.vision_tools")
-    stub.vision_analyze_tool = vision_analyze_tool
-    sys.modules["tools.vision_tools"] = stub
-
-
-def test_preprocess_images_with_vision_repairs_malformed_json(tmp_path):
-    _stub_auxiliary_client()
-    from cli import HermesCLI
-
-    cli_obj = HermesCLI.__new__(HermesCLI)
-    image_path = tmp_path / "test.png"
-    image_path.write_bytes(b"fake-image-bytes")
-
-    async def fake_vision(**kwargs):
-        return "{'success': true, 'analysis': 'Recovered image description',}"
-
-    _stub_vision_tools(fake_vision)
-    result = HermesCLI._preprocess_images_with_vision(
-        cli_obj,
-        "Describe this",
-        [image_path],
-        announce=False,
-    )
-
-    assert "Recovered image description" in result
-    assert "Describe this" in result
-    assert str(image_path) in result
-
-
-def test_handle_cron_command_repairs_malformed_json(capsys):
-    _stub_auxiliary_client()
-    from cli import HermesCLI
-
-    cli_obj = HermesCLI.__new__(HermesCLI)
-    malformed_result = """{'success': true, 'jobs': [{'job_id': 'job-1234567890ab', 'name': 'Nightly Check', 'state': 'scheduled', 'schedule': 'every 1h', 'repeat': 'forever', 'prompt_preview': 'Check server status', 'skills': ['blogwatcher',], 'next_run_at': '2026-04-22T01:00:00Z',},],}"""
-
-    with patch("tools.cronjob_tools.cronjob", return_value=malformed_result):
-        HermesCLI._handle_cron_command(cli_obj, "/cron list")
-
-    out = capsys.readouterr().out
-    assert "Scheduled Jobs:" in out
-    assert "job-1234567890ab" in out
-    assert "Nightly Check" in out
-    assert "blogwatcher" in out

tests/tools/test_browser_json_repair.py

@@ -1,108 +0,0 @@
-import io
-import json
-import sys
-import types
-from unittest.mock import MagicMock, patch
-
-
-def _stub_auxiliary_client():
-    stub = types.ModuleType("agent.auxiliary_client")
-    stub.call_llm = lambda *args, **kwargs: None
-    stub.resolve_provider_client = lambda *args, **kwargs: (None, None)
-    stub.get_text_auxiliary_client = lambda *args, **kwargs: (None, None)
-    stub.async_call_llm = lambda *args, **kwargs: None
-    stub.extract_content_or_reasoning = lambda *args, **kwargs: ""
-    stub._OR_HEADERS = {}
-    stub._get_task_timeout = lambda *args, **kwargs: 30
-    sys.modules["agent.auxiliary_client"] = stub
-
-
-def test_run_browser_command_repairs_malformed_stdout_envelope(tmp_path):
-    _stub_auxiliary_client()
-    from tools.browser_tool import _run_browser_command
-
-    mock_proc = MagicMock()
-    mock_proc.returncode = 0
-    mock_proc.wait.return_value = 0
-    fake_session = {
-        "session_name": "test-session",
-        "session_id": "test-id",
-        "cdp_url": None,
-    }
-    malformed_stdout = "{'success': true, 'data': {'url': 'https://example.com',},}"
-
-    def fake_open(path, mode="r", *args, **kwargs):
-        path = str(path)
-        if path.endswith("_stdout_navigate"):
-            return io.StringIO(malformed_stdout)
-        if path.endswith("_stderr_navigate"):
-            return io.StringIO("")
-        raise FileNotFoundError(path)
-
-    with (
-        patch("tools.browser_tool._find_agent_browser", return_value="/usr/bin/agent-browser"),
-        patch("tools.browser_tool._get_session_info", return_value=fake_session),
-        patch("tools.browser_tool._socket_safe_tmpdir", return_value=str(tmp_path)),
-        patch("tools.browser_tool._merge_browser_path", side_effect=lambda p: p),
-        patch("tools.interrupt.is_interrupted", return_value=False),
-        patch("subprocess.Popen", return_value=mock_proc),
-        patch("os.open", return_value=99),
-        patch("os.close"),
-        patch("os.unlink"),
-        patch("builtins.open", side_effect=fake_open),
-    ):
-        result = _run_browser_command("task-1", "navigate", ["https://example.com"])
-
-    assert result["success"] is True
-    assert result["data"]["url"] == "https://example.com"
-
-
-def test_agent_browser_eval_repairs_malformed_json_result():
-    _stub_auxiliary_client()
-    from tools.browser_tool import _browser_eval
-
-    with patch(
-        "tools.browser_tool._run_browser_command",
-        return_value={"success": True, "data": {"result": "{'items': ['a', 'b',],}"}},
-    ):
-        result = json.loads(_browser_eval("document.body.innerText", task_id="test"))
-
-    assert result["success"] is True
-    assert result["result"] == {"items": ["a", "b"]}
-    assert result["result_type"] == "dict"
-
-
-def test_camofox_eval_repairs_malformed_json_result():
-    _stub_auxiliary_client()
-    from tools.browser_tool import _camofox_eval
-
-    with (
-        patch("tools.browser_camofox._ensure_tab", return_value={"tab_id": "tab-1", "user_id": "user-1"}),
-        patch("tools.browser_camofox._post", return_value={"result": "{'count': 3,}"}),
-    ):
-        result = json.loads(_camofox_eval("2+1", task_id="test"))
-
-    assert result["success"] is True
-    assert result["result"] == {"count": 3}
-    assert result["result_type"] == "dict"
-
-
-def test_browser_get_images_repairs_malformed_json_result():
-    _stub_auxiliary_client()
-    from tools.browser_tool import browser_get_images
-
-    with patch(
-        "tools.browser_tool._run_browser_command",
-        return_value={
-            "success": True,
-            "data": {
-                "result": "[{\"src\": \"https://example.com/cat.png\", \"alt\": \"cat\",}]"
-            },
-        },
-    ):
-        result = json.loads(browser_get_images(task_id="test"))
-
-    assert result["success"] is True
-    assert result["count"] == 1
-    assert result["images"] == [{"src": "https://example.com/cat.png", "alt": "cat"}]
-    assert "warning" not in result

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/browser_tool.py

@@ -67,7 +67,6 @@ from typing import Dict, Any, Optional, List
 from pathlib import Path
 from agent.auxiliary_client import call_llm
 from hermes_constants import get_hermes_home
-from utils import repair_and_load_json
 
 try:
     from tools.website_policy import check_website_access
@@ -1172,12 +1171,8 @@ def _run_browser_command(
             return {"success": False, "error": f"Browser command '{command}' returned no output"}
 
         if stdout_text:
-            parsed = repair_and_load_json(
-                stdout_text,
-                default=None,
-                context=f"browser_{command}_stdout",
-            )
-            if isinstance(parsed, dict):
+            try:
+                parsed = json.loads(stdout_text)
                 # Warn if snapshot came back empty (common sign of daemon/CDP issues)
                 if command == "snapshot" and parsed.get("success"):
                     snap_data = parsed.get("data", {})
@@ -1186,35 +1181,35 @@ def _run_browser_command(
                                        "Possible stale daemon or CDP connection issue. "
                                        "returncode=%s", returncode)
                 return parsed
+            except json.JSONDecodeError:
+                raw = stdout_text[:2000]
+                logger.warning("browser '%s' returned non-JSON output (rc=%s): %s",
+                               command, returncode, raw[:500])
 
-            raw = stdout_text[:2000]
-            logger.warning("browser '%s' returned non-JSON output (rc=%s): %s",
-                           command, returncode, raw[:500])
-
-            if command == "screenshot":
-                stderr_text = (stderr or "").strip()
-                combined_text = "\n".join(
-                    part for part in [stdout_text, stderr_text] if part
-                )
-                recovered_path = _extract_screenshot_path_from_text(combined_text)
-
-                if recovered_path and Path(recovered_path).exists():
-                    logger.info(
-                        "browser 'screenshot' recovered file from non-JSON output: %s",
-                        recovered_path,
+                if command == "screenshot":
+                    stderr_text = (stderr or "").strip()
+                    combined_text = "\n".join(
+                        part for part in [stdout_text, stderr_text] if part
                     )
-                    return {
-                        "success": True,
-                        "data": {
-                            "path": recovered_path,
-                            "raw": raw,
-                        },
-                    }
+                    recovered_path = _extract_screenshot_path_from_text(combined_text)
 
-            return {
-                "success": False,
-                "error": f"Non-JSON output from agent-browser for '{command}': {raw}"
-            }
+                    if recovered_path and Path(recovered_path).exists():
+                        logger.info(
+                            "browser 'screenshot' recovered file from non-JSON output: %s",
+                            recovered_path,
+                        )
+                        return {
+                            "success": True,
+                            "data": {
+                                "path": recovered_path,
+                                "raw": raw,
+                            },
+                        }
+
+                return {
+                    "success": False,
+                    "error": f"Non-JSON output from agent-browser for '{command}': {raw}"
+                }
         
         # Check for errors
         if returncode != 0:
@@ -1782,11 +1777,10 @@ def _browser_eval(expression: str, task_id: Optional[str] = None) -> str:
     # is valid JSON, parse it so the model gets structured data.
     parsed = raw_result
     if isinstance(raw_result, str):
-        parsed = repair_and_load_json(
-            raw_result,
-            default=raw_result,
-            context="browser_eval_result",
-        )
+        try:
+            parsed = json.loads(raw_result)
+        except (json.JSONDecodeError, ValueError):
+            pass  # keep as string
 
     return json.dumps({
         "success": True,
@@ -1807,11 +1801,10 @@ def _camofox_eval(expression: str, task_id: Optional[str] = None) -> str:
         raw_result = resp.get("result") if isinstance(resp, dict) else resp
         parsed = raw_result
         if isinstance(raw_result, str):
-            parsed = repair_and_load_json(
-                raw_result,
-                default=raw_result,
-                context="camofox_eval_result",
-            )
+            try:
+                parsed = json.loads(raw_result)
+            except (json.JSONDecodeError, ValueError):
+                pass
 
         return json.dumps({
             "success": True,
@@ -1911,29 +1904,26 @@ def browser_get_images(task_id: Optional[str] = None) -> str:
     if result.get("success"):
         data = result.get("data", {})
         raw_result = data.get("result", "[]")
-
-        warning = None
-        if isinstance(raw_result, str):
-            images = repair_and_load_json(
-                raw_result,
-                default=None,
-                context="browser_get_images_result",
-            )
-        else:
-            images = raw_result
-
-        if not isinstance(images, list):
-            images = []
-            warning = "Could not parse image data"
-
-        payload = {
-            "success": True,
-            "images": images,
-            "count": len(images),
-        }
-        if warning:
-            payload["warning"] = warning
-        return json.dumps(payload, ensure_ascii=False)
+        
+        try:
+            # Parse the JSON string returned by JavaScript
+            if isinstance(raw_result, str):
+                images = json.loads(raw_result)
+            else:
+                images = raw_result
+            
+            return json.dumps({
+                "success": True,
+                "images": images,
+                "count": len(images)
+            }, ensure_ascii=False)
+        except json.JSONDecodeError:
+            return json.dumps({
+                "success": True,
+                "images": [],
+                "count": 0,
+                "warning": "Could not parse image data"
+            }, ensure_ascii=False)
     else:
         return json.dumps({
             "success": False,

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,
         }