feat: add profile-scoped local Hindsight eval path

Closes #1010

agent/auxiliary_client.py

@@ -1,4 +1,4 @@
-"""Shared auxiliary client router for side tasks.
+from agent.telemetry_logger import log_token_usage\n"""Shared auxiliary client router for side tasks.
 
 Provides a single resolution chain so every consumer (context compression,
 session search, web extraction, vision analysis, browser vision) picks up
@@ -34,8 +34,6 @@ Payment / credit exhaustion fallback:
   their OpenRouter balance but has Codex OAuth or another provider available.
 """
 
-from agent.telemetry_logger import log_token_usage
-
 import json
 import logging
 import os
@@ -398,8 +396,7 @@ class _CodexCompletionsAdapter:
                     prompt_tokens=getattr(resp_usage, "input_tokens", 0),
                     completion_tokens=getattr(resp_usage, "output_tokens", 0),
                     total_tokens=getattr(resp_usage, "total_tokens", 0),
-                )
-                log_token_usage(usage.prompt_tokens, usage.completion_tokens, model)
+                )\n        log_token_usage(usage.prompt_tokens, usage.completion_tokens, model)
         except Exception as exc:
             logger.debug("Codex auxiliary Responses API call failed: %s", exc)
             raise
@@ -532,8 +529,7 @@ class _AnthropicCompletionsAdapter:
                 prompt_tokens=prompt_tokens,
                 completion_tokens=completion_tokens,
                 total_tokens=total_tokens,
-            )
-            log_token_usage(usage.prompt_tokens, usage.completion_tokens, model)
+            )\n    log_token_usage(usage.prompt_tokens, usage.completion_tokens, model)
 
         choice = SimpleNamespace(
             index=0,

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")

cli.py

@@ -7254,40 +7254,6 @@ class HermesCLI:
             "Use your best judgement to make the choice and proceed."
         )
 
-    def _handle_clarify_selection(self) -> None:
-        """Process the currently selected clarify choice."""
-        state = self._clarify_state
-        if not state or self._clarify_freetext:
-            return
-
-        selected = state.get("selected", 0)
-        choices = state.get("choices") or []
-        if selected < len(choices):
-            state["response_queue"].put(choices[selected])
-            self._clarify_state = None
-            self._clarify_freetext = False
-            self._invalidate()
-            return
-
-        if selected == len(choices):
-            self._clarify_freetext = True
-            self._invalidate()
-
-    def _handle_clarify_number_shortcut(self, number: int) -> bool:
-        """Select a clarify option by number key."""
-        state = self._clarify_state
-        if not state or self._clarify_freetext:
-            return False
-
-        choices = state.get("choices") or []
-        max_option = len(choices) + 1
-        if number < 1 or number > max_option:
-            return False
-
-        state["selected"] = number - 1
-        self._handle_clarify_selection()
-        return True
-
     def _sudo_password_callback(self) -> str:
         """
         Prompt for sudo password through the prompt_toolkit UI.
@@ -7396,20 +7362,6 @@ class HermesCLI:
             choices.append("view")
         return choices
 
-    def _handle_approval_number_shortcut(self, number: int) -> bool:
-        """Select an approval option by number key."""
-        state = self._approval_state
-        if not state:
-            return False
-
-        choices = state.get("choices") or []
-        if number < 1 or number > len(choices):
-            return False
-
-        state["selected"] = number - 1
-        self._handle_approval_selection()
-        return True
-
     def _handle_approval_selection(self) -> None:
         """Process the currently selected dangerous-command approval choice."""
         state = self._approval_state
@@ -7485,9 +7437,8 @@ class HermesCLI:
         preview_lines.extend(_wrap_panel_text(cmd_display, 60))
         for i, choice in enumerate(choices):
             prefix = '❯ ' if i == selected else '  '
-            label = f"{i + 1}. {choice_labels.get(choice, choice)}"
             preview_lines.extend(_wrap_panel_text(
-                f"{prefix}{label}",
+                f"{prefix}{choice_labels.get(choice, choice)}",
                 60,
                 subsequent_indent="  ",
             ))
@@ -7505,7 +7456,7 @@ class HermesCLI:
             _append_panel_line(lines, 'class:approval-border', 'class:approval-cmd', wrapped, box_width)
         _append_blank_panel_line(lines, 'class:approval-border', box_width)
         for i, choice in enumerate(choices):
