fix: make session search initialize session db

This commit is contained in:
HenkDz 2026-05-03 17:43:53 +01:00 committed by Teknium
parent 9c26297c80
commit 840ebe063e
8 changed files with 220 additions and 11 deletions

View file

@ -1,4 +1,5 @@
from types import SimpleNamespace
import sys
from types import ModuleType, SimpleNamespace
import pytest
from acp.schema import TextContentBlock
@ -66,6 +67,53 @@ def make_agent_and_state():
return acp_agent, state, fake, conn
def test_acp_real_agent_gets_session_db_for_recall(monkeypatch):
"""ACP sessions persist to SessionDB; recall must receive the same DB handle."""
captured = {}
sentinel_db = NoopDb()
class CapturingAgent(FakeAgent):
def __init__(self, **kwargs):
super().__init__()
captured.update(kwargs)
def mod(name, **attrs):
module = ModuleType(name)
for key, value in attrs.items():
setattr(module, key, value)
return module
monkeypatch.setitem(sys.modules, "run_agent", mod("run_agent", AIAgent=CapturingAgent))
monkeypatch.setitem(
sys.modules,
"hermes_cli.config",
mod("hermes_cli.config", load_config=lambda: {"model": {"default": "m", "provider": "p"}}),
)
monkeypatch.setitem(
sys.modules,
"hermes_cli.runtime_provider",
mod(
"hermes_cli.runtime_provider",
resolve_runtime_provider=lambda **_kwargs: {
"provider": "p",
"api_mode": "chat_completions",
"base_url": "u",
"api_key": "k",
"command": None,
"args": [],
},
),
)
manager = SessionManager(db=sentinel_db)
agent = manager._make_agent(session_id="acp-session", cwd=".")
assert isinstance(agent, CapturingAgent)
assert captured["session_db"] is sentinel_db
assert captured["platform"] == "acp"
assert captured["session_id"] == "acp-session"
@pytest.mark.asyncio
async def test_acp_steer_slash_command_injects_into_running_agent():
acp_agent, state, fake, _conn = make_agent_and_state()

View file

@ -419,6 +419,72 @@ def test_oneshot_distinguishes_disabled_mcp_from_unknown(monkeypatch, capsys):
assert "mcp-off" in err
def test_oneshot_wires_session_db_for_recall(monkeypatch):
"""hermes -z bypasses HermesCLI, but recall still needs SessionDB."""
from hermes_cli.oneshot import _run_agent
captured = {}
sentinel_db = object()
class FakeAgent:
def __init__(self, **kwargs):
captured.update(kwargs)
self.suppress_status_output = False
self.stream_delta_callback = object()
self.tool_gen_callback = object()
def chat(self, prompt):
captured["prompt"] = prompt
return "ok"
class FakeSessionDB:
def __new__(cls):
return sentinel_db
def mod(name, **attrs):
module = types.ModuleType(name)
for key, value in attrs.items():
setattr(module, key, value)
return module
monkeypatch.setitem(sys.modules, "run_agent", mod("run_agent", AIAgent=FakeAgent))
monkeypatch.setitem(sys.modules, "hermes_state", mod("hermes_state", SessionDB=FakeSessionDB))
monkeypatch.setitem(
sys.modules,
"hermes_cli.config",
mod("hermes_cli.config", load_config=lambda: {"model": {"default": "m"}}),
)
monkeypatch.setitem(
sys.modules,
"hermes_cli.models",
mod("hermes_cli.models", detect_provider_for_model=lambda *_args, **_kwargs: None),
)
monkeypatch.setitem(
sys.modules,
"hermes_cli.runtime_provider",
mod(
"hermes_cli.runtime_provider",
resolve_runtime_provider=lambda **_kwargs: {
"api_key": "k",
"base_url": "u",
"provider": "p",
"api_mode": "chat_completions",
"credential_pool": None,
},
),
)
monkeypatch.setitem(
sys.modules,
"hermes_cli.tools_config",
mod("hermes_cli.tools_config", _get_platform_tools=lambda *_args, **_kwargs: {"session_search"}),
)
assert _run_agent("recall this") == "ok"
assert captured["session_db"] is sentinel_db
assert captured["enabled_toolsets"] == ["session_search"]
assert captured["prompt"] == "recall this"
def test_launch_tui_exports_model_provider_and_toolsets(monkeypatch, main_mod):
captured = {}
active_path_during_call = None

View file

@ -1,5 +1,7 @@
from types import SimpleNamespace
from types import ModuleType, SimpleNamespace
from unittest.mock import MagicMock, patch
import json
import sys
from run_agent import AIAgent
@ -61,3 +63,33 @@ def test_run_conversation_persists_tokens_for_cron_sessions():
assert result["final_response"] == "done"
session_db.update_token_counts.assert_called_once()
assert session_db.update_token_counts.call_args.args[0] == "cron-session"
def test_session_search_lazily_opens_db_when_entrypoint_did_not_pass_one(monkeypatch):
sentinel_db = object()
captured = {}
class FakeSessionDB:
def __new__(cls):
return sentinel_db
hermes_state = ModuleType("hermes_state")
hermes_state.SessionDB = FakeSessionDB
monkeypatch.setitem(sys.modules, "hermes_state", hermes_state)
session_search_mod = ModuleType("tools.session_search_tool")
def fake_session_search(**kwargs):
captured.update(kwargs)
return json.dumps({"success": True, "results": []})
session_search_mod.session_search = fake_session_search
monkeypatch.setitem(sys.modules, "tools.session_search_tool", session_search_mod)
agent = _make_agent(None, platform="acp")
result = json.loads(agent._invoke_tool("session_search", {"query": "Hermes"}, "task-id"))
assert result["success"] is True
assert captured["db"] is sentinel_db
assert captured["query"] == "Hermes"
assert agent._session_db is sentinel_db

View file

@ -309,11 +309,27 @@ class TestRecentSessionListing:
# =========================================================================
class TestSessionSearch:
def test_no_db_returns_error(self):
def test_no_db_lazily_opens_default_session_db(self, monkeypatch):
from unittest.mock import MagicMock
from tools.session_search_tool import session_search
mock_db = MagicMock()
mock_db.search_messages.return_value = []
class FakeSessionDB:
def __new__(cls):
return mock_db
import types
import sys
fake_state = types.ModuleType("hermes_state")
fake_state.SessionDB = FakeSessionDB
monkeypatch.setitem(sys.modules, "hermes_state", fake_state)
result = json.loads(session_search(query="test"))
assert result["success"] is False
assert "not available" in result["error"].lower()
assert result["success"] is True
mock_db.search_messages.assert_called_once()
def test_empty_query_returns_error(self):
from tools.session_search_tool import session_search