"""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 datetime import datetime
from unittest.mock import patch
import hermes_yaml as yaml

from hermes_constants import reset_hermes_home_override, set_hermes_home_override
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_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"
        )


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


    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


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

    def test_cron_shared_adapter_owner_is_the_launch_profile(self, monkeypatch, tmp_path):
        """A ``--profile rex`` multiplexer owns ``runner.adapters``; its ticker must name rex (not the
        literal ``default``) as the shared-adapter owner or rex's own jobs fall to the fail-closed map."""
        import asyncio
        from types import SimpleNamespace
        from gateway import run as run_mod
        from cron.scheduler_provider import InProcessCronScheduler

        captured = {}

        class _Ticker(InProcessCronScheduler):
            def start(self, stop_event, **kwargs):
                captured.update(kwargs)

        monkeypatch.setattr("cron.scheduler_provider.resolve_cron_scheduler", lambda: _Ticker())
        monkeypatch.setattr(run_mod, "_cron_tick_profile_homes", lambda cfg: [("rex", tmp_path)])
        monkeypatch.setattr(run_mod, "_start_gateway_housekeeping", lambda *a, **k: None)
        runner = SimpleNamespace(
            config=GatewayConfig(multiplex_profiles=True), adapters={}, _profile_adapters={},
            _primary_profile_name="rex", _draining=False, _external_drain_active=False)

        async def _go():
            return run_mod._start_gateway_start_cron_and_housekeeping(runner)

        cron_stop, _provider, cron_thread, hk = asyncio.run(_go())
        cron_stop.set()
        cron_thread.join(timeout=5)
        hk.join(timeout=5)
        assert captured["default_profile"] == "rex"

    def test_unset_is_undecided_and_reads_as_off(self):
        """The default (on) is applied by the boot guard, not the dataclass: an unset flag stays
        ``None`` so the guard can tell it from an explicit choice, and every reader treats it as off."""
        assert GatewayConfig().multiplex_profiles is None
        assert not GatewayConfig().multiplex_profiles
        assert GatewayConfig.from_dict({}).multiplex_profiles is None


    def test_from_dict_top_level(self):
        cfg = GatewayConfig.from_dict({"multiplex_profiles": True})
        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)


class _RecoveringDB:
    def __init__(self, row):
        self.row = row
        self.reopened = []

    def find_latest_gateway_session_for_peer(self, **_kwargs):
        return self.row

    def reopen_session(self, session_id):
        self.reopened.append(session_id)


class TestSessionStoreUnmultiplexedRecovery:
    """Turning multiplexing off must not recover another profile's session."""

    def _store_with_row(self, tmp_path, row, **cfg_kw):
        config = GatewayConfig(**cfg_kw)
        with patch("gateway.session.SessionStore._ensure_loaded"):
            store = SessionStore(sessions_dir=tmp_path, config=config)
        store._db = _RecoveringDB(row)
        store._loaded = True
        return store


    def test_flag_off_allows_active_profile_peer_fallback(self, tmp_path):
        row = {
            "id": "sess-coder",
            "started_at": 1700000000,
            "session_key": "agent:coder:telegram:dm:99",
        }
        store = self._store_with_row(tmp_path, row)
        source = _src(chat_id="99", chat_type="dm")

        with patch("hermes_cli.profiles.get_active_profile_name", return_value="coder"):
            recovered = store._recover_session_from_db(
                session_key="agent:main:telegram:dm:99",
                source=source,
                now=datetime.fromtimestamp(1700000001),
            )

        assert recovered is not None
        assert recovered.session_id == "sess-coder"
        assert recovered.session_key == "agent:main:telegram:dm:99"
        assert store._db.reopened == ["sess-coder"]

    @pytest.mark.parametrize(
        ("recovered_key", "adopted"),
        [
            ("agent:coder:telegram:dm:99", False),  # sibling namespace → fail closed
            ("agent:main:telegram:dm:99:v1", True),  # same namespace → adoptable
        ],
        ids=["sibling-profile", "same-profile"],
    )
    def test_flag_on_fences_recovery_by_requested_namespace(
        self, tmp_path, recovered_key, adopted
    ):
        """#74285: under multiplexing the guard compares the recovered row's
        ``agent:<ns>:`` against the REQUESTED key, never the active profile."""
        row = {"id": "sess", "started_at": 1700000000, "session_key": recovered_key}
        store = self._store_with_row(tmp_path, row, multiplex_profiles=True)
        store._db_pinned = store._db
        with patch("hermes_cli.profiles.get_active_profile_name", return_value="coder"):
            recovered = store._recover_session_from_db(
                session_key="agent:main:telegram:dm:99",
                source=_src(chat_id="99", chat_type="dm"),
                now=datetime.fromtimestamp(1700000001),
            )
        assert (recovered is not None) is adopted