-            label = f"{i + 1}. {choice_labels.get(choice, choice)}"
+            label = choice_labels.get(choice, choice)
             style = 'class:approval-selected' if i == selected else 'class:approval-choice'
             prefix = '❯ ' if i == selected else '  '
             for wrapped in _wrap_panel_text(f"{prefix}{label}", inner_text_width, subsequent_indent="  "):
@@ -7514,97 +7465,6 @@ class HermesCLI:
         lines.append(('class:approval-border', '╰' + ('─' * box_width) + '╯\n'))
         return lines
 
-    def _get_clarify_display_fragments(self):
-        """Render the clarify panel for the prompt_toolkit UI."""
-        state = self._clarify_state
-        if not state:
-            return []
-
-        def _panel_box_width(title: str, content_lines: list[str], min_width: int = 46, max_width: int = 76) -> int:
-            term_cols = shutil.get_terminal_size((100, 20)).columns
-            longest = max([len(title)] + [len(line) for line in content_lines] + [min_width - 4])
-            inner = min(max(longest + 4, min_width - 2), max_width - 2, max(24, term_cols - 6))
-            return inner + 2
-
-        def _wrap_panel_text(text: str, width: int, subsequent_indent: str = "") -> list[str]:
-            wrapped = textwrap.wrap(
-                text,
-                width=max(8, width),
-                break_long_words=False,
-                break_on_hyphens=False,
-                subsequent_indent=subsequent_indent,
-            )
-            return wrapped or [""]
-
-        def _append_panel_line(lines, border_style: str, content_style: str, text: str, box_width: int) -> None:
-            inner_width = max(0, box_width - 2)
-            lines.append((border_style, "│ "))
-            lines.append((content_style, text.ljust(inner_width)))
-            lines.append((border_style, " │\n"))
-
-        def _append_blank_panel_line(lines, border_style: str, box_width: int) -> None:
-            lines.append((border_style, "│" + (" " * box_width) + "│\n"))
-
-        question = state["question"]
-        choices = state.get("choices") or []
-        selected = state.get("selected", 0)
-        preview_lines = _wrap_panel_text(question, 60)
-        for i, choice in enumerate(choices):
-            prefix = "❯ " if i == selected and not self._clarify_freetext else "  "
-            label = f"{i + 1}. {choice}"
-            preview_lines.extend(_wrap_panel_text(f"{prefix}{label}", 60, subsequent_indent="  "))
-        other_number = len(choices) + 1
-        other_label = (
-            f"❯ {other_number}. Other (type below)" if self._clarify_freetext
-            else f"❯ {other_number}. Other (type your answer)" if selected == len(choices)
-            else f"  {other_number}. Other (type your answer)"
-        )
-        preview_lines.extend(_wrap_panel_text(other_label, 60, subsequent_indent="  "))
-        box_width = _panel_box_width("Hermes needs your input", preview_lines)
-        inner_text_width = max(8, box_width - 2)
-
-        lines = []
-        lines.append(('class:clarify-border', '╭─ '))
-        lines.append(('class:clarify-title', 'Hermes needs your input'))
-        lines.append(('class:clarify-border', ' ' + ('─' * max(0, box_width - len("Hermes needs your input") - 3)) + '╮\n'))
-        _append_blank_panel_line(lines, 'class:clarify-border', box_width)
-
-        for wrapped in _wrap_panel_text(question, inner_text_width):
-            _append_panel_line(lines, 'class:clarify-border', 'class:clarify-question', wrapped, box_width)
-        _append_blank_panel_line(lines, 'class:clarify-border', box_width)
-
-        if self._clarify_freetext and not choices:
-            guidance = "Type your answer in the prompt below, then press Enter."
-            for wrapped in _wrap_panel_text(guidance, inner_text_width):
-                _append_panel_line(lines, 'class:clarify-border', 'class:clarify-choice', wrapped, box_width)
-            _append_blank_panel_line(lines, 'class:clarify-border', box_width)
-
-        if choices:
-            for i, choice in enumerate(choices):
-                style = 'class:clarify-selected' if i == selected and not self._clarify_freetext else 'class:clarify-choice'
-                prefix = '❯ ' if i == selected and not self._clarify_freetext else '  '
-                label = f"{i + 1}. {choice}"
-                wrapped_lines = _wrap_panel_text(f"{prefix}{label}", inner_text_width, subsequent_indent="  ")
-                for wrapped in wrapped_lines:
-                    _append_panel_line(lines, 'class:clarify-border', style, wrapped, box_width)
-
-            other_idx = len(choices)
-            if selected == other_idx and not self._clarify_freetext:
-                other_style = 'class:clarify-selected'
-                other_label = f'❯ {other_number}. Other (type your answer)'
-            elif self._clarify_freetext:
-                other_style = 'class:clarify-active-other'
-                other_label = f'❯ {other_number}. Other (type below)'
-            else:
-                other_style = 'class:clarify-choice'
-                other_label = f'  {other_number}. Other (type your answer)'
-            for wrapped in _wrap_panel_text(other_label, inner_text_width, subsequent_indent="  "):
-                _append_panel_line(lines, 'class:clarify-border', other_style, wrapped, box_width)
-
-        _append_blank_panel_line(lines, 'class:clarify-border', box_width)
-        lines.append(('class:clarify-border', '╰' + ('─' * box_width) + '╯\n'))
-        return lines
-
     def _secret_capture_callback(self, var_name: str, prompt: str, metadata=None) -> dict:
         return prompt_for_secret(self, var_name, prompt, metadata)
 
