"""Hermes provider lifecycle tests. These tests exercise the Python provider the way the Hermes host calls it, but with a fake JSON-RPC bridge so they stay deterministic and do not spawn Node, Hermes, or the HTTP viewer. """ from __future__ import annotations import json import sys import tempfile import unittest from pathlib import Path from unittest.mock import patch _ADAPTER_ROOT = Path(__file__).resolve().parent.parent.parent / "adapters" / "hermes" _PLUGIN_DIR = _ADAPTER_ROOT / "memos_provider" for _p in (_ADAPTER_ROOT, _PLUGIN_DIR): if str(_p) not in sys.path: sys.path.insert(0, str(_p)) import memos_provider # noqa: E402 class FakeBridge: def __init__(self) -> None: self.calls: list[tuple[str, dict]] = [] self.closed = False self.host_handlers: dict[str, object] = {} def register_host_handler(self, method: str, handler: object) -> None: self.host_handlers[method] = handler def request(self, method: str, params: dict | None = None, **_kwargs: object) -> dict: payload = params or {} self.calls.append((method, payload)) if method == "session.open": return {"sessionId": payload.get("sessionId") or "hermes:test-session"} if method == "turn.start": return { "query": { "sessionId": payload.get("sessionId") or "hermes:test-session", "episodeId": "episode-from-turn-start", }, "injectedContext": "remembered HERMES_MEMOS_E2E_0428", } if method == "turn.end": return {"traceId": "trace-1", "episodeId": payload.get("episodeId")} if method == "core.health": return {"ok": True} if method in {"episode.close", "session.close", "subagent.record"}: return {"ok": True} raise AssertionError(f"unexpected bridge method: {method}") def close(self) -> None: self.closed = True class FailingSessionOpenBridge(FakeBridge): def request(self, method: str, params: dict | None = None, **_kwargs: object) -> dict: if method == "session.open": self.closed = True raise RuntimeError("session.open did not respond") return super().request(method, params, **_kwargs) class HermesProviderPipelineTests(unittest.TestCase): def test_lifecycle_persists_turn_and_closes_real_episode(self) -> None: bridge = FakeBridge() with ( patch("memos_provider.ensure_bridge_running", return_value=True), patch("memos_provider.ensure_viewer_daemon", return_value=True), patch("memos_provider.MemosBridgeClient", return_value=bridge), ): provider = memos_provider.MemTensorProvider() provider.initialize( "host-session", hermes_home="/tmp/hermes-test-home", platform="cli", agent_identity="hermes-test", ) provider.on_turn_start(1, "Remember project HERMES_MEMOS_E2E_0428") prefetch = provider.prefetch("HERMES_MEMOS_E2E_0428") self.assertIn("remembered HERMES_MEMOS_E2E_0428", prefetch) self.assertEqual(provider._episode_id, "episode-from-turn-start") provider._on_post_tool_call( tool_name="terminal", args={"cmd": "npm test"}, result="all green", tool_call_id="tool-1", ) provider.sync_turn( "Remember project HERMES_MEMOS_E2E_0428", "Recorded the Hermes MemOS test fact.", ) provider.on_session_end([]) provider.shutdown() methods = [method for method, _params in bridge.calls] self.assertEqual( methods, [ "session.open", "turn.start", "turn.end", "session.close", ], ) turn_end = next(params for method, params in bridge.calls if method == "turn.end") self.assertEqual(turn_end["agent"], "hermes") self.assertEqual(turn_end["sessionId"], "host-session") self.assertEqual(turn_end["episodeId"], "episode-from-turn-start") self.assertIn("HERMES_MEMOS_E2E_0428", turn_end["userText"]) self.assertIn("Recorded", turn_end["agentText"]) self.assertEqual(turn_end["toolCalls"][0]["name"], "terminal") self.assertIn("npm test", turn_end["toolCalls"][0]["input"]) self.assertTrue(bridge.closed) def test_sync_turn_recovers_when_initial_bridge_open_timed_out(self) -> None: failed_bridge = FailingSessionOpenBridge() recovered_bridge = FakeBridge() bridge_attempts = [failed_bridge, recovered_bridge] def bridge_factory() -> FakeBridge: return bridge_attempts.pop(0) with ( patch("memos_provider.ensure_bridge_running", return_value=True), patch("memos_provider.ensure_viewer_daemon", return_value=True), patch("memos_provider.MemosBridgeClient", side_effect=bridge_factory), ): provider = memos_provider.MemTensorProvider() provider.initialize("slow-start-session") self.assertIsNone(provider._bridge) self.assertTrue(failed_bridge.closed) provider.on_turn_start(1, "检查 package.json") provider._on_post_tool_call( tool_name="read_file", args={"path": "package.json"}, result='{"content":"{}"}', tool_call_id="tool-1", ) provider.sync_turn("检查 package.json", "检查完成") methods = [method for method, _params in recovered_bridge.calls] self.assertEqual(methods, ["session.open", "turn.start", "turn.end"]) turn_end = next(params for method, params in recovered_bridge.calls if method == "turn.end") self.assertEqual(turn_end["sessionId"], "slow-start-session") self.assertEqual(turn_end["episodeId"], "episode-from-turn-start") self.assertEqual(turn_end["toolCalls"][0]["name"], "read_file") def test_delegation_recovers_when_initial_bridge_open_timed_out(self) -> None: recovered_bridge = FakeBridge() bridge_attempts = [FailingSessionOpenBridge(), recovered_bridge] with ( patch("memos_provider.ensure_bridge_running", return_value=True), patch("memos_provider.ensure_viewer_daemon", return_value=True), patch("memos_provider.MemosBridgeClient", side_effect=lambda: bridge_attempts.pop(0)), ): provider = memos_provider.MemTensorProvider() provider.initialize("slow-parent-session") provider.on_turn_start(1, "请派一个子代理检查 package.json") provider.on_delegation( "检查 package.json scripts", "当前目录没有 package.json", child_session_id="child-session", ) methods = [method for method, _params in recovered_bridge.calls] self.assertEqual(methods, ["session.open", "turn.start", "subagent.record"]) record = next( params for method, params in recovered_bridge.calls if method == "subagent.record" ) self.assertEqual(record["sessionId"], "slow-parent-session") self.assertEqual(record["episodeId"], "episode-from-turn-start") self.assertEqual(record["childSessionId"], "child-session") def test_sync_turn_lazily_starts_turn_when_prefetch_was_skipped(self) -> None: bridge = FakeBridge() with ( patch("memos_provider.ensure_bridge_running", return_value=True), patch("memos_provider.ensure_viewer_daemon", return_value=True), patch("memos_provider.MemosBridgeClient", return_value=bridge), ): provider = memos_provider.MemTensorProvider() provider.initialize("host-session") provider.on_turn_start(1, "继续处理 Hermes viewer 端口") provider.sync_turn( "继续处理 Hermes viewer 端口", "已继续检查 viewer 端口配置。", ) methods = [method for method, _params in bridge.calls] self.assertEqual(methods, ["session.open", "turn.start", "turn.end"]) turn_end = next(params for method, params in bridge.calls if method == "turn.end") self.assertEqual(turn_end["episodeId"], "episode-from-turn-start") def test_internal_hermes_review_prompt_is_not_persisted_as_user_turn(self) -> None: bridge = FakeBridge() review_prompt = ( "Review the conversation above and consider whether a skill should be " "saved or updated. Work in this order -- do not skip." ) with ( patch("memos_provider.ensure_bridge_running", return_value=True), patch("memos_provider.ensure_viewer_daemon", return_value=True), patch("memos_provider.MemosBridgeClient", return_value=bridge), ): provider = memos_provider.MemTensorProvider() provider.initialize("host-session") provider.on_turn_start(10, review_prompt) self.assertEqual(provider.prefetch(review_prompt), "") provider._on_post_tool_call( tool_name="memos_search", args={"query": "conversation"}, result="[]", tool_call_id="tool-1", ) provider.sync_turn(review_prompt, "Nothing to save.") provider.on_session_end([]) methods = [method for method, _params in bridge.calls] self.assertEqual(methods, ["session.open", "session.close"]) self.assertFalse(any(method == "turn.start" for method, _ in bridge.calls)) self.assertFalse(any(method == "turn.end" for method, _ in bridge.calls)) def test_on_pre_compress_reuses_last_user_text_for_snapshot(self) -> None: bridge = FakeBridge() with ( patch("memos_provider.ensure_bridge_running", return_value=True), patch("memos_provider.ensure_viewer_daemon", return_value=True), patch("memos_provider.MemosBridgeClient", return_value=bridge), ): provider = memos_provider.MemTensorProvider() provider.initialize("compress-session") provider.on_turn_start(2, "compress HERMES_MEMOS_E2E_0428 context") snapshot = provider.on_pre_compress([{"role": "user", "content": "x"}]) self.assertIn("MemOS memory snapshot", snapshot) self.assertIn("remembered HERMES_MEMOS_E2E_0428", snapshot) self.assertEqual(bridge.calls[-1][0], "turn.start") self.assertIn("HERMES_MEMOS_E2E_0428", bridge.calls[-1][1]["userText"]) def test_prefetch_suppresses_memory_injection_for_explicit_delegation(self) -> None: bridge = FakeBridge() with ( patch("memos_provider.ensure_bridge_running", return_value=True), patch("memos_provider.ensure_viewer_daemon", return_value=True), patch("memos_provider.MemosBridgeClient", return_value=bridge), ): provider = memos_provider.MemTensorProvider() provider.initialize("parent-session") provider.on_turn_start(1, "请派一个子代理检查 package.json") prefetch = provider.prefetch("请派一个子代理检查 package.json") self.assertEqual(prefetch, "") self.assertEqual(provider._episode_id, "episode-from-turn-start") self.assertEqual(bridge.calls[-1][0], "turn.start") self.assertIn("子代理", bridge.calls[-1][1]["userText"]) def test_tool_hook_ignores_other_sessions(self) -> None: bridge = FakeBridge() with ( patch("memos_provider.ensure_bridge_running", return_value=True), patch("memos_provider.ensure_viewer_daemon", return_value=True), patch("memos_provider.MemosBridgeClient", return_value=bridge), ): provider = memos_provider.MemTensorProvider() provider.initialize("parent-session") provider.on_turn_start(1, "parent task") provider.prefetch("parent task") provider._on_post_tool_call( tool_name="read_file", args={"path": "child-only.txt"}, result="child output", tool_call_id="child-tool", session_id="child-session", ) provider._on_post_tool_call( tool_name="terminal", args={"cmd": "npm test"}, result="parent output", tool_call_id="parent-tool", session_id="parent-session", ) provider.sync_turn("parent task", "parent done") turn_end = next(params for method, params in bridge.calls if method == "turn.end") self.assertEqual([tc["name"] for tc in turn_end["toolCalls"]], ["terminal"]) def test_on_delegation_targets_parent_episode(self) -> None: bridge = FakeBridge() with ( patch("memos_provider.ensure_bridge_running", return_value=True), patch("memos_provider.ensure_viewer_daemon", return_value=True), patch("memos_provider.MemosBridgeClient", return_value=bridge), ): provider = memos_provider.MemTensorProvider() provider.initialize("parent-session") provider.on_turn_start(1, "delegate task") provider.prefetch("delegate task") provider.on_delegation( "check package", "no package.json", child_session_id="child-session" ) method, params = bridge.calls[-1] self.assertEqual(method, "subagent.record") self.assertEqual(params["sessionId"], "parent-session") self.assertEqual(params["episodeId"], "episode-from-turn-start") self.assertEqual(params["childSessionId"], "child-session") def test_on_delegation_backfills_child_session_tool_calls(self) -> None: bridge = FakeBridge() with tempfile.TemporaryDirectory() as tmp: sessions_dir = Path(tmp) / "sessions" sessions_dir.mkdir() (sessions_dir / "session_child-session.json").write_text( json.dumps( { "messages": [ { "role": "assistant", "tool_calls": [ { "id": "tool-1", "type": "function", "function": { "name": "read_file", "arguments": json.dumps( {"path": "package.json", "limit": 20} ), }, } ], }, { "role": "tool", "tool_call_id": "tool-1", "content": json.dumps({"content": "1|{}", "total_lines": 1}), }, ] }, ensure_ascii=False, ), encoding="utf-8", ) with ( patch("memos_provider.ensure_bridge_running", return_value=True), patch("memos_provider.ensure_viewer_daemon", return_value=True), patch("memos_provider.MemosBridgeClient", return_value=bridge), ): provider = memos_provider.MemTensorProvider() provider.initialize("parent-session", hermes_home=tmp) provider.on_turn_start(1, "delegate task") provider.prefetch("delegate task") provider.on_delegation( "check package", "package exists", child_session_id="child-session", ) method, params = bridge.calls[-1] self.assertEqual(method, "subagent.record") self.assertEqual(params["toolCalls"][0]["name"], "read_file") self.assertEqual(params["toolCalls"][0]["input"]["path"], "package.json") self.assertIn("total_lines", params["toolCalls"][0]["output"]) def test_post_llm_call_backfills_tool_calls_without_post_tool_hook(self) -> None: bridge = FakeBridge() with ( patch("memos_provider.ensure_bridge_running", return_value=True), patch("memos_provider.ensure_viewer_daemon", return_value=True), patch("memos_provider.MemosBridgeClient", return_value=bridge), ): provider = memos_provider.MemTensorProvider() provider.initialize("host-session") provider.on_turn_start(1, "东京房产投资分析") provider.prefetch("东京房产投资分析") provider._on_post_llm_call( conversation_history=[ {"role": "user", "content": "东京房产投资分析"}, { "role": "assistant", "content": "好的，我来逐步完成这个分析。", "reasoning": "先列计划，再查汇率和房源。", "tool_calls": [ { "id": "call_todo_1", "call_id": "call_todo_1", "response_item_id": "fc_todo_1", "type": "function", "function": { "name": "todo", "arguments": '{"todos": [{"id": "1"}]}', }, } ], }, ] ) provider.sync_turn("东京房产投资分析", "好的，我来逐步完成这个分析。") turn_end = next(params for method, params in bridge.calls if method == "turn.end") self.assertEqual(turn_end["toolCalls"][0]["name"], "todo") self.assertIn('"todos"', turn_end["toolCalls"][0]["input"]) self.assertEqual(turn_end["toolCalls"][0]["thinkingBefore"], "先列计划，再查汇率和房源。") self.assertEqual( turn_end["toolCalls"][0]["assistantTextBefore"], "好的，我来逐步完成这个分析。", ) def test_post_tool_call_merges_with_llm_tool_aliases(self) -> None: bridge = FakeBridge() with ( patch("memos_provider.ensure_bridge_running", return_value=True), patch("memos_provider.ensure_viewer_daemon", return_value=True), patch("memos_provider.MemosBridgeClient", return_value=bridge), ): provider = memos_provider.MemTensorProvider() provider.initialize("host-session") provider.on_turn_start(1, "查汇率") provider.prefetch("查汇率") provider._on_post_llm_call( conversation_history=[ {"role": "user", "content": "查汇率"}, { "role": "assistant", "reasoning": "用 terminal 调 API。", "tool_calls": [ { "id": "call_terminal_1", "call_id": "call_terminal_1", "response_item_id": "fc_terminal_1", "type": "function", "function": { "name": "terminal", "arguments": '{"command": "curl example"}', }, } ], }, ] ) provider._on_post_tool_call( tool_name="terminal", args={"command": "curl example"}, result="1 JPY = 0.006 USD", tool_call_id="call_terminal_1", ) provider.sync_turn("查汇率", "查到了。") turn_end = next(params for method, params in bridge.calls if method == "turn.end") self.assertEqual(len(turn_end["toolCalls"]), 1) self.assertEqual(turn_end["toolCalls"][0]["name"], "terminal") self.assertIn("0.006 USD", turn_end["toolCalls"][0]["output"]) self.assertEqual(turn_end["toolCalls"][0]["thinkingBefore"], "用 terminal 调 API。") def test_post_llm_call_preserves_visible_text_before_tool_call(self) -> None: bridge = FakeBridge() with ( patch("memos_provider.ensure_bridge_running", return_value=True), patch("memos_provider.ensure_viewer_daemon", return_value=True), patch("memos_provider.MemosBridgeClient", return_value=bridge), ): provider = memos_provider.MemTensorProvider() provider.initialize("host-session") provider.on_turn_start(1, "分析房价数据集") provider.prefetch("分析房价数据集") provider._on_post_llm_call( conversation_history=[ {"role": "user", "content": "分析房价数据集"}, { "role": "assistant", "content": "好的，这是经典的 Kaggle 房价预测数据集。先创建计划。", "reasoning": "用户要元数据清单，先列 todo。", "tool_calls": [ { "id": "call_todo_1", "call_id": "call_todo_1", "type": "function", "function": { "name": "todo", "arguments": '{"todos": [{"id": "step0"}]}', }, } ], }, ] ) provider.sync_turn("分析房价数据集", "计划已创建。") turn_end = next(params for method, params in bridge.calls if method == "turn.end") tool = turn_end["toolCalls"][0] self.assertEqual(tool["name"], "todo") self.assertEqual(tool["thinkingBefore"], "用户要元数据清单，先列 todo。") self.assertEqual( tool["assistantTextBefore"], "好的，这是经典的 Kaggle 房价预测数据集。先创建计划。", ) def test_transform_tool_result_appends_memos_search_hint_after_three_failures(self) -> None: provider = memos_provider.MemTensorProvider() provider.on_turn_start(1, "run failing command") self.assertIsNone( provider._on_transform_tool_result( tool_name="terminal", result="boom", is_error=True, ) ) self.assertIsNone( provider._on_transform_tool_result( tool_name="terminal", result="boom again", is_error=True, ) ) third = provider._on_transform_tool_result( tool_name="terminal", result="boom third", is_error=True, ) self.assertIsNotNone(third) self.assertIn("failed multiple times in a row", third or "") self.assertIn("memos_search", third or "") provider._on_transform_tool_result( tool_name="terminal", result="ok", is_error=False, ) self.assertIsNone( provider._on_transform_tool_result( tool_name="terminal", result="boom after reset", is_error=True, ) ) def test_transform_tool_result_detects_plain_error_text(self) -> None: provider = memos_provider.MemTensorProvider() provider.on_turn_start(1, "run failing command") self.assertIsNone( provider._on_transform_tool_result( tool_name="terminal", result="Error: command failed", ) ) self.assertIsNone( provider._on_transform_tool_result( tool_name="terminal", result="Error: command failed again", ) ) third = provider._on_transform_tool_result( tool_name="terminal", result="Error: command failed third time", ) self.assertIsNotNone(third) self.assertIn("memos_search", third or "") def test_post_llm_call_orders_backfilled_tools_before_later_tool_results(self) -> None: bridge = FakeBridge() with ( patch("memos_provider.ensure_bridge_running", return_value=True), patch("memos_provider.ensure_viewer_daemon", return_value=True), patch("memos_provider.MemosBridgeClient", return_value=bridge), ): provider = memos_provider.MemTensorProvider() provider.initialize("host-session") provider.on_turn_start(1, "规划北欧旅行") provider.prefetch("规划北欧旅行") # A later executed tool may be reported before post_llm_call # backfills planner/todo calls from conversation_history. provider._on_post_tool_call( tool_name="terminal", args={"command": "search flights"}, result="PVG-CPH 4200 RMB", tool_call_id="call_terminal_1", ) provider._on_post_llm_call( conversation_history=[ {"role": "user", "content": "规划北欧旅行"}, { "role": "assistant", "reasoning": "先列计划，再查机票。", "tool_calls": [ { "id": "call_todo_1", "call_id": "call_todo_1", "type": "function", "function": { "name": "todo", "arguments": '{"todos": [{"id": "1"}]}', }, }, { "id": "call_terminal_1", "call_id": "call_terminal_1", "type": "function", "function": { "name": "terminal", "arguments": '{"command": "search flights"}', }, }, ], }, ] ) provider.sync_turn("规划北欧旅行", "路线和预算整理好了。") turn_end = next(params for method, params in bridge.calls if method == "turn.end") self.assertEqual([tc["name"] for tc in turn_end["toolCalls"]], ["todo", "terminal"]) self.assertIn('"todos"', turn_end["toolCalls"][0]["input"]) self.assertEqual(turn_end["toolCalls"][0]["thinkingBefore"], "先列计划，再查机票。") self.assertIn("PVG-CPH", turn_end["toolCalls"][1]["output"]) self.assertEqual(turn_end["toolCalls"][1]["thinkingBefore"], "先列计划，再查机票。") if __name__ == "__main__": unittest.main()