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

hermes_cli/tools_config.py

@@ -57,7 +57,7 @@ CONFIGURABLE_TOOLSETS = [
     ("moa",             "🧠 Mixture of Agents",         "mixture_of_agents"),
     ("tts",             "🔊 Text-to-Speech",            "text_to_speech"),
     ("skills",          "📚 Skills",                    "list, view, manage"),
-    ("todo",            "📋 Task Planning",             "todo, ultraplan"),
+    ("todo",            "📋 Task Planning",             "todo"),
     ("memory",          "💾 Memory",                    "persistent memory across sessions"),
     ("session_search",  "🔎 Session Search",            "search past conversations"),
     ("clarify",         "❓ Clarifying Questions",      "clarify"),

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_registry.py

@@ -294,32 +294,22 @@ class TestBuiltinDiscovery:
             "tools.browser_tool",
             "tools.clarify_tool",
             "tools.code_execution_tool",
-            "tools.crisis_tool",
             "tools.cronjob_tools",
             "tools.delegate_tool",
             "tools.file_tools",
             "tools.homeassistant_tool",
             "tools.image_generation_tool",
-            "tools.local_inference_tool",
             "tools.memory_tool",
             "tools.mixture_of_agents_tool",
             "tools.process_registry",
             "tools.rl_training_tool",
-            "tools.scavenger_fixer",
             "tools.send_message_tool",
             "tools.session_search_tool",
             "tools.skill_manager_tool",
             "tools.skills_tool",
-            "tools.sovereign_router",
-            "tools.sovereign_scavenger",
-            "tools.sovereign_teleport",
-            "tools.static_analyzer",
-            "tools.symbolic_verify",
             "tools.terminal_tool",
             "tools.todo_tool",
             "tools.tts_tool",
-            "tools.ultraplan",
-            "tools.verify_tool",
             "tools.vision_tools",
             "tools.web_tools",
         }

tests/tools/test_ultraplan_tool.py

@@ -1,81 +0,0 @@
-import json
-from pathlib import Path
-
-from toolsets import resolve_toolset
-from tools.registry import registry
-
-
-def test_create_action_saves_markdown_and_json(tmp_path):
-    from tools.ultraplan import ultraplan_tool
-
-    result = json.loads(
-        ultraplan_tool(
-            action="create",
-            mission="Daily autonomous planning",
-            streams=[
-                {
-                    "id": "A",
-                    "name": "Backlog burn",
-                    "phases": [
-                        {"id": "A1", "name": "Triage", "artifact": "issue list"},
-                        {"id": "A2", "name": "Ship", "dependencies": ["A1"], "artifact": "PR"},
-                    ],
-                }
-            ],
-            base_dir=str(tmp_path),
-        )
-    )
-
-    assert result["success"] is True
-    assert Path(result["file_path"]).exists()
-    assert Path(result["json_path"]).exists()
-    assert "Work Streams" in Path(result["file_path"]).read_text(encoding="utf-8")
-
-
-def test_load_action_returns_saved_plan(tmp_path):
-    from tools.ultraplan import ultraplan_tool
-
-    created = json.loads(
-        ultraplan_tool(
-            action="create",
-            date="20260422",
-            mission="Mission from saved plan",
-            base_dir=str(tmp_path),
-        )
-    )
-    loaded = json.loads(
-        ultraplan_tool(
-            action="load",
-            date="20260422",
-            base_dir=str(tmp_path),
-        )
-    )
-
-    assert created["success"] is True
-    assert loaded["success"] is True
-    assert loaded["plan"]["mission"] == "Mission from saved plan"
-    assert loaded["file_path"].endswith("ultraplan_20260422.md")
-
-
-def test_cron_spec_returns_daily_schedule_and_prompt():
-    from tools.ultraplan import ultraplan_tool
-
-    result = json.loads(ultraplan_tool(action="cron_spec"))
-
-    assert result["success"] is True
-    assert result["schedule"] == "0 6 * * *"
-    assert "Ultraplan" in result["prompt"]
-    assert "ultraplan_YYYYMMDD.md" in result["prompt"]
-
-
-def test_registry_registers_ultraplan_tool():
-    import tools.ultraplan  # noqa: F401
-
-    entry = registry.get_entry("ultraplan")
-    assert entry is not None
-    assert entry.toolset == "todo"
-
-
-def test_default_toolsets_include_ultraplan():
-    assert "ultraplan" in resolve_toolset("todo")
-    assert "ultraplan" in resolve_toolset("hermes-cli")

tools/ultraplan.py

@@ -290,9 +290,6 @@ def load_ultraplan(date: str, base_dir: Path = None) -> Optional[Ultraplan]:
         return None
 
 
-DEFAULT_ULTRAPLAN_SCHEDULE = "0 6 * * *"
-
-
 def generate_daily_cron_prompt() -> str:
     """Generate the prompt for the daily ultraplan cron job."""
     return """Generate today's Ultraplan.
@@ -301,9 +298,9 @@ Steps:
 1. Check open Gitea issues assigned to you
 2. Check open PRs needing review
 3. Check fleet health status
-4. Decompose work into parallel streams with concrete phases and artifacts
-5. Use the ultraplan tool to save ~/.timmy/cron/ultraplan_YYYYMMDD.md and the matching JSON sidecar
-6. Optionally file a Gitea issue with the plan summary
+4. Decompose work into parallel streams
+5. Generate ultraplan_YYYYMMDD.md
+6. File Gitea issue with the plan
 
 Output format:
 - Mission statement
@@ -311,176 +308,3 @@ Output format:
 - Dependency map
 - Success metrics
 """
-
-
-def generate_daily_cron_job_spec(schedule: str = DEFAULT_ULTRAPLAN_SCHEDULE) -> Dict[str, str]:
-    """Return a reusable cron job spec for daily Ultraplan generation."""
-    return {
-        "name": "Daily Ultraplan",
-        "schedule": schedule,
-        "prompt": generate_daily_cron_prompt(),
-        "path_pattern": "~/.timmy/cron/ultraplan_YYYYMMDD.md",
-    }
-
-
-def _resolve_base_dir(base_dir: Optional[str | Path]) -> Path:
-    """Normalize the requested Ultraplan base directory."""
-    if base_dir is None:
-        return Path.home() / ".timmy" / "cron"
-    return Path(base_dir).expanduser()
-
-
-def ultraplan_tool(
-    action: str,
-    date: Optional[str] = None,
-    mission: str = "",
-    streams: Optional[List[Dict[str, Any]]] = None,
-    metrics: Optional[Dict[str, Any]] = None,
-    notes: str = "",
-    base_dir: Optional[str] = None,
-) -> str:
-    """Create/load Ultraplan artifacts and expose a daily cron spec."""
-    from tools.registry import tool_error, tool_result
-
-    action = (action or "").strip().lower()
-    resolved_base_dir = _resolve_base_dir(base_dir)
-
-    try:
-        if action == "create":
-            plan = create_ultraplan(date=date, mission=mission, streams=streams or [])
-            if metrics:
-                plan.metrics = metrics
-            if notes:
-                plan.notes = notes
-            md_path = save_ultraplan(plan, base_dir=resolved_base_dir)
-            json_path = resolved_base_dir / f"ultraplan_{plan.date}.json"
-            return tool_result(
-                success=True,
-                action="create",
-                date=plan.date,
-                file_path=str(md_path),
-                json_path=str(json_path),
-                plan=plan.to_dict(),
-            )
-
-        if action == "load":
-            plan_date = date or datetime.now().strftime("%Y%m%d")
-            plan = load_ultraplan(plan_date, base_dir=resolved_base_dir)
-            if plan is None:
-                return tool_error(
-                    f"No Ultraplan found for {plan_date}",
-                    success=False,
-                    action="load",
-                    date=plan_date,
-                )
-            return tool_result(
-                success=True,
-                action="load",
-                date=plan.date,
-                file_path=str(resolved_base_dir / f"ultraplan_{plan.date}.md"),
-                json_path=str(resolved_base_dir / f"ultraplan_{plan.date}.json"),
-                plan=plan.to_dict(),
-                markdown=plan.to_markdown(),
-            )
-
-        if action == "cron_spec":
-            spec = generate_daily_cron_job_spec()
-            return tool_result(success=True, action="cron_spec", **spec)
-
-        return tool_error(
-            f"Unknown Ultraplan action: {action}",
-            success=False,
-            action=action,
-        )
-    except Exception as e:
-        return tool_error(f"Ultraplan {action or 'tool'} failed: {e}", success=False, action=action)
-
-
-ULTRAPLAN_SCHEMA = {
-    "name": "ultraplan",
-    "description": (
-        "Create or load daily Ultraplan planning artifacts under ~/.timmy/cron/ and "
-        "return a reusable cron spec for autonomous planning. Use this when you want "
-        "a concrete markdown/json plan file with streams, phases, dependencies, and metrics."
-    ),
-    "parameters": {
-        "type": "object",
-        "properties": {
-            "action": {
-                "type": "string",
-                "enum": ["create", "load", "cron_spec"],
-                "description": "Operation to perform",
-            },
-            "date": {
-                "type": "string",
-                "description": "Plan date as YYYYMMDD. Defaults to today for create/load.",
-            },
-            "mission": {
-                "type": "string",
-                "description": "High-level mission statement for today's plan.",
-            },
-            "streams": {
-                "type": "array",
-                "description": "Optional work streams with phases/artifacts/dependencies for create.",
-                "items": {
-                    "type": "object",
-                    "properties": {
-                        "id": {"type": "string"},
-                        "name": {"type": "string"},
-                        "phases": {
-                            "type": "array",
-                            "items": {
-                                "type": "object",
-                                "properties": {
-                                    "id": {"type": "string"},
-                                    "name": {"type": "string"},
-                                    "description": {"type": "string"},
-                                    "artifact": {"type": "string"},
-                                    "dependencies": {
-                                        "type": "array",
-                                        "items": {"type": "string"},
-                                    },
-                                },
-                                "required": ["name"],
-                            },
-                        },
-                    },
-                    "required": ["name"],
-                },
-            },
-            "metrics": {
-                "type": "object",
-                "description": "Optional success metrics to store on the plan.",
-                "additionalProperties": True,
-            },
-            "notes": {
-                "type": "string",
-                "description": "Optional free-form notes appended to the saved plan.",
-            },
-            "base_dir": {
-                "type": "string",
-                "description": "Optional override for the Ultraplan storage directory.",
-            },
-        },
-        "required": ["action"],
-    },
-}
-
-
-from tools.registry import registry
-
-registry.register(
-    name="ultraplan",
-    toolset="todo",
-    schema=ULTRAPLAN_SCHEMA,
-    handler=lambda args, **_kw: ultraplan_tool(
-        action=args.get("action", ""),
-        date=args.get("date"),
-        mission=args.get("mission", ""),
-        streams=args.get("streams"),
-        metrics=args.get("metrics"),
-        notes=args.get("notes", ""),
-        base_dir=args.get("base_dir"),
-    ),
-    emoji="🗺️",
-)

toolsets.py

@@ -47,7 +47,7 @@ _HERMES_CORE_TOOLS = [
     # Text-to-speech
     "text_to_speech",
     # Planning & memory
-    "todo", "ultraplan", "memory",
+    "todo", "memory",
     # Session history search
     "session_search",
     # Clarifying questions
@@ -157,8 +157,8 @@ TOOLSETS = {
     },
     
     "todo": {
-        "description": "Task planning and tracking for multi-step work, including daily Ultraplan artifacts",
-        "tools": ["todo", "ultraplan"],
+        "description": "Task planning and tracking for multi-step work",
+        "tools": ["todo"],
         "includes": []
     },