@@ -8511,8 +8371,17 @@ class HermesCLI:
 
             # --- Clarify choice mode: confirm the highlighted selection ---
             if self._clarify_state and not self._clarify_freetext:
-                self._handle_clarify_selection()
-                event.app.invalidate()
+                state = self._clarify_state
+                selected = state["selected"]
+                choices = state.get("choices") or []
+                if selected < len(choices):
+                    state["response_queue"].put(choices[selected])
+                    self._clarify_state = None
+                    event.app.invalidate()
+                else:
+                    # "Other" selected → switch to freetext
+                    self._clarify_freetext = True
+                    event.app.invalidate()
                 return
 
             # --- Normal input routing ---
@@ -8632,19 +8501,6 @@ class HermesCLI:
                 self._approval_state["selected"] = min(max_idx, self._approval_state["selected"] + 1)
                 event.app.invalidate()
 
-        # --- Numbered shortcuts for clarify / approval modal prompts ---
-        for _digit in '123456789':
-            @kb.add(_digit, filter=Condition(lambda: bool(self._approval_state) or (bool(self._clarify_state) and not self._clarify_freetext)))
-            def _handle_modal_number(event, digit=_digit):
-                number = int(digit)
-                handled = False
-                if self._approval_state:
-                    handled = self._handle_approval_number_shortcut(number)
-                elif self._clarify_state and not self._clarify_freetext:
-                    handled = self._handle_clarify_number_shortcut(number)
-                if handled:
-                    event.app.invalidate()
-
         # --- /model picker: arrow-key navigation ---
         @kb.add('up', filter=Condition(lambda: bool(self._model_picker_state)))
         def model_picker_up(event):
@@ -9139,7 +8995,7 @@ class HermesCLI:
             if cli_ref._approval_state:
                 remaining = max(0, int(cli_ref._approval_deadline - _time.monotonic()))
                 return [
-                    ('class:hint', '  1-9 or ↑/↓ to select, Enter to confirm'),
+                    ('class:hint', '  ↑/↓ to select, Enter to confirm'),
                     ('class:clarify-countdown', f'  ({remaining}s)'),
                 ]
 
@@ -9152,7 +9008,7 @@ class HermesCLI:
                         ('class:clarify-countdown', countdown),
                     ]
                 return [
-                    ('class:hint', '  1-9 or ↑/↓ to select, Enter to confirm'),
+                    ('class:hint', '  ↑/↓ to select, Enter to confirm'),
                     ('class:clarify-countdown', countdown),
                 ]
 
@@ -9230,7 +9086,71 @@ class HermesCLI:
             lines.append((border_style, "│" + (" " * box_width) + "│\n"))
 
         def _get_clarify_display():
