hermes-agent/tests/acp_adapter/test_session.py

"""Tests for acp_adapter.session — SessionManager and SessionState."""

import contextlib
import io
import json
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.<platform>``,
        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
        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"