hermes-agent/tests/gateway/test_multiplex_phase0.py

"""Phase 0 foundations for multi-profile gateway multiplexing.

Covers the three Phase 0 deliverables:
  1. ``gateway.multiplex_profiles`` config flag (default False, round-trips).
  2. ``hermes_cli.profiles.profiles_to_serve`` enumeration.
  3. Profile-stamped ``build_session_key`` that is BYTE-IDENTICAL when the
     flag is off (the orphan-every-session guard) and namespace-segmented when
     on, without disturbing the positional key layout downstream parsers rely
     on.
"""
import pytest
from unittest.mock import patch

from gateway.config import GatewayConfig, Platform
from gateway.session import SessionSource, SessionStore, build_session_key


def _src(**kw) -> SessionSource:
    kw.setdefault("platform", Platform.TELEGRAM)
    kw.setdefault("chat_id", "99")
    kw.setdefault("chat_type", "dm")
    return SessionSource(**kw)


class TestSessionKeyByteIdenticalWhenOff:
    """The non-negotiable guard: with no profile (or 'default'), every key is
    byte-for-byte what it was before Phase 0. A diff here orphans every
    existing session on upgrade."""

    @pytest.mark.parametrize("profile", [None, "default"])
    def test_dm_with_chat_id(self, profile):
        s = _src(chat_id="99", chat_type="dm")
        assert build_session_key(s, profile=profile) == "agent:main:telegram:dm:99"

    @pytest.mark.parametrize("profile", [None, "default"])
    def test_dm_with_thread(self, profile):
        s = _src(chat_id="99", chat_type="dm", thread_id="t1")
        assert build_session_key(s, profile=profile) == "agent:main:telegram:dm:99:t1"

    @pytest.mark.parametrize("profile", [None, "default"])
    def test_dm_without_chat_id_falls_back_to_user(self, profile):
        s = _src(chat_id="", chat_type="dm", user_id="jordan")
        assert build_session_key(s, profile=profile) == "agent:main:telegram:dm:jordan"

    @pytest.mark.parametrize("profile", [None, "default"])
    def test_group_per_user(self, profile):
        s = _src(platform=Platform.DISCORD, chat_id="g1", chat_type="group", user_id="alice")
        assert (
            build_session_key(s, profile=profile)
            == "agent:main:discord:group:g1:alice"
        )

    @pytest.mark.parametrize("profile", [None, "default"])
    def test_group_shared_when_disabled(self, profile):
        s = _src(platform=Platform.DISCORD, chat_id="g1", chat_type="group", user_id="alice")
        assert (
            build_session_key(s, group_sessions_per_user=False, profile=profile)
            == "agent:main:discord:group:g1"
        )


class TestSessionKeyNamespacedWhenOn:
    """A named profile occupies the namespace slot, isolating its sessions."""

    def test_named_profile_dm(self):
        s = _src(chat_id="99", chat_type="dm")
        assert build_session_key(s, profile="coder") == "agent:coder:telegram:dm:99"

    def test_named_profile_group_per_user(self):
        s = _src(platform=Platform.DISCORD, chat_id="g1", chat_type="group", user_id="alice")
        assert (
            build_session_key(s, profile="coder")
            == "agent:coder:discord:group:g1:alice"
        )

    def test_two_profiles_same_chat_do_not_collide(self):
        s = _src(chat_id="99", chat_type="dm")
        a = build_session_key(s, profile="default")
        b = build_session_key(s, profile="coder")
        c = build_session_key(s, profile="writer")
        assert a != b != c and a != c

    def test_positional_layout_preserved_for_parsers(self):
        """Downstream parsers split on ':' and read parts[2]=platform,
        parts[3]=chat_type, parts[4]=chat_id (see qqbot adapter
        _parse_gateway_session_key). The profile must occupy parts[1] only."""
        s = _src(platform=Platform.DISCORD, chat_id="g1", chat_type="group", user_id="alice")
        parts = build_session_key(s, profile="coder").split(":")
        assert parts[0] == "agent"
        assert parts[1] == "coder"  # namespace slot (was always 'main')
        assert parts[2] == "discord"  # platform — unchanged offset
        assert parts[3] == "group"  # chat_type — unchanged offset
        assert parts[4] == "g1"  # chat_id — unchanged offset

    def test_default_namespace_layout_matches_named(self):
        """Default and named keys differ ONLY in parts[1]."""
        s = _src(platform=Platform.SLACK, chat_id="c1", chat_type="channel", user_id="u1")
        d = build_session_key(s, profile="default").split(":")
        n = build_session_key(s, profile="coder").split(":")
        assert d[0] == n[0] == "agent"
        assert d[1] == "main" and n[1] == "coder"
        assert d[2:] == n[2:]  # everything after the namespace is identical


class TestMultiplexConfigFlag:
    """gateway.multiplex_profiles defaults off and round-trips."""

    def test_default_is_false(self):
        assert GatewayConfig().multiplex_profiles is False

    def test_to_dict_includes_flag(self):
        assert GatewayConfig().to_dict()["multiplex_profiles"] is False

    def test_from_dict_top_level(self):
        cfg = GatewayConfig.from_dict({"multiplex_profiles": True})
        assert cfg.multiplex_profiles is True

    def test_from_dict_nested_gateway(self):
        cfg = GatewayConfig.from_dict({"gateway": {"multiplex_profiles": True}})
        assert cfg.multiplex_profiles is True

    def test_from_dict_coerces_truthy_string(self):
        cfg = GatewayConfig.from_dict({"multiplex_profiles": "true"})
        assert cfg.multiplex_profiles is True

    def test_roundtrip(self):
        cfg = GatewayConfig.from_dict(GatewayConfig(multiplex_profiles=True).to_dict())
        assert cfg.multiplex_profiles is True


class TestSessionStoreProfileResolution:
    """SessionStore._generate_session_key honors the flag: legacy namespace
    when off, active-profile namespace when on."""

    def _store(self, tmp_path, **cfg_kw):
        config = GatewayConfig(**cfg_kw)
        with patch("gateway.session.SessionStore._ensure_loaded"):
            s = SessionStore(sessions_dir=tmp_path, config=config)
        s._db = None
        s._loaded = True
        return s

    def test_flag_off_uses_legacy_namespace(self, tmp_path):
        store = self._store(tmp_path)  # multiplex_profiles defaults False
        s = _src(chat_id="99", chat_type="dm")
        assert store._generate_session_key(s) == "agent:main:telegram:dm:99"
        assert store._generate_session_key(s) == build_session_key(s)

    def test_flag_off_resolve_profile_is_none(self, tmp_path):
        store = self._store(tmp_path)
        assert store._resolve_profile_for_key() is None

    def test_flag_on_uses_active_profile_namespace(self, tmp_path):
        store = self._store(tmp_path, multiplex_profiles=True)
        s = _src(chat_id="99", chat_type="dm")
        with patch("hermes_cli.profiles.get_active_profile_name", return_value="coder"):
            assert store._generate_session_key(s) == "agent:coder:telegram:dm:99"

    def test_flag_on_default_profile_stays_legacy(self, tmp_path):
        store = self._store(tmp_path, multiplex_profiles=True)
        s = _src(chat_id="99", chat_type="dm")
        with patch("hermes_cli.profiles.get_active_profile_name", return_value="default"):
            assert store._generate_session_key(s) == "agent:main:telegram:dm:99"