-            return cli_ref._get_clarify_display_fragments()
+            """Build styled text for the clarify question/choices panel."""
+            state = cli_ref._clarify_state
+            if not state:
+                return []
+
+            question = state["question"]
+            choices = state.get("choices") or []
+            selected = state.get("selected", 0)
+            preview_lines = _wrap_panel_text(question, 60)
+            for i, choice in enumerate(choices):
+                prefix = "❯ " if i == selected and not cli_ref._clarify_freetext else "  "
+                preview_lines.extend(_wrap_panel_text(f"{prefix}{choice}", 60, subsequent_indent="  "))
+            other_label = (
+                "❯ Other (type below)" if cli_ref._clarify_freetext
+                else "❯ Other (type your answer)" if selected == len(choices)
+                else "  Other (type your answer)"
+            )
+            preview_lines.extend(_wrap_panel_text(other_label, 60, subsequent_indent="  "))
+            box_width = _panel_box_width("Hermes needs your input", preview_lines)
+            inner_text_width = max(8, box_width - 2)
+
+            lines = []
+            # Box top border
+            lines.append(('class:clarify-border', '╭─ '))
+            lines.append(('class:clarify-title', 'Hermes needs your input'))
+            lines.append(('class:clarify-border', ' ' + ('─' * max(0, box_width - len("Hermes needs your input") - 3)) + '╮\n'))
+            _append_blank_panel_line(lines, 'class:clarify-border', box_width)
+
+            # Question text
+            for wrapped in _wrap_panel_text(question, inner_text_width):
+                _append_panel_line(lines, 'class:clarify-border', 'class:clarify-question', wrapped, box_width)
+            _append_blank_panel_line(lines, 'class:clarify-border', box_width)
+
+            if cli_ref._clarify_freetext and not choices:
+                guidance = "Type your answer in the prompt below, then press Enter."
+                for wrapped in _wrap_panel_text(guidance, inner_text_width):
+                    _append_panel_line(lines, 'class:clarify-border', 'class:clarify-choice', wrapped, box_width)
+                _append_blank_panel_line(lines, 'class:clarify-border', box_width)
+
+            if choices:
+                # Multiple-choice mode: show selectable options
+                for i, choice in enumerate(choices):
+                    style = 'class:clarify-selected' if i == selected and not cli_ref._clarify_freetext else 'class:clarify-choice'
+                    prefix = '❯ ' if i == selected and not cli_ref._clarify_freetext else '  '
+                    wrapped_lines = _wrap_panel_text(f"{prefix}{choice}", inner_text_width, subsequent_indent="  ")
+                    for wrapped in wrapped_lines:
+                        _append_panel_line(lines, 'class:clarify-border', style, wrapped, box_width)
+
+                # "Other" option (5th line, only shown when choices exist)
+                other_idx = len(choices)
+                if selected == other_idx and not cli_ref._clarify_freetext:
+                    other_style = 'class:clarify-selected'
+                    other_label = '❯ Other (type your answer)'
+                elif cli_ref._clarify_freetext:
+                    other_style = 'class:clarify-active-other'
+                    other_label = '❯ Other (type below)'
+                else:
+                    other_style = 'class:clarify-choice'
+                    other_label = '  Other (type your answer)'
+                for wrapped in _wrap_panel_text(other_label, inner_text_width, subsequent_indent="  "):
+                    _append_panel_line(lines, 'class:clarify-border', other_style, wrapped, box_width)
+
+            _append_blank_panel_line(lines, 'class:clarify-border', box_width)
+            lines.append(('class:clarify-border', '╰' + ('─' * box_width) + '╯\n'))
+            return lines
 
         clarify_widget = ConditionalContainer(
             Window(

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/cli/test_cli_numbered_prompt_shortcuts.py

@@ -1,172 +0,0 @@
-import queue
-import threading
-from types import SimpleNamespace
-from unittest.mock import MagicMock
-
-from cli import HermesCLI
-
-
-class _FakeBuffer:
-    def __init__(self, text="", cursor_position=None):
-        self.text = text
-        self.cursor_position = len(text) if cursor_position is None else cursor_position
-
-    def reset(self, append_to_history=False):
-        self.text = ""
-        self.cursor_position = 0
-
-
-def _make_cli_stub():
-    cli = HermesCLI.__new__(HermesCLI)
-    cli._approval_state = None
-    cli._approval_deadline = 0
-    cli._approval_lock = threading.Lock()
-    cli._clarify_state = None
-    cli._clarify_freetext = False
-    cli._clarify_deadline = 0
-    cli._sudo_state = None
-    cli._sudo_deadline = 0
-    cli._secret_state = None
-    cli._secret_deadline = 0
-    cli._modal_input_snapshot = None
-    cli._invalidate = MagicMock()
-    cli._app = SimpleNamespace(invalidate=MagicMock(), current_buffer=_FakeBuffer())
-    return cli
-
-
-def test_approval_display_numbers_choices():
-    cli = _make_cli_stub()
-    cli._approval_state = {
-        "command": "sudo rm -rf /tmp/example",
-        "description": "dangerous command",
-        "choices": ["once", "session", "always", "deny"],
-        "selected": 0,
-        "response_queue": queue.Queue(),
-    }
-
-    rendered = "".join(text for _style, text in cli._get_approval_display_fragments())
-
-    assert "❯ 1. Allow once" in rendered
-    assert "2. Allow for this session" in rendered
-    assert "3. Add to permanent allowlist" in rendered
-    assert "4. Deny" in rendered
-
-
-def test_approval_number_shortcut_submits_choice():
-    cli = _make_cli_stub()
-    response_queue = queue.Queue()
-    cli._approval_state = {
-        "command": "sudo rm -rf /tmp/example",
-        "description": "dangerous command",
-        "choices": ["once", "session", "always", "deny"],
-        "selected": 0,
-        "response_queue": response_queue,
-    }
-
-    assert cli._handle_approval_number_shortcut(2) is True
-    assert response_queue.get_nowait() == "session"
-    assert cli._approval_state is None
-
-
-def test_approval_selection_still_submits_selected_choice():
-    cli = _make_cli_stub()
-    response_queue = queue.Queue()
-    cli._approval_state = {
-        "command": "sudo rm -rf /tmp/example",
-        "description": "dangerous command",
-        "choices": ["once", "session", "always", "deny"],
-        "selected": 1,
-        "response_queue": response_queue,
-    }
-
-    cli._handle_approval_selection()
-
-    assert response_queue.get_nowait() == "session"
-    assert cli._approval_state is None
-
-
-def test_approval_number_shortcut_handles_view_in_place():
-    cli = _make_cli_stub()
-    response_queue = queue.Queue()
-    cli._approval_state = {
-        "command": "sudo dd if=/tmp/in of=/usr/share/keyrings/githubcli-archive-keyring.gpg bs=4M status=progress",
-        "description": "disk copy",
-        "choices": ["once", "session", "always", "deny", "view"],
-        "selected": 0,
-        "response_queue": response_queue,
-    }
-
-    assert cli._handle_approval_number_shortcut(5) is True
-    assert cli._approval_state is not None
-    assert cli._approval_state["show_full"] is True
-    assert "view" not in cli._approval_state["choices"]
-    assert cli._approval_state["selected"] == 3
-    assert response_queue.empty()
-
-
-def test_clarify_display_numbers_choices_and_other():
-    cli = _make_cli_stub()
-    cli._clarify_state = {
-        "question": "Pick the best option",
-        "choices": ["Alpha", "Beta", "Gamma", "Delta"],
-        "selected": 1,
-        "response_queue": queue.Queue(),
-    }
-
-    rendered = "".join(text for _style, text in cli._get_clarify_display_fragments())
-
-    assert "1. Alpha" in rendered
-    assert "❯ 2. Beta" in rendered
-    assert "3. Gamma" in rendered
-    assert "4. Delta" in rendered
-    assert "5. Other (type your answer)" in rendered
-
-
-def test_clarify_number_shortcut_submits_choice():
-    cli = _make_cli_stub()
-    response_queue = queue.Queue()
-    cli._clarify_state = {
-        "question": "Pick the best option",
-        "choices": ["Alpha", "Beta", "Gamma"],
-        "selected": 0,
-        "response_queue": response_queue,
-    }
-
-    assert cli._handle_clarify_number_shortcut(3) is True
-    assert response_queue.get_nowait() == "Gamma"
-    assert cli._clarify_state is None
-    assert cli._clarify_freetext is False
-
-
-def test_clarify_selection_still_submits_selected_choice():
-    cli = _make_cli_stub()
-    response_queue = queue.Queue()
-    cli._clarify_state = {
-        "question": "Pick the best option",
-        "choices": ["Alpha", "Beta", "Gamma"],
-        "selected": 1,
-        "response_queue": response_queue,
-    }
-
-    cli._handle_clarify_selection()
-
-    assert response_queue.get_nowait() == "Beta"
-    assert cli._clarify_state is None
-    assert cli._clarify_freetext is False
-
-
-def test_clarify_number_shortcut_activates_other_freetext():
-    cli = _make_cli_stub()
-    response_queue = queue.Queue()
-    cli._clarify_state = {
-        "question": "Pick the best option",
-        "choices": ["Alpha", "Beta", "Gamma"],
-        "selected": 0,
-        "response_queue": response_queue,
-    }
-
-    assert cli._handle_clarify_number_shortcut(4) is True
-    assert cli._clarify_state is not None
-    assert cli._clarify_state["selected"] == 3
-    assert cli._clarify_freetext is True
-    assert response_queue.empty()

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