"""Tests for acp_adapter.session — SessionManager and SessionState.""" import contextlib import io import json import re from types import SimpleNamespace import pytest from unittest.mock import MagicMock, patch from acp_adapter import session as acp_session from acp_adapter.session import SessionManager, SessionState from hermes_state import SessionDB def _mock_agent(): return MagicMock(name="MockAIAgent") @pytest.fixture() def manager(): """SessionManager with a mock agent factory (avoids needing API keys).""" return SessionManager(agent_factory=_mock_agent) # --------------------------------------------------------------------------- # create / get # --------------------------------------------------------------------------- class TestCreateSession: def test_create_session_returns_state(self, manager): state = manager.create_session(cwd="/tmp/work") assert isinstance(state, SessionState) assert state.cwd == "/tmp/work" assert state.session_id assert state.history == [] assert state.agent is not None def test_register_task_cwd_translates_windows_drive_for_wsl_tools(self, monkeypatch): captured = {} def fake_register_task_env_overrides(task_id, overrides): captured["task_id"] = task_id captured["overrides"] = overrides monkeypatch.setattr("hermes_platform.host.runtime._wsl_detected", True) monkeypatch.setattr( "tools.terminal_tool.register_task_env_overrides", fake_register_task_env_overrides, ) acp_session._register_task_cwd("session-1", r"E:\Projects\AI\paperclip") assert captured == { "task_id": "session-1", "overrides": {"cwd": "/mnt/e/Projects/AI/paperclip"}, } def test_make_agent_uses_session_cwd_during_init_and_stamps_runtime( self, monkeypatch, tmp_path ): workspace = tmp_path / "workspace" workspace.mkdir() observed = {} class FakeAgent: model = "fake-model" def __init__(self, **kwargs): self.kwargs = kwargs observed["cwd"] = kwargs.get("cwd") monkeypatch.setattr("run_agent.AIAgent", FakeAgent) monkeypatch.setattr( "acp_adapter.session.load_config", lambda: { "model": { "default": "fake-model", "provider": "fake-provider", }, "mcp_servers": {}, }, raising=False, ) monkeypatch.setattr( "hermes_cli.config.load_config", lambda: { "model": { "default": "fake-model", "provider": "fake-provider", }, "mcp_servers": {}, }, ) monkeypatch.setattr( "hermes_cli.runtime_provider.resolve_runtime_provider", lambda requested=None: { "provider": requested, "api_mode": "codex_app_server", "base_url": "https://example.invalid", "api_key": "test-key", }, ) monkeypatch.setattr("acp_adapter.session._register_task_cwd", lambda task_id, cwd: None) monkeypatch.setattr("hermes_cli.mcp_startup.ensure_mcp_discovery_before_agent_build", lambda **_kw: None) SessionManager(db=None).create_session(cwd=str(workspace)) assert observed["cwd"] == str(workspace) def test_make_agent_prefers_passed_toolsets_over_config_servers(self, monkeypatch): """#42719: a rebuild (model switch) passes the live session's toolsets and they are kept verbatim; a fresh session still derives them from the config-declared MCP servers.""" seen: list[dict] = [] class FakeAgent: def __init__(self, **kwargs): seen.append(kwargs) config = {"model": {"default": "m", "provider": "p"}, "mcp_servers": {"cfg-server": {}}} monkeypatch.setattr("run_agent.AIAgent", FakeAgent) monkeypatch.setattr("hermes_cli.config.load_config", lambda: config) monkeypatch.setattr("hermes_cli.runtime_provider.resolve_runtime_provider", lambda **_kw: {}) monkeypatch.setattr("hermes_cli.mcp_startup.ensure_mcp_discovery_before_agent_build", lambda **_kw: None) monkeypatch.setattr("acp_adapter.session._register_task_cwd", lambda task_id, cwd: None) manager = SessionManager(db=None) manager._make_agent(session_id="fresh", cwd=".") manager._make_agent( session_id="rebuilt", cwd=".", enabled_toolsets=["hermes-acp", "mcp-acp-server"], disabled_toolsets=["browser"], ) assert "mcp-cfg-server" in seen[0]["enabled_toolsets"] and seen[0]["disabled_toolsets"] is None assert (seen[1]["enabled_toolsets"], seen[1]["disabled_toolsets"]) == (["hermes-acp", "mcp-acp-server"], ["browser"]) @pytest.mark.parametrize("config, offered, withheld", [ # agent.disabled_toolsets, in the JSON-string shape `hermes config set` stores (#74582). ({"agent": {"disabled_toolsets": "['code_execution']"}}, "file", "code_execution"), # platform_toolsets.acp narrows the surface like every other platform (#79516). ({"platform_toolsets": {"acp": ["file"]}}, "file", "code_execution"), ]) def test_fresh_agent_tool_surface_honours_toolset_config(self, monkeypatch, config, offered, withheld): """A fresh ACP agent resolves its tools like the gateway/cron: the real tool surface built from its kwargs carries the offered toolset and none of the withheld one.""" from model_tools import get_tool_definitions from toolsets import resolve_toolset seen: list[dict] = [] class FakeAgent: def __init__(self, **kwargs): seen.append(kwargs) monkeypatch.setattr("run_agent.AIAgent", FakeAgent) monkeypatch.setattr("hermes_cli.config.load_config", lambda: {"model": {"default": "m"}, **config}) monkeypatch.setattr("hermes_cli.runtime_provider.resolve_runtime_provider", lambda **_kw: {}) monkeypatch.setattr("hermes_cli.mcp_startup.ensure_mcp_discovery_before_agent_build", lambda **_kw: None) monkeypatch.setattr("acp_adapter.session._register_task_cwd", lambda task_id, cwd: None) SessionManager(db=None)._make_agent(session_id="fresh", cwd=".") def surface(enabled, disabled=None) -> set: return {t["function"]["name"] for t in get_tool_definitions( enabled_toolsets=enabled, disabled_toolsets=disabled, quiet_mode=True)} assert set(resolve_toolset(withheld)) <= surface(["hermes-acp"]) # non-vacuous: offered by default names = surface(seen[0]["enabled_toolsets"], seen[0]["disabled_toolsets"]) assert set(resolve_toolset(offered)) <= names assert not names & set(resolve_toolset(withheld)) @pytest.mark.parametrize("acp_toolsets, expected_mcp", [ (None, {"mcp-alpha", "mcp-beta"}), # default: every enabled config server (["hermes-acp", "alpha"], {"mcp-alpha"}), # listed server names are an allowlist (["hermes-acp", "no_mcp"], set()), # the no_mcp sentinel drops them all ]) def test_fresh_agent_mcp_servers_follow_platform_toolsets(self, monkeypatch, acp_toolsets, expected_mcp): """Config MCP servers reach a fresh ACP agent by the gateway's rules for ``platform_toolsets.``, not unconditionally; a disabled server never does.""" seen: list[dict] = [] class FakeAgent: def __init__(self, **kwargs): seen.append(kwargs) config = {"model": {"default": "m"}, "mcp_servers": {"alpha": {"command": "a"}, "beta": {"command": "b"}, "off": {"enabled": False}}} if acp_toolsets is not None: config["platform_toolsets"] = {"acp": acp_toolsets} monkeypatch.setattr("run_agent.AIAgent", FakeAgent) monkeypatch.setattr("hermes_cli.config.load_config", lambda: config) monkeypatch.setattr("hermes_cli.runtime_provider.resolve_runtime_provider", lambda **_kw: {}) monkeypatch.setattr("hermes_cli.mcp_startup.ensure_mcp_discovery_before_agent_build", lambda **_kw: None) monkeypatch.setattr("acp_adapter.session._register_task_cwd", lambda task_id, cwd: None) SessionManager(db=None)._make_agent(session_id="fresh", cwd=".") enabled = seen[0]["enabled_toolsets"] assert {t for t in enabled if t.startswith("mcp-")} == expected_mcp assert not {"alpha", "beta", "no_mcp"} & set(enabled) def test_make_agent_surfaces_the_provider_resolution_failure(self, monkeypatch): """#91090: when ``resolve_runtime_provider`` fails, the bare-AIAgent fallback dies with the first-run "No LLM provider configured" text; the operator must get the swallowed cause instead. The fallback still stands when the bare build succeeds.""" def _no_creds(**_kw): raise RuntimeError("No Codex credentials stored. Run `hermes auth add openai-codex`") class BareFails: def __init__(self, **kwargs): raise RuntimeError("No LLM provider configured. Run `hermes setup`") class BareWorks: def __init__(self, **kwargs): self.kwargs = kwargs monkeypatch.setattr("hermes_cli.config.load_config", lambda: {"model": {"default": "m", "provider": "openai-codex"}}) monkeypatch.setattr("hermes_cli.runtime_provider.resolve_runtime_provider", _no_creds) monkeypatch.setattr("hermes_cli.mcp_startup.ensure_mcp_discovery_before_agent_build", lambda **_kw: None) monkeypatch.setattr("acp_adapter.session._register_task_cwd", lambda task_id, cwd: None) manager = SessionManager(db=None) monkeypatch.setattr("run_agent.AIAgent", BareFails) with pytest.raises(RuntimeError, match="No Codex credentials stored") as exc: manager._make_agent(session_id="rebuilt", cwd=".", requested_provider="openai-codex") assert "No LLM provider configured" in str(exc.value.__cause__) monkeypatch.setattr("run_agent.AIAgent", BareWorks) assert "provider" not in manager._make_agent(session_id="fresh", cwd=".").kwargs def test_make_agent_forwards_resolved_credential_pool(self, monkeypatch): """#70292: the provider-scoped credential pool selected by resolve_runtime_provider reaches the ACP agent by identity, so a long-lived session can refresh/rotate on 401 instead of needing a restart.""" seen: list[dict] = [] sentinel_pool = object() class FakeAgent: def __init__(self, **kwargs): seen.append(kwargs) monkeypatch.setattr("run_agent.AIAgent", FakeAgent) monkeypatch.setattr("hermes_cli.config.load_config", lambda: {"model": {"default": "m", "provider": "openai-codex"}}) monkeypatch.setattr("hermes_cli.runtime_provider.resolve_runtime_provider", lambda **_kw: { "provider": "openai-codex", "api_mode": "codex_app_server", "api_key": "test-key", "credential_pool": sentinel_pool, }) monkeypatch.setattr("hermes_cli.mcp_startup.ensure_mcp_discovery_before_agent_build", lambda **_kw: None) monkeypatch.setattr("acp_adapter.session._register_task_cwd", lambda task_id, cwd: None) SessionManager(db=None)._make_agent(session_id="s", cwd=".") assert seen[0]["credential_pool"] is sentinel_pool # --------------------------------------------------------------------------- # WSL cwd translation # --------------------------------------------------------------------------- class TestWslCwdTranslation: def test_translate_acp_cwd_converts_windows_drive_path_when_wsl(self, monkeypatch): monkeypatch.setattr("hermes_platform.host.runtime._wsl_detected", True) assert acp_session._translate_acp_cwd(r"E:\Projects\AI\paperclip") == "/mnt/e/Projects/AI/paperclip" def test_fork_session_stores_translated_cwd_on_wsl(self, manager, monkeypatch): monkeypatch.setattr("hermes_platform.host.runtime._wsl_detected", True) original = manager.create_session(cwd="/tmp/base") forked = manager.fork_session(original.session_id, cwd=r"D:\work\project") assert forked is not None assert forked.cwd == "/mnt/d/work/project" def test_update_cwd_stores_translated_cwd_on_wsl(self, manager, monkeypatch): monkeypatch.setattr("hermes_platform.host.runtime._wsl_detected", True) state = manager.create_session(cwd="/tmp/old") updated = manager.update_cwd(state.session_id, cwd=r"C:\Users\foo\project") assert updated is not None assert updated.cwd == "/mnt/c/Users/foo/project" # --------------------------------------------------------------------------- # fork # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # list / cleanup / remove # --------------------------------------------------------------------------- class TestSymlinkAliasNormalization: """Ported from PrimeIntellect-ai/prime-agent#628 — symlink aliases of the same directory (macOS ``/var`` vs ``/private/var``, ``/tmp`` vs ``/private/tmp``) must compare equal, or ACP history filters silently drop a workspace's own sessions.""" @pytest.mark.require_symlinks def test_symlink_alias_compares_equal(self, tmp_path): real = tmp_path / "real" real.mkdir() alias = tmp_path / "alias" alias.symlink_to(real) assert acp_session._normalize_cwd_for_compare( str(alias) ) == acp_session._normalize_cwd_for_compare(str(real)) def test_distinct_dirs_still_compare_different(self, tmp_path): a = tmp_path / "a" b = tmp_path / "b" a.mkdir() b.mkdir() assert acp_session._normalize_cwd_for_compare( str(a) ) != acp_session._normalize_cwd_for_compare(str(b)) def test_missing_path_keeps_lexical_normalization(self): # realpath(strict=False) is lexical for nonexistent paths, so cwds # that don't exist on this host (e.g. WSL-translated drives) behave # exactly as the old normpath comparison did. assert acp_session._normalize_cwd_for_compare( "/nonexistent-hermes-test/x/../y" ) == acp_session._normalize_cwd_for_compare("/nonexistent-hermes-test/y") @pytest.mark.require_symlinks def test_list_sessions_matches_symlink_alias_cwd(self, manager, tmp_path): real = tmp_path / "proj" real.mkdir() alias = tmp_path / "link" alias.symlink_to(real) state = manager.create_session(cwd=str(real)) state.history.append({"role": "user", "content": "hello"}) listed = manager.list_sessions(cwd=str(alias)) assert [s["session_id"] for s in listed] == [state.session_id] # --------------------------------------------------------------------------- # list / cleanup # --------------------------------------------------------------------------- class TestListAndCleanup: def test_list_sessions_empty(self, manager): assert manager.list_sessions() == [] def test_save_session_preserves_existing_messages_on_encode_failure(self, manager): """Regression for #13675: a bad message in state.history must not clobber the previously-persisted transcript. replace_messages() wraps DELETE + INSERT in a single rolled-back-on-exception txn. """ state = manager.create_session() state.history.append({"role": "user", "content": "original"}) manager.save_session(state.session_id) # Now swap history with a message whose tool_calls is non-JSON-serializable. # _execute_write rolls back; the previously persisted "original" stays. state.history = [ {"role": "user", "content": "replacement"}, { "role": "assistant", "content": None, "tool_calls": [{"bad": object()}], }, ] manager.save_session(state.session_id) db = manager._get_db() messages = db.get_messages_as_conversation(state.session_id) assert len(messages) == 1 assert messages[0]["role"] == "user" assert messages[0]["content"] == "original" assert isinstance(messages[0].get("timestamp"), (int, float)) # --------------------------------------------------------------------------- # persistence — sessions survive process restarts (via SessionDB) # --------------------------------------------------------------------------- class TestPersistence: """Verify that sessions are persisted to SessionDB and can be restored.""" def test_first_persist_keeps_provider_snapshot(self, tmp_path): """The FIRST row written for an ACP session carries provider/base_url/api_mode, so a restart before any later save restores the same route (#9812).""" agent = SimpleNamespace( model="test-model", provider="anthropic", base_url="https://anthropic.example/v1", api_mode="anthropic_messages", ) db = SessionDB(tmp_path / "state.db") manager = SessionManager(agent_factory=lambda: agent, db=db) state = manager.create_session(cwd="/work") state.history.append({"role": "user", "content": "hello"}) manager.save_session(state.session_id) mc = json.loads(db.get_session(state.session_id)["model_config"]) assert mc == {"cwd": "/work", "provider": "anthropic", "base_url": "https://anthropic.example/v1", "api_mode": "anthropic_messages"} def test_only_restores_acp_sessions(self, manager): """get_session should not restore non-ACP sessions from DB.""" db = manager._get_db() # Manually create a CLI session in the DB. db.create_session(session_id="cli-session-123", source="cli", model="test") # Should not be found via ACP SessionManager. assert manager.get_session("cli-session-123") is None def test_end_all_sessions_stamps_ended_and_unlocks_prune(self, tmp_path): """#118216: the adapter's stdio shutdown is the ACP session end. Without the ended_at writer, source='acp' rows are invisible to prune/archive forever (the maintenance filter only ever selects ended sessions) and the desktop sidebar accumulates one auto-titled row per editor wake.""" agent = SimpleNamespace(model="test-model", provider=None, base_url=None, api_mode=None) db = SessionDB(tmp_path / "state.db") manager = SessionManager(agent_factory=lambda: agent, db=db) ended_earlier = manager.create_session(cwd="/work") live_one = manager.create_session(cwd="/work") live_two = manager.create_session(cwd="/work") for state in (ended_earlier, live_one, live_two): state.history.append({"role": "user", "content": "hello"}) manager.save_session(state.session_id) # A row already ended by an earlier boundary keeps its first end_reason # (end_session is first-writer-wins). db.end_session(ended_earlier.session_id, "compression") # Before the shutdown writer, only the already-ended row is a prune # candidate — the two live acp rows are invisible to maintenance. assert {row["id"] for row in db.list_prune_candidates(source="acp", older_than_days=0)} == { ended_earlier.session_id} ended = manager.end_all_sessions() assert ended == 3 rows = {sid: db.get_session(sid) for sid in (ended_earlier.session_id, live_one.session_id, live_two.session_id)} assert all(row is not None for row in rows.values()) assert rows[ended_earlier.session_id]["end_reason"] == "compression" for sid in (live_one.session_id, live_two.session_id): assert rows[sid]["ended_at"] is not None assert rows[sid]["end_reason"] == "acp_disconnect" candidates = db.list_prune_candidates(source="acp", older_than_days=0) assert {row["id"] for row in candidates} == { ended_earlier.session_id, live_one.session_id, live_two.session_id} def test_restore_reopens_a_session_ended_by_previous_process(self, tmp_path): """#118216: an ACP row stamped ended at a previous adapter's shutdown is reopened when a later process resumes it — the same contract as the TUI gateway's cold resume.""" agent = SimpleNamespace(model="test-model", provider=None, base_url=None, api_mode=None) db = SessionDB(tmp_path / "state.db") first = SessionManager(agent_factory=lambda: agent, db=db) state = first.create_session(cwd="/work") state.history.append({"role": "user", "content": "hello"}) first.save_session(state.session_id) first.end_all_sessions() assert db.get_session(state.session_id)["ended_at"] is not None second = SessionManager(agent_factory=lambda: agent, db=db) restored = second.get_session(state.session_id) assert restored is not None row = db.get_session(state.session_id) assert row is not None assert row["ended_at"] is None assert row["end_reason"] is None def test_end_all_sessions_without_db_is_a_noop(self): """Teardown must never raise; a DB-unavailable process just ends nothing.""" manager = SessionManager( agent_factory=_mock_agent, db=None, ) manager._db_instance = None with patch.object(manager, "_get_db", return_value=None): assert manager.end_all_sessions() == 0 def test_sessions_searchable_via_fts(self, manager): """ACP sessions stored in SessionDB are searchable via FTS5.""" state = manager.create_session() state.history.append({"role": "user", "content": "how do I configure nginx"}) state.history.append({"role": "assistant", "content": "Here is the nginx config..."}) manager.save_session(state.session_id) db = manager._get_db() results = db.search_messages("nginx") assert len(results) > 0 session_ids = {r["session_id"] for r in results} assert state.session_id in session_ids def test_assistant_reasoning_fields_persisted(self, manager): """ACP session restore should preserve assistant reasoning context.""" state = manager.create_session() state.history.append({ "role": "assistant", "content": "hello", "reasoning": "step-by-step", "reasoning_details": [ {"type": "thinking", "thinking": "first thought"}, ], "codex_reasoning_items": [ {"type": "reasoning", "id": "rs_123", "encrypted_content": "enc_blob"}, ], }) manager.save_session(state.session_id) with manager._lock: del manager._sessions[state.session_id] restored = manager.get_session(state.session_id) assert restored is not None msg = restored.history[0] assert isinstance(msg.pop("timestamp", None), (int, float)) # Load-time durability stamp (#92231): rows materialized from the DB # are marked persisted so a later flush can't re-append them. assert msg.pop("_db_persisted", None) is True # The durable per-message id rides on every restored row, like the timestamp. assert re.fullmatch(r"[0-9a-f]{32}", msg.pop("message_uid", "")) # Repair bookkeeping (underscore-prefixed, stripped before the wire) rides restored rows too. assert isinstance(msg.pop("_db_row_snapshot", None), str) assert restored.history == [{ "role": "assistant", "content": "hello", "reasoning": "step-by-step", "reasoning_details": [ {"type": "thinking", "thinking": "first thought"}, ], "codex_reasoning_items": [ {"type": "reasoning", "id": "rs_123", "encrypted_content": "enc_blob"}, ], }] def test_acp_agents_route_human_output_to_stderr(self, tmp_path, monkeypatch): """ACP agents must keep stdout clean for JSON-RPC stdio transport.""" def fake_resolve_runtime_provider(requested=None, **kwargs): return { "provider": "openrouter", "api_mode": "chat_completions", "base_url": "https://openrouter.example/v1", "api_key": "test-key", "command": None, "args": [], } def fake_agent(**kwargs): return SimpleNamespace(model=kwargs.get("model"), _print_fn=None) monkeypatch.setattr("hermes_cli.config.load_config", lambda: { "model": {"provider": "openrouter", "default": "test-model"} }) monkeypatch.setattr( "hermes_cli.runtime_provider.resolve_runtime_provider", fake_resolve_runtime_provider, ) db = SessionDB(tmp_path / "state.db") with patch("run_agent.AIAgent", side_effect=fake_agent): manager = SessionManager(db=db) state = manager.create_session(cwd="/work") stdout_buf = io.StringIO() stderr_buf = io.StringIO() with contextlib.redirect_stdout(stdout_buf), contextlib.redirect_stderr(stderr_buf): state.agent._print_fn("ACP noise") assert stdout_buf.getvalue() == "" assert stderr_buf.getvalue() == "ACP noise\n"