0ff36548e0
* feat(hermes): file-based config + updated docs Replace the old dataclass/configure() singleton with a plain dict config loaded from ~/.hindsight/hermes.json — same field names and conventions as the openclaw and claude-code integrations. Loading order: defaults → config file → env var overrides. - config.py: rewritten with load_config() returning a plain dict, DEFAULTS matching openclaw/claude-code fields, ENV_OVERRIDES with typed casting - tools.py: register() uses load_config() instead of raw env vars - __init__.py: clean exports (removed configure/get_config/reset_config) - README.md: full rewrite with config file examples, tables by category - docs/hermes.md: full rewrite with quick start, architecture, all config tables, gateway section, troubleshooting - tests: updated for new config pattern, 46 tests pass * ci: add test job for hermes integration * chore: regenerate docs skill for hermes integration
381 lines
15 KiB
Python
381 lines
15 KiB
Python
"""Tests for hindsight_hermes.tools module."""
|
|
|
|
import json
|
|
import sys
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from hindsight_hermes.config import DEFAULTS
|
|
from hindsight_hermes.errors import HindsightError
|
|
from hindsight_hermes.tools import (
|
|
RECALL_SCHEMA,
|
|
REFLECT_SCHEMA,
|
|
RETAIN_SCHEMA,
|
|
_resolve_bank_id,
|
|
_resolve_client,
|
|
get_tool_definitions,
|
|
memory_instructions,
|
|
register,
|
|
register_tools,
|
|
)
|
|
|
|
|
|
# --- Fixtures ---
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clean_env(monkeypatch):
|
|
"""Ensure no stale env vars leak between tests."""
|
|
for key in ("HINDSIGHT_API_URL", "HINDSIGHT_API_KEY", "HINDSIGHT_API_TOKEN",
|
|
"HINDSIGHT_BANK_ID", "HINDSIGHT_AUTO_RETAIN", "HINDSIGHT_RECALL_BUDGET"):
|
|
monkeypatch.delenv(key, raising=False)
|
|
|
|
|
|
@pytest.fixture()
|
|
def mock_client():
|
|
client = MagicMock()
|
|
# Sync methods (used by memory_instructions and register_tools sync path)
|
|
client.create_bank = MagicMock()
|
|
client.retain = MagicMock()
|
|
client.recall = MagicMock(
|
|
return_value=SimpleNamespace(
|
|
results=[
|
|
SimpleNamespace(text="Memory 1"),
|
|
SimpleNamespace(text="Memory 2"),
|
|
]
|
|
)
|
|
)
|
|
client.reflect = MagicMock(return_value=SimpleNamespace(text="Synthesized answer"))
|
|
# Async methods (used by tool handlers and hooks)
|
|
client.acreate_bank = AsyncMock()
|
|
client.aretain = AsyncMock()
|
|
client.arecall = AsyncMock(
|
|
return_value=SimpleNamespace(
|
|
results=[
|
|
SimpleNamespace(text="Memory 1"),
|
|
SimpleNamespace(text="Memory 2"),
|
|
]
|
|
)
|
|
)
|
|
client.areflect = AsyncMock(return_value=SimpleNamespace(text="Synthesized answer"))
|
|
return client
|
|
|
|
|
|
@pytest.fixture()
|
|
def mock_registry():
|
|
"""Patch tools.registry.registry so register_tools() can import it."""
|
|
mock_reg = MagicMock()
|
|
mock_module = MagicMock()
|
|
mock_module.registry = mock_reg
|
|
with patch.dict(sys.modules, {"tools": MagicMock(), "tools.registry": mock_module}):
|
|
yield mock_reg
|
|
|
|
|
|
# --- Schema tests ---
|
|
|
|
|
|
class TestSchemas:
|
|
def test_retain_schema_has_content(self):
|
|
assert RETAIN_SCHEMA["name"] == "hindsight_retain"
|
|
assert "content" in RETAIN_SCHEMA["parameters"]["properties"]
|
|
assert "content" in RETAIN_SCHEMA["parameters"]["required"]
|
|
|
|
def test_recall_schema_has_query(self):
|
|
assert RECALL_SCHEMA["name"] == "hindsight_recall"
|
|
assert "query" in RECALL_SCHEMA["parameters"]["properties"]
|
|
assert "query" in RECALL_SCHEMA["parameters"]["required"]
|
|
|
|
def test_reflect_schema_has_query(self):
|
|
assert REFLECT_SCHEMA["name"] == "hindsight_reflect"
|
|
assert "query" in REFLECT_SCHEMA["parameters"]["properties"]
|
|
|
|
def test_get_tool_definitions(self):
|
|
defs = get_tool_definitions()
|
|
assert len(defs) == 3
|
|
names = {d["name"] for d in defs}
|
|
assert names == {"hindsight_retain", "hindsight_recall", "hindsight_reflect"}
|
|
|
|
|
|
# --- Bank resolution tests ---
|
|
|
|
|
|
class TestResolveBankId:
|
|
def test_bank_resolver_takes_priority(self):
|
|
resolver = lambda args: "resolved-bank"
|
|
assert _resolve_bank_id({}, "static-bank", resolver) == "resolved-bank"
|
|
|
|
def test_static_bank_id(self):
|
|
assert _resolve_bank_id({}, "static-bank", None) == "static-bank"
|
|
|
|
def test_env_var_fallback(self, monkeypatch):
|
|
monkeypatch.setenv("HINDSIGHT_BANK_ID", "env-bank")
|
|
assert _resolve_bank_id({}, None, None) == "env-bank"
|
|
|
|
def test_raises_when_no_bank(self, monkeypatch):
|
|
monkeypatch.delenv("HINDSIGHT_BANK_ID", raising=False)
|
|
with pytest.raises(HindsightError, match="No bank_id available"):
|
|
_resolve_bank_id({}, None, None)
|
|
|
|
|
|
# --- Client resolution tests ---
|
|
|
|
|
|
class TestResolveClient:
|
|
def test_returns_provided_client(self, mock_client):
|
|
assert _resolve_client(mock_client, None, None) is mock_client
|
|
|
|
def test_creates_client_from_args(self):
|
|
with patch("hindsight_hermes.tools.Hindsight") as MockH:
|
|
_resolve_client(None, "http://localhost:9999", "key")
|
|
MockH.assert_called_once_with(base_url="http://localhost:9999", timeout=30.0, api_key="key")
|
|
|
|
def test_raises_without_url(self):
|
|
with pytest.raises(HindsightError, match="No Hindsight API URL"):
|
|
_resolve_client(None, None, None)
|
|
|
|
|
|
# --- register_tools tests ---
|
|
|
|
|
|
class TestRegisterTools:
|
|
def test_registers_three_tools(self, mock_client, mock_registry):
|
|
register_tools(bank_id="b", client=mock_client)
|
|
assert mock_registry.register.call_count == 3
|
|
names = {call.kwargs["name"] for call in mock_registry.register.call_args_list}
|
|
assert names == {"hindsight_retain", "hindsight_recall", "hindsight_reflect"}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_retain_handler_success(self, mock_client, mock_registry):
|
|
register_tools(bank_id="b", client=mock_client)
|
|
handler = mock_registry.register.call_args_list[0].kwargs["handler"]
|
|
result = json.loads(await handler({"content": "hello"}))
|
|
assert result["result"] == "Memory stored successfully."
|
|
mock_client.aretain.assert_called_once_with(bank_id="b", content="hello")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_retain_handler_with_tags(self, mock_client, mock_registry):
|
|
register_tools(bank_id="b", client=mock_client, tags=["tag1"])
|
|
handler = mock_registry.register.call_args_list[0].kwargs["handler"]
|
|
await handler({"content": "hello"})
|
|
mock_client.aretain.assert_called_once_with(bank_id="b", content="hello", tags=["tag1"])
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_recall_handler_success(self, mock_client, mock_registry):
|
|
register_tools(bank_id="b", client=mock_client)
|
|
handler = mock_registry.register.call_args_list[1].kwargs["handler"]
|
|
result = json.loads(await handler({"query": "test"}))
|
|
assert "Memory 1" in result["result"]
|
|
assert "Memory 2" in result["result"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_recall_handler_no_results(self, mock_client, mock_registry):
|
|
mock_client.arecall.return_value = SimpleNamespace(results=[])
|
|
register_tools(bank_id="b", client=mock_client)
|
|
handler = mock_registry.register.call_args_list[1].kwargs["handler"]
|
|
result = json.loads(await handler({"query": "test"}))
|
|
assert result["result"] == "No relevant memories found."
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reflect_handler_success(self, mock_client, mock_registry):
|
|
register_tools(bank_id="b", client=mock_client)
|
|
handler = mock_registry.register.call_args_list[2].kwargs["handler"]
|
|
result = json.loads(await handler({"query": "test"}))
|
|
assert result["result"] == "Synthesized answer"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handler_returns_error_on_exception(self, mock_client, mock_registry):
|
|
mock_client.aretain.side_effect = RuntimeError("boom")
|
|
register_tools(bank_id="b", client=mock_client)
|
|
handler = mock_registry.register.call_args_list[0].kwargs["handler"]
|
|
result = json.loads(await handler({"content": "hello"}))
|
|
assert "error" in result
|
|
assert "boom" in result["error"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_bank_called(self, mock_client, mock_registry):
|
|
register_tools(bank_id="b", client=mock_client)
|
|
handler = mock_registry.register.call_args_list[0].kwargs["handler"]
|
|
await handler({"content": "hello"})
|
|
mock_client.acreate_bank.assert_called_once_with(bank_id="b", name="b")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_bank_idempotent(self, mock_client, mock_registry):
|
|
register_tools(bank_id="b", client=mock_client)
|
|
handler = mock_registry.register.call_args_list[0].kwargs["handler"]
|
|
await handler({"content": "first"})
|
|
await handler({"content": "second"})
|
|
# acreate_bank should only be called once
|
|
mock_client.acreate_bank.assert_called_once()
|
|
|
|
|
|
# --- register (plugin entry point) tests ---
|
|
|
|
|
|
class TestRegisterPlugin:
|
|
def test_register_calls_ctx_register_tool(self, monkeypatch, mock_client):
|
|
monkeypatch.setenv("HINDSIGHT_API_URL", "http://localhost:8888")
|
|
monkeypatch.setenv("HINDSIGHT_BANK_ID", "test-bank")
|
|
ctx = MagicMock()
|
|
with patch("hindsight_hermes.tools._resolve_client", return_value=mock_client):
|
|
register(ctx)
|
|
assert ctx.register_tool.call_count == 3
|
|
|
|
def test_register_skips_without_config(self):
|
|
empty_cfg = dict(DEFAULTS) # no apiUrl, no apiToken
|
|
ctx = MagicMock()
|
|
with patch("hindsight_hermes.tools.load_config", return_value=empty_cfg):
|
|
register(ctx)
|
|
ctx.register_tool.assert_not_called()
|
|
|
|
def test_register_hooks(self, monkeypatch, mock_client):
|
|
monkeypatch.setenv("HINDSIGHT_API_URL", "http://localhost:8888")
|
|
monkeypatch.setenv("HINDSIGHT_BANK_ID", "test-bank")
|
|
ctx = MagicMock()
|
|
with patch("hindsight_hermes.tools._resolve_client", return_value=mock_client):
|
|
register(ctx)
|
|
hook_names = {call.args[0] for call in ctx.register_hook.call_args_list}
|
|
assert hook_names == {"pre_llm_call", "post_llm_call"}
|
|
|
|
|
|
# --- lifecycle hook tests ---
|
|
|
|
|
|
class TestLifecycleHooks:
|
|
"""Tests for pre_llm_call and post_llm_call hook callbacks."""
|
|
|
|
def _get_hook(self, ctx_mock, hook_name: str):
|
|
"""Extract the registered hook callback by name from the mock ctx."""
|
|
for call in ctx_mock.register_hook.call_args_list:
|
|
if call.args[0] == hook_name:
|
|
return call.args[1]
|
|
raise AssertionError(f"Hook {hook_name!r} not registered")
|
|
|
|
def _register_with_hooks(self, monkeypatch, mock_client, **env_overrides):
|
|
monkeypatch.setenv("HINDSIGHT_API_URL", "http://localhost:8888")
|
|
monkeypatch.setenv("HINDSIGHT_BANK_ID", "test-bank")
|
|
for k, v in env_overrides.items():
|
|
monkeypatch.setenv(k, v)
|
|
ctx = MagicMock()
|
|
with patch("hindsight_hermes.tools._resolve_client", return_value=mock_client):
|
|
register(ctx)
|
|
return ctx
|
|
|
|
# -- pre_llm_call --
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pre_llm_call_returns_context(self, monkeypatch, mock_client):
|
|
ctx = self._register_with_hooks(monkeypatch, mock_client)
|
|
hook = self._get_hook(ctx, "pre_llm_call")
|
|
result = await hook(
|
|
session_id="s1",
|
|
user_message="what color do I like?",
|
|
conversation_history=[],
|
|
is_first_turn=True,
|
|
model="test",
|
|
)
|
|
assert result is not None
|
|
assert "context" in result
|
|
assert "Memory 1" in result["context"]
|
|
assert "Memory 2" in result["context"]
|
|
mock_client.arecall.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pre_llm_call_returns_none_on_no_results(self, monkeypatch, mock_client):
|
|
mock_client.arecall.return_value = SimpleNamespace(results=[])
|
|
ctx = self._register_with_hooks(monkeypatch, mock_client)
|
|
hook = self._get_hook(ctx, "pre_llm_call")
|
|
result = await hook(user_message="hello")
|
|
assert result is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pre_llm_call_returns_none_on_empty_message(self, monkeypatch, mock_client):
|
|
ctx = self._register_with_hooks(monkeypatch, mock_client)
|
|
hook = self._get_hook(ctx, "pre_llm_call")
|
|
result = await hook(user_message="")
|
|
assert result is None
|
|
mock_client.arecall.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pre_llm_call_returns_none_on_error(self, monkeypatch, mock_client):
|
|
mock_client.arecall.side_effect = RuntimeError("connection failed")
|
|
ctx = self._register_with_hooks(monkeypatch, mock_client)
|
|
hook = self._get_hook(ctx, "pre_llm_call")
|
|
result = await hook(user_message="hello")
|
|
assert result is None
|
|
|
|
# -- post_llm_call --
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_post_llm_call_retains_turn(self, monkeypatch, mock_client):
|
|
ctx = self._register_with_hooks(monkeypatch, mock_client)
|
|
hook = self._get_hook(ctx, "post_llm_call")
|
|
await hook(
|
|
session_id="s1",
|
|
user_message="remember I like green",
|
|
assistant_response="Got it, you like green!",
|
|
model="test",
|
|
)
|
|
mock_client.aretain.assert_called_once()
|
|
content = mock_client.aretain.call_args.kwargs["content"]
|
|
assert "remember I like green" in content
|
|
assert "Got it, you like green!" in content
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_post_llm_call_skips_empty_messages(self, monkeypatch, mock_client):
|
|
ctx = self._register_with_hooks(monkeypatch, mock_client)
|
|
hook = self._get_hook(ctx, "post_llm_call")
|
|
await hook(user_message="", assistant_response="hello")
|
|
mock_client.aretain.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_post_llm_call_skips_when_disabled(self, monkeypatch, mock_client):
|
|
ctx = self._register_with_hooks(
|
|
monkeypatch, mock_client, HINDSIGHT_AUTO_RETAIN="false"
|
|
)
|
|
hook = self._get_hook(ctx, "post_llm_call")
|
|
await hook(user_message="hi", assistant_response="hello")
|
|
mock_client.aretain.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_post_llm_call_does_not_raise_on_error(self, monkeypatch, mock_client):
|
|
mock_client.aretain.side_effect = RuntimeError("boom")
|
|
ctx = self._register_with_hooks(monkeypatch, mock_client)
|
|
hook = self._get_hook(ctx, "post_llm_call")
|
|
# Should not raise
|
|
await hook(user_message="hi", assistant_response="hello")
|
|
|
|
|
|
# --- memory_instructions tests ---
|
|
|
|
|
|
class TestMemoryInstructions:
|
|
def test_returns_formatted_memories(self, mock_client):
|
|
result = memory_instructions(bank_id="b", client=mock_client)
|
|
assert "Relevant memories:" in result
|
|
assert "1. Memory 1" in result
|
|
assert "2. Memory 2" in result
|
|
|
|
def test_returns_empty_on_no_results(self, mock_client):
|
|
mock_client.recall.return_value = SimpleNamespace(results=[])
|
|
result = memory_instructions(bank_id="b", client=mock_client)
|
|
assert result == ""
|
|
|
|
def test_returns_empty_on_exception(self, mock_client):
|
|
mock_client.recall.side_effect = RuntimeError("fail")
|
|
result = memory_instructions(bank_id="b", client=mock_client)
|
|
assert result == ""
|
|
|
|
def test_returns_empty_on_no_client(self):
|
|
result = memory_instructions(bank_id="b")
|
|
assert result == ""
|
|
|
|
def test_respects_max_results(self, mock_client):
|
|
result = memory_instructions(bank_id="b", client=mock_client, max_results=1)
|
|
assert "1. Memory 1" in result
|
|
assert "Memory 2" not in result
|
|
|
|
def test_custom_prefix(self, mock_client):
|
|
result = memory_instructions(bank_id="b", client=mock_client, prefix="Context:\n")
|
|
assert result.startswith("Context:")
|