feat: add profile-scoped local Hindsight eval path

Closes #1010

agent/context_snapshots.py

@@ -0,0 +1,69 @@
+"""First-class context snapshot artifacts for live runtime memory evaluation."""
+
+from __future__ import annotations
+
+import json
+import re
+from pathlib import Path
+from typing import Any
+
+from hermes_constants import get_hermes_home
+
+
+_SAFE_SEGMENT_RE = re.compile(r"[^A-Za-z0-9_.-]+")
+
+
+class ContextSnapshotRecorder:
+    """Write per-call prompt-composition artifacts for a Hermes session."""
+
+    def __init__(self, session_id: str, *, enabled: bool = False, base_dir: str | Path | None = None):
+        self.session_id = session_id or "session"
+        self.enabled = bool(enabled)
+        self.base_dir = Path(base_dir) if base_dir else get_hermes_home() / "reports" / "context_snapshots"
+
+    @property
+    def session_dir(self) -> Path:
+        safe_session = _SAFE_SEGMENT_RE.sub("_", self.session_id).strip("._") or "session"
+        return self.base_dir / safe_session
+
+    def record_call(
+        self,
+        api_call_count: int,
+        *,
+        system_prompt: str,
+        memory_provider_system_prompt: str = "",
+        memory_prefetch_raw: str = "",
+        memory_context_block: str = "",
+        api_user_message: str = "",
+        api_messages: list[dict[str, Any]] | None = None,
+        metadata: dict[str, Any] | None = None,
+    ) -> Path | None:
+        if not self.enabled:
+            return None
+
+        call_dir = self.session_dir / f"call_{api_call_count:03d}"
+        call_dir.mkdir(parents=True, exist_ok=True)
+
+        self._write_text(call_dir / "system_prompt.txt", system_prompt or "")
+        self._write_text(call_dir / "memory_provider_system_prompt.txt", memory_provider_system_prompt or "")
+        self._write_text(call_dir / "memory_prefetch_raw.txt", memory_prefetch_raw or "")
+        self._write_text(call_dir / "memory_context_block.txt", memory_context_block or "")
+        self._write_text(call_dir / "api_user_message.txt", api_user_message or "")
+        self._write_json(call_dir / "api_messages.json", api_messages or [])
+        self._write_json(
+            call_dir / "metadata.json",
+            {
+                "session_id": self.session_id,
+                "api_call_count": api_call_count,
+                **(metadata or {}),
+            },
+        )
+        return call_dir
+
+    @staticmethod
+    def _write_text(path: Path, content: str) -> None:
+        path.write_text(content, encoding="utf-8")
+
+    @staticmethod
+    def _write_json(path: Path, payload: Any) -> None:
+        path.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8")

docs/hindsight-local-eval.md

@@ -0,0 +1,132 @@
+# Hindsight local eval homes for live Hermes runtime testing
+
+Issue: #1010
+Parent: #985
+
+This document defines a reproducible, profile-scoped evaluation layout for baseline / MemPalace / Hindsight comparisons without requiring Hindsight Cloud.
+
+## Eval home layout
+
+Use three separate `HERMES_HOME` directories so each run has isolated config, memory, sessions, and artifacts.
+
+```text
+~/.hermes/profiles/atlas-baseline/
+  config.yaml
+  .env
+  MEMORY.md
+  USER.md
+  reports/context_snapshots/
+
+~/.hermes/profiles/atlas-mempalace/
+  config.yaml
+  .env
+  MEMORY.md
+  USER.md
+  reports/context_snapshots/
+  plugins/           # if a local MemPalace plugin is installed for this eval lane
+
+~/.hermes/profiles/atlas-hindsight/
+  config.yaml
+  .env
+  MEMORY.md
+  USER.md
+  hindsight/config.json
+  reports/context_snapshots/
+```
+
+## Hindsight local config
+
+The Hindsight provider already loads config from `$HERMES_HOME/hindsight/config.json` first. For the local eval lane, prefer `local_embedded` so Hermes can bring up a local Hindsight daemon without cloud signup.
+
+Example `~/.hermes/profiles/atlas-hindsight/hindsight/config.json`:
+
+```json
+{
+  "mode": "local_embedded",
+  "memory_mode": "context",
+  "recall_prefetch_method": "recall",
+  "llm_provider": "ollama",
+  "llm_model": "gemma3:12b",
+  "api_url": "http://localhost:8888"
+}
+```
+
+Notes:
+- `local_embedded` avoids any Hindsight Cloud dependency.
+- If `profile` is omitted, Hermes now derives a stable local Hindsight profile name from the active profile identity / `HERMES_HOME` instead of collapsing all local runs into the shared legacy `hermes` profile.
+- `local_external` remains valid if you already run a local Hindsight server yourself.
+
+## Runtime switching procedure
+
+Switch by exporting `HERMES_HOME` before launching Hermes.
+
+### 1. Baseline
+
+```bash
+export HERMES_HOME="$HOME/.hermes/profiles/atlas-baseline"
+unset HERMES_CONTEXT_SNAPSHOTS
+hermes chat
+```
+
+### 2. MemPalace lane
+
+```bash
+export HERMES_HOME="$HOME/.hermes/profiles/atlas-mempalace"
+export HERMES_CONTEXT_SNAPSHOTS=1
+hermes chat
+```
+
+### 3. Hindsight lane
+
+```bash
+export HERMES_HOME="$HOME/.hermes/profiles/atlas-hindsight"
+export HERMES_CONTEXT_SNAPSHOTS=1
+hermes chat
+```
+
+## Raw artifact capture
+
+When `HERMES_CONTEXT_SNAPSHOTS=1` is enabled, Hermes writes first-class prompt-composition artifacts under the active home by default.
+
+Artifact tree:
+
+```text
+$HERMES_HOME/reports/context_snapshots/<session-id>/call_001/
+  system_prompt.txt
+  memory_provider_system_prompt.txt
+  memory_prefetch_raw.txt
+  memory_context_block.txt
+  api_user_message.txt
+  api_messages.json
+  metadata.json
+```
+
+Minimum files a benchmark should inspect:
+- `system_prompt.txt`
+- `memory_prefetch_raw.txt`
+- `memory_context_block.txt`
+- `api_user_message.txt`
+- `api_messages.json`
+
+These prove:
+- what the system prompt was
+- what the provider prefetched
+- what entered `<memory-context>`
+- what the final API user message looked like
+- what full payload reached the model
+
+## Follow-on benchmark workflow
+
+A benchmark issue can now consume this path without redoing integration work:
+1. pick one eval home (`atlas-baseline`, `atlas-mempalace`, `atlas-hindsight`)
+2. export the corresponding `HERMES_HOME`
+3. run Hermes on the same prompt set
+4. compare the snapshot artifacts in `reports/context_snapshots/`
+5. score recall quality and answer quality separately
+
+## Why this is sovereign
+
+- no hosted Hindsight Cloud dependency is required
+- the Hindsight config is profile-scoped under `hindsight/config.json`
+- the runtime artifacts stay under the active `HERMES_HOME`
+- switching between baseline / MemPalace / Hindsight is just a `HERMES_HOME` swap

plugins/memory/hindsight/__init__.py

@@ -178,6 +178,25 @@ def _load_config() -> dict:
     }
 
 
+def _derive_local_profile_name(agent_identity: str = "", hermes_home: str = "") -> str:
+    """Return a stable profile name for local embedded Hindsight storage.
+
+    Prefer the active Hermes profile identity when available, otherwise fall back
+    to the basename of the active HERMES_HOME path. This prevents all local
+    Hindsight eval homes from sharing the legacy default profile name "hermes".
+    """
+    from pathlib import Path
+    import re
+
+    raw = (agent_identity or "").strip()
+    if not raw and hermes_home:
+        raw = Path(hermes_home).name.strip()
+    if not raw:
+        raw = "hermes"
+    safe = re.sub(r"[^A-Za-z0-9_.-]+", "-", raw).strip(".-_")
+    return safe or "hermes"
+
+
 # ---------------------------------------------------------------------------
 # MemoryProvider implementation
 # ---------------------------------------------------------------------------
@@ -468,6 +487,8 @@ class HindsightMemoryProvider(MemoryProvider):
 
     def initialize(self, session_id: str, **kwargs) -> None:
         self._session_id = session_id
+        hermes_home = str(kwargs.get("hermes_home") or "")
+        agent_identity = str(kwargs.get("agent_identity") or "")
 
         # Check client version and auto-upgrade if needed
         try:
@@ -500,6 +521,11 @@ class HindsightMemoryProvider(MemoryProvider):
         # "local" is a legacy alias for "local_embedded"
         if self._mode == "local":
             self._mode = "local_embedded"
+        if self._mode == "local_embedded" and not self._config.get("profile"):
+            self._config["profile"] = _derive_local_profile_name(
+                agent_identity=agent_identity,
+                hermes_home=hermes_home,
+            )
         self._api_key = self._config.get("apiKey") or self._config.get("api_key") or os.environ.get("HINDSIGHT_API_KEY", "")
         default_url = _DEFAULT_LOCAL_URL if self._mode in ("local_embedded", "local_external") else _DEFAULT_API_URL
         self._api_url = self._config.get("api_url") or os.environ.get("HINDSIGHT_API_URL", default_url)

run_agent.py

@@ -604,6 +604,8 @@ class AIAgent:
         checkpoint_max_snapshots: int = 50,
         pass_session_id: bool = False,
         persist_session: bool = True,
+        context_snapshots_enabled: bool | None = None,
+        context_snapshots_dir: str | None = None,
     ):
         """
         Initialize the AI Agent.
@@ -1129,6 +1131,43 @@ class AIAgent:
         except Exception:
             _agent_cfg = {}
 
+        def _is_enabled(value):
+            if isinstance(value, bool):
+                return value
+            return str(value).strip().lower() in {"1", "true", "yes", "on"}
+
+        _debug_cfg = _agent_cfg.get("debug", {}) if isinstance(_agent_cfg, dict) else {}
+        if not isinstance(_debug_cfg, dict):
+            _debug_cfg = {}
+        _snapshot_cfg = _debug_cfg.get("context_snapshots", {})
+        if not isinstance(_snapshot_cfg, dict):
+            _snapshot_cfg = {}
+        _snapshots_env = os.getenv("HERMES_CONTEXT_SNAPSHOTS")
+        _snapshots_dir_env = os.getenv("HERMES_CONTEXT_SNAPSHOTS_DIR")
+        if context_snapshots_enabled is None:
+            if _snapshots_env is not None:
+                self._context_snapshots_enabled = _is_enabled(_snapshots_env)
+            else:
+                self._context_snapshots_enabled = _is_enabled(_snapshot_cfg.get("enabled", False))
+        else:
+            self._context_snapshots_enabled = bool(context_snapshots_enabled)
+        self._context_snapshots_dir = (
+            context_snapshots_dir
+            or _snapshots_dir_env
+            or _snapshot_cfg.get("dir")
+            or None
+        )
+        try:
+            from agent.context_snapshots import ContextSnapshotRecorder
+            self._context_snapshot_recorder = ContextSnapshotRecorder(
+                session_id=self.session_id,
+                enabled=self._context_snapshots_enabled,
+                base_dir=self._context_snapshots_dir,
+            )
+        except Exception as _snapshot_err:
+            logger.debug("Context snapshot recorder init failed: %s", _snapshot_err)
+            self._context_snapshot_recorder = None
+
         # Persistent memory (MEMORY.md + USER.md) -- loaded from disk
         self._memory_store = None
         self._memory_enabled = False
@@ -8144,12 +8183,17 @@ class AIAgent:
         # Use original_user_message (clean input) — user_message may contain
         # injected skill content that bloats / breaks provider queries.
         _ext_prefetch_cache = ""
+        _memory_provider_prompt_cache = ""
         if self._memory_manager:
             try:
                 _query = original_user_message if isinstance(original_user_message, str) else ""
                 _ext_prefetch_cache = self._memory_manager.prefetch_all(_query) or ""
             except Exception:
                 pass
+            try:
+                _memory_provider_prompt_cache = self._memory_manager.build_system_prompt() or ""
+            except Exception:
+                pass
 
         while (api_call_count < self.max_iterations and self.iteration_budget.remaining > 0) or self._budget_grace_call:
             # Reset per-turn checkpoint dedup so each iteration can take one snapshot
@@ -8217,6 +8261,8 @@ class AIAgent:
             # However, providers like Moonshot AI require a separate 'reasoning_content' field
             # on assistant messages with tool_calls. We handle both cases here.
             api_messages = []
+            _current_api_user_message = ""
+            _current_memory_context_block = ""
             for idx, msg in enumerate(messages):
                 api_msg = msg.copy()
 
@@ -8231,12 +8277,15 @@ class AIAgent:
                         _fenced = build_memory_context_block(_ext_prefetch_cache)
                         if _fenced:
                             _injections.append(_fenced)
+                            _current_memory_context_block = _fenced
                     if _plugin_user_context:
                         _injections.append(_plugin_user_context)
                     if _injections:
                         _base = api_msg.get("content", "")
                         if isinstance(_base, str):
                             api_msg["content"] = _base + "\n\n" + "\n\n".join(_injections)
+                    if isinstance(api_msg.get("content"), str):
+                        _current_api_user_message = api_msg["content"]
 
                 # For ALL assistant messages, pass reasoning back to the API
                 # This ensures multi-turn reasoning context is preserved
@@ -8271,7 +8320,13 @@ class AIAgent:
                 from agent.privacy_filter import PrivacyFilter
                 pf = PrivacyFilter()
                 # Sanitize messages before they reach the provider
-                api_messages = pf.sanitize_messages(api_messages)
+                _pf_result = pf.sanitize_messages(api_messages)
+                if isinstance(_pf_result, tuple):
+                    api_messages, _pf_report = _pf_result
+                    if getattr(pf, "last_report", None) is None:
+                        pf.last_report = _pf_report
+                else:
+                    api_messages = _pf_result
                 if pf.last_report and pf.last_report.had_redactions:
                     logger.info(f"Privacy Filter: Redacted sensitive data from turn payload. Details: {pf.last_report.summary()}")
             except Exception as e:
@@ -8342,6 +8397,27 @@ class AIAgent:
                     new_tcs.append(tc)
                 am["tool_calls"] = new_tcs
 
+            if self._context_snapshot_recorder:
+                try:
+                    self._context_snapshot_recorder.record_call(
+                        api_call_count,
+                        system_prompt=effective_system,
+                        memory_provider_system_prompt=_memory_provider_prompt_cache,
+                        memory_prefetch_raw=_ext_prefetch_cache,
+                        memory_context_block=_current_memory_context_block,
+                        api_user_message=_current_api_user_message,
+                        api_messages=api_messages,
+                        metadata={
+                            "model": self.model,
+                            "provider": self.provider,
+                            "platform": self.platform or "",
+                            "api_mode": self.api_mode,
+                            "memory_providers": [p.name for p in getattr(self._memory_manager, "providers", [])],
+                        },
+                    )
+                except Exception as _snapshot_err:
+                    logger.debug("Context snapshot capture failed: %s", _snapshot_err)
+
             # Calculate approximate request size for logging
             total_chars = sum(len(str(msg)) for msg in api_messages)
             approx_tokens = estimate_messages_tokens_rough(api_messages)

tests/agent/test_context_snapshots.py

@@ -0,0 +1,43 @@
+from pathlib import Path
+
+from agent.context_snapshots import ContextSnapshotRecorder
+
+
+def test_disabled_recorder_writes_nothing(tmp_path):
+    recorder = ContextSnapshotRecorder(session_id="session-1", enabled=False, base_dir=tmp_path)
+
+    out = recorder.record_call(
+        1,
+        system_prompt="system",
+        api_messages=[{"role": "user", "content": "hello"}],
+    )
+
+    assert out is None
+    assert not (tmp_path / "session-1").exists()
+
+
+def test_enabled_recorder_writes_expected_artifacts(tmp_path):
+    recorder = ContextSnapshotRecorder(session_id="session-1", enabled=True, base_dir=tmp_path)
+
+    out = recorder.record_call(
+        1,
+        system_prompt="system prompt",
+        memory_provider_system_prompt="# Hindsight Memory\nActive.",
+        memory_prefetch_raw="- remembered fact",
+        memory_context_block="<memory-context>\nremembered\n</memory-context>",
+        api_user_message="What do I prefer?\n\n<memory-context>\nremembered\n</memory-context>",
+        api_messages=[
+            {"role": "system", "content": "system prompt"},
+            {"role": "user", "content": "What do I prefer?"},
+        ],
+        metadata={"provider": "openai", "memory_providers": ["builtin", "hindsight"]},
+    )
+
+    assert out == tmp_path / "session-1" / "call_001"
+    assert (out / "system_prompt.txt").read_text(encoding="utf-8") == "system prompt"
+    assert (out / "memory_provider_system_prompt.txt").read_text(encoding="utf-8").startswith("# Hindsight Memory")
+    assert (out / "memory_prefetch_raw.txt").read_text(encoding="utf-8") == "- remembered fact"
+    assert "<memory-context>" in (out / "memory_context_block.txt").read_text(encoding="utf-8")
+    assert "What do I prefer?" in (out / "api_user_message.txt").read_text(encoding="utf-8")
+    assert (out / "api_messages.json").read_text(encoding="utf-8").startswith("[")
+    assert '"hindsight"' in (out / "metadata.json").read_text(encoding="utf-8")

tests/plugins/memory/test_hindsight_provider.py

@@ -596,3 +596,26 @@ class TestAvailability:
         monkeypatch.setenv("HINDSIGHT_MODE", "local")
         p = HindsightMemoryProvider()
         assert p.is_available()
+
+    def test_local_embedded_profile_defaults_to_agent_identity(self, tmp_path, monkeypatch):
+        config_path = tmp_path / "hindsight" / "config.json"
+        config_path.parent.mkdir(parents=True, exist_ok=True)
+        config_path.write_text(json.dumps({
+            "mode": "local_embedded",
+            "llm_provider": "ollama",
+            "llm_model": "gemma3:12b",
+        }))
+        monkeypatch.setattr(
+            "plugins.memory.hindsight.get_hermes_home",
+            lambda: tmp_path,
+        )
+
+        p = HindsightMemoryProvider()
+        p.initialize(
+            session_id="test-session",
+            hermes_home=str(tmp_path / "profiles" / "atlas-hindsight"),
+            platform="cli",
+            agent_identity="atlas-hindsight",
+        )
+
+        assert p._config["profile"] == "atlas-hindsight"

tests/run_agent/test_run_agent_context_snapshots.py

@@ -0,0 +1,94 @@
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+import importlib
+import sys
+import types
+
+
+
+def _make_tool_defs(*names: str) -> list:
+    return [
+        {
+            "type": "function",
+            "function": {
+                "name": n,
+                "description": f"{n} tool",
+                "parameters": {"type": "object", "properties": {}},
+            },
+        }
+        for n in names
+    ]
+
+
+
+def _mock_response(content="Done", finish_reason="stop"):
+    msg = SimpleNamespace(content=content, tool_calls=None)
+    choice = SimpleNamespace(message=msg, finish_reason=finish_reason)
+    return SimpleNamespace(choices=[choice], usage=SimpleNamespace(prompt_tokens=1, completion_tokens=1, total_tokens=2))
+
+
+
+def _load_ai_agent():
+    sys.modules.setdefault("agent.auxiliary_client", types.SimpleNamespace(call_llm=lambda *a, **k: ""))
+    run_agent = importlib.import_module("run_agent")
+    return run_agent.AIAgent
+
+
+
+def test_run_conversation_writes_context_snapshot_artifacts(tmp_path):
+    AIAgent = _load_ai_agent()
+
+    class _FakePrivacyFilter:
+        def __init__(self):
+            self.last_report = None
+
+        def sanitize_messages(self, messages):
+            return list(messages)
+
+    with (
+        patch("run_agent.get_tool_definitions", return_value=_make_tool_defs("web_search")),
+        patch("run_agent.check_toolset_requirements", return_value={}),
+        patch("run_agent.OpenAI"),
+        patch("hermes_cli.plugins.invoke_hook", return_value=[]),
+        patch.dict(sys.modules, {"agent.privacy_filter": types.SimpleNamespace(PrivacyFilter=_FakePrivacyFilter)}),
+    ):
+        agent = AIAgent(
+            api_key="test-key-1234567890",
+            base_url="https://example.com/v1",
+            quiet_mode=True,
+            skip_context_files=True,
+            skip_memory=True,
+            context_snapshots_enabled=True,
+            context_snapshots_dir=str(tmp_path),
+        )
+
+    agent.client = MagicMock()
+    agent.client.chat.completions.create.return_value = _mock_response(content="Done")
+    agent._build_system_prompt = MagicMock(return_value="Core system prompt")
+    agent._memory_manager = MagicMock()
+    agent._memory_manager.prefetch_all.return_value = "- remembered preference"
+    agent._memory_manager.build_system_prompt.return_value = "# Hindsight Memory\nActive."
+    agent._memory_manager.providers = [
+        SimpleNamespace(name="builtin"),
+        SimpleNamespace(name="hindsight"),
+    ]
+
+    result = agent.run_conversation("What do I prefer?")
+
+    assert result["final_response"] == "Done"
+
+    call_dir = tmp_path / agent.session_id / "call_001"
+    assert call_dir.exists()
+    assert (call_dir / "system_prompt.txt").read_text(encoding="utf-8") == "Core system prompt"
+    assert (call_dir / "memory_provider_system_prompt.txt").read_text(encoding="utf-8").startswith("# Hindsight Memory")
+    assert (call_dir / "memory_prefetch_raw.txt").read_text(encoding="utf-8") == "- remembered preference"
+    assert "<memory-context>" in (call_dir / "memory_context_block.txt").read_text(encoding="utf-8")
+    api_user_message = (call_dir / "api_user_message.txt").read_text(encoding="utf-8")
+    assert "What do I prefer?" in api_user_message
+    assert "remembered preference" in api_user_message
+    api_messages = (call_dir / "api_messages.json").read_text(encoding="utf-8")
+    assert '"role": "system"' in api_messages
+    assert '"role": "user"' in api_messages
+    metadata = (call_dir / "metadata.json").read_text(encoding="utf-8")
+    assert '"hindsight"' in metadata

tests/test_hindsight_local_eval_doc.py

@@ -0,0 +1,22 @@
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+DOC = ROOT / "docs" / "hindsight-local-eval.md"
+
+
+def test_hindsight_local_eval_doc_exists_and_covers_switching():
+    assert DOC.exists(), "missing Hindsight local eval doc"
+    text = DOC.read_text(encoding="utf-8")
+    for snippet in (
+        "atlas-baseline",
+        "atlas-mempalace",
+        "atlas-hindsight",
+        "HERMES_HOME",
+        "HERMES_CONTEXT_SNAPSHOTS",
+        "memory_prefetch_raw.txt",
+        "api_user_message.txt",
+        "local_embedded",
+        "hindsight/config.json",
+    ):
+        assert snippet in text

tests/tools/test_fuzzy_match.py

@@ -148,184 +148,3 @@ 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

@@ -1,114 +0,0 @@
-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,14 +757,12 @@ class ShellFileOperations(FileOperations):
             content, old_string, new_string, replace_all
         )
         
-        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)
+        if error:
+            return PatchResult(error=error)
+        
+        if match_count == 0:
+            return PatchResult(error=f"Could not find match for old_string in {path}")
+        
         # Write back
         write_result = self.write_file(path, new_content)
         if write_result.error:

tools/file_tools.py

@@ -8,7 +8,6 @@ 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
@@ -691,11 +690,8 @@ 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"]):
-            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.]"
+            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,21 +93,6 @@ 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
@@ -116,46 +101,6 @@ 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.
@@ -619,86 +564,3 @@ 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,16 +290,10 @@ def _validate_operations(
                 )
                 if count == 0:
                     label = f"'{hunk.context_hint}'" if hunk.context_hint else "(no hint)"
-                    msg = (
+                    errors.append(
                         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.
@@ -543,13 +537,7 @@ def _apply_update(op: PatchOperation, file_ops: Any) -> Tuple[bool, str]:
                             error = None
                 
                 if 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
+                    return False, f"Could not apply hunk: {error}"
         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,15 +575,9 @@ 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": err_msg,
+            "error": match_error,
             "file_preview": preview,
         }