"""Tests for hermes_cli.web_server and related config utilities."""

import asyncio
import os
import json
import re
import shutil
import sys
import threading
import time
from pathlib import Path
from unittest.mock import patch, MagicMock

import pytest
import hermes_yaml as yaml

from hermes_cli.config import (
    reload_env,
    redact_key,
    OPTIONAL_ENV_VARS,
)
import gateway.status as _gw_status
import hermes_cli.config as _cfg_mod
import hermes_cli.web_routers.chat_ws as _rt_chat_ws
import hermes_cli.web_server_chat as _web_server_chat
import hermes_cli.web_server_config as _web_server_config
import hermes_cli.web_server_dashboard as _web_server_dashboard
import hermes_cli.web_server_files as _web_server_files
import hermes_cli.web_server_gateway as _web_server_gateway
import hermes_cli.web_server_lifecycle as _web_server_lifecycle
import hermes_cli.web_server_memory as _web_server_memory
import hermes_cli.web_server_messaging as _web_server_messaging
import hermes_cli.web_server_sessions as _web_server_sessions


# ---------------------------------------------------------------------------
# Shared fixtures
# ---------------------------------------------------------------------------


# Path to the test-only example-dashboard plugin. Lives under
# tests/fixtures/ so the bundled-plugins directory stays clean — stock
# installs no longer ship a dummy "Example" sidebar tab. Tests that
# depend on its routes opt in via the `_install_example_plugin` fixture
# below.
_EXAMPLE_PLUGIN_FIXTURE = (
    Path(__file__).resolve().parent.parent / "fixtures" / "plugins" / "example-dashboard"
)


@pytest.fixture
def _install_example_plugin(_isolate_hermes_home):
    """Drop the example-dashboard fixture into the per-test HERMES_HOME
    user-plugins directory and force the web_server's dashboard plugin
    cache + API mount to rediscover it.

    The plugin used to live under ``<repo>/plugins/example-dashboard/``
    and was loaded for every install, putting an "Example" tab in every
    user's sidebar. It is now a tests-only fixture: any test that needs
    ``/api/plugins/example/hello`` or ``/dashboard-plugins/example/...``
    requests this fixture so the plugin appears only for that test's
    isolated ``HERMES_HOME``.

    The user-plugin source is preferred over a transient
    ``HERMES_BUNDLED_PLUGINS`` override because the bundled dir is
    resolved per-call (other tests in the suite implicitly rely on the
    real bundled plugins — kanban, hermes-achievements, model providers
    — being available, and globally swapping that root would yank them
    all). User plugins are first in the discovery search order, so
    laying down the fixture here is enough.
    """
    from hermes_constants import get_hermes_home
    from hermes_cli import web_server

    user_plugins_dir = get_hermes_home() / "plugins"
    user_plugins_dir.mkdir(parents=True, exist_ok=True)
    dst = user_plugins_dir / "example-dashboard"
    if dst.exists():
        shutil.rmtree(dst)
    shutil.copytree(_EXAMPLE_PLUGIN_FIXTURE, dst)

    # The dashboard now gates user-plugin asset serving + backend import
    # behind the ``plugins.enabled`` allow-list (GHSA-mcfc-hp25-cjv7).
    # An installed-but-not-enabled user plugin has its API mount skipped
    # and its assets 404'd — which is the whole point of the gate. These
    # fixtures exist to exercise the *serving* paths, so opt the example
    # plugin in exactly as a real operator would with `hermes plugins
    # enable example`.
    from hermes_cli.config import load_config, save_config
    _cfg = load_config()
    _plugins_cfg = _cfg.setdefault("plugins", {})
    _enabled = _plugins_cfg.get("enabled")
    if not isinstance(_enabled, list):
        _enabled = []
    if "example" not in _enabled:
        _enabled.append("example")
    _plugins_cfg["enabled"] = _enabled
    save_config(_cfg)

    # Snapshot the existing routes BEFORE mounting so we can:
    #   1. Identify the routes the mount call appends.
    #   2. Restore the original list on teardown — otherwise leftover
    #      ``/api/plugins/example/*`` routes leak into subsequent tests
    #      and start serving requests against a torn-down HERMES_HOME.
    app = web_server.app
    original_routes = list(app.router.routes)

    # Bust the module-level cache and re-discover so the example plugin
    # shows up in `_get_dashboard_plugins()`. `_mount_plugin_api_routes`
    # imports the plugin's `plugin_api.py` and ``include_router``s its
    # FastAPI router under ``/api/plugins/example/*``. The static-asset
    # route at ``/dashboard-plugins/<name>/<path>`` reads the plugins
    # list dynamically per request, so the rescan alone is enough for
    # the static-asset tests; the API auth tests additionally need the
    # route reorder below.
    web_server._dashboard_plugins_cache = None
    web_server._get_dashboard_plugins(force_rescan=True)
    _web_server_dashboard._mount_plugin_api_routes()

    # ``include_router`` appends the new routes to the END of
    # ``app.router.routes``. That works fine at import time — the SPA
    # catch-all ``mount_spa(app)`` registers AFTER the initial mount
    # call — but when we mount mid-flight the catch-all is already in
    # place, so the new ``/api/plugins/example/*`` route loses the
    # match-order race and we get a 404. Move the newly-appended routes
    # to the front of the list so FastAPI matches them first. They're
    # path-prefixed to ``/api/plugins/example/`` and can't shadow
    # anything else.
    new_routes = [r for r in app.router.routes if r not in original_routes]
    for route in new_routes:
        app.router.routes.remove(route)
    for offset, route in enumerate(new_routes):
        app.router.routes.insert(offset, route)

    try:
        yield
    finally:
        # Restore the original route list — drops the example plugin's
        # routes so the next test sees a clean app — and clear the
        # cache for the same reason.
        app.router.routes[:] = original_routes
        web_server._dashboard_plugins_cache = None


# ---------------------------------------------------------------------------
# reload_env tests
# ---------------------------------------------------------------------------


class TestReloadEnv:
    """Tests for reload_env() — re-reads .env into os.environ."""

    def test_adds_new_vars(self, tmp_path):
        """reload_env() adds vars from .env that are not in os.environ."""
        env_file = tmp_path / ".env"
        env_file.write_text("TEST_RELOAD_VAR=hello123\n", encoding="utf-8")
        with patch.dict(reload_env.__globals__, {"get_env_path": lambda: env_file}):
            os.environ.pop("TEST_RELOAD_VAR", None)
            count = reload_env()
            assert count >= 1
            assert os.environ.get("TEST_RELOAD_VAR") == "hello123"
        os.environ.pop("TEST_RELOAD_VAR", None)


    def test_removes_deleted_known_vars(self, tmp_path):
        """reload_env() removes known Hermes vars not present in .env."""
        env_file = tmp_path / ".env"
        env_file.write_text("")  # empty .env
        # Pick a known key from OPTIONAL_ENV_VARS
        known_key = next(iter(OPTIONAL_ENV_VARS.keys()))
        with patch.dict(reload_env.__globals__, {"get_env_path": lambda: env_file}):
            os.environ[known_key] = "stale_value"
            count = reload_env()
            assert known_key not in os.environ
            assert count >= 1


# ---------------------------------------------------------------------------
# redact_key tests
# ---------------------------------------------------------------------------


class TestRedactKey:
    def test_long_key_shows_prefix_suffix(self):
        result = redact_key("sk-1234567890abcdef")
        assert result.startswith("sk-1")
        assert result.endswith("cdef")
        assert "..." in result

    def test_short_key_fully_masked(self):
        assert redact_key("short") == "***"


class TestSessionTokenInjection:
    """The desktop shell mints HERMES_DASHBOARD_SESSION_TOKEN and signs its
    /api + /api/ws calls with it. The backend must adopt that token, else every
    desktop request 401s ("gateway is offline"). A main-merge once silently
    dropped this read — this guards the contract, not a literal value.
    """

    def test_honors_injected_token(self, monkeypatch):
        import hermes_cli.web_server as ws

        original_app = ws.app
        original_token = ws._SESSION_TOKEN
        monkeypatch.setenv("HERMES_DASHBOARD_SESSION_TOKEN", "desktop-seeded-token")
        assert ws._resolve_session_token() == "desktop-seeded-token"
        # No module reload: the loaded app and its adopted token are untouched.
        assert ws.app is original_app
        assert ws._SESSION_TOKEN == original_token


    def test_session_token_resolution_preserves_loaded_app_auth(self, monkeypatch):
        import hermes_cli.web_server as ws
        from starlette.testclient import TestClient

        original_app = ws.app
        original_header_name = ws._SESSION_HEADER_NAME
        original_token = ws._SESSION_TOKEN
        monkeypatch.setenv("HERMES_DASHBOARD_SESSION_TOKEN", "desktop-seeded-token")
        assert ws._resolve_session_token() == "desktop-seeded-token"
        monkeypatch.delenv("HERMES_DASHBOARD_SESSION_TOKEN", raising=False)
        with patch.object(ws.secrets, "token_urlsafe", return_value="generated-token"):
            assert ws._resolve_session_token() == "generated-token"

        client = TestClient(original_app)
        client.headers[original_header_name] = original_token
        assert client.get("/api/__session_token_probe").status_code == 404
        assert ws.app is original_app
        assert ws._SESSION_HEADER_NAME == original_header_name
        assert ws._SESSION_TOKEN == original_token


# ---------------------------------------------------------------------------
# web_server tests (FastAPI endpoints)
# ---------------------------------------------------------------------------


class TestWebServerEndpoints:
    """Test the FastAPI REST endpoints using Starlette TestClient."""

    @pytest.fixture(autouse=True)
    def _setup_test_client(self, monkeypatch, _isolate_hermes_home):
        """Create a TestClient and isolate the state DB under the test HERMES_HOME."""
        try:
            from starlette.testclient import TestClient
        except ImportError:
            pytest.skip("fastapi/starlette not installed")

        import hermes_state
        from hermes_constants import get_hermes_home
        from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN

        monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db")

        self.client = TestClient(app)
        self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN

    @pytest.mark.requires_wal
    def test_get_sessions_poll_preserves_pending_wal(self):
        """Repeated GET-only polls must not checkpoint another writer's WAL."""
        import sqlite3

        from hermes_constants import get_hermes_home
        from hermes_state import SessionDB

        _web_server_sessions._last_auto_archive_check.clear()
        db_path = get_hermes_home() / "state.db"
        wal_path = Path(f"{db_path}-wal")
        writer = SessionDB(db_path=db_path)
        monitor = None
        try:
            writer._conn.execute("PRAGMA wal_autocheckpoint=0")
            writer.create_session("poll-wal", source="cli")
            writer.append_message(
                "poll-wal",
                role="user",
                content="pending writer frame " + ("x" * 65_536),
            )

            monitor = sqlite3.connect(str(db_path), isolation_level=None)
            wal_bytes_before = wal_path.stat().st_size
            data_version_before = monitor.execute(
                "PRAGMA data_version"
            ).fetchone()[0]
            counts_before = monitor.execute(
                "SELECT (SELECT COUNT(*) FROM sessions), "
                "(SELECT COUNT(*) FROM messages)"
            ).fetchone()

            responses = [
                self.client.get(
                    "/api/sessions?limit=50&offset=0&order=created"
                )
                for _ in range(3)
            ]

            wal_bytes_after = wal_path.stat().st_size
            data_version_after = monitor.execute(
                "PRAGMA data_version"
            ).fetchone()[0]
            counts_after = monitor.execute(
                "SELECT (SELECT COUNT(*) FROM sessions), "
                "(SELECT COUNT(*) FROM messages)"
            ).fetchone()

            assert all(response.status_code == 200 for response in responses)
            assert all(response.json()["total"] == 1 for response in responses)
            assert wal_bytes_before > 0
            assert wal_bytes_after == wal_bytes_before
            assert data_version_after == data_version_before
            assert counts_after == counts_before == (1, 1)
        finally:
            if monitor is not None:
                monitor.close()
            writer.close()

    def test_get_sessions_transient_ioerr_is_503(self, monkeypatch):
        """Busy store, not a gone store: the desktop keeps the list it has."""
        import sqlite3


        def boom(*_args, **_kwargs):
            raise sqlite3.OperationalError("disk I/O error")

        monkeypatch.setattr(_web_server_sessions, "_open_session_db_for_profile", boom)
        assert self.client.get("/api/sessions?limit=1&offset=0").status_code == 503

    def test_get_sessions_non_transient_operational_error_is_500(self, monkeypatch):
        import sqlite3


        def boom(*_args, **_kwargs):
            raise sqlite3.OperationalError("no such table: sessions")

        monkeypatch.setattr(_web_server_sessions, "_open_session_db_for_profile", boom)
        assert self.client.get("/api/sessions?limit=1&offset=0").status_code == 500


    def test_get_sessions_auto_archive_uses_maintenance_writer(self):
        from hermes_cli.config import load_config, save_config
        from hermes_constants import get_hermes_home
        from hermes_state import SessionDB

        db_path = get_hermes_home() / "state.db"
        seed = SessionDB(db_path=db_path)
        try:
            seed.create_session("stale", source="cli")
            seed.create_session("fresh", source="cli")
            seed._conn.execute(
                "UPDATE sessions SET started_at = ? WHERE id = ?",
                (time.time() - 30 * 86400, "stale"),
            )
        finally:
            seed.close()

        config = load_config()
        config.setdefault("sessions", {}).update(
            {
                "auto_archive": True,
                "auto_archive_days": 3,
                "min_interval_hours": 0,
            }
        )
        save_config(config)
        _web_server_sessions._last_auto_archive_check.clear()

        response = self.client.get("/api/sessions?limit=50&offset=0")

        assert response.status_code == 200
        assert [row["id"] for row in response.json()["sessions"]] == ["fresh"]
        verify = SessionDB(db_path=db_path, read_only=True)
        try:
            assert verify.get_session("stale")["archived"] == 1
            assert verify.get_meta("last_auto_archive")
        finally:
            verify.close()

    def test_get_sessions_fresh_store_returns_empty_list(self):
        response = self.client.get("/api/sessions?limit=50&offset=0")

        assert response.status_code == 200
        assert response.json()["sessions"] == []
        assert response.json()["total"] == 0

    @pytest.mark.parametrize(
        "missing_column", ["archived", "pinned", "last_activity_at"]
    )
    def test_get_sessions_heals_stale_schema_store(self, missing_column):
        import sqlite3

        from hermes_constants import get_hermes_home
        from hermes_state import SessionDB

        db_path = get_hermes_home() / "state.db"
        seed = SessionDB(db_path=db_path)
        try:
            seed.create_session("stale-schema", source="cli")
        finally:
            seed.close()

        legacy = sqlite3.connect(str(db_path))
        try:
            # SQLite refuses DROP COLUMN while an index references the
            # column; a pre-column legacy store has neither.
            legacy.execute("DROP INDEX IF EXISTS idx_sessions_effective_activity")
            legacy.execute(f"ALTER TABLE sessions DROP COLUMN {missing_column}")
            legacy.commit()
        finally:
            legacy.close()

        response = self.client.get("/api/sessions?limit=50&offset=0")

        assert response.status_code == 200
        assert [row["id"] for row in response.json()["sessions"]] == [
            "stale-schema"
        ]
        healed = sqlite3.connect(str(db_path))
        try:
            columns = {
                row[1] for row in healed.execute("PRAGMA table_info(sessions)")
            }
        finally:
            healed.close()
        assert missing_column in columns

    def test_profiles_sidebar_heals_stale_schema_store(self):
        """The desktop's batched sidebar route must heal a stale store too.

        The shipped regression (#72424 aftermath): a store predating
        ``sessions.last_activity_at`` made every per-profile read raise
        "no such column", which this endpoint swallowed into its ``errors``
        array — the desktop rendered "No sessions yet" after `hermes update`
        until the user's first message forced a writable open elsewhere.
        """
        import sqlite3

        from hermes_constants import get_hermes_home
        from hermes_state import SessionDB

        db_path = get_hermes_home() / "state.db"
        seed = SessionDB(db_path=db_path)
        try:
            seed.create_session("sidebar-stale", source="cli")
            seed.append_message(
                session_id="sidebar-stale", role="user", content="hi"
            )
        finally:
            seed.close()

        legacy = sqlite3.connect(str(db_path))
        try:
            legacy.execute("DROP INDEX IF EXISTS idx_sessions_effective_activity")
            legacy.execute("ALTER TABLE sessions DROP COLUMN last_activity_at")
            legacy.commit()
        finally:
            legacy.close()

        response = self.client.get("/api/profiles/sessions/sidebar")

        assert response.status_code == 200
        payload = response.json()
        assert payload["errors"] == []
        assert [row["id"] for row in payload["recents"]["sessions"]] == [
            "sidebar-stale"
        ]

    def test_startup_eager_reconcile_heals_stale_store(self):
        """The lifespan's eager reconcile brings a stale store current.

        #79531/#80037: after `hermes update` an old-schema state.db used to
        stay stale until the first NEW session forced a writable open —
        every /api/sessions poll 500ed with "no such column" in between.
        The lifespan now schedules one writable open at startup; this
        exercises that worker directly against a store missing
        sessions.last_read_at and asserts the schema is brought current.
        """
        import sqlite3

        from hermes_constants import get_hermes_home
        from hermes_state import SessionDB

        db_path = get_hermes_home() / "state.db"
        seed = SessionDB(db_path=db_path)
        try:
            seed.create_session("eager-stale", source="cli")
        finally:
            seed.close()

        legacy = sqlite3.connect(str(db_path))
        try:
            legacy.execute("ALTER TABLE sessions DROP COLUMN last_read_at")
            legacy.commit()
        finally:
            legacy.close()

        _web_server_lifecycle._eager_reconcile_own_session_db()

        healed = sqlite3.connect(str(db_path))
        try:
            columns = {
                row[1] for row in healed.execute("PRAGMA table_info(sessions)")
            }
        finally:
            healed.close()
        assert "last_read_at" in columns

        # The healed store serves the full rich listing.
        db = SessionDB(db_path=db_path, read_only=True)
        try:
            rows = db.list_sessions_rich(limit=10, compact_rows=True)
        finally:
            db.close()
        assert [r["id"] for r in rows] == ["eager-stale"]

    def test_startup_eager_reconcile_is_read_only_on_a_healthy_store(self, monkeypatch):
        """A current-schema store gets NO writable open from the dashboard (#107688).

        The gateway owns the writer; a second writable SessionDB from the
        dashboard (close-time checkpoint, possible FTS rebuild) is the
        two-writer corruption vector. Only the stale-schema heal may write.
        """
        import hermes_state
        from hermes_constants import get_hermes_home
        from hermes_state import SessionDB

        SessionDB(db_path=get_hermes_home() / "state.db").close()

        writable_opens = []
        real_init = SessionDB.__init__

        def spy(self, *args, **kwargs):
            if not kwargs.get("read_only"):
                writable_opens.append(kwargs)
            return real_init(self, *args, **kwargs)

        monkeypatch.setattr(hermes_state.SessionDB, "__init__", spy)
        _web_server_lifecycle._eager_reconcile_own_session_db()

        assert writable_opens == []

    def test_startup_eager_reconcile_never_raises(self, monkeypatch):
        """A store the eager reconcile cannot open must not break startup."""
        import sqlite3 as sqlite3_module

        import hermes_state


        def boom(*args, **kwargs):
            raise sqlite3_module.OperationalError("database is locked")

        monkeypatch.setattr(hermes_state, "SessionDB", boom)
        # Must swallow — reads fall back to the per-poll probe heal.
        _web_server_lifecycle._eager_reconcile_own_session_db()

    def test_heal_gives_up_when_reconcile_cannot_fix_the_store(self, monkeypatch):
        """A probe failure reconciliation can't cure must not retry forever.

        The writable heal is a full SessionDB init against a possibly-live
        DB. If the store is STILL behind the probe afterwards (schema problem
        ADD COLUMN can't express), retrying that init on every sidebar poll
        would hammer the DB for nothing: serve reads probe-less instead, warn
        once, and never pay the writable open for that store again.
        """
        from hermes_constants import get_hermes_home
        from hermes_state import SessionDB

        db_path = get_hermes_home() / "state.db"
        seed = SessionDB(db_path=db_path)
        try:
            seed.create_session("unfixable", source="cli")
        finally:
            seed.close()

        # A column no SCHEMA_SQL declares: the heal's writable reconcile
        # cannot add it, so the re-probe keeps failing.
        monkeypatch.setattr(
            _web_server_sessions,
            "_session_db_read_probe_statements",
            lambda: ('SELECT "sessions"."not_a_real_column" FROM "sessions" LIMIT 0',),
        )
        monkeypatch.setattr(_web_server_sessions, "_session_db_heal_exhausted", set())
        monkeypatch.setattr(_web_server_sessions, "_session_db_heal_warned", set())

        writable_opens = []

        import hermes_state

        original_init = hermes_state.SessionDB.__init__

        def counting_init(self, *args, **kwargs):
            if not kwargs.get("read_only", False):
                writable_opens.append(1)
            return original_init(self, *args, **kwargs)

        # web_server imports SessionDB inside the function body, so patching
        # the class on hermes_state covers every open the helper makes.
        monkeypatch.setattr(hermes_state.SessionDB, "__init__", counting_init)

        # First open: probe fails -> one writable heal -> re-probe fails ->
        # exhausted. Still returns a usable read-only handle.
        db = _web_server_sessions._open_session_db_for_profile(None, read_only=True)
        try:
            assert db.list_sessions_rich(limit=10, compact_rows=True)
        finally:
            db.close()
        assert len(writable_opens) == 1
        assert str(db_path) in _web_server_sessions._session_db_heal_exhausted

        # Second open: probe skipped, NO further writable opens.
        db = _web_server_sessions._open_session_db_for_profile(None, read_only=True)
        try:
            assert db.list_sessions_rich(limit=10, compact_rows=True)
        finally:
            db.close()
        assert len(writable_opens) == 1

    def test_generic_corruption_does_not_trigger_writable_heal(
        self, tmp_path, monkeypatch
    ):
        """Unscoped SQLITE_CORRUPT must not escalate a dashboard read to writes."""
        import sqlite3

        import hermes_state

        db_path = tmp_path / "state.db"
        db_path.write_bytes(b"not-empty")
        opens = []

        def corrupt_open(*_args, **kwargs):
            opens.append(kwargs.get("read_only", False))
            raise sqlite3.DatabaseError("database disk image is malformed")

        monkeypatch.setattr(hermes_state, "SessionDB", corrupt_open)

        with pytest.raises(sqlite3.DatabaseError, match="disk image is malformed"):
            _web_server_sessions._open_session_db_at_path(db_path, read_only=True)

        assert opens == [True]

    def test_decode_error_triggers_writable_heal(self, tmp_path, monkeypatch):
        """UnicodeDecodeError — pysqlite failing to decode SQLite's own error
        message over corrupt file bytes (#98924) — must route through the
        same one-writable-open heal as malformed schema."""
        import hermes_state

        db_path = tmp_path / "state.db"
        db_path.write_bytes(b"not-empty")
        opens = []

        class _OkDB:
            _conn = None

            def close(self):
                pass

        def scripted_open(*_args, **kwargs):
            opens.append(kwargs.get("read_only", False))
            if opens == [True]:
                raise UnicodeDecodeError("utf-8", b"\x81", 0, 1, "invalid start byte")
            return _OkDB()

        monkeypatch.setattr(hermes_state, "SessionDB", scripted_open)

        db = _web_server_sessions._open_session_db_at_path(db_path, read_only=True)

        assert isinstance(db, _OkDB)
        assert opens == [True, False, True]

    def test_get_sessions_zero_byte_store_returns_empty_list(self):
        from hermes_constants import get_hermes_home

        db_path = get_hermes_home() / "state.db"
        db_path.parent.mkdir(parents=True, exist_ok=True)
        db_path.touch()

        response = self.client.get("/api/sessions?limit=50&offset=0")

        assert response.status_code == 200
        assert response.json()["sessions"] == []
        assert response.json()["total"] == 0

    def test_concurrent_first_load_reads_all_succeed_on_fresh_store(self):
        from concurrent.futures import ThreadPoolExecutor

        paths = [
            "/api/sessions?limit=50&offset=0",
            "/api/sessions/stats",
            "/api/sessions/empty/count",
            "/api/sessions?limit=10&offset=0&order=recent",
        ] * 2
        with ThreadPoolExecutor(max_workers=8) as pool:
            responses = list(pool.map(self.client.get, paths))

        assert [response.status_code for response in responses] == [
            200
        ] * len(paths)


    def test_messaging_platforms_profile_scopes_gateway_reads(self, monkeypatch):
        """?profile=<name> must resolve liveness from the profile's own home.

        The gateway status readers resolve process-level paths and ignore the
        HERMES_HOME contextvar override (#56986), so /api/messaging/platforms
        has to pass the profile directory explicitly — otherwise it reports a
        DIFFERENT profile's gateway as this profile's, which hides a real
        outage behind a false "connected" (issue #71211).
        """
        import hermes_cli.web_server as web_server
        from hermes_cli import profiles as profiles_mod

        worker_home = profiles_mod.get_profile_dir("worker")
        worker_home.mkdir(parents=True)
        (worker_home / "config.yaml").touch()  # identity marker: bare dirs are not profiles

        seen = {}

        def _pid(pid_path=None, **kw):
            # The served-profile probe also verifies the DEFAULT home's gateway identity; the
            # contract here is that the worker's OWN pid file is what the scoped rung reads.
            seen.setdefault("pid_paths", []).append(pid_path)
            return None

        def _runtime(path=None):
            seen.setdefault("status_paths", []).append(path)
            return None

        def _runtime_pid(runtime=None, *, expected_home=None):
            seen.setdefault("expected_homes", []).append(expected_home)
            return None

        monkeypatch.setattr(_gw_status, "get_running_pid_cached", _pid)
        monkeypatch.setattr(_gw_status, "get_running_pid", _pid)
        monkeypatch.setattr(_gw_status, "read_runtime_status", _runtime)
        monkeypatch.setattr(_gw_status, "get_runtime_status_running_pid", _runtime_pid)
        monkeypatch.setattr(web_server, "_GATEWAY_HEALTH_URL", None)

        resp = self.client.get("/api/messaging/platforms?profile=worker")

        assert resp.status_code == 200
        assert worker_home / "gateway.pid" in seen["pid_paths"]
        assert worker_home / "gateway_state.json" in seen["status_paths"]
        assert worker_home in seen["expected_homes"]


    def test_gateway_drain_bad_action_400(self):
        resp = self.client.post("/api/gateway/drain", json={"action": "explode"})
        assert resp.status_code == 400


    @staticmethod
    def _provider_field_map(payload):
        return {field["key"]: field for field in payload["fields"]}


    def test_openviking_dashboard_persists_typed_recall_values(self):
        from hermes_cli.config import load_config

        resp = self.client.put(
            "/api/memory/providers/openviking/config",
            json={
                "values": {
                    "endpoint": "http://127.0.0.1:1933",
                    "recall_limit": "12",
                    "recall_score_threshold": "0.42",
                    "recall_max_injected_chars": "8000",
                    "profile_token_budget": "7000",
                    "recall_timeout_seconds": "2.5",
                    "recall_request_timeout_seconds": "1.5",
                    "recall_full_read_limit": "5",
                    "recall_prefer_abstract": True,
                    "recall_resources": False,
                }
            },
        )

        assert resp.status_code == 200
        config = load_config()["memory"]["openviking"]
        assert config["recall_limit"] == 12
        assert config["recall_score_threshold"] == 0.42
        assert config["profile_token_budget"] == 7000
        assert config["recall_prefer_abstract"] is True
        assert config["recall_resources"] is False

    def test_openviking_dashboard_rejects_out_of_range_recall_value(self):
        resp = self.client.put(
            "/api/memory/providers/openviking/config",
            json={
                "values": {
                    "endpoint": "http://127.0.0.1:1933",
                    "recall_limit": 101,
                }
            },
        )

        assert resp.status_code == 400

    def test_openviking_dashboard_rejects_blocked_endpoint_before_saving(self):
        from hermes_cli.config import load_config

        resp = self.client.put(
            "/api/memory/providers/openviking/config",
            json={
                "values": {
                    "endpoint": "http://169.254.169.254/latest/meta-data/credential",
                }
            },
        )

        assert resp.status_code == 400
        assert "credential" not in resp.json()["detail"]
        memory_config = load_config().get("memory", {})
        assert "openviking" not in memory_config


    # A user-installed memory provider with a DECLARED config surface (``config_schema.py``, flat
    # ``<home>/<name>/config.json`` storage) and a live ``get_config_schema``/``save_config`` pair.
    # Bundled providers no longer ship a flat-storage declared schema (hindsight moved to the
    # plugin catalog), so the generic router paths are exercised against this synthetic one.
    _FLATPROV_INIT = """
import json
from pathlib import Path
from agent.memory_provider import MemoryProvider


class FlatProvMemoryProvider(MemoryProvider):
    @property
    def name(self):
        return "flatprov"

    def is_available(self):
        return True

    def initialize(self, session_id, **kwargs):
        pass

    def get_tool_schemas(self):
        return []

    def get_config_schema(self):
        return [
            {"key": "mode", "label": "Mode", "choices": ["cloud", "local_external"], "default": "cloud"},
            {"key": "api_url", "label": "API URL", "default": ""},
            {"key": "api_key", "label": "API key", "secret": True, "env_var": "FLATPROV_API_KEY"},
            {"key": "bank_id", "label": "Bank", "default": "hermes"},
            {"key": "recall_budget", "label": "Budget", "choices": ["low", "mid", "high"], "default": "mid"},
        ]

    def save_config(self, values, hermes_home):
        path = Path(hermes_home) / "flatprov" / "config.json"
        path.parent.mkdir(parents=True, exist_ok=True)
        existing = json.loads(path.read_text()) if path.exists() else {}
        existing.update(values)
        path.write_text(json.dumps(existing))
"""
    _FLATPROV_SCHEMA = """
from plugins.memory.config_schema import (
    KIND_SECRET, KIND_SELECT, KIND_TEXT, ProviderConfigSchema, ProviderField, ProviderFieldOption,
)

CONFIG_SCHEMA = ProviderConfigSchema(
    name="flatprov",
    label="Flat Provider",
    fields=(
        ProviderField(key="mode", label="Mode", kind=KIND_SELECT, description="", default="cloud",
                      options=(ProviderFieldOption("cloud", "Cloud"), ProviderFieldOption("local_external", "Local"))),
        ProviderField(key="api_url", label="API URL", kind=KIND_TEXT, description=""),
        ProviderField(key="api_key", label="API key", kind=KIND_SECRET, description="", env_key="FLATPROV_API_KEY"),
    ),
)
"""

    def _install_flatprov(self):
        from hermes_constants import get_hermes_home

        plugin_dir = get_hermes_home() / "plugins" / "flatprov"
        plugin_dir.mkdir(parents=True, exist_ok=True)
        (plugin_dir / "__init__.py").write_text(self._FLATPROV_INIT, encoding="utf-8")
        (plugin_dir / "config_schema.py").write_text(self._FLATPROV_SCHEMA, encoding="utf-8")
        return plugin_dir

    def test_declared_surface_put_writes_config_and_secret(self):
        from hermes_constants import get_hermes_home
        from hermes_cli.config import load_env

        self._install_flatprov()
        resp = self.client.put(
            "/api/memory/providers/flatprov/config?surface=declared",
            json={
                "values": {
                    "mode": "local_external",
                    "api_url": "http://localhost:8888",
                    "api_key": "fp-declared-key",
                }
            },
        )

        assert resp.status_code == 200
        assert resp.json() == {"ok": True}
        assert load_env()["FLATPROV_API_KEY"] == "fp-declared-key"

        config_path = get_hermes_home() / "flatprov" / "config.json"
        provider_config = json.loads(config_path.read_text(encoding="utf-8"))
        assert provider_config["mode"] == "local_external"
        assert provider_config["api_url"] == "http://localhost:8888"
        assert "api_key" not in provider_config


    def test_post_memory_provider_setup_routes_python_deps_through_pm(self, monkeypatch):
        """Dashboard dependency setup publishes through PM, never direct pip."""
        import subprocess as _subprocess

        import hermes_cli.web_server as web_server
        from hermes_cli import memory_setup

        prepared = []
        monkeypatch.setattr(
            memory_setup,
            "prepare_memory_provider_dependencies",
            lambda name: (prepared.append(name) or ({}, "installed")),
        )

        # Any direct pip/uv subprocess from the memory-provider pip path is
        # a regression; external-dep checks may still run subprocess, so only
        # trip on pip-flavored commands.
        real_run = _subprocess.run

        def guarded_run(command, **kwargs):
            flat = command if isinstance(command, str) else " ".join(map(str, command))
            assert "pip install" not in flat, f"direct pip call leaked: {flat}"
            return real_run(command, **kwargs)

        monkeypatch.setattr(web_server.subprocess, "run", guarded_run)

        resp = self.client.post("/api/memory/providers/honcho/setup", json={"values": {}})

        assert resp.status_code == 200
        data = resp.json()
        pip_rows = [row for row in data["results"] if row["kind"] == "pip"]
        assert pip_rows and pip_rows[0]["status"] == "installed"
        assert pip_rows[0]["command"] == "hermes pm install"
        assert prepared == ["honcho"]


    def test_put_memory_provider_config_writes_config_and_secret(self):
        from hermes_constants import get_hermes_home
        from hermes_cli.config import load_config, load_env

        self._install_flatprov()
        resp = self.client.put(
            "/api/memory/providers/flatprov/config",
            json={
                "values": {
                    "mode": "local_external",
                    "api_url": "http://localhost:8888",
                    "api_key": "fp-test-key",
                    "bank_id": "ben-bank",
                    "recall_budget": "high",
                }
            },
        )

        assert resp.status_code == 200
        assert resp.json() == {"ok": True, "active": "flatprov"}
        assert load_config()["memory"]["provider"] == "flatprov"
        assert load_env()["FLATPROV_API_KEY"] == "fp-test-key"

        config_path = get_hermes_home() / "flatprov" / "config.json"
        provider_config = json.loads(config_path.read_text(encoding="utf-8"))
        assert provider_config["mode"] == "local_external"
        assert provider_config["api_url"] == "http://localhost:8888"
        assert provider_config["bank_id"] == "ben-bank"
        assert provider_config["recall_budget"] == "high"
        assert "api_key" not in provider_config


    def test_get_memory_provider_config_does_not_return_secret(self):
        self._install_flatprov()
        self.client.put(
            "/api/memory/providers/flatprov/config",
            json={
                "values": {
                    "mode": "cloud",
                    "api_url": "https://api.example.invalid",
                    "api_key": "secret-value",
                    "bank_id": "hermes",
                    "recall_budget": "mid",
                }
            },
        )

        resp = self.client.get("/api/memory/providers/flatprov/config")

        assert resp.status_code == 200
        data = resp.json()
        fields = self._provider_field_map(data)
        assert fields["api_key"]["is_set"] is True
        assert fields["api_key"]["value"] == ""
        assert "secret-value" not in json.dumps(data)


    # ── Memory provider config (Honcho host-block backend) ──────────────

    @pytest.fixture(autouse=True)
    def _isolate_honcho_config(self):
        # Honcho tests write the suite-wide HERMES_HOME honcho.json; snapshot and
        # restore it so provider status/config state never leaks across tests.
        from hermes_constants import get_hermes_home

        path = get_hermes_home() / "honcho.json"
        before = path.read_bytes() if path.exists() else None
        yield
        if before is None:
            path.unlink(missing_ok=True)
        else:
            path.write_bytes(before)

    @staticmethod
    def _seed_local_honcho(cfg=None):
        from hermes_constants import get_hermes_home

        path = get_hermes_home() / "honcho.json"
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(cfg if cfg is not None else {}), encoding="utf-8")
        return path


    def test_put_honcho_writes_host_block_root_and_secret(self, monkeypatch, tmp_path):
        monkeypatch.setenv("HOME", str(tmp_path))
        monkeypatch.setenv("HONCHO_API_KEY", "guard")
        monkeypatch.delenv("HONCHO_API_KEY")
        self._seed_local_honcho()
        from hermes_constants import get_hermes_home
        from hermes_cli.config import load_config, load_env

        resp = self.client.put(
            "/api/memory/providers/honcho/config?surface=declared",
            json={
                "values": {
                    "apiKey": "hch-test-key",
                    "baseUrl": "https://honcho.example.dev",
                    "environment": "local",
                    "workspace": "myws",
                    "peerName": "eri",
                    "aiPeer": "hermes",
                    "sessionStrategy": "per-repo",
                }
            },
        )

        assert resp.status_code == 200
        assert resp.json() == {"ok": True}
        assert load_config()["memory"]["provider"] == "honcho"
        assert load_env()["HONCHO_API_KEY"] == "hch-test-key"

        cfg = json.loads((get_hermes_home() / "honcho.json").read_text(encoding="utf-8"))
        # baseUrl is root-scoped; the rest live in the active host block.
        assert cfg["baseUrl"] == "https://honcho.example.dev"
        assert cfg["hosts"]["hermes"]["workspace"] == "myws"
        assert cfg["hosts"]["hermes"]["peerName"] == "eri"
        assert cfg["hosts"]["hermes"]["environment"] == "local"
        assert cfg["hosts"]["hermes"]["sessionStrategy"] == "per-repo"
        # The key lands where the client reads first; GET keeps it write-only.
        assert cfg["hosts"]["hermes"]["apiKey"] == "hch-test-key"


    def test_get_honcho_config_does_not_return_secret(self, monkeypatch, tmp_path):
        monkeypatch.setenv("HOME", str(tmp_path))
        monkeypatch.setenv("HONCHO_API_KEY", "guard")
        monkeypatch.delenv("HONCHO_API_KEY")
        self._seed_local_honcho()

        self.client.put(
            "/api/memory/providers/honcho/config?surface=declared",
            json={"values": {"apiKey": "secret-value"}},
        )

        resp = self.client.get("/api/memory/providers/honcho/config?surface=declared")

        assert resp.status_code == 200
        data = resp.json()
        fields = self._provider_field_map(data)
        assert fields["apiKey"]["is_set"] is True
        assert fields["apiKey"]["value"] == ""
        assert "secret-value" not in json.dumps(data)


    # ── GET /api/media (remote image display) ───────────────────────────


    def test_get_media_requires_auth(self):
        from hermes_cli.web_server import _SESSION_HEADER_NAME

        resp = self.client.get(
            "/api/media",
            params={"path": "/tmp/x.png"},
            headers={_SESSION_HEADER_NAME: "wrong-token"},
        )
        assert resp.status_code == 401

    # ── GET /api/media/proxy (client-blocked CDN fallback, #74564) ──────


    def test_media_proxy_requires_auth(self):
        from hermes_cli.web_server import _SESSION_HEADER_NAME

        resp = self.client.get(
            "/api/media/proxy",
            params={"url": "https://v3.fal.media/x.png"},
            headers={_SESSION_HEADER_NAME: "wrong-token"},
        )
        assert resp.status_code == 401

    def test_media_proxy_rejects_disallowed_hosts_and_schemes(self):
        for bad in (
            "https://evil.example.com/img.png",
            "https://sub.fal.media.evil.com/img.png",
            "file:///etc/passwd",
            "not a url",
            "",
        ):
            resp = self.client.get("/api/media/proxy", params={"url": bad})
            assert resp.status_code in (400, 403), (bad, resp.status_code)

    def test_media_proxy_fetches_allowlisted_image_and_returns_data_url(self, monkeypatch):
        png_bytes = b"\x89PNG\r\n\x1a\n" + b"0" * 8

        class _Resp:
            status_code = 200
            headers = {"content-type": "image/png"}
            content = png_bytes

        class _Client:
            def __init__(self, *a, **k):
                pass

            async def __aenter__(self):
                return self

            async def __aexit__(self, *a):
                return False

            async def get(self, url):
                assert url == "https://v3.fal.media/media/abc123"
                return _Resp()

        import hermes_cli.web_routers.files as files_router

        monkeypatch.setattr(files_router, "_require_token", lambda request: None, raising=False)
        # The route imports httpx locally; patch the module it resolves from.
        import httpx

        monkeypatch.setattr(httpx, "AsyncClient", _Client, raising=False)

        resp = self.client.get(
            "/api/media/proxy", params={"url": "https://v3.fal.media/media/abc123"}
        )
        assert resp.status_code == 200
        import base64

        assert resp.json()["data_url"] == (
            "data:image/png;base64," + base64.b64encode(png_bytes).decode("ascii")
        )

    def test_media_proxy_rejects_non_image_content_type(self, monkeypatch):
        class _Resp:
            status_code = 200
            headers = {"content-type": "text/html"}
            content = b"<html>nope</html>"

        class _Client:
            def __init__(self, *a, **k):
                pass

            async def __aenter__(self):
                return self

            async def __aexit__(self, *a):
                return False

            async def get(self, url):
                return _Resp()

        import httpx

        monkeypatch.setattr(httpx, "AsyncClient", _Client, raising=False)

        resp = self.client.get(
            "/api/media/proxy", params={"url": "https://fal.media/media/abc123"}
        )
        assert resp.status_code == 415

    # ── POST /api/chat/image-upload (browser clipboard/drop images) ─────


    # ── Dashboard font override ─────────────────────────────────────────


    def test_import_sessions_endpoint_imports_exported_json(self):
        from hermes_state import SessionDB

        payload = {
            "id": "imported-web-session",
            "source": "cli",
            "title": "Imported from dashboard",
            "started_at": 100.0,
            "ended_at": 110.0,
            "end_reason": "complete",
            "messages": [
                {"role": "user", "content": "hello", "timestamp": 101.0},
                {"role": "assistant", "content": "hi", "timestamp": 102.0},
            ],
        }

        resp = self.client.post("/api/sessions/import", json={"sessions": [payload]})
        assert resp.status_code == 200
        data = resp.json()
        assert data["imported"] == 1
        assert data["skipped"] == 0

        db = SessionDB()
        try:
            session = db.get_session("imported-web-session")
            assert session["title"] == "Imported from dashboard"
            assert session["message_count"] == 2
            assert [m["content"] for m in db.get_messages("imported-web-session")] == [
                "hello",
                "hi",
            ]
        finally:
            db.close()

        duplicate = self.client.post("/api/sessions/import", json={"sessions": [payload]})
        assert duplicate.status_code == 200
        assert duplicate.json()["skipped_ids"] == ["imported-web-session"]

        invalid = self.client.post(
            "/api/sessions/import",
            json={"sessions": [{"source": "cli", "messages": []}]},
        )
        assert invalid.status_code == 400
        errors = invalid.json()["detail"]["errors"]
        assert [e["index"] for e in errors] == [0] and errors[0]["error"]


    def test_latest_descendant_survives_parent_cycle(self):
        """Regression for the #39140 CTE salvage: a corrupted parent chain
        that loops (a -> b -> a) must terminate (UNION dedup) instead of
        recursing forever like UNION ALL would."""
        from hermes_state import SessionDB

        db = SessionDB()
        try:
            db.create_session(session_id="cyc-a", source="cli")
            db.create_session(
                session_id="cyc-b", source="cli", parent_session_id="cyc-a"
            )
            db._conn.execute(
                "UPDATE sessions SET parent_session_id='cyc-b' WHERE id='cyc-a'"
            )
            db._conn.commit()
        finally:
            db.close()

        resp = self.client.get("/api/sessions/cyc-a/latest-descendant")
        assert resp.status_code == 200
        assert resp.json()["session_id"] == "cyc-b"

    def test_latest_descendant_never_resumes_into_a_subagent_or_branch_child(self):
        """#115092: after a ws_orphan_reap the dashboard resumes the predecessor's newest descendant. A
        subagent run (``_delegate_from``) or a /branch fork (``_branched_from``) is its own conversation and
        never listed as a continuation, so following it parks the user's chat in a hidden row; only
        compression continuations are followed."""
        from hermes_state import SessionDB

        db = SessionDB()
        try:
            db.create_session(session_id="primary", source="tui")
            db.create_session(session_id="primary-sub", source="tui", parent_session_id="primary",
                              model_config={"_delegate_from": "primary"})
            db.create_session(session_id="primary-fork", source="tui", parent_session_id="primary",
                              model_config={"_branched_from": "primary"})
            db.end_session("primary", "ws_orphan_reap")
            assert self.client.get("/api/sessions/primary/latest-descendant").json()["session_id"] == "primary"

            db._conn.execute("UPDATE sessions SET end_reason='compression' WHERE id='primary'")
            db._conn.commit()
            db.create_session(session_id="primary-cont", source="tui", parent_session_id="primary")
        finally:
            db.close()

        resp = self.client.get("/api/sessions/primary/latest-descendant")
        assert resp.status_code == 200
        assert resp.json()["session_id"] == "primary-cont"


    def test_update_hermes_returns_docker_guidance_without_spawning(self, monkeypatch):

        spawned = False

        def fail_spawn(*_args, **_kwargs):
            nonlocal spawned
            spawned = True
            raise AssertionError("docker update guard should not spawn hermes update")

        # Bypass the managed-externally gate so we reach the docker install check.
        monkeypatch.setattr(_web_server_files, "_dashboard_local_update_managed_externally", lambda: False)
        # The shared admission gate (#91277 Phase 3) resolves the install
        # method through hermes_cli.config directly.
        monkeypatch.setattr(
            "hermes_cli.config.detect_install_method", lambda *_a, **_k: "docker"
        )
        monkeypatch.setattr(_cfg_mod, "detect_install_method", lambda _root: "docker")
        monkeypatch.setattr(_web_server_gateway, "_spawn_hermes_action", fail_spawn)
        _web_server_gateway._ACTION_PROCS.pop("hermes-update", None)
        _web_server_gateway._ACTION_RESULTS.pop("hermes-update", None)

        resp = self.client.post("/api/hermes/update")

        assert resp.status_code == 200
        data = resp.json()
        assert data["ok"] is False
        assert data["name"] == "hermes-update"
        assert data["pid"] is None
        assert data["error"] == "docker_update_unsupported"
        assert spawned is False

        status = self.client.get("/api/actions/hermes-update/status")
        assert status.status_code == 200
        status_data = status.json()
        assert status_data["running"] is False
        assert status_data["exit_code"] == 1
        assert status_data["pid"] is None

    def test_update_hermes_returns_apt_guidance_without_spawning(self, monkeypatch):

        spawned = False

        def fail_spawn(*_args, **_kwargs):
            nonlocal spawned
            spawned = True
            raise AssertionError("APT-managed update guard should not spawn hermes update")

        monkeypatch.setattr(_web_server_files, "_dashboard_local_update_managed_externally", lambda: False)
        # The shared admission gate (#91277 Phase 3) resolves the install
        # method through hermes_cli.config directly, so patch it there (the
        # web_server module alias only feeds the /update/check endpoint).
        monkeypatch.setattr(
            "hermes_cli.config.detect_install_method", lambda *_a, **_k: "apt"
        )
        monkeypatch.setattr(_cfg_mod, "detect_install_method", lambda _root: "apt")
        monkeypatch.setattr(_web_server_gateway, "_spawn_hermes_action", fail_spawn)
        _web_server_gateway._ACTION_PROCS.pop("hermes-update", None)
        _web_server_gateway._ACTION_RESULTS.pop("hermes-update", None)

        resp = self.client.post("/api/hermes/update")

        assert resp.status_code == 200
        data = resp.json()
        assert data["ok"] is False
        assert data["pid"] is None
        assert data["error"] == "apt_update_required"
        assert data["update_command"]
        assert spawned is False

        check = self.client.get("/api/hermes/update/check")
        assert check.status_code == 200
        check_data = check.json()
        assert check_data["install_method"] == "apt"
        assert check_data["can_apply"] is False
        assert check_data["update_command"] == data["update_command"]

    def test_update_status_recovers_completed_result_after_dashboard_restart(self, monkeypatch, tmp_path):

        action_id = "c" * 32
        (tmp_path / "hermes-update.log").write_text(
            "=== hermes-update started 2026-08-17 11:19:34 ===\n"
            "pulling updates...\n",
            encoding="utf-8",
        )
        (tmp_path / "update.log").write_text(
            "=== hermes update started 2026-08-17T11:19:35 ===\n"
            "✓ Update complete!\n"
            f"=== hermes-update completed {action_id} ===\n",
            encoding="utf-8",
        )
        monkeypatch.setattr(_web_server_gateway, "_ACTION_LOG_DIR", tmp_path)
        monkeypatch.setattr(_web_server_gateway, "_ACTION_PROCS", {})
        monkeypatch.setattr(_web_server_gateway, "_ACTION_RESULTS", {})
        monkeypatch.setattr(_web_server_gateway, "_ACTION_COMMANDS", {})
        monkeypatch.setattr(_web_server_gateway, "_ACTION_IDS", {})

        status = self.client.get("/api/actions/hermes-update/status?lines=2000")

        assert status.status_code == 200
        data = status.json()
        assert data["running"] is False
        assert data["exit_code"] == 0
        assert data["action_id"] == action_id
        assert f"=== hermes-update completed {action_id} ===" in data["lines"]

    def test_update_hermes_spawns_with_action_id(self, monkeypatch):
        import hermes_cli.web_server as web_server

        class Proc:
            pid = 12345

        calls = []

        def fake_spawn(subcommand, name, *, env_overrides=None):
            calls.append((subcommand, name, env_overrides))
            return Proc()

        monkeypatch.setattr(_web_server_files, "_dashboard_local_update_managed_externally", lambda: False)
        monkeypatch.setattr(_cfg_mod, "detect_install_method", lambda _root: "git")
        monkeypatch.setattr(web_server.secrets, "token_hex", lambda _size: "a" * 32)
        monkeypatch.setattr(_web_server_gateway, "_spawn_hermes_action", fake_spawn)
        _web_server_gateway._ACTION_PROCS.pop("hermes-update", None)
        _web_server_gateway._ACTION_RESULTS.pop("hermes-update", None)

        resp = self.client.post("/api/hermes/update")

        assert resp.status_code == 200
        assert resp.json() == {
            "ok": True,
            "pid": 12345,
            "name": "hermes-update",
            "action_id": "a" * 32,
        }
        assert calls == [
            (["update"], "hermes-update", {"HERMES_ACTION_ID": "a" * 32})
        ]

    def test_update_hermes_reuses_running_action(self, monkeypatch):

        class Proc:
            pid = 24680

            def poll(self):
                return None

        monkeypatch.setattr(_web_server_files, "_dashboard_local_update_managed_externally", lambda: False)
        monkeypatch.setattr(_cfg_mod, "detect_install_method", lambda _root: "git")
        monkeypatch.setattr(
            _web_server_gateway,
            "_spawn_hermes_action",
            lambda *_args, **_kwargs: pytest.fail("must not spawn a duplicate update"),
        )
        _web_server_gateway._ACTION_PROCS["hermes-update"] = Proc()
        _web_server_gateway._ACTION_IDS["hermes-update"] = "b" * 32

        try:
            resp = self.client.post("/api/hermes/update")
        finally:
            _web_server_gateway._ACTION_PROCS.pop("hermes-update", None)
            _web_server_gateway._ACTION_IDS.pop("hermes-update", None)

        assert resp.status_code == 200
        assert resp.json() == {
            "ok": True,
            "pid": 24680,
            "name": "hermes-update",
            "already_running": True,
            "action_id": "b" * 32,
        }


    def test_model_set_maps_unknown_vendor_to_aggregator(self, monkeypatch):
        """A bare vendor name from analytics rows (no billing_provider) is not
        a Hermes provider — keep the user's aggregator instead of writing a
        provider that can never resolve credentials."""
        monkeypatch.setattr(
            "hermes_cli.model_cost_guard.expensive_model_warning",
            lambda *_args, **_kwargs: None,
        )
        from hermes_cli.config import load_config, save_config
        cfg = load_config()
        cfg["model"] = {"provider": "openrouter", "default": "openai/gpt-5.5"}
        save_config(cfg)

        resp = self.client.post(
            "/api/model/set",
            json={
                "scope": "main",
                "provider": "moonshotai",  # vendor prefix, not a provider
                "model": "moonshotai/kimi-k2.6",
            },
        )
        assert resp.status_code == 200
        data = resp.json()
        assert data["ok"] is True
        assert data["provider"] == "openrouter"
        assert data["model"] == "moonshotai/kimi-k2.6"


    def test_model_set_flips_a_stale_setup_record(self, monkeypatch):
        """POST /api/model/set landed a provider on disk; the serve process's boot record
        (``provider_configured: false`` since a failed boot-time mint) must follow at once, with
        the ``setup.ready`` broadcast, or the web chat stays gated on "need setup" until a restart
        (setup.status answers from the record)."""
        from hermes_cli import free_tier_bootstrap as fb

        fb.reset_for_tests()
        monkeypatch.setattr("hermes_cli.model_cost_guard.expensive_model_warning", lambda *_a, **_k: None)
        monkeypatch.setattr("agent.bedrock_adapter.has_aws_credentials", lambda: False)
        broadcasts = []
        monkeypatch.setattr(fb, "_broadcast", broadcasts.append)
        with fb._lock:
            fb._record = fb.SetupRecord(provider_configured=False, inference_provider="", free_tier_account=False,
                                        has_identity=False, other_providers=False)
            fb._started = True
            fb._done.set()
        try:
            resp = self.client.post(
                "/api/model/set",
                json={"scope": "main", "provider": "custom", "model": "local-model",
                      "base_url": "http://127.0.0.1:8081/v1", "api_key": "sk-local"},
            )
            assert resp.status_code == 200 and resp.json()["ok"] is True
            record = fb.current_record()
            assert record.provider_configured is True and record.other_providers is True
            assert record.inference_provider == "custom"
            assert broadcasts == [record]
        finally:
            fb.reset_for_tests()


    def test_reveal_env_var(self, tmp_path):
        """POST /api/env/reveal should return the real unredacted value."""
        from hermes_cli.config import save_env_value
        from hermes_cli.web_server import _SESSION_HEADER_NAME, _SESSION_TOKEN
        save_env_value("TEST_REVEAL_KEY", "super-secret-value-12345")
        resp = self.client.post(
            "/api/env/reveal",
            json={"key": "TEST_REVEAL_KEY"},
            headers={_SESSION_HEADER_NAME: _SESSION_TOKEN},
        )
        assert resp.status_code == 200
        data = resp.json()
        assert data["key"] == "TEST_REVEAL_KEY"
        assert data["value"] == "super-secret-value-12345"


    def test_reveal_env_var_custom_session_header_ignores_proxy_authorization(self, tmp_path):
        """A valid dashboard session header should coexist with proxy auth."""
        from hermes_cli.config import save_env_value
        from hermes_cli.web_server import _SESSION_HEADER_NAME, _SESSION_TOKEN

        save_env_value("TEST_REVEAL_PROXY_AUTH", "secret-value")
        resp = self.client.post(
            "/api/env/reveal",
            json={"key": "TEST_REVEAL_PROXY_AUTH"},
            headers={
                _SESSION_HEADER_NAME: _SESSION_TOKEN,
                "Authorization": "Basic dXNlcjpwYXNz",
            },
        )

        assert resp.status_code == 200
        assert resp.json()["value"] == "secret-value"

    def test_reveal_env_var_legacy_authorization_header_still_works(self, tmp_path):
        """Keep old dashboard bundles working while the new header rolls out."""
        from hermes_cli.config import save_env_value
        from hermes_cli.web_server import _SESSION_TOKEN

        save_env_value("TEST_REVEAL_LEGACY_AUTH", "secret-value")
        resp = self.client.post(
            "/api/env/reveal",
            json={"key": "TEST_REVEAL_LEGACY_AUTH"},
            headers={"Authorization": f"Bearer {_SESSION_TOKEN}"},
        )

        assert resp.status_code == 200


    def test_messaging_catalog_prefers_plugin_label_over_enum_pseudo_member(self):
        """A plugin platform that leaked into Platform.__members__ as a pseudo-
        member must still render with its plugin label, not a title-cased id.

        Regression: Platform("<plugin id>") caches a pseudo-member in the enum;
        the catalog iterated the enum FIRST and claimed the id with no plugin
        metadata, so bundled plugin platforms (irc, ntfy, photon, …) rendered
        as nameless "Irc"/"Ntfy" cards with empty descriptions.
        """
        from gateway.config import Platform
        from gateway.platform_registry import PlatformEntry, platform_registry

        entry = PlatformEntry(
            name="pseudofake",
            label="Pseudo Fake (plugin label)",
            adapter_factory=lambda cfg: None,
            check_fn=lambda: True,
            source="plugin",
        )
        platform_registry.register(entry)
        try:
            # Materialize the enum pseudo-member the way any earlier config
            # read would (Platform(value) on a registered plugin platform).
            member = Platform("pseudofake")
            assert member.value == "pseudofake"
            assert "PSEUDOFAKE" in Platform.__members__

            resp = self.client.get("/api/messaging/platforms")
            ids = {row["id"]: row for row in resp.json()["platforms"]}
            assert "pseudofake" in ids
            assert ids["pseudofake"]["name"] == "Pseudo Fake (plugin label)"
        finally:
            platform_registry.unregister("pseudofake")
            Platform._value2member_map_.pop("pseudofake", None)
            Platform._member_map_.pop("PSEUDOFAKE", None)


    def test_telegram_onboarding_apply_reports_restart_failure_after_save(
        self, monkeypatch
    ):
        from hermes_cli.config import load_config, load_env

        with _web_server_messaging._telegram_onboarding_lock:
            _web_server_messaging._telegram_onboarding_pairings.clear()

        def fake_request(method, path, *, body=None, bearer_token=None):
            if method == "POST":
                return {
                    "pairing_id": "pair-restart-fails",
                    "poll_token": "poll-secret",
                    "suggested_username": "hermes_pair_restart_fails_bot",
                    "deep_link": "https://t.me/newbot/HermesSetupBot/hermes_pair_restart_fails_bot",
                    "qr_payload": "https://t.me/newbot/HermesSetupBot/hermes_pair_restart_fails_bot",
                    "expires_at": "2027-05-18T00:00:00.000Z",
                }
            assert method == "GET"
            assert path == "/v1/telegram/pairings/pair-restart-fails"
            assert bearer_token == "poll-secret"
            return {
                "status": "ready",
                "bot_username": "hermes_pair_restart_fails_bot",
                "owner_user_id": 123456789,
                "token": "123456:SECRET",
            }

        monkeypatch.setattr(_web_server_messaging, "_telegram_onboarding_request_sync", fake_request)
        _web_server_gateway._ACTION_PROCS.pop("gateway-restart", None)

        def fail_spawn_action(subcommand, name):
            # The default home is named explicitly: a bare child would re-read the sticky active_profile.
            assert subcommand == ["-p", "default", "gateway", "restart"]
            assert name == "gateway-restart"
            raise RuntimeError("supervisor unavailable")

        monkeypatch.setattr(_web_server_gateway, "_spawn_hermes_action", fail_spawn_action)

        start = self.client.post("/api/messaging/telegram/onboarding/start", json={})
        assert start.status_code == 200
        ready = self.client.get("/api/messaging/telegram/onboarding/pair-restart-fails")
        assert ready.status_code == 200
        assert ready.json()["status"] == "ready"

        applied = self.client.post(
            "/api/messaging/telegram/onboarding/pair-restart-fails/apply",
            json={"allowed_user_ids": ["123456789"]},
        )

        assert applied.status_code == 200
        applied_data = applied.json()
        assert applied_data["ok"] is True
        assert applied_data["needs_restart"] is True
        assert applied_data["restart_started"] is False
        assert "supervisor unavailable" in applied_data["restart_error"]
        assert "token" not in applied_data
        env = load_env()
        assert env["TELEGRAM_BOT_TOKEN"] == "123456:SECRET"
        assert env["TELEGRAM_ALLOWED_USERS"] == "123456789"
        assert load_config()["platforms"]["telegram"]["enabled"] is True


    def test_unauthenticated_api_blocked(self):
        """API requests without the session token should be rejected."""
        from starlette.testclient import TestClient
        from hermes_cli.web_server import app
        # Create a client WITHOUT the dashboard session header
        unauth_client = TestClient(app)
        resp = unauth_client.get("/api/env")
        assert resp.status_code == 401
        resp = unauth_client.get("/api/config")
        assert resp.status_code == 401
        # Public endpoints should still work
        resp = unauth_client.get("/api/status")
        assert resp.status_code == 200
        resp = unauth_client.get("/api/dashboard/plugins")
        assert resp.status_code == 200
        resp = unauth_client.get("/api/dashboard/plugins/rescan")
        assert resp.status_code == 401
        resp = self.client.get("/api/dashboard/plugins/rescan")
        assert resp.status_code == 200


    def test_parse_model_ids_handles_openai_and_bare_shapes(self):
        """Model discovery must tolerate the common /v1/models shapes and
        never raise (so a slightly non-standard local endpoint still works)."""
        from hermes_cli.web_server_profiles import _parse_model_ids

        class FakeResp:
            def __init__(self, payload, ok=True):
                self._payload = payload
                self.is_success = ok

            def json(self):
                if isinstance(self._payload, Exception):
                    raise self._payload
                return self._payload

        # OpenAI / vLLM / llama.cpp shape.
        assert _parse_model_ids(
            FakeResp({"data": [{"id": "llama-3.1-8b"}, {"id": "qwen2.5-7b"}]})
        ) == ["llama-3.1-8b", "qwen2.5-7b"]
        # Bare list of ids.
        assert _parse_model_ids(FakeResp({"data": ["m1", "m2"]})) == ["m1", "m2"]
        # Top-level list.
        assert _parse_model_ids(FakeResp([{"id": "x"}])) == ["x"]
        # Non-success / malformed / exception → [] (never raises).
        assert _parse_model_ids(FakeResp({"data": []}, ok=False)) == []
        assert _parse_model_ids(FakeResp({"nope": 1})) == []
        assert _parse_model_ids(FakeResp(ValueError("bad json"))) == []


    def test_set_model_main_custom_persists_api_key_and_registers_provider(self):
        """A custom endpoint that requires auth must persist model.api_key (where
        the runtime reads it) AND register a named custom_providers entry so the
        endpoint reappears as a ready row in the picker — matching the
        ``hermes model`` custom flow. Regression for the desktop loop where a
        keyed custom endpoint could never be configured from the GUI."""
        from hermes_cli.config import load_config

        resp = self.client.post(
            "/api/model/set",
            json={
                "scope": "main",
                "provider": "custom",
                "model": "gpt-oss-120b",
                "base_url": "https://text.example.com/v1",
                "api_key": "sk-secret",
            },
        )
        assert resp.status_code == 200
        assert resp.json()["ok"] is True

        cfg = load_config()
        model_cfg = cfg.get("model")
        assert isinstance(model_cfg, dict)
        assert model_cfg["provider"] == "custom"
        assert model_cfg["base_url"] == "https://text.example.com/v1"
        assert model_cfg["api_key"] == "sk-secret"

        # Registered in custom_providers (dedup by base_url) so the picker shows
        # a proper ready row instead of the "needs setup" dead-end.
        custom = cfg.get("custom_providers") or []
        assert any(
            isinstance(e, dict)
            and e.get("base_url") == "https://text.example.com/v1"
            and e.get("api_key") == "sk-secret"
            and e.get("model") == "gpt-oss-120b"
            for e in custom
        )


    def test_deleting_the_active_custom_endpoint_clears_its_model_mirror(self):
        """Deleting an endpoint must not leave its credential running the agent.

        ``activate`` mirrors the endpoint's base_url + credential reference
        onto ``model``, and that mirror outranks the environment at client
        construction (#62269). Without clearing it the agent keeps
        authenticating to the deleted host, and the credential the operator
        just removed through the dashboard survives the delete.
        """
        from hermes_cli.config import custom_endpoint_key_env, get_env_value, load_config

        self.client.post(
            "/api/providers/custom-endpoints",
            json={
                "id": "acme",
                "name": "Acme",
                "base_url": "https://llm.acme.corp/v1",
                "model": "acme/model-1",
                "api_key": "sk-acme-secret",
            },
        )
        assert self.client.post(
            "/api/providers/custom-endpoints/acme/activate", json={}
        ).status_code == 200

        env_var = custom_endpoint_key_env("acme")
        cfg = load_config()
        assert cfg["model"]["key_env"] == env_var
        assert get_env_value(env_var) == "sk-acme-secret"

        assert self.client.request(
            "DELETE", "/api/providers/custom-endpoints/acme"
        ).status_code == 200

        cfg = load_config()
        assert "acme" not in (cfg.get("providers") or {})
        model_cfg = cfg.get("model") or {}
        assert not model_cfg.get("api_key"), "deleted endpoint's key still in config.yaml"
        assert not model_cfg.get("key_env"), "deleted endpoint's key ref still in config.yaml"
        assert not model_cfg.get("base_url"), "deleted endpoint's host still routed to"
        assert not model_cfg.get("provider")
        assert not get_env_value(env_var), "deleted endpoint's key still in .env"


    def test_numeric_yaml_provider_key_can_be_activated_and_deleted(self):
        """Hand-edited `providers: 2070:` (YAML int key) must still activate.

        PyYAML loads unquoted 2070 as int; string lookup then 404ed, so
        Desktop could list the endpoint but not assign or delete it.
        """
        from hermes_cli.config import get_config_path, load_config

        get_config_path().write_text(
            "model:\n"
            "  provider: 2070\n"
            "  default: Qwen.gguf\n"
            "  base_url: http://127.0.0.1:1/v1\n"
            "providers:\n"
            "  2070:\n"
            "    name: 2070\n"
            "    base_url: http://127.0.0.1:1/v1\n"
            "    model: Qwen.gguf\n",
            encoding="utf-8",
        )

        listed = self.client.get("/api/providers/custom-endpoints")
        assert listed.status_code == 200
        assert "2070" in [e["id"] for e in listed.json()["endpoints"]]

        activate = self.client.post(
            "/api/providers/custom-endpoints/2070/activate", json={}
        )
        assert activate.status_code == 200, activate.text
        assert activate.json()["provider"] == "2070"

        deleted = self.client.request(
            "DELETE", "/api/providers/custom-endpoints/2070"
        )
        assert deleted.status_code == 200, deleted.text
        providers = load_config().get("providers") or {}
        assert 2070 not in providers
        assert "2070" not in providers

    def test_punctuated_provider_key_round_trips_through_activate_edit_and_delete(self):
        """A stored ``providers.<key>`` with dots/colons or mixed case is what the
        list route returns as ``id``; the same spelling must reach the entry on
        activate, save (edit) and delete instead of being slugified into a
        non-existent twin (404 / duplicate row), and delete must still clear
        the model mirror ``switch_model`` wrote for it.
        """
        from urllib.parse import quote

        from hermes_cli.config import get_config_path, load_config

        get_config_path().write_text(
            "model:\n"
            "  provider: openrouter\n"
            "  default: some/model\n"
            "providers:\n"
            "  local-127.0.0.1:8283:\n"
            "    name: Local (127.0.0.1:8283)\n"
            "    base_url: http://127.0.0.1:8283/v1\n"
            "    model: Qwen.gguf\n"
            "  EXllamav3:\n"
            "    name: EXllamav3\n"
            "    base_url: http://127.0.0.1:8290/v1\n"
            "    model: Qwen3-27B\n",
            encoding="utf-8",
        )
        dotted = "local-127.0.0.1:8283"
        listed = [e["id"] for e in self.client.get("/api/providers/custom-endpoints").json()["endpoints"]]
        assert dotted in listed and "EXllamav3" in listed

        # Edit by the listed id updates the entry in place — no slugged twin.
        saved = self.client.post(
            "/api/providers/custom-endpoints",
            json={"id": dotted, "name": "Local (127.0.0.1:8283)",
                  "base_url": "http://127.0.0.1:8283/v1", "model": "Qwen2.gguf"},
        )
        assert saved.status_code == 200, saved.text
        providers = load_config()["providers"]
        assert providers[dotted]["model"] == "Qwen2.gguf"
        assert "local-127-0-0-1-8283" not in providers

        for key in (dotted, "EXllamav3"):
            path = f"/api/providers/custom-endpoints/{quote(key, safe='')}"
            activate = self.client.post(f"{path}/activate", json={})
            assert activate.status_code == 200, activate.text
            assert load_config()["model"].get("base_url"), key
            current = [e["id"] for e in self.client.get("/api/providers/custom-endpoints").json()["endpoints"]
                       if e["is_current"]]
            assert current == [key], f"{key}: list does not mark the endpoint just activated as current: {current}"
            deleted = self.client.request("DELETE", path)
            assert deleted.status_code == 200, deleted.text
            cfg = load_config()
            assert key not in (cfg.get("providers") or {})
            assert not cfg["model"].get("base_url"), f"{key}: deleted endpoint's host still routed to"
            assert not cfg["model"].get("provider"), key

    def test_unslugged_display_name_still_resolves_to_its_slug_key(self):
        """Compatibility fallback: a caller sending the display name reaches the
        dashboard-minted slug key; an unknown id is still a 404."""
        from hermes_cli.config import get_config_path, load_config

        get_config_path().write_text(
            "providers:\n"
            "  local-8000:\n"
            "    name: Local 8000\n"
            "    base_url: http://127.0.0.1:8000/v1\n"
            "    model: m\n",
            encoding="utf-8",
        )
        assert self.client.request("DELETE", "/api/providers/custom-endpoints/nope.nope").status_code == 404
        assert self.client.request("DELETE", "/api/providers/custom-endpoints/Local%208000").status_code == 200
        assert "local-8000" not in (load_config().get("providers") or {})


    def test_custom_endpoint_save_scopes_to_the_requested_profile(self):
        """``?profile=<name>`` must write into that profile's config.yaml.

        The desktop settings UI targets the active profile, so a custom
        endpoint saved while a non-default profile is selected has to land in
        that profile's config — not the dashboard process's default home.
        Before this fix the handlers ran bare ``load_config``/``save_config``,
        so every custom provider silently landed in the default profile and
        never appeared for the profile the user was actually configuring.
        """
        from hermes_cli import profiles as profiles_mod
        from hermes_cli.config import custom_endpoint_key_env
        from hermes_constants import get_hermes_home

        default_home = get_hermes_home()
        worker_home = profiles_mod.get_profile_dir("worker")
        worker_home.mkdir(parents=True)
        (worker_home / "config.yaml").touch()  # identity marker: bare dirs are not profiles

        assert self.client.post(
            "/api/providers/custom-endpoints?profile=worker",
            json={
                "id": "worker-proxy",
                "name": "Worker Proxy",
                "base_url": "https://llm.worker.example/v1",
                "model": "worker/model-1",
                "api_key": "sk-worker-secret",
            },
        ).status_code == 200

        # Assert against the files on disk rather than load_config()/
        # get_env_value(): save_env_value also mirrors the key into the shared
        # os.environ, so a reader-based check can't tell WHICH profile's store
        # actually received the write.
        env_var = custom_endpoint_key_env("worker-proxy")

        worker_cfg = (worker_home / "config.yaml").read_text()
        assert "worker-proxy" in worker_cfg
        assert env_var in worker_cfg
        assert "sk-worker-secret" in (worker_home / ".env").read_text()

        for leaked in (default_home / "config.yaml", default_home / ".env"):
            text = leaked.read_text() if leaked.exists() else ""
            assert "worker-proxy" not in text, f"endpoint leaked into default profile ({leaked.name})"
            assert "sk-worker-secret" not in text, f"credential leaked into default profile ({leaked.name})"

        # And it comes back through the scoped GET, not the unscoped one.
        scoped = self.client.get("/api/providers/custom-endpoints?profile=worker").json()
        assert any(e["id"] == "worker-proxy" for e in scoped["endpoints"])
        default_list = self.client.get("/api/providers/custom-endpoints").json()
        assert not any(e["id"] == "worker-proxy" for e in default_list["endpoints"])


    def test_custom_endpoint_save_keeps_the_api_key_out_of_config(self):
        """The key belongs in .env behind key_env, never in config.yaml (#69449)."""
        from hermes_cli.config import custom_endpoint_key_env, get_env_value, load_config

        self.client.post(
            "/api/providers/custom-endpoints",
            json={
                "id": "proxy",
                "name": "Proxy",
                "base_url": "https://llm.example.com/v1",
                "model": "m",
                "api_key": "sk-super-secret",
                "make_default": True,
            },
        )

        cfg = load_config()
        entry = cfg["providers"]["proxy"]
        env_var = custom_endpoint_key_env("proxy")
        assert entry["key_env"] == env_var
        assert "api_key" not in entry
        assert "api_key" not in cfg["model"]
        assert get_env_value(env_var) == "sk-super-secret"
        assert "sk-super-secret" not in yaml.safe_dump(cfg)


    def test_custom_endpoint_save_pins_api_mode_and_resolves_reasoning_alias(self):
        """Desktop's Custom Endpoints form pins the transport and keeps alias metadata (#93622).

        A Responses-only host 404s on the runtime's Chat Completions default, so the chosen
        ``api_mode`` must land on the providers entry and read back; a discovered reasoning
        alias resolves to its canonical model + ``agent.reasoning_overrides`` instead of being
        saved as a literal upstream model id.
        """
        from hermes_cli.config import load_config

        response = self.client.post(
            "/api/providers/custom-endpoints",
            json={
                "id": "custom-responses", "name": "custom-responses",
                "base_url": "https://responses-gateway.example.com/v1",
                "model": "gpt-5.6-sol-high", "api_mode": "codex_responses", "make_default": True,
                "models": ["gpt-5.6-sol", "gpt-5.6-sol-high"],
                "model_details": [
                    {"id": "gpt-5.6-sol"},
                    {"id": "gpt-5.6-sol-high", "canonical_model": "gpt-5.6-sol", "reasoning_effort": "high"},
                ],
            },
        )
        assert response.status_code == 200
        row = next(e for e in response.json()["endpoints"] if e["id"] == "custom-responses")
        assert row["api_mode"] == "codex_responses"
        assert row["model"] == "gpt-5.6-sol"

        cfg = load_config()
        entry = cfg["providers"]["custom-responses"]
        assert entry["api_mode"] == "codex_responses"
        assert entry["model"] == "gpt-5.6-sol"
        assert entry["models"]["gpt-5.6-sol-high"] == {"canonical_model": "gpt-5.6-sol", "reasoning_effort": "high"}
        assert cfg["model"]["default"] == "gpt-5.6-sol"
        assert cfg["agent"]["reasoning_overrides"]["gpt-5.6-sol"] == "high"

        # An older UI payload (no api_mode) leaves the pinned transport alone; "" clears it.
        self.client.post("/api/providers/custom-endpoints", json={
            "id": "custom-responses", "name": "custom-responses",
            "base_url": "https://responses-gateway.example.com/v1", "model": "gpt-5.6-sol"})
        assert load_config()["providers"]["custom-responses"]["api_mode"] == "codex_responses"
        self.client.post("/api/providers/custom-endpoints", json={
            "id": "custom-responses", "name": "custom-responses", "api_mode": "",
            "base_url": "https://responses-gateway.example.com/v1", "model": "gpt-5.6-sol"})
        listed = self.client.get("/api/providers/custom-endpoints").json()["endpoints"]
        assert next(e for e in listed if e["id"] == "custom-responses")["api_mode"] == ""
        assert "api_mode" not in load_config()["providers"]["custom-responses"]

    def test_custom_endpoint_validate_keeps_model_alias_metadata(self, monkeypatch):
        """``validate`` returns the bare id list older clients read AND ``model_details`` with
        the ``canonical_model`` / ``reasoning_effort`` a gateway advertises (#93622)."""
        import contextlib

        from hermes_cli.web_routers import config_env

        class FakeResp:
            status_code = 200
            is_success = True

            def json(self):
                return {"data": [
                    {"id": "gpt-5.6-sol", "object": "model"},
                    {"id": "gpt-5.6-sol-high", "canonical_model": "gpt-5.6-sol", "reasoning_effort": "high"},
                ]}

        class FakeClient:
            async def get(self, url, headers=None):
                return FakeResp()

            async def post(self, url, json=None, headers=None):
                return FakeResp()

        @contextlib.asynccontextmanager
        async def fake_probe_client(url, timeout):
            yield FakeClient()

        monkeypatch.setattr(config_env, "_endpoint_probe_client", fake_probe_client)
        body = self.client.post("/api/providers/custom-endpoints/validate", json={
            "name": "x", "base_url": "https://responses-gateway.example.com/v1", "model": ""}).json()
        assert body["ok"] is True
        assert body["models"] == ["gpt-5.6-sol", "gpt-5.6-sol-high"]
        assert body["model_details"] == [
            {"id": "gpt-5.6-sol"},
            {"id": "gpt-5.6-sol-high", "canonical_model": "gpt-5.6-sol", "reasoning_effort": "high"},
        ]

    @staticmethod
    def _responses_only_host(monkeypatch, posted):
        """A gateway that lists models on GET /models and serves POST /responses but 404s
        POST /chat/completions — the #93622 reporter's host."""
        import contextlib

        from hermes_cli.web_routers import config_env

        class Resp:
            def __init__(self, status):
                self.status_code, self.is_success = status, status < 400

            def json(self):
                return {"data": [{"id": "gpt-5.6-sol"}]}

        class Client:
            async def get(self, url, headers=None):
                return Resp(200)

            async def post(self, url, json=None, headers=None):
                posted.append((url, json))
                return Resp(400 if url.endswith("/responses") else 404)

        @contextlib.asynccontextmanager
        async def probe_client(url, timeout):
            yield Client()

        monkeypatch.setattr(config_env, "_endpoint_probe_client", probe_client)

    def test_custom_endpoint_validate_fails_when_the_transport_route_is_missing(self, monkeypatch):
        """Test must exercise the leg the runtime will use: a Responses-only host answers /models
        fine, so validation also POSTs the resolved transport's route and fails on 404 (#93622)."""
        posted = []
        self._responses_only_host(monkeypatch, posted)
        for api_mode in ("", "chat_completions"):  # auto-detect resolves to chat_completions here
            body = self.client.post("/api/providers/custom-endpoints/validate", json={
                "name": "x", "base_url": "https://gw.example.com/v1", "model": "", "api_mode": api_mode}).json()
            assert body["ok"] is False and body["reachable"] is True
            assert body["transport_checked"] == "chat_completions"
            assert body["message"]
            assert body["models"] == ["gpt-5.6-sol"], "discovered models still returned so the user can re-pick"
        assert posted[-1][0] == "https://gw.example.com/v1/chat/completions"
        assert posted[-1][1]["model"] == "gpt-5.6-sol"

    def test_custom_endpoint_validate_passes_when_the_pinned_transport_is_served(self, monkeypatch):
        posted = []
        self._responses_only_host(monkeypatch, posted)
        body = self.client.post("/api/providers/custom-endpoints/validate", json={
            "name": "x", "base_url": "https://gw.example.com/v1", "model": "", "api_mode": "codex_responses"}).json()
        assert body["ok"] is True and body["message"] == ""
        assert body["transport_checked"] == "codex_responses"
        assert [url for url, _ in posted] == ["https://gw.example.com/v1/responses"]
        assert posted[0][1]["model"] == "gpt-5.6-sol"

    def test_custom_endpoint_save_leaves_a_hand_written_env_ref_alone(self, monkeypatch):
        """``api_key: ${MY_KEY}`` is already safe — don't copy it elsewhere.

        load_config() expands env refs, so such an entry looks like a literal
        secret by the time Save sees it. Migrating it would duplicate the
        user's secret into a second env var they never asked for.
        """
        import hermes_yaml as yaml

        from hermes_cli.config import custom_endpoint_key_env, get_config_path, get_env_value

        monkeypatch.setenv("MY_PROXY_KEY", "sk-user-managed")
        get_config_path().write_text(
            yaml.safe_dump({
                "providers": {
                    "proxy": {
                        "name": "Proxy",
                        "base_url": "https://llm.example.com/v1",
                        "model": "m",
                        "api_key": "${MY_PROXY_KEY}",
                    }
                },
            }),
            encoding="utf-8",
        )

        self.client.post(
            "/api/providers/custom-endpoints",
            json={
                "id": "proxy",
                "name": "Proxy",
                "base_url": "https://llm.example.com/v1",
                "model": "m",
            },
        )

        raw = yaml.safe_load(get_config_path().read_text(encoding="utf-8"))
        assert raw["providers"]["proxy"]["api_key"] == "${MY_PROXY_KEY}"
        assert not get_env_value(custom_endpoint_key_env("proxy"))


    def test_two_endpoints_on_one_host_keep_separate_credentials(self):
        """Two local servers must not share an .env slot.

        Deriving the env var from the hostname collapses ``127.0.0.1:8000``
        and ``:8001`` onto one name, so saving the second silently overwrites
        the first's key.
        """
        from hermes_cli.config import custom_endpoint_key_env, get_env_value

        for port, key in ((8000, "sk-first"), (8001, "sk-second")):
            self.client.post(
                "/api/providers/custom-endpoints",
                json={
                    "id": f"local-{port}",
                    "name": f"Local {port}",
                    "base_url": f"http://127.0.0.1:{port}/v1",
                    "model": "m",
                    "api_key": key,
                },
            )

        assert get_env_value(custom_endpoint_key_env("local-8000")) == "sk-first"
        assert get_env_value(custom_endpoint_key_env("local-8001")) == "sk-second"

    def test_custom_endpoint_response_reports_a_key_held_in_env(self):
        """has_api_key must follow key_env, not just a plaintext api_key.

        Reading only ``api_key`` made the panel report "no API key" for every
        endpoint whose credential had been moved to .env.
        """
        resp = self.client.post(
            "/api/providers/custom-endpoints",
            json={
                "id": "proxy",
                "name": "Proxy",
                "base_url": "https://llm.example.com/v1",
                "model": "m",
                "api_key": "sk-in-env",
            },
        )

        endpoint = next(e for e in resp.json()["endpoints"] if e["id"] == "proxy")
        assert endpoint["has_api_key"] is True
        assert "sk-in-env" not in (endpoint["api_key_preview"] or "")

    def test_env_rejects_its_redacted_preview(self):
        """Invariant: a GET preview (sentinel or legacy bare mask) never gains write
        authority, even after another actor rotates the secret behind it."""
        from hermes_cli.config import load_env, save_env_value

        key = "OPENAI_API_KEY"
        real = "sk-live-secret-abcdef1234567890"
        save_env_value(key, real)
        preview = self.client.get("/api/env").json()[key]["redacted_value"]
        assert preview.startswith("«redacted")

        response = self.client.put("/api/env", json={"key": key, "value": preview})
        assert response.status_code == 400
        assert load_env()[key] == real

        rotated = "sk-rotated-secret-0987654321"
        save_env_value(key, rotated)
        for stale in (preview, redact_key(real)):
            response = self.client.put("/api/env", json={"key": key, "value": stale})
            assert response.status_code == 400
            assert load_env()[key] == rotated

    def test_messaging_and_custom_endpoint_reject_stale_previews(self):
        """Invariant: preview rejection runs before any mutation (messaging clear+set),
        and custom-endpoint display strings (``${KEY_ENV}`` / legacy plaintext preview)
        are refused even after the entry rotated underneath them."""
        from hermes_cli.config import load_config, load_env, save_config, save_env_value

        key = "DISCORD_BOT_TOKEN"
        real = "discord-live-secret-abcdef1234567890"
        save_env_value(key, real)
        response = self.client.put(
            "/api/messaging/platforms/discord",
            json={"clear_env": [key], "env": {key: redact_key(real)}},
        )
        assert response.status_code == 400
        assert load_env()[key] == real

        save_env_value("OLD_ENDPOINT_KEY", "old-secret-1234567890")
        save_env_value("NEW_ENDPOINT_KEY", "new-secret-0987654321")
        cfg = load_config()
        cfg["providers"] = {
            "env-preview": {"name": "Env Preview", "base_url": "https://env-preview.example.com/v1",
                            "model": "m", "key_env": "OLD_ENDPOINT_KEY", "models": {"m": {}}},
            "legacy-preview": {"name": "Legacy Preview", "base_url": "https://legacy-preview.example.com/v1",
                               "model": "m", "api_key": "legacy-secret-A-1234567890", "models": {"m": {}}},
        }
        save_config(cfg)
        endpoints = {e["id"]: e for e in self.client.get("/api/providers/custom-endpoints").json()["endpoints"]}
        assert endpoints["env-preview"]["api_key_preview"] == "${OLD_ENDPOINT_KEY}"
        assert endpoints["legacy-preview"]["api_key_preview"].startswith("«redacted")

        cfg = load_config()
        cfg["providers"]["env-preview"]["key_env"] = "NEW_ENDPOINT_KEY"
        cfg["providers"]["legacy-preview"]["api_key"] = "legacy-secret-B-0987654321"
        save_config(cfg)
        for endpoint_id, base_url in (("env-preview", "https://env-preview.example.com/v1"),
                                      ("legacy-preview", "https://legacy-preview.example.com/v1")):
            response = self.client.post("/api/providers/custom-endpoints", json={
                "id": endpoint_id, "name": "x", "base_url": base_url, "model": "m",
                "api_key": endpoints[endpoint_id]["api_key_preview"],
            })
            assert response.status_code == 400
        providers = load_config()["providers"]
        assert providers["env-preview"]["key_env"] == "NEW_ENDPOINT_KEY"
        assert providers["legacy-preview"]["api_key"] == "legacy-secret-B-0987654321"

    def test_activating_an_endpoint_carries_its_credential_either_way(self):
        """Activate must work for both key_env and pre-#69449 plaintext entries."""
        from hermes_cli.config import load_config, save_config

        cfg = load_config()
        cfg["providers"] = {
            "legacy": {
                "name": "Legacy",
                "base_url": "https://llm.legacy.com/v1",
                "model": "m",
                "api_key": "sk-legacy",
                "models": {"m": {}},
            },
            "modern": {
                "name": "Modern",
                "base_url": "https://llm.modern.com/v1",
                "model": "m",
                "key_env": "MODERN_API_KEY",
                "models": {"m": {}},
            },
        }
        save_config(cfg)

        self.client.post("/api/providers/custom-endpoints/modern/activate", json={})
        model_cfg = load_config()["model"]
        assert model_cfg["key_env"] == "MODERN_API_KEY"
        assert "api_key" not in model_cfg

        self.client.post("/api/providers/custom-endpoints/legacy/activate", json={})
        model_cfg = load_config()["model"]
        assert model_cfg["api_key"] == "sk-legacy"

    def test_legacy_custom_providers_entries_get_a_row_and_can_be_deleted(self):
        """A post-migration ``custom_providers:`` list entry is still routed by the
        runtime (``get_compatible_custom_providers``), so Custom Endpoints must show
        it — and Delete must remove it from the legacy list, not 404 (#114471)."""
        from hermes_cli.config import load_config, save_config

        cfg = load_config()
        cfg["providers"] = {
            "modern": {"name": "Modern", "base_url": "https://llm.modern.com/v1", "model": "m"},
        }
        cfg["custom_providers"] = [
            {"name": "Old Box", "base_url": "http://10.0.0.5:8080/v1", "model": "qwen"},
        ]
        save_config(cfg)

        rows = {e["id"]: e for e in self.client.get("/api/providers/custom-endpoints").json()["endpoints"]}
        assert set(rows) == {"modern", "old-box"}
        assert rows["old-box"]["source"] == "custom_providers"
        assert rows["old-box"]["base_url"] == "http://10.0.0.5:8080/v1"
        assert rows["old-box"]["model"] == "qwen"

        deleted = self.client.request("DELETE", "/api/providers/custom-endpoints/old-box")
        assert deleted.status_code == 200, deleted.text
        assert [e["id"] for e in deleted.json()["endpoints"]] == ["modern"]
        cfg = load_config()
        assert cfg.get("custom_providers") == []
        assert "modern" in cfg["providers"]

    def test_activating_a_legacy_custom_providers_entry_promotes_it(self):
        """Use on a legacy row moves the entry under ``providers:`` (the v12+ shape the
        main slot names by key) instead of 404ing on a row the list just rendered."""
        from hermes_cli.config import load_config, save_config

        cfg = load_config()
        cfg["custom_providers"] = [
            {"name": "Old Box", "base_url": "http://127.0.0.1:1/v1", "model": "qwen", "api_key": "sk-old"},
        ]
        save_config(cfg)

        activated = self.client.post("/api/providers/custom-endpoints/old-box/activate", json={})
        assert activated.status_code == 200, activated.text
        cfg = load_config()
        assert cfg.get("custom_providers") == []
        assert cfg["providers"]["old-box"]["api"] == "http://127.0.0.1:1/v1"
        assert cfg["model"]["provider"] == "old-box"
        assert cfg["model"]["default"] == "qwen"

    def test_get_sessions_rejects_negative_limit(self):
        """limit=-1 must be rejected (422), not passed through to SQLite as
        LIMIT -1 (unbounded) — issue #74316."""
        resp = self.client.get("/api/sessions?limit=-1")
        assert resp.status_code == 422


    def test_get_sessions_positive_limit_still_works(self):
        from hermes_state import SessionDB

        db = SessionDB()
        try:
            for i in range(5):
                db.create_session(session_id=f"pos-limit-{i}", source="cli")
                db.append_message(session_id=f"pos-limit-{i}", role="user", content="hi")
        finally:
            db.close()

        resp = self.client.get("/api/sessions?limit=3&offset=0")
        assert resp.status_code == 200
        payload = resp.json()
        assert payload["limit"] == 3
        assert len(payload["sessions"]) == 3

    def test_profiles_sessions_rejects_negative_limit(self):
        """Same guard on the cross-profile aggregate route — negative limit
        previously bypassed the per-profile 500-row clamp entirely."""
        resp = self.client.get("/api/profiles/sessions?limit=-1")
        assert resp.status_code == 422


    def test_profiles_sessions_positive_limit_still_works(self):
        from hermes_state import SessionDB

        db = SessionDB()
        try:
            for i in range(5):
                db.create_session(session_id=f"pos-plimit-{i}", source="cli")
                db.append_message(session_id=f"pos-plimit-{i}", role="user", content="hi")
        finally:
            db.close()

        resp = self.client.get("/api/profiles/sessions?limit=3&offset=0")
        assert resp.status_code == 200
        payload = resp.json()
        assert payload["limit"] == 3
        assert len(payload["sessions"]) == 3

    def test_get_session_messages_rejects_negative_limit(self):
        """limit=-1 previously bypassed the documented 500-row clamp because
        min(-1, 500) == -1, which SQLite treats as 'no limit'."""
        from hermes_state import SessionDB

        db = SessionDB()
        try:
            db.create_session(session_id="neg-limit-messages", source="cli")
            for i in range(60):
                db.append_message(
                    session_id="neg-limit-messages", role="user", content=f"msg {i}"
                )
        finally:
            db.close()

        resp = self.client.get("/api/sessions/neg-limit-messages/messages?limit=-1")
        assert resp.status_code == 422


    def test_get_session_messages_limit_above_500_is_capped_not_rejected(self):
        """A limit above the documented 500-row cap is silently clamped
        (existing ``min(limit, 500)`` behaviour), not rejected — the request
        still succeeds."""
        from hermes_state import SessionDB

        db = SessionDB()
        try:
            db.create_session(session_id="many-messages", source="cli")
            for i in range(60):
                db.append_message(session_id="many-messages", role="user", content=f"msg {i}")
        finally:
            db.close()

        resp = self.client.get("/api/sessions/many-messages/messages?limit=1000")
        assert resp.status_code == 200
        assert resp.json()["pagination"]["limit"] == 500

    def test_get_session_messages_default_hides_compacted_rows(self):
        """The endpoint default matches get_messages: active rows only.

        Guards the #80680 contract — display reads opt into compacted history
        explicitly; the dashboard default view stays as it was.
        """
        from hermes_state import SessionDB

        db = SessionDB()
        try:
            db.create_session(session_id="compacted-default", source="cli")
            db.append_messages_batch(
                "compacted-default",
                [
                    {"role": "user", "content": "old q"},
                    {"role": "assistant", "content": "old a"},
                ],
            )
            db.archive_and_compact(
                "compacted-default",
                [
                    {"role": "assistant", "content": "summary"},
                    {"role": "user", "content": "live q"},
                    {"role": "assistant", "content": "live a"},
                ],
            )
        finally:
            db.close()

        resp = self.client.get("/api/sessions/compacted-default/messages")
        assert resp.status_code == 200
        contents = [m["content"] for m in resp.json()["messages"]]
        assert contents == ["summary", "live q", "live a"]

    def test_get_session_messages_include_compacted_surfaces_archived_rows(self):
        """include_compacted=true returns the full display history: archived
        (active=0, compacted=1) rows plus live rows, in insertion order.
        """
        from hermes_state import SessionDB

        db = SessionDB()
        try:
            db.create_session(session_id="compacted-visible", source="cli")
            db.append_messages_batch(
                "compacted-visible",
                [
                    {"role": "user", "content": "old q"},
                    {"role": "assistant", "content": "old a"},
                ],
            )
            db.archive_and_compact(
                "compacted-visible",
                [
                    {"role": "assistant", "content": "summary"},
                    {"role": "user", "content": "live q"},
                    {"role": "assistant", "content": "live a"},
                ],
            )
        finally:
            db.close()

        resp = self.client.get(
            "/api/sessions/compacted-visible/messages?include_compacted=true"
        )
        assert resp.status_code == 200
        contents = [m["content"] for m in resp.json()["messages"]]
        assert contents == ["old q", "old a", "summary", "live q", "live a"]

    def test_get_session_messages_projects_and_dedupes_composite_carrier(self):
        from agent.context_compressor import (
            HISTORICAL_TASK_HEADING,
            SUMMARY_PREFIX,
            _MERGED_PRIOR_CONTEXT_HEADER,
            _MERGED_SUMMARY_DELIMITER,
            _SUMMARY_END_MARKER,
        )
        from hermes_state import SessionDB

        handoff = (
            f"{SUMMARY_PREFIX}\n{HISTORICAL_TASK_HEADING}\nold task\n\n"
            f"{_SUMMARY_END_MARKER}"
        )
        carrier = f"{handoff}\n\nREAL ASK"
        assistant_carrier = (
            f"{_MERGED_PRIOR_CONTEXT_HEADER}\n"
            "real completed answer\n\n"
            f"{_MERGED_SUMMARY_DELIMITER}\n\n{handoff}"
        )
        db = SessionDB()
        try:
            db.create_session(session_id="compacted-carrier-display", source="desktop")
            db.append_message(
                "compacted-carrier-display",
                "user",
                "REAL ASK",
                timestamp=123.0,
            )
            db.archive_and_compact(
                "compacted-carrier-display",
                [{"role": "user", "content": carrier, "timestamp": 123.0}],
            )
            db.append_message(
                "compacted-carrier-display",
                "user",
                handoff,
                timestamp=124.0,
            )
            db.append_message(
                "compacted-carrier-display",
                "assistant",
                assistant_carrier,
                timestamp=125.0,
            )
            active_id = db.get_messages("compacted-carrier-display")[0]["id"]
        finally:
            db.close()

        resp = self.client.get(
            "/api/sessions/compacted-carrier-display/messages"
            "?include_compacted=true"
        )
        assert resp.status_code == 200
        messages = resp.json()["messages"]
        assert len(messages) == 3
        assert messages[0]["id"] == active_id
        assert messages[0]["content"] == carrier
        assert messages[0]["display_content"] == "REAL ASK"
        assert not messages[0].get("display_kind")
        assert messages[1]["content"] == handoff
        assert messages[1]["display_kind"] == "hidden"
        assert messages[2]["content"] == assistant_carrier
        assert messages[2]["display_content"] == "real completed answer"

    def test_get_session_messages_latest_page_with_compacted_rows(self):
        """The desktop's real read path (getLatestSessionMessages: limit +
        order=latest + include_compacted=true) pages back from the newest
        message and returns the window in chronological order.
        """
        from hermes_state import SessionDB

        db = SessionDB()
        try:
            db.create_session(session_id="compacted-latest", source="cli")
            db.append_messages_batch(
                "compacted-latest",
                [
                    {"role": "user", "content": "old q"},
                    {"role": "assistant", "content": "old a"},
                ],
            )
            db.archive_and_compact(
                "compacted-latest",
                [
                    {"role": "assistant", "content": "summary"},
                    {"role": "user", "content": "live q"},
                    {"role": "assistant", "content": "live a"},
                ],
            )
        finally:
            db.close()

        # Display history: old q, old a, summary, live q, live a (5 rows).
        resp = self.client.get(
            "/api/sessions/compacted-latest/messages"
            "?include_compacted=true&limit=2&offset=1&order=latest"
        )
        assert resp.status_code == 200
        contents = [m["content"] for m in resp.json()["messages"]]
        # Newest-first window of 2, skipping the newest (live a):
        # summary, live q — chronological order, matching the non-compacted path.
        assert contents == ["summary", "live q"]

    def test_get_session_messages_omitted_limit_defaults_to_500(self):
        """The dashboard must never load an entire unbounded transcript."""
        from hermes_state import SessionDB

        db = SessionDB()
        try:
            db.create_session(session_id="default-limit-messages", source="cli")
            db.append_messages_batch(
                "default-limit-messages",
                [
                    {"role": "user", "content": f"msg {i}"}
                    for i in range(501)
                ],
            )
        finally:
            db.close()

        resp = self.client.get("/api/sessions/default-limit-messages/messages")
        assert resp.status_code == 200
        payload = resp.json()
        assert payload["pagination"] == {
            "limit": 500,
            "offset": 0,
            "order": "latest",
            "returned": 500,
        }
        assert len(payload["messages"]) == 500
        assert payload["messages"][0]["content"] == "msg 1"
        assert payload["messages"][-1]["content"] == "msg 500"

        explicit = self.client.get(
            "/api/sessions/default-limit-messages/messages?limit=2&offset=1"
        ).json()
        assert explicit["pagination"]["order"] == "oldest"
        assert [message["content"] for message in explicit["messages"]] == [
            "msg 1",
            "msg 2",
        ]

        latest = self.client.get(
            "/api/sessions/default-limit-messages/messages"
            "?limit=2&offset=1&order=latest"
        ).json()
        assert latest["pagination"]["order"] == "latest"
        assert [message["content"] for message in latest["messages"]] == [
            "msg 498",
            "msg 499",
        ]

    def test_export_session_streams_bounded_message_pages(self, monkeypatch):
        from hermes_state import SessionDB

        db = SessionDB()
        try:
            db.create_session(session_id="stream-export", source="cli")
            db.append_messages_batch(
                "stream-export",
                [
                    {"role": "user", "content": f"msg {i}"}
                    for i in range(501)
                ],
            )
        finally:
            db.close()

        calls = []
        original_get_messages = SessionDB.get_messages

        def tracked_get_messages(self, session_id, *args, **kwargs):
            calls.append((kwargs.get("limit"), kwargs.get("after_id"), kwargs.get("include_inactive")))
            return original_get_messages(self, session_id, *args, **kwargs)

        monkeypatch.setattr(SessionDB, "get_messages", tracked_get_messages)
        response = self.client.get("/api/sessions/stream-export/export")

        assert response.status_code == 200
        payload = response.json()
        assert payload["id"] == "stream-export"
        assert len(payload["messages"]) == 501
        assert payload["messages"][0]["content"] == "msg 0"
        assert payload["messages"][-1]["content"] == "msg 500"
        # Transfer projection: archived rows ride along with their flags (import re-archives them).
        assert calls == [(500, 0, True), (500, 500, True)]


# ---------------------------------------------------------------------------
# _build_schema_from_config tests
# ---------------------------------------------------------------------------


class TestBuildSchemaFromConfig:


    def test_timezone_field_is_searchable_select(self):
        """timezone must ship as a searchable, clearable select of IANA ids.

        Desktop renders this via SearchableSelect (Popover + cmdk); the old
        free-text input let users type invalid timezone strings (#68970).
        Invariants, not snapshots: valid IANA entries present, sorted, no
        blank entry server-side (the clear item is client-side via
        ``clearable``), and never empty even without tzdata (UTC fallback).
        """
        from hermes_cli.web_server_config import CONFIG_SCHEMA, _timezone_options

        entry = CONFIG_SCHEMA["timezone"]
        assert entry["type"] == "select"
        assert entry.get("searchable") is True
        assert entry.get("clearable") is True
        options = entry["options"]
        assert len(options) >= 1
        assert options == sorted(options)
        assert "" not in options
        assert "UTC" in options
        # Fallback path: never returns an empty list.
        assert len(_timezone_options()) >= 1

    def test_dynamic_merge_recomputes_memory_provider_options(self, monkeypatch):
        """The per-request schema merge re-discovers memory providers.

        The import-time _SCHEMA_OVERRIDES freezes the list at server start;
        _schema_with_dynamic_provider_options must recompute it so a provider
        installed mid-session is selectable without a restart.
        """

        monkeypatch.setattr(_cfg_mod, "load_config", lambda: {"memory": {"provider": "honcho"}})
        monkeypatch.setattr(
            _web_server_config,
            "_memory_provider_options",
            lambda: ["", "honcho", "mem0", "freshly_installed"],
        )

        fields = _web_server_config._schema_with_dynamic_provider_options()

        assert "freshly_installed" in fields["memory.provider"]["options"]
        # The entry is copied, not mutated in place, and keeps its select type.
        assert fields["memory.provider"]["type"] == "select"
        assert _web_server_config.CONFIG_SCHEMA["memory.provider"] is not fields["memory.provider"]


# ---------------------------------------------------------------------------
# Config round-trip tests
# ---------------------------------------------------------------------------


class TestConfigRoundTrip:
    """Verify config survives GET → edit → PUT without data loss."""

    @pytest.fixture(autouse=True)
    def _setup(self):
        try:
            from starlette.testclient import TestClient
        except ImportError:
            pytest.skip("fastapi/starlette not installed")
        from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
        self.client = TestClient(app)
        self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN


    def test_round_trip_preserves_schema_invisible_nested_keys(self):
        """Nested keys that aren't in CONFIG_SCHEMA must also survive a
        round-trip. Deep-merge is required — a shallow merge would drop
        ``agent.<custom_key>`` when the frontend sends a partial ``agent``
        dict containing only schema-known sub-fields."""
        from hermes_cli.config import read_raw_config, save_config

        # Seed config with a key under `agent` that isn't in the schema.
        # Use a sentinel name to avoid colliding with future schema fields.
        save_config({
            "agent": {
                "max_turns": 50,
                "x_dashboard_invisible_test_key": {"nested": "value"},
            },
        })

        # PUT only schema-known agent fields, exactly like the dashboard.
        web_config = self.client.get("/api/config").json()
        web_config.setdefault("agent", {})
        web_config["agent"]["max_turns"] = 75
        # Strip our sentinel so we're sending what the schema-driven form
        # would send.
        web_config["agent"].pop("x_dashboard_invisible_test_key", None)

        resp = self.client.put("/api/config", json={"config": web_config})
        assert resp.status_code == 200

        on_disk = read_raw_config()
        assert on_disk.get("agent", {}).get("max_turns") == 75
        assert on_disk.get("agent", {}).get("x_dashboard_invisible_test_key") \
            == {"nested": "value"}, \
            "Shallow-merge regression: agent.x_dashboard_invisible_test_key " \
            "was wiped when the frontend sent a partial agent dict."

    def test_schema_types_match_config_values(self):
        """Every schema field should have a matching-type value in the config."""
        config = self.client.get("/api/config").json()
        schema_resp = self.client.get("/api/config/schema").json()
        schema = schema_resp["fields"]

        def get_nested(obj, path):
            parts = path.split(".")
            cur = obj
            for p in parts:
                if cur is None or not isinstance(cur, dict):
                    return None
                cur = cur.get(p)
            return cur

        mismatches = []
        for key, entry in schema.items():
            val = get_nested(config, key)
            if val is None:
                continue  # not set in user config — fine
            expected = entry["type"]
            if expected in {"string", "select"} and not isinstance(val, str):
                mismatches.append(f"{key}: expected str, got {type(val).__name__}")
            elif expected == "number" and not isinstance(val, (int, float)):
                mismatches.append(f"{key}: expected number, got {type(val).__name__}")
            elif expected == "boolean" and not isinstance(val, bool):
                mismatches.append(f"{key}: expected bool, got {type(val).__name__}")
            elif expected == "list" and not isinstance(val, list):
                mismatches.append(f"{key}: expected list, got {type(val).__name__}")
        assert not mismatches, "Type mismatches:\n" + "\n".join(mismatches)

    def test_desktop_terminal_font_round_trip_preserves_terminal_config(self):
        """The Appearance picker persists a font without replacing sibling settings."""
        from hermes_cli.config import load_config

        web_config = self.client.get("/api/config").json()
        terminal_before = dict(web_config.get("terminal", {}))
        web_config.setdefault("terminal", {})["font_family"] = "MesloLGS NF"

        response = self.client.put("/api/config", json={"config": web_config})

        assert response.status_code == 200
        persisted = load_config()["terminal"]
        assert persisted["font_family"] == "MesloLGS NF"
        for key, value in terminal_before.items():
            if key != "font_family":
                assert persisted[key] == value

        reloaded = self.client.get("/api/config").json()
        assert reloaded["terminal"]["font_family"] == "MesloLGS NF"


# ---------------------------------------------------------------------------
# New feature endpoint tests
# ---------------------------------------------------------------------------


class TestNewEndpoints:
    """Tests for session detail, logs, cron, skills, tools, raw config, analytics."""

    @pytest.fixture(autouse=True)
    def _setup(self, monkeypatch, _isolate_hermes_home):
        try:
            from starlette.testclient import TestClient
        except ImportError:
            pytest.skip("fastapi/starlette not installed")

        import hermes_state
        from hermes_constants import get_hermes_home
        from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN

        monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db")

        self.client = TestClient(app)
        self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN


    # --- Automation Blueprints ---


    # --- Profiles ---


    def test_profiles_create_builder_mcp_auth_is_profile_scoped(
        self, monkeypatch
    ):
        from hermes_constants import get_hermes_home
        import hermes_cli.profiles as profiles_mod

        monkeypatch.setattr(profiles_mod, "create_wrapper_script", lambda name: None)

        secret = "profile-builder-secret"
        resp = self.client.post(
            "/api/profiles",
            json={
                "name": "builder-auth",
                "mcp_servers": [
                    {
                        "name": "Bearer Server",
                        "url": "https://example.com/mcp",
                        "auth": "header",
                        "bearer_token": f"Bearer {secret}",
                    },
                    {
                        "name": "oauth-server",
                        "url": "https://example.com/oauth-mcp",
                        "auth": "oauth",
                    },
                    {
                        "name": "local-server",
                        "command": "uvx",
                        "args": ["mcp-server", "--debug"],
                        "env": {"API_KEY": "stdio-secret"},
                    },
                    {
                        "name": "missing-token",
                        "url": "https://example.com/bad",
                        "auth": "header",
                    },
                    {
                        "name": "http-with-env",
                        "url": "https://example.com/bad-env",
                        "env": {"NOT_SUPPORTED": "value"},
                    },
                ],
            },
        )

        assert resp.status_code == 200
        assert resp.json()["mcp_written"] == 3

        root = get_hermes_home()
        profile_dir = root / "profiles" / "builder-auth"
        config_text = (profile_dir / "config.yaml").read_text(encoding="utf-8")
        config = yaml.safe_load(config_text)
        servers = config["mcp_servers"]

        assert sorted(servers) == [
            "Bearer Server",
            "local-server",
            "oauth-server",
        ]
        assert servers["Bearer Server"] == {
            "url": "https://example.com/mcp",
            "headers": {
                "Authorization": "Bearer ${MCP_BEARER_SERVER_API_KEY}",
            },
        }
        assert servers["oauth-server"] == {
            "url": "https://example.com/oauth-mcp",
            "auth": "oauth",
        }
        assert servers["local-server"] == {
            "command": "uvx",
            "args": ["mcp-server", "--debug"],
            "env": {"API_KEY": "stdio-secret"},
        }

        assert secret not in config_text
        profile_env = (profile_dir / ".env").read_text(encoding="utf-8")
        assert f"MCP_BEARER_SERVER_API_KEY={secret}" in profile_env
        assert "Bearer Bearer" not in profile_env
        assert not (root / ".env").exists()


    # --- New profiles endpoints: active / description / model / describe-auto ---


    def test_discord_toolsets_read_and_write_discord_platform(self):
        """Platform-restricted toolsets must not be saved as successful CLI no-ops."""
        from hermes_cli.config import load_config

        listing = {t["name"]: t for t in self.client.get("/api/tools/toolsets").json()}
        assert listing["discord"]["platform"] == "discord"
        assert listing["discord"]["platform_label"] == "Discord"
        assert listing["discord"]["enabled"] is False

        resp = self.client.put("/api/tools/toolsets/discord", json={"enabled": True})
        assert resp.status_code == 200
        assert resp.json() == {
            "ok": True,
            "name": "discord",
            "platform": "discord",
            "enabled": True,
            # Install-on-enable: no provider post_setup pending for discord.
            "post_setup_started": None,
        }

        config = load_config()
        assert "discord" in config["platform_toolsets"]["discord"]
        assert "discord" not in config["platform_toolsets"].get("cli", [])

        listing = {t["name"]: t for t in self.client.get("/api/tools/toolsets").json()}
        assert listing["discord"]["enabled"] is True
        assert listing["discord_admin"]["enabled"] is False

        resp = self.client.put(
            "/api/tools/toolsets/discord_admin", json={"enabled": True}
        )
        assert resp.status_code == 200
        config = load_config()
        assert {"discord", "discord_admin"} <= set(
            config["platform_toolsets"]["discord"]
        )


    def test_get_toolset_config_returns_provider_matrix(self):
        """GET .../config returns provider rows with structured env_vars."""
        resp = self.client.get("/api/tools/toolsets/tts/config")
        assert resp.status_code == 200
        data = resp.json()
        assert data["name"] == "tts"
        assert data["has_category"] is True
        assert isinstance(data["providers"], list)
        assert data["providers"], "tts always has at least the built-in providers"
        # active_provider is part of the contract so the GUI can highlight the
        # provider actually written to config (else it falls back to the first
        # keyless one). It's either None or the name of one listed provider.
        assert "active_provider" in data
        names = {p["name"] for p in data["providers"]}
        assert data["active_provider"] is None or data["active_provider"] in names
        for prov in data["providers"]:
            assert "name" in prov
            assert "is_active" in prov
            assert "env_vars" in prov
            assert isinstance(prov["env_vars"], list)
            for ev in prov["env_vars"]:
                assert "key" in ev
                assert "is_set" in ev
        # active_provider summarizes the first provider flagged is_active
        # (some catalogs list two rows backed by the same config value, e.g.
        # Firecrawl cloud + self-hosted both map to web.backend=firecrawl).
        active = [p["name"] for p in data["providers"] if p["is_active"]]
        if active:
            assert data["active_provider"] == active[0]
        else:
            assert data["active_provider"] is None

    def test_get_toolset_config_reports_truthful_provider_status(self, monkeypatch):
        """Each provider row carries a server-computed readiness `status`.

        Regression: the GUI pilled every zero-env-var row "Ready" — including
        logged-out Nous Subscription rows, xAI TTS without Grok OAuth, and
        never-installed KittenTTS/Piper. The endpoint now reports the honest
        state so keyless ≠ ready.
        """
        import hermes_cli.tools_config as tools_config
        from hermes_cli.nous_account import NousPortalAccountInfo

        # Logged out of Nous Portal → managed subscription rows need sign-in.
        monkeypatch.setattr(
            "hermes_cli.nous_subscription.get_nous_portal_account_info",
            lambda *a, **k: NousPortalAccountInfo(
                logged_in=False, source="none", fresh=False, paid_service_access=None
            ),
        )
        # No xAI credentials → the Grok OAuth-backed row needs sign-in.
        import hermes_cli.tools_config_post_setup as tools_config_post_setup

        monkeypatch.setattr(tools_config, "_xai_credentials_present", lambda: False)
        # Local TTS engines not installed → their rows need setup.
        monkeypatch.setattr(tools_config_post_setup, "_module_installed", lambda name: False)
        monkeypatch.delenv("ELEVENLABS_API_KEY", raising=False)

        resp = self.client.get("/api/tools/toolsets/tts/config")
        assert resp.status_code == 200
        data = resp.json()
        by_name = {p["name"]: p for p in data["providers"]}

        valid = {"ready", "needs_keys", "needs_auth", "needs_setup"}
        assert all(p["status"] in valid for p in data["providers"])
        # Genuinely-free keyless row stays Ready.
        assert by_name["Microsoft Edge TTS"]["status"] == "ready"
        # Keyless ≠ ready for gated rows:
        assert by_name["Nous Subscription"]["status"] == "needs_auth"
        assert by_name["xAI TTS"]["status"] == "needs_auth"
        assert by_name["KittenTTS"]["status"] == "needs_setup"
        assert by_name["Piper"]["status"] == "needs_setup"
        # Keyed row with the key unset:
        assert by_name["ElevenLabs"]["status"] == "needs_keys"


    def test_select_managed_nous_provider_reports_needs_nous_auth(self, monkeypatch):
        """Selecting a managed Nous row while logged out flags needs_nous_auth.

        Regression: the GUI PUT wrote browser.cloud_provider + use_gateway
        but skipped the Portal entitlement handshake the CLI runs inline
        (ensure_nous_portal_access) — so the row never activated and nothing
        told the user to sign in. The endpoint now reports the entitlement
        gap so the client can drive the existing Nous OAuth flow.
        """
        from hermes_cli.nous_account import NousPortalAccountInfo

        monkeypatch.setattr(
            "hermes_cli.nous_subscription.get_nous_portal_account_info",
            lambda *a, **k: NousPortalAccountInfo(
                logged_in=False, source="none", fresh=False, paid_service_access=None
            ),
        )

        resp = self.client.put(
            "/api/tools/toolsets/browser/provider",
            json={"provider": "Nous Subscription (Browser Use cloud)"},
        )
        assert resp.status_code == 200
        data = resp.json()
        assert data["ok"] is True
        assert data["needs_nous_auth"] is True
        assert data["feature"] == "browser"
        # The selection is still persisted — activation is what's gated.
        # Managed rows store the single 'nous' provider string (the runtime
        # maps it to the Browser Use cloud through the Nous Tool Gateway).
        from hermes_cli.config import load_config
        cfg = load_config()
        assert cfg["browser"]["cloud_provider"] == "nous"
        assert "use_gateway" not in cfg["browser"]


    # -- Web capability split (search vs extract backends) ------------------


    def test_select_web_search_backend_matches_runtime_resolution(self, monkeypatch):
        """PUT provider with capability=search writes web.search_backend and the
        runtime search dispatcher resolves to it — while extract is untouched."""
        # Make SearXNG available so both the endpoint gate and the runtime
        # availability check agree it's usable.
        monkeypatch.setenv("SEARXNG_URL", "http://localhost:8888")
        # Give extract an explicit shared backend so the assertion isn't
        # hostage to whatever creds exist on the machine running the tests.
        monkeypatch.setenv("FIRECRAWL_API_URL", "http://localhost:3002")
        base = self.client.put(
            "/api/tools/toolsets/web/provider",
            json={"provider": "Firecrawl Self-Hosted"},
        )
        assert base.status_code == 200

        resp = self.client.put(
            "/api/tools/toolsets/web/provider",
            json={"provider": "SearXNG", "capability": "search"},
        )
        assert resp.status_code == 200
        body = resp.json()
        assert body["ok"] is True
        assert body["capability"] == "search"

        from hermes_cli.config import load_config
        cfg = load_config()
        assert cfg["web"]["search_backend"] == "searxng"
        # The shared backend selected first must be preserved for extract.
        assert cfg["web"]["backend"] == "firecrawl"

        # The REAL runtime resolution — not a parallel reimplementation.
        from tools.web_tools import _get_extract_backend, _get_search_backend
        assert _get_search_backend() == "searxng"
        assert _get_extract_backend() == "firecrawl"

        # And the config endpoint reports the same split.
        data = self.client.get("/api/tools/toolsets/web/config").json()
        assert data["active_search_backend"] == "searxng"
        assert data["active_extract_backend"] == "firecrawl"


    # -- Terminal execution backend picker ---------------------------------


    def test_terminal_ssh_probe_ready_when_configured(self, monkeypatch):
        """SSH host + user in config.yaml -> ready."""
        from hermes_cli.config import load_config, save_config

        monkeypatch.setattr(shutil, "which", lambda name: None)
        config = load_config()
        config.setdefault("terminal", {})
        config["terminal"]["ssh_host"] = "devbox.example.com"
        config["terminal"]["ssh_user"] = "hermes"
        save_config(config)

        body = self.client.get("/api/tools/terminal/backends").json()
        ssh = next(r for r in body["backends"] if r["name"] == "ssh")
        assert ssh["status"] == "ready"
        assert "hermes@devbox.example.com" in ssh["detail"]


    def test_analytics_usage_includes_skill_breakdown(self):
        from hermes_state import SessionDB

        db = SessionDB()
        try:
            db.create_session(
                session_id="skills-analytics-test",
                source="cli",
                model="anthropic/claude-sonnet-4",
            )
            db.update_token_counts(
                "skills-analytics-test",
                input_tokens=120,
                output_tokens=45,
            )
            db.append_message(
                "skills-analytics-test",
                role="assistant",
                content="Loading and updating skills.",
                tool_calls=[
                    {
                        "function": {
                            "name": "skill_view",
                            "arguments": '{"name":"github-pr-workflow"}',
                        }
                    },
                    {
                        "function": {
                            "name": "skill_manage",
                            "arguments": '{"name":"github-code-review"}',
                        }
                    },
                ],
            )
        finally:
            db.close()

        resp = self.client.get("/api/analytics/usage?days=7")
        assert resp.status_code == 200

        data = resp.json()
        assert data["skills"]["summary"] == {
            "total_skill_loads": 1,
            "total_skill_edits": 1,
            "total_skill_actions": 2,
            "distinct_skills_used": 2,
        }
        assert len(data["skills"]["top_skills"]) == 2

        top_skill = data["skills"]["top_skills"][0]
        assert top_skill["skill"] == "github-pr-workflow"
        assert top_skill["view_count"] == 1
        assert top_skill["manage_count"] == 0
        assert top_skill["total_count"] == 1
        assert top_skill["last_used_at"] is not None

    def _daily_for_local_starts(self, tz_name, local_starts):
        """Seed one session per naive local start in ``tz_name``; return the daily buckets."""
        from datetime import datetime
        from zoneinfo import ZoneInfo

        from hermes_state import SessionDB

        db = SessionDB()
        try:
            for i, local in enumerate(local_starts):
                db.create_session(session_id=f"day-bucket-{i}", source="cli")
                db._conn.execute("UPDATE sessions SET started_at = ? WHERE id = ?",
                                 (local.replace(tzinfo=ZoneInfo(tz_name)).timestamp(), f"day-bucket-{i}"))
            db._conn.commit()
        finally:
            db.close()
        original_tz = os.environ.get("TZ")
        try:
            os.environ["TZ"] = tz_name
            time.tzset()
            resp = self.client.get("/api/analytics/usage?days=365")
        finally:
            if original_tz is None:
                os.environ.pop("TZ", None)
            else:
                os.environ["TZ"] = original_tz
            time.tzset()
        assert resp.status_code == 200
        return {row["day"]: row["sessions"] for row in resp.json()["daily"]}

    @pytest.mark.skipif(not hasattr(time, "tzset"), reason="needs a POSIX process timezone switch")
    def test_analytics_daily_buckets_use_local_day(self):
        """A session at 02:00 IST is counted on that local day, as /insights counts it (not the UTC day before)."""
        from datetime import datetime, timedelta

        yesterday = (datetime.now() - timedelta(days=1)).replace(hour=2, minute=0, second=0, microsecond=0)
        daily = self._daily_for_local_starts("Asia/Kolkata", [yesterday])
        assert daily == {yesterday.strftime("%Y-%m-%d"): 1}

    @pytest.mark.skipif(not hasattr(time, "tzset"), reason="needs a POSIX process timezone switch")
    def test_analytics_daily_buckets_follow_dst_offset_per_session(self):
        """23:30 local stays on its own day in both EST and EDT; one fixed offset for the range misplaces one."""
        from datetime import datetime, timedelta

        # The latest past day in January (EST) and in July (EDT), both inside the 365-day window.
        recent = [(datetime.now() - timedelta(days=n)).replace(hour=23, minute=30, second=0, microsecond=0) for n in range(2, 360)]
        starts = [next(d for d in recent if d.month == m) for m in (1, 7)]
        daily = self._daily_for_local_starts("America/New_York", starts)
        assert daily == {s.strftime("%Y-%m-%d"): 1 for s in starts}


# ---------------------------------------------------------------------------
# Desktop-owned loopback backends are not gated by dashboard.public_url (#96490)
# ---------------------------------------------------------------------------


class TestDesktopLoopbackAuthExemption:
    """``_desktop_loopback_auth_exempt`` decides the #96490 exemption."""

    def test_exempt_with_desktop_env_and_session_token_on_loopback(self, monkeypatch):
        import hermes_cli.web_server as web_server

        monkeypatch.setenv("HERMES_DESKTOP", "1")
        monkeypatch.setenv("HERMES_DASHBOARD_SESSION_TOKEN", "desktop-minted")
        assert web_server._desktop_loopback_auth_exempt("127.0.0.1") is True
        assert web_server._desktop_loopback_auth_exempt("::1") is True

    def test_exempt_via_ssh_spawn_credentials_without_env_token(self, monkeypatch):
        import hermes_cli.web_server as web_server

        monkeypatch.setenv("HERMES_DESKTOP", "1")
        monkeypatch.delenv("HERMES_DASHBOARD_SESSION_TOKEN", raising=False)
        assert web_server._desktop_loopback_auth_exempt(
            "127.0.0.1", ssh_session_token="tok"
        )
        assert web_server._desktop_loopback_auth_exempt(
            "127.0.0.1", ssh_owner_nonce="nonce"
        )

    def test_not_exempt_without_desktop_env(self, monkeypatch):
        import hermes_cli.web_server as web_server

        monkeypatch.delenv("HERMES_DESKTOP", raising=False)
        monkeypatch.setenv("HERMES_DASHBOARD_SESSION_TOKEN", "tok")
        assert web_server._desktop_loopback_auth_exempt("127.0.0.1") is False

    def test_not_exempt_without_any_credential(self, monkeypatch):
        import hermes_cli.web_server as web_server

        # HERMES_DESKTOP=1 alone is not enough: a plain serve with the env var
        # exported must stay gated.
        monkeypatch.setenv("HERMES_DESKTOP", "1")
        monkeypatch.delenv("HERMES_DASHBOARD_SESSION_TOKEN", raising=False)
        assert web_server._desktop_loopback_auth_exempt("127.0.0.1") is False

    def test_not_exempt_on_non_loopback_bind(self, monkeypatch):
        import hermes_cli.web_server as web_server

        monkeypatch.setenv("HERMES_DESKTOP", "1")
        monkeypatch.setenv("HERMES_DASHBOARD_SESSION_TOKEN", "tok")
        assert web_server._desktop_loopback_auth_exempt("0.0.0.0") is False
        assert web_server._desktop_loopback_auth_exempt("192.168.1.10") is False

    def test_public_url_engages_gate_for_non_desktop_loopback(self, monkeypatch):
        import hermes_cli.web_server as web_server

        # Sanity: the base behaviour is untouched — a non-Desktop loopback
        # serve with a public_url configured stays ticket-gated.
        monkeypatch.delenv("HERMES_DESKTOP", raising=False)
        assert web_server.should_require_dashboard_auth(
            "127.0.0.1", frozenset({"dash.example.com"})
        ) is True


class TestDesktopHostRendezvousIsolation:
    """Desktop pool children have a private lifecycle, not a host ownership role."""

    def test_desktop_backend_does_not_claim_the_host_serve_record(self, monkeypatch, tmp_path):
        """A Desktop child must not block a separately supervised public dashboard, yet a
        terminal `hermes plugins install` on a Desktop-only box must still find it (#119644):
        it publishes under its OWN role, which the attach ladder never reads."""
        import io
        import urllib.request
        from gateway import host_rendezvous as hr
        import hermes_cli.web_server as web_server
        from hermes_cli.main_dashboard import _host_backend_attachment
        from hermes_cli.plugins_activation import notify_serve_backend

        monkeypatch.setenv("HERMES_GATEWAY_LOCK_DIR", str(tmp_path / "locks"))
        monkeypatch.setenv("HERMES_DESKTOP", "1")
        monkeypatch.setenv("HERMES_DASHBOARD_SESSION_TOKEN", "desktop-spawn-token")
        monkeypatch.setattr(web_server, "_SESSION_TOKEN", "desktop-spawn-token")
        monkeypatch.setattr(hr, "cleanup_on_exit", lambda role: None)
        dialed = []

        class _Reply(io.BytesIO):
            def __enter__(self):
                return self

            def __exit__(self, *exc):
                return False

        def _fake_urlopen(request, timeout=None):
            dialed.append((request.full_url, request.get_header("X-hermes-session-token")))
            return _Reply(b'{"ok": true}')

        monkeypatch.setattr(urllib.request, "urlopen", _fake_urlopen)
        try:
            web_server._publish_host_rendezvous("127.0.0.1", 9231)

            # Not a host owner: the supervised public dashboard's attach ladder sees nobody.
            assert hr.read_record(hr.ROLE_SERVE) is None
            assert _host_backend_attachment() is None
            # ...but a terminal `hermes plugins install` still lights up its open chats.
            assert notify_serve_backend("demo", tmp_path) == {"ok": True}
            assert dialed == [("http://127.0.0.1:9231/api/dashboard/agent-plugins/activate",
                               "desktop-spawn-token")]
        finally:
            hr.clear_record(hr.ROLE_DESKTOP_SERVE)
            hr.release_host_lock(hr.ROLE_DESKTOP_SERVE)

    def test_standalone_backend_still_claims_the_host_serve_record(self, monkeypatch):
        """The Desktop exclusion must not alter standalone dashboard discovery — including a
        supervised service whose shell merely inherited HERMES_DESKTOP=1 without the token."""
        from gateway import host_rendezvous as hr
        import hermes_cli.web_server as web_server

        monkeypatch.setenv("HERMES_DESKTOP", "1")
        monkeypatch.delenv("HERMES_DASHBOARD_SESSION_TOKEN", raising=False)
        claimed = []
        published = []
        monkeypatch.setattr(
            hr,
            "claim_host_lock",
            lambda role: (claimed.append(role) or (hr.HostLockOutcome.ACQUIRED, None)),
        )
        monkeypatch.setattr(hr, "publish_record", lambda *args, **kwargs: published.append((args, kwargs)))
        monkeypatch.setattr(hr, "cleanup_on_exit", lambda role: None)

        web_server._publish_host_rendezvous("0.0.0.0", 9119)

        assert claimed == [hr.ROLE_SERVE]
        assert published[0][0] == (hr.ROLE_SERVE,)
        assert published[0][1]["host"] == "0.0.0.0"
        assert published[0][1]["port"] == 9119


# ---------------------------------------------------------------------------
# Model context length: normalize/denormalize + /api/model/info
# ---------------------------------------------------------------------------


class TestModelContextLength:
    """Tests for model_context_length in normalize/denormalize and /api/model/info."""

    def test_normalize_extracts_context_length_from_dict(self):
        """normalize should surface context_length from model dict."""
        from hermes_cli.web_server_config import _normalize_config_for_web

        cfg = {
            "model": {
                "default": "anthropic/claude-opus-4.6",
                "provider": "openrouter",
                "context_length": 200000,
            }
        }
        result = _normalize_config_for_web(cfg)
        assert result["model"] == "anthropic/claude-opus-4.6"
        assert result["model_context_length"] == 200000

    def test_normalize_bare_string_model_yields_zero(self):
        """normalize should set model_context_length=0 for bare string model."""
        from hermes_cli.web_server_config import _normalize_config_for_web

        result = _normalize_config_for_web({"model": "anthropic/claude-sonnet-4"})
        assert result["model"] == "anthropic/claude-sonnet-4"
        assert result["model_context_length"] == 0


    def test_denormalize_writes_context_length_into_model_dict(self):
        """denormalize should write model_context_length back into model dict."""
        from hermes_cli.web_server_config import _denormalize_config_from_web
        from hermes_cli.config import save_config

        # Set up disk config with model as a dict
        save_config({
            "model": {"default": "anthropic/claude-opus-4.6", "provider": "openrouter"}
        })

        result = _denormalize_config_from_web({
            "model": "anthropic/claude-opus-4.6",
            "model_context_length": 100000,
        })
        assert isinstance(result["model"], dict)
        assert result["model"]["context_length"] == 100000
        assert "model_context_length" not in result  # virtual field removed

    def test_denormalize_context_length_alone_is_applied(self):
        """The Settings autosave now sends a diff, not the full draft: editing
        only the Context Window control must not omit ``model`` and thereby
        drop the context_length edit on the floor (#89597 review)."""
        from hermes_cli.web_server_config import _denormalize_config_from_web
        from hermes_cli.config import save_config

        save_config({
            "model": {"default": "anthropic/claude-sonnet-4", "provider": "anthropic",
                      "context_length": 100000}
        })

        result = _denormalize_config_from_web({"model_context_length": 200000})
        assert isinstance(result["model"], dict)
        assert result["model"]["context_length"] == 200000
        assert result["model"]["default"] == "anthropic/claude-sonnet-4"

    def test_denormalize_model_alone_preserves_context_length(self):
        """The mirror case: editing only the Model field must not silently
        wipe an existing context_length override just because the diff omits
        the unrelated model_context_length key (#89597 review).

        No ``provider`` on disk here on purpose: that keeps this test isolated
        to the diff-omission bug rather than the separate, pre-existing (and
        intentional, see ``_apply_main_model_assignment``) behavior where a
        real provider switch drops the context_length override."""
        from hermes_cli.web_server_config import _denormalize_config_from_web
        from hermes_cli.config import save_config

        save_config({
            "model": {"default": "anthropic/claude-sonnet-4", "context_length": 150000}
        })

        result = _denormalize_config_from_web({"model": "anthropic/claude-opus-4.6"})
        assert result["model"]["context_length"] == 150000
        assert result["model"]["default"] == "anthropic/claude-opus-4.6"


class TestDenormalizeProviderSwitch:
    """The flat Config-page Model field carries no provider info. When the
    model string changes to one served by a different provider, the saved
    provider must follow it (issue #14058)."""

    def test_vendor_slug_switches_off_non_aggregator_provider(self):
        """ollama-local + a vendor/model slug → switch to openrouter and drop
        the stale local base_url (the issue's exact repro)."""
        from hermes_cli.web_server_config import _denormalize_config_from_web
        from unittest.mock import patch as _patch
        from hermes_cli.config import save_config

        save_config({
            "model": {
                "default": "llama3.2",
                "provider": "ollama-local",
                "base_url": "http://localhost:11434/v1",
                "api_mode": "chat_completions",
            }
        })

        with _patch("hermes_cli.models_detect.provider_has_credentials", lambda p: p == "openrouter"):
            result = _denormalize_config_from_web({"model": "google/gemini-2.5-flash"})
        model = result["model"]
        assert model["provider"] == "openrouter"
        assert model["default"] == "google/gemini-2.5-flash"
        # The old ollama-local endpoint must not carry over to openrouter (the switch resolves
        # the aggregator's own endpoint instead of leaving the field blank or stale).
        assert model.get("base_url") != "http://localhost:11434/v1"


    def test_context_length_override_survives_provider_switch(self):
        """An explicit context-length override must persist alongside a
        provider switch."""
        from hermes_cli.web_server_config import _denormalize_config_from_web
        from unittest.mock import patch as _patch
        from hermes_cli.config import save_config

        save_config({"model": {"default": "llama3.2", "provider": "ollama-local"}})

        with _patch("hermes_cli.models_detect.provider_has_credentials", lambda p: p == "openrouter"):
            result = _denormalize_config_from_web({
                "model": "google/gemini-2.5-flash",
                "model_context_length": 128000,
            })
        model = result["model"]
        assert model["provider"] == "openrouter"
        assert model["context_length"] == 128000

    def test_rejected_switch_is_400_and_leaves_the_model_block_byte_identical(self, monkeypatch):
        """``switch_model`` rejecting the inferred provider must surface as 400 from
        ``PUT /api/config`` — not fall back to the flat string, which the deep-merge would
        write OVER the on-disk ``model:`` dict (provider/base_url/api_mode/slots destroyed)."""
        from starlette.testclient import TestClient
        from hermes_constants import get_hermes_home
        from hermes_cli.model_switch import ModelSwitchResult
        from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN

        cfg_path = get_hermes_home() / "config.yaml"
        cfg_path.write_text(
            "model:\n"
            "  default: llama3.2\n"
            "  provider: ollama-local\n"
            "  base_url: http://localhost:11434/v1\n"
            "  api_mode: chat_completions\n"
            "  context_length: 32000\n"
            "  model_slots:\n"
            "    fast: qwen3\n",
            encoding="utf-8")
        before = cfg_path.read_bytes()
        monkeypatch.setattr("hermes_cli.models_detect.provider_has_credentials", lambda p: p == "openrouter")
        monkeypatch.setattr("hermes_cli.model_switch.switch_model",
                            lambda **_kw: ModelSwitchResult(success=False, error_message="models.dev offline"))

        client = TestClient(app)
        client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
        resp = client.put("/api/config", json={"config": {"model": "openai/gpt-5.5-zzz"}})

        assert resp.status_code == 400 and "models.dev offline" in resp.json()["detail"]
        assert cfg_path.read_bytes() == before


class TestModelInfoEndpoint:
    """Tests for GET /api/model/info endpoint."""

    @pytest.fixture(autouse=True)
    def _setup(self):
        try:
            from starlette.testclient import TestClient
        except ImportError:
            pytest.skip("fastapi/starlette not installed")
        from hermes_cli.web_server import app
        self.client = TestClient(app)


    def test_model_info_with_dict_config(self, monkeypatch):

        monkeypatch.setattr(_cfg_mod, "load_config", lambda: {
            "model": {
                "default": "anthropic/claude-opus-4.6",
                "provider": "openrouter",
                "context_length": 100000,
            }
        })

        with patch("agent.model_metadata.get_model_context_length", return_value=200000):
            resp = self.client.get("/api/model/info")

        data = resp.json()
        assert data["model"] == "anthropic/claude-opus-4.6"
        assert data["provider"] == "openrouter"
        assert data["auto_context_length"] == 200000
        assert data["config_context_length"] == 100000
        assert data["effective_context_length"] == 100000  # override wins


    def test_model_info_graceful_on_metadata_error(self, monkeypatch):
        """Endpoint should return zeros on import/resolution errors, not 500."""

        monkeypatch.setattr(_cfg_mod, "load_config", lambda: {
            "model": "some/obscure-model"
        })

        with patch("agent.model_metadata.get_model_context_length", side_effect=Exception("boom")):
            resp = self.client.get("/api/model/info")

        assert resp.status_code == 200
        data = resp.json()
        assert data["auto_context_length"] == 0


# ---------------------------------------------------------------------------
# Gateway health probe tests
# ---------------------------------------------------------------------------


class TestProbeGatewayHealth:
    """Tests for _probe_gateway_health() — cross-container gateway detection."""


    def test_probe_uses_configured_short_timeout(self, monkeypatch):
        """The HTTP probe must not fall through to the OS TCP timeout."""
        import hermes_cli.web_server as ws

        monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642")
        monkeypatch.setattr(ws, "_GATEWAY_HEALTH_TIMEOUT", 0.75)
        timeouts = []

        def mock_urlopen(req, **kwargs):
            timeouts.append(kwargs.get("timeout"))
            raise TimeoutError("mock timeout")

        monkeypatch.setattr(ws.urllib.request, "urlopen", mock_urlopen)

        alive, body = _web_server_gateway._probe_gateway_health()

        assert alive is False
        assert body is None
        assert timeouts == [0.75, 0.75]


    def test_detailed_fails_falls_back_to_simple_health(self, monkeypatch):
        """If /health/detailed fails, falls back to /health."""
        import hermes_cli.web_server as ws
        monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642")
        monkeypatch.setattr(ws, "_GATEWAY_HEALTH_TIMEOUT", 1)

        call_count = [0]

        def mock_urlopen(req, **kwargs):
            call_count[0] += 1
            if call_count[0] == 1:
                raise ConnectionError("detailed failed")
            mock_resp = MagicMock()
            mock_resp.status = 200
            mock_resp.read.return_value = json.dumps({"status": "ok"}).encode()
            mock_resp.__enter__ = MagicMock(return_value=mock_resp)
            mock_resp.__exit__ = MagicMock(return_value=False)
            return mock_resp

        monkeypatch.setattr(ws.urllib.request, "urlopen", mock_urlopen)
        alive, body = _web_server_gateway._probe_gateway_health()
        assert alive is True
        assert body["status"] == "ok"
        assert call_count[0] == 2


class TestStatusRemoteGateway:
    """Tests for /api/status with remote gateway health fallback."""

    @pytest.fixture(autouse=True)
    def _setup_test_client(self):
        try:
            from starlette.testclient import TestClient
        except ImportError:
            pytest.skip("fastapi/starlette not installed")

        from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
        self.client = TestClient(app)
        self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN

    def test_status_falls_back_to_remote_probe(self, monkeypatch):
        """When local PID check fails and remote probe succeeds, gateway shows running."""
        import hermes_cli.web_server as ws

        monkeypatch.setattr(_gw_status, "get_running_pid_cached", lambda: None)
        monkeypatch.setattr(_gw_status, "read_runtime_status", lambda: None)
        monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642")
        monkeypatch.setattr(_web_server_gateway, "_probe_gateway_health", lambda: (True, {
            "status": "ok",
            "gateway_state": "running",
            "platforms": {"telegram": {"state": "connected"}},
            "pid": 999,
        }))

        resp = self.client.get("/api/status")
        assert resp.status_code == 200
        data = resp.json()
        assert data["gateway_running"] is True
        assert data["gateway_pid"] == 999
        assert data["gateway_state"] == "running"
        assert data["gateway_health_url"] == "http://gw:8642"


    def test_status_remote_running_null_pid(self, monkeypatch):
        """Remote gateway running but PID not in response — pid should be None."""
        import hermes_cli.web_server as ws

        monkeypatch.setattr(_gw_status, "get_running_pid_cached", lambda: None)
        monkeypatch.setattr(_gw_status, "read_runtime_status", lambda: None)
        monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642")
        monkeypatch.setattr(_web_server_gateway, "_probe_gateway_health", lambda: (True, {
            "status": "ok",
        }))

        resp = self.client.get("/api/status")
        assert resp.status_code == 200
        data = resp.json()
        assert data["gateway_running"] is True
        assert data["gateway_pid"] is None
        assert data["gateway_state"] == "running"


class TestStatusInstallId:
    """Stable per-install identity on /api/status.

    Behaviour contracts: the id is minted once, persisted under the ROOT
    Hermes home (not the profile home), survives a fresh process-cache read,
    and is byte-identical for every profile served by the same install — the
    desktop uses it to collapse duplicate roster rows when one backend is
    registered under two addresses.
    """

    @pytest.fixture(autouse=True)
    def _setup_test_client(self, monkeypatch):
        try:
            from starlette.testclient import TestClient
        except ImportError:
            pytest.skip("fastapi/starlette not installed")

        import hermes_cli.web_server as ws
        from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN

        # Fresh process cache per test: the cache is process-global by design
        # (stability), so tests must not observe a previous test's id.
        monkeypatch.setattr(ws, "_INSTALL_ID_CACHE", {"value": None})
        self.client = TestClient(app)
        self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN

    def test_status_reports_persistent_install_id(self, monkeypatch):
        from hermes_constants import get_default_hermes_root

        monkeypatch.setattr(_gw_status, "get_running_pid_cached", lambda: None)
        monkeypatch.setattr(_gw_status, "read_runtime_status", lambda: None)

        first = self.client.get("/api/status")
        assert first.status_code == 200
        install_id = first.json().get("install_id")
        assert isinstance(install_id, str)
        # Opaque random hex only — no hardware/user-derived material.
        assert re.fullmatch(r"[0-9a-f]{32}", install_id)

        # Persisted under the ROOT home, and stable across calls.
        id_file = get_default_hermes_root() / "install_id"
        assert id_file.is_file()
        assert id_file.read_text(encoding="utf-8").strip() == install_id

        second = self.client.get("/api/status")
        assert second.json().get("install_id") == install_id

    def test_install_id_survives_process_cache_reset(self, monkeypatch):
        """A restart (fresh cache) re-reads the SAME persisted id."""
        import hermes_cli.web_server as ws

        first = ws.get_install_id()
        assert first
        monkeypatch.setattr(ws, "_INSTALL_ID_CACHE", {"value": None})
        assert ws.get_install_id() == first

    def test_all_profiles_of_one_install_share_the_id(self, monkeypatch, tmp_path):
        """HERMES_HOME=<root> and HERMES_HOME=<root>/profiles/<name> resolve to
        the same id file — profiles share one physical install identity."""
        import hermes_cli.web_server as ws

        root = tmp_path / "hermes-root"
        profile_home = root / "profiles" / "research"
        profile_home.mkdir(parents=True)

        monkeypatch.setenv("HERMES_HOME", str(root))
        monkeypatch.setattr(ws, "_INSTALL_ID_CACHE", {"value": None})
        root_id = ws.get_install_id()
        assert root_id

        monkeypatch.setenv("HERMES_HOME", str(profile_home))
        monkeypatch.setattr(ws, "_INSTALL_ID_CACHE", {"value": None})
        assert ws.get_install_id() == root_id
        # Exactly one id file exists — under the root, not the profile home.
        assert (root / "install_id").is_file()
        assert not (profile_home / "install_id").exists()

    def test_corrupt_id_file_is_replaced_not_propagated(self, monkeypatch, tmp_path):
        import hermes_cli.web_server as ws

        root = tmp_path / "hermes-root"
        root.mkdir()
        (root / "install_id").write_text("not-a-valid-id\n", encoding="utf-8")
        monkeypatch.setenv("HERMES_HOME", str(root))
        monkeypatch.setattr(ws, "_INSTALL_ID_CACHE", {"value": None})

        value = ws.get_install_id()
        assert value and re.fullmatch(r"[0-9a-f]{32}", value)
        assert (root / "install_id").read_text(encoding="utf-8").strip() == value


class TestGatewayBusyReadout:
    """Tests for the NAS busy/drainable readout on /api/status.

    Behaviour contracts (not snapshots): assert how gateway_busy / gateway_drainable
    must RELATE to gateway_running + gateway_state + active_agents, and that every
    field degrades to a safe falsy value when the gateway is down or its status
    file is absent. Liveness must key off gateway_running, NEVER gateway_updated_at.
    """

    @pytest.fixture(autouse=True)
    def _setup_test_client(self):
        try:
            from starlette.testclient import TestClient
        except ImportError:
            pytest.skip("fastapi/starlette not installed")

        from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
        self.client = TestClient(app)
        self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN


    def test_draining_state_is_neither_busy_nor_drainable(self, monkeypatch):
        """While draining, the gateway is not a fresh begin-drain target, and
        busy is False even with a stale active_agents>0 in the file — the state
        gate dominates."""

        monkeypatch.setattr(_gw_status, "get_running_pid_cached", lambda: 1234)
        monkeypatch.setattr(_gw_status, "read_runtime_status", lambda: {
            "gateway_state": "draining",
            "platforms": {},
            "active_agents": 3,
        })

        data = self.client.get("/api/status").json()
        assert data["gateway_busy"] is False
        assert data["gateway_drainable"] is False


    def test_active_agents_unparseable_in_file_degrades_to_zero(self, monkeypatch):
        """A corrupt active_agents value in the status file must not 500 or
        produce a spurious busy — it degrades to 0/not-busy."""

        monkeypatch.setattr(_gw_status, "get_running_pid_cached", lambda: 1234)
        monkeypatch.setattr(_gw_status, "read_runtime_status", lambda: {
            "gateway_state": "running",
            "platforms": {},
            "active_agents": "garbage",
        })

        data = self.client.get("/api/status").json()
        assert data["active_agents"] == 0
        assert data["gateway_busy"] is False


class TestStatusMemoryBlock:
    """NS-656: /api/status must always carry a `memory` block."""

    @pytest.fixture(autouse=True)
    def _setup_test_client(self):
        try:
            from starlette.testclient import TestClient
        except ImportError:
            pytest.skip("fastapi/starlette not installed")

        from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
        self.client = TestClient(app)
        self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN


    def test_memory_block_degrades_when_collector_raises(self, monkeypatch):
        """A broken collector must never take down the status endpoint —
        the block degrades to pressure=unknown."""
        import gateway.memory_status as ms

        def _boom(*_a, **_k):
            raise RuntimeError("collector exploded")

        monkeypatch.setattr(ms, "collect_memory_status", _boom)
        resp = self.client.get("/api/status")
        assert resp.status_code == 200
        assert resp.json()["memory"] == {"pressure": "unknown"}


    def test_disk_block_degrades_when_collector_raises(self, monkeypatch):
        """Same contract as the memory block: a broken collector must never
        take down the status endpoint."""
        import gateway.disk_status as ds

        def _boom(*_a, **_k):
            raise RuntimeError("collector exploded")

        monkeypatch.setattr(ds, "collect_disk_status", _boom)
        resp = self.client.get("/api/status")
        assert resp.status_code == 200
        assert resp.json()["disk"] == {"pressure": "unknown"}


class TestGatewayUpdatedAtContract:
    """Contract tests for /api/status ``gateway_updated_at``.

    The field is promised to consumers (web/src/lib/api.ts declares
    ``string | null``) as an RFC3339 timestamp or null — NEVER a number.
    Legacy gateways wrote epoch floats into gateway_state.json and the file
    is hand-editable, so the endpoint must normalize whatever it reads.
    """

    @pytest.fixture(autouse=True)
    def _setup_test_client(self):
        try:
            from starlette.testclient import TestClient
        except ImportError:
            pytest.skip("fastapi/starlette not installed")

        from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
        self.client = TestClient(app)
        self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN

    @staticmethod
    def _assert_contract(value):
        """gateway_updated_at is None or a tz-aware-parseable ISO string."""
        from datetime import datetime

        assert not isinstance(value, bool), f"bool leaked: {value!r}"
        assert not isinstance(value, (int, float)), f"number leaked: {value!r}"
        if value is not None:
            assert isinstance(value, str)
            parsed = datetime.fromisoformat(value)
            assert parsed.tzinfo is not None, f"naive timestamp leaked: {value!r}"


    def test_local_runtime_valid_epoch_becomes_iso_string(self, monkeypatch):
        """A plausible legacy epoch value is converted, not dropped."""
        from datetime import datetime, timezone

        epoch = 1750000000
        monkeypatch.setattr(_gw_status, "get_running_pid_cached", lambda: 1234)
        monkeypatch.setattr(_gw_status, "read_runtime_status", lambda: {
            "gateway_state": "running",
            "platforms": {},
            "active_agents": 0,
            "updated_at": epoch,
        })

        value = self.client.get("/api/status").json()["gateway_updated_at"]
        assert isinstance(value, str)
        parsed = datetime.fromisoformat(value)
        assert parsed.tzinfo is not None
        assert parsed == datetime.fromtimestamp(epoch, tz=timezone.utc)


    def test_remote_health_numeric_updated_at_normalized(self, monkeypatch):
        """Cross-container path: the remote /health/detailed body is the
        runtime source, and a numeric updated_at from an older gateway build
        must still come out as string|null."""
        import hermes_cli.web_server as ws

        monkeypatch.setattr(_gw_status, "get_running_pid_cached", lambda: None)
        monkeypatch.setattr(_gw_status, "read_runtime_status", lambda: None)
        monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642")
        monkeypatch.setattr(_web_server_gateway, "_probe_gateway_health", lambda: (True, {
            "status": "ok",
            "gateway_state": "running",
            "platforms": {},
            "updated_at": 1750000000.25,
            "pid": 999,
        }))

        resp = self.client.get("/api/status")
        assert resp.status_code == 200
        data = resp.json()
        assert data["gateway_running"] is True
        self._assert_contract(data["gateway_updated_at"])
        # A plausible epoch is converted, not nulled.
        assert isinstance(data["gateway_updated_at"], str)


# ---------------------------------------------------------------------------
# Dashboard theme normaliser tests
# ---------------------------------------------------------------------------


class TestNormaliseThemeDefinition:
    """Tests for _normalise_theme_definition() — parses YAML theme files."""


    def test_rejects_non_dict(self):
        from hermes_cli.web_server_dashboard import _normalise_theme_definition
        assert _normalise_theme_definition("string") is None
        assert _normalise_theme_definition(None) is None
        assert _normalise_theme_definition([1, 2, 3]) is None

    def test_loose_colors_shorthand(self):
        """Bare hex strings under `colors` parse as {hex, alpha=1.0}."""
        from hermes_cli.web_server_dashboard import _normalise_theme_definition
        result = _normalise_theme_definition({
            "name": "loose",
            "colors": {"background": "#000000", "midground": "#ffffff"},
        })
        assert result is not None
        assert result["palette"]["background"] == {"hex": "#000000", "alpha": 1.0}
        assert result["palette"]["midground"] == {"hex": "#ffffff", "alpha": 1.0}
        # foreground falls back to default (transparent white)
        assert result["palette"]["foreground"]["hex"] == "#ffffff"
        assert result["palette"]["foreground"]["alpha"] == 0.0


class TestDiscoverUserThemes:
    """Tests for _discover_user_themes() — scans ~/.hermes/dashboard-themes/."""

    def test_returns_empty_when_dir_missing(self, tmp_path, monkeypatch):
        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
        assert _web_server_dashboard._discover_user_themes() == []

    def test_loads_and_normalises_yaml(self, tmp_path, monkeypatch):
        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
        themes_dir = tmp_path / "dashboard-themes"
        themes_dir.mkdir()
        (themes_dir / "ocean.yaml").write_text(
            "name: ocean\n"
            "label: Ocean\n"
            "palette:\n"
            "  background:\n"
            "    hex: \"#0a1628\"\n"
            "    alpha: 1.0\n"
            "layout:\n"
            "  density: spacious\n"
        )
        results = _web_server_dashboard._discover_user_themes()
        assert len(results) == 1
        assert results[0]["name"] == "ocean"
        assert results[0]["label"] == "Ocean"
        assert results[0]["palette"]["background"]["hex"] == "#0a1628"
        assert results[0]["layout"]["density"] == "spacious"
        # defaults filled in
        assert "fontSans" in results[0]["typography"]


    def test_ignores_transient_profile_override(self, tmp_path, monkeypatch):
        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
        themes_dir = tmp_path / "dashboard-themes"
        themes_dir.mkdir()
        (themes_dir / "mine.yaml").write_text("name: mine\n", encoding="utf-8")

        other = tmp_path / "other-profile"
        other.mkdir()

        from hermes_constants import (
            reset_hermes_home_override,
            set_hermes_home_override,
        )

        token = set_hermes_home_override(str(other))
        try:
            results = _web_server_dashboard._discover_user_themes()
        finally:
            reset_hermes_home_override(token)

        assert [r["name"] for r in results] == ["mine"]


class TestThemeBootstrapCSS:
    """Tests for _render_active_theme_bootstrap_css() and its injection
    into index.html via _serve_index() — the critical-CSS shim that kills
    the default-teal first-paint flash for user YAML themes."""

    @staticmethod
    def _write_theme(hermes_home, name="ocean"):
        themes_dir = hermes_home / "dashboard-themes"
        themes_dir.mkdir(exist_ok=True)
        (themes_dir / f"{name}.yaml").write_text(
            f"name: {name}\n"
            "label: Ocean\n"
            "palette:\n"
            "  background:\n"
            "    hex: \"#0a1628\"\n"
            "  midground:\n"
            "    hex: \"#dbe4f0\"\n"
            "typography:\n"
            "  fontSans: \"Inter, sans-serif\"\n"
            "  baseSize: \"17px\"\n",
            encoding="utf-8",
        )


    @staticmethod
    def _mount_spa_client(tmp_path, monkeypatch):
        from fastapi import FastAPI
        from starlette.testclient import TestClient
        import hermes_cli.web_server as ws

        dist = tmp_path / "web_dist"
        (dist / "assets").mkdir(parents=True)
        (dist / "index.html").write_text(
            "<html><head><title>t</title></head><body>SPA</body></html>",
            encoding="utf-8",
        )
        monkeypatch.setattr(ws, "WEB_DIST", dist)
        spa_app = FastAPI()
        _web_server_dashboard.mount_spa(spa_app)
        return TestClient(spa_app)

    def test_serve_index_injects_bootstrap_for_user_theme(self, tmp_path, monkeypatch):
        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
        self._write_theme(tmp_path)
        monkeypatch.setattr(
            _cfg_mod, "load_config", lambda: {"dashboard": {"theme": "ocean"}}
        )
        client = self._mount_spa_client(tmp_path, monkeypatch)
        resp = client.get("/chat")
        assert resp.status_code == 200
        assert '<style id="hermes-theme-bootstrap">' in resp.text
        assert "--background-base:#0a1628;" in resp.text
        # Injected inside <head>, before the closing tag.
        head = resp.text.split("</head>")[0]
        assert "hermes-theme-bootstrap" in head


class TestNormaliseThemeExtensions:
    """Tests for the extended normaliser fields (assets, customCSS,
    componentStyles, layoutVariant) — the surfaces themes use to reskin
    the dashboard without shipping code."""


    def test_custom_css_passthrough_and_capped(self):
        from hermes_cli.web_server_dashboard import _normalise_theme_definition
        # Small CSS passes through verbatim.
        r = _normalise_theme_definition({
            "name": "t",
            "customCSS": "body { color: red; }",
        })
        assert r["customCSS"] == "body { color: red; }"

        # 40 KiB of CSS gets clipped to the 32 KiB cap.
        huge = "/* x */ " * (40 * 1024 // 8 + 10)
        r2 = _normalise_theme_definition({"name": "t", "customCSS": huge})
        assert len(r2["customCSS"]) <= 32 * 1024


    def test_component_styles_per_bucket(self):
        from hermes_cli.web_server_dashboard import _normalise_theme_definition
        r = _normalise_theme_definition({
            "name": "t",
            "componentStyles": {
                "card": {
                    "clipPath": "polygon(0 0, 100% 0, 100% 100%, 0 100%)",
                    "boxShadow": "inset 0 0 0 1px red",
                    "bad prop!": "ignored",  # non-alnum prop rejected
                },
                "header": {"background": "linear-gradient(red, blue)"},
                "rogueBucket": {"foo": "bar"},  # not a known bucket — rejected
            },
        })
        assert r["componentStyles"]["card"] == {
            "clipPath": "polygon(0 0, 100% 0, 100% 100%, 0 100%)",
            "boxShadow": "inset 0 0 0 1px red",
        }
        assert r["componentStyles"]["header"]["background"].startswith("linear-gradient")
        assert "rogueBucket" not in r["componentStyles"]


class TestDeleteSessionEndpoint:
    """Tests for ``DELETE /api/sessions/{session_id}`` — the single-row delete
    behind the desktop sidebar's per-session delete.

    The desktop optimistically removes the row, then RESTORES it on any error
    and surfaces the message. So a 404 on a row that is already gone (reaped by
    empty-session hygiene, or removed by a concurrent client — both common amid
    /goal + auto-compression churn that leaves transient empty rows) resurrected
    a ghost row and showed "session not found". DELETE must be idempotent and
    resolve ids like every other session endpoint.
    """

    @pytest.fixture(autouse=True)
    def _setup_test_client(self, monkeypatch, _isolate_hermes_home):
        try:
            from starlette.testclient import TestClient
        except ImportError:
            pytest.skip("fastapi/starlette not installed")

        import hermes_state
        from hermes_constants import get_hermes_home
        from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN

        monkeypatch.setattr(
            hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db"
        )

        self.auth_client = TestClient(app)
        self.auth_client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN

    def test_delete_absent_session_is_idempotent(self):
        # PREMISE / regression: deleting a row that no longer exists must NOT
        # 404 — the desktop would resurrect the ghost row and show
        # "session not found". DELETE's contract is "ensure it's gone".
        resp = self.auth_client.delete("/api/sessions/never_existed")
        assert resp.status_code == 200
        assert resp.json().get("ok") is True

    def test_delete_existing_session_scrubs_row_and_disk(self):
        # The CLI delete path threads the sessions dir so transcript
        # artifacts are removed with the row; the endpoint historically
        # didn't, leaving secret-bearing session_<id>.json snapshots and
        # request dumps orphaned on disk after a UI delete.
        from hermes_constants import get_hermes_home
        from hermes_state import SessionDB

        db_path = get_hermes_home() / "state.db"
        db = SessionDB(db_path=db_path)
        try:
            db.create_session("disk-scrub", source="cli")
        finally:
            db.close()

        sessions_dir = get_hermes_home() / "sessions"
        sessions_dir.mkdir(parents=True, exist_ok=True)
        for name, body in (
            ("session_disk-scrub.json", '{"messages": [{"content": "secret-token"}]}'),
            ("disk-scrub.jsonl", "{}\n"),
            ("request_dump_disk-scrub_001.json", "{}"),
        ):
            (sessions_dir / name).write_text(body, encoding="utf-8")
        # Another session's artifacts must survive.
        (sessions_dir / "session_disk-scrub-neighbour.json").write_text("{}", encoding="utf-8")

        resp = self.auth_client.delete("/api/sessions/disk-scrub")

        assert resp.status_code == 200
        assert resp.json().get("ok") is True
        db = SessionDB(db_path=db_path)
        try:
            assert db.get_session("disk-scrub") is None
        finally:
            db.close()
        assert not (sessions_dir / "session_disk-scrub.json").exists()
        assert not (sessions_dir / "disk-scrub.jsonl").exists()
        assert not (sessions_dir / "request_dump_disk-scrub_001.json").exists()
        assert (sessions_dir / "session_disk-scrub-neighbour.json").exists()

    def test_delete_named_profile_session_scrubs_profile_disk(self):
        from hermes_cli import profiles as profiles_mod
        from hermes_state import SessionDB

        profile_home = profiles_mod.get_profile_dir("worker")
        profile_home.mkdir(parents=True)
        (profile_home / "config.yaml").touch()  # identity marker: bare dirs are not profiles
        sessions_dir = profile_home / "sessions"
        sessions_dir.mkdir(parents=True, exist_ok=True)
        db_path = profile_home / "state.db"
        db = SessionDB(db_path=db_path)
        try:
            db.create_session("profile-scrub", source="cli")
        finally:
            db.close()
        (sessions_dir / "session_profile-scrub.json").write_text(
            '{"messages": [{"content": "secret-token"}]}', encoding="utf-8"
        )

        resp = self.auth_client.delete("/api/sessions/profile-scrub?profile=worker")

        assert resp.status_code == 200
        assert resp.json().get("ok") is True
        db = SessionDB(db_path=db_path)
        try:
            assert db.get_session("profile-scrub") is None
        finally:
            db.close()
        assert not (sessions_dir / "session_profile-scrub.json").exists()


class TestBulkDeleteSessionsEndpoint:
    """Tests for ``POST /api/sessions/bulk-delete`` — backs the
    dashboard's "Delete N selected" flow on the sessions page.

    Locks in four things:

    1. Route-ordering: ``/api/sessions/bulk-delete`` must shadow the
       templated ``/api/sessions/{session_id}`` route below it (see
       the block comment in ``hermes_cli/web_server.py``).
    2. Behaviour parity with :meth:`SessionDB.delete_sessions` — real
       deleted count, archive/active sessions deleted on explicit
       selection.
    3. The 500-ID payload cap is enforced.
    4. Auth gating (issue #19533 contract).
    """

    @pytest.fixture(autouse=True)
    def _setup_test_client(self, monkeypatch, _isolate_hermes_home):
        try:
            from starlette.testclient import TestClient
        except ImportError:
            pytest.skip("fastapi/starlette not installed")

        import hermes_state
        from hermes_constants import get_hermes_home
        from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN

        monkeypatch.setattr(
            hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db"
        )

        self.client = TestClient(app)
        self.auth_client = TestClient(app)
        self.auth_client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN

    def _seed(self, ids):
        from hermes_state import SessionDB

        db = SessionDB()
        try:
            for sid in ids:
                db.create_session(session_id=sid, source="cli")
        finally:
            db.close()


    def test_deletes_listed_sessions_only(self):
        from hermes_state import SessionDB

        self._seed(["a", "b", "c"])
        resp = self.auth_client.post(
            "/api/sessions/bulk-delete", json={"ids": ["a", "b"]}
        )
        assert resp.status_code == 200
        assert resp.json() == {"ok": True, "deleted": 2, "skipped_active": []}

        db = SessionDB()
        try:
            assert db.get_session("a") is None
            assert db.get_session("b") is None
            assert db.get_session("c") is not None
        finally:
            db.close()


class TestDeleteEmptySessionsEndpoint:
    """Tests for ``GET /api/sessions/empty/count`` and
    ``DELETE /api/sessions/empty`` — the bulk-delete endpoints backing
    the dashboard's "Delete empty" button.

    Locks in three things the implementation has to get right:

    1. Route-ordering: the literal ``/api/sessions/empty[/count]`` paths
       must shadow the templated ``/api/sessions/{session_id}`` route
       above them. A regression here would route ``DELETE /api/sessions/
       empty`` to the single-session handler with ``session_id="empty"``
       (which 404s instead of bulk-deleting).
    2. Behaviour parity with :meth:`SessionDB.delete_empty_sessions`:
       active sessions and archived sessions are both preserved.
    3. Auth gating: both routes require the session token like every
       other ``/api/*`` endpoint (issue #19533 contract).
    """

    @pytest.fixture(autouse=True)
    def _setup_test_client(self, monkeypatch, _isolate_hermes_home):
        try:
            from starlette.testclient import TestClient
        except ImportError:
            pytest.skip("fastapi/starlette not installed")

        import hermes_state
        from hermes_constants import get_hermes_home
        from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN

        # Pin the SessionDB to the isolated HERMES_HOME so each test
        # starts with a clean state.db.
        monkeypatch.setattr(
            hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db"
        )

        self.client = TestClient(app)
        self.auth_client = TestClient(app)
        self.auth_client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN

    def _seed(self):
        """Build the standard test corpus:

        * ``empty1`` / ``empty2`` — ended, no messages → should delete
        * ``hasmsg``  — ended, has one message → must survive
        * ``live``    — un-ended, empty → must survive (active)
        * ``archived``— ended, empty, archived → must survive
        """
        from hermes_state import SessionDB

        db = SessionDB()
        try:
            db.create_session(session_id="empty1", source="cli")
            db.end_session("empty1", end_reason="done")
            db.create_session(session_id="empty2", source="cli")
            db.end_session("empty2", end_reason="done")

            db.create_session(session_id="hasmsg", source="cli")
            db.append_message("hasmsg", role="user", content="hello")
            db.end_session("hasmsg", end_reason="done")

            db.create_session(session_id="live", source="cli")

            db.create_session(session_id="archived", source="cli")
            db.end_session("archived", end_reason="done")
            db.set_session_archived("archived", True)
        finally:
            db.close()


    def test_delete_endpoint_requires_auth(self):
        """DELETE /api/sessions/empty must 401 without the session token.

        Regression guard for issue #19533 — the bulk-delete is a strictly
        destructive primitive, the middleware must gate it even if a
        future refactor introduces a non-auth path."""
        resp = self.client.delete("/api/sessions/empty")
        assert resp.status_code == 401


    def test_delete_returns_count_and_removes_only_empties(self):
        """DELETE returns the deleted count and removes only the
        empty-ended-unarchived rows — same shape contract as the
        DB-level method's unit tests."""
        from hermes_state import SessionDB

        self._seed()
        resp = self.auth_client.delete("/api/sessions/empty")
        assert resp.status_code == 200
        assert resp.json() == {"ok": True, "deleted": 2}

        db = SessionDB()
        try:
            assert db.get_session("empty1") is None
            assert db.get_session("empty2") is None
            # Survivors: hasmsg has a message, live is active, archived
            # is archived. All three must still be there.
            assert db.get_session("hasmsg") is not None
            assert db.get_session("live") is not None
            assert db.get_session("archived") is not None
            # And the count endpoint now reports 0.
            assert db.count_empty_sessions() == 0
        finally:
            db.close()

    def test_delete_removes_on_disk_files_of_deleted_sessions_only(self):
        """Deleting an empty session also removes its files in ``sessions/``.
        A kept session's files stay."""
        from hermes_constants import get_hermes_home

        self._seed()
        sessions_dir = get_hermes_home() / "sessions"
        sessions_dir.mkdir(parents=True, exist_ok=True)
        deleted_files = [sessions_dir / "session_empty1.json", sessions_dir / "request_dump_empty2_1.json"]
        kept_file = sessions_dir / "session_hasmsg.json"
        for path in (*deleted_files, kept_file):
            path.write_text("{}", encoding="utf-8")

        resp = self.auth_client.delete("/api/sessions/empty")

        assert resp.json() == {"ok": True, "deleted": 2}
        assert [p for p in deleted_files if p.exists()] == []
        assert kept_file.exists()


class TestPluginAPIAuth:
    """Tests that plugin API routes require the session token (issue #19533)."""

    @pytest.fixture(autouse=True)
    def _setup_test_client(self, monkeypatch, _isolate_hermes_home, _install_example_plugin):
        """Create a TestClient without the session token header.

        Pulls in ``_install_example_plugin`` so ``test_plugin_route_allows_auth``
        has the ``/api/plugins/example/hello`` endpoint available — the
        example plugin is no longer a bundled plugin, so the fixture
        installs it into the per-test ``HERMES_HOME``.
        """
        try:
            from starlette.testclient import TestClient
        except ImportError:
            pytest.skip("fastapi/starlette not installed")

        import hermes_state
        from hermes_constants import get_hermes_home
        from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN

        monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db")

        self.client = TestClient(app)
        self.auth_client = TestClient(app)
        self.auth_client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN


    def test_plugin_route_allows_auth(self):
        """Plugin API routes should work with a valid session token.

        Uses ``/api/plugins/example/hello`` from the example-dashboard
        test fixture (installed into HERMES_HOME by the class-level
        ``_install_example_plugin`` fixture) — a stable, side-effect-free
        GET that's only loaded for tests. With a valid token the handler
        should run (200); without one the middleware should 401 before
        the handler is reached.
        """
        # Without auth: middleware blocks before reaching the handler.
        resp = self.client.get("/api/plugins/example/hello")
        assert resp.status_code == 401

        # With auth: handler runs.
        resp = self.auth_client.get("/api/plugins/example/hello")
        assert resp.status_code == 200


    def test_plugin_patch_requires_auth(self):
        """Plugin PATCH routes should return 401 without a valid session token.

        PATCH is the mutation method most commonly used by the dashboard for
        kanban task edits — explicitly cover it so a future middleware
        regression that whitelists non-GET methods can't sneak through.
        """
        resp = self.client.patch(
            "/api/plugins/kanban/tasks/t_fake",
            json={"title": "renamed"},
        )
        assert resp.status_code == 401


    def test_non_kanban_plugin_route_requires_auth(self):
        """Auth must be plugin-agnostic, not kanban-specific.

        The middleware fix is at the gate level (no per-plugin allowlist),
        so any plugin's API surface — kanban, hermes-achievements, future
        plugins — must require the session token. Hit a non-kanban plugin
        path to lock that in.
        """
        # Real plugin path (hermes-achievements is loaded by default).
        resp = self.client.get("/api/plugins/hermes-achievements/overview")
        assert resp.status_code == 401
        # Same for an arbitrary plugin namespace that doesn't even exist —
        # the middleware should 401 before routing decides 404, so an
        # attacker can't fingerprint plugin names by status codes.
        resp = self.client.get("/api/plugins/_definitely_not_a_plugin_/anything")
        assert resp.status_code == 401


class TestPluginAPISecretScopeProductionMount:
    """#120310: a plugin API handler's ``get_secret()`` must resolve the *requested*
    profile's credentials under multi-profile hosting, verified through the REAL mount
    path — discovery → import → ``_mount_plugin_api_routes()`` (which attaches
    ``_plugin_route_secret_scope``) → a live request against ``app`` — not a hand-built
    ``include_router``. The dedicated ``test_plugin_api_secret_scope.py`` suite proves the
    dependency in isolation; this closes the actual reported surface end-to-end and pins
    that the launch profile and a ``?profile=`` request read distinct secrets, that the
    request profile does not leak back into the launch profile, and that an unknown
    profile is rejected in the dependency before the handler runs.
    """

    _PROBE_KEY = "EXAMPLE_PLUGIN_PROBE_KEY"

    @pytest.fixture(autouse=True)
    def _setup(self, monkeypatch, _isolate_hermes_home, _install_example_plugin):
        try:
            from starlette.testclient import TestClient
        except ImportError:
            pytest.skip("fastapi/starlette not installed")

        import hermes_state
        from hermes_constants import get_hermes_home
        from hermes_cli import profiles
        from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN

        default_home = get_hermes_home()
        monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", default_home / "state.db")

        # Anchor the named-profiles root to the isolated home so ``?profile=workerb``
        # resolves inside the test sandbox (mirrors test_web_server_skills_profiles).
        profiles_root = default_home / "profiles"
        monkeypatch.setattr(profiles, "_get_default_hermes_home", lambda: default_home)
        monkeypatch.setattr(profiles, "_get_profiles_root", lambda: profiles_root)

        # Named profile B: a live profile (``.env`` is both an identity marker and the
        # secret source) with its OWN value for the probe key.
        worker_home = profiles_root / "workerb"
        worker_home.mkdir(parents=True, exist_ok=True)
        (worker_home / ".env").write_text(f"{self._PROBE_KEY}=sk-workerb\n", encoding="utf-8")

        # Launch profile A: an env-only credential (systemd ``Environment=`` style), frozen
        # into the launch scope when multi-profile hosting activates.
        monkeypatch.setenv(self._PROBE_KEY, "sk-launch-a")

        import tui_gateway.launch_profile_policy as lpp
        from agent.secret_scope import is_multiplex_active, set_multiplex_active

        monkeypatch.setattr(lpp, "_snapshot", None)  # freeze the launch env fresh
        self._previous_multiplex = is_multiplex_active()
        lpp.activate_multi_profile_hosting()  # freezes os.environ + flips multiplex on

        self.client = TestClient(app)
        self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
        try:
            yield
        finally:
            set_multiplex_active(self._previous_multiplex)

    def _probe(self, profile=None):
        params = {"profile": profile} if profile is not None else None
        return self.client.get("/api/plugins/example/whoami", params=params)

    def test_launch_and_requested_profile_read_distinct_secrets(self):
        # No ``?profile=`` → the launch profile's frozen env-only credential.
        resp = self._probe()
        assert resp.status_code == 200
        assert resp.json() == {"ok": True, "key": "sk-launch-a"}

        # ``?profile=workerb`` → that profile's own credential, through the real mount.
        resp = self._probe("workerb")
        assert resp.status_code == 200
        assert resp.json() == {"ok": True, "key": "sk-workerb"}

        # Back to the launch profile: the request scope reset, so B never leaks into A.
        resp = self._probe()
        assert resp.status_code == 200
        assert resp.json() == {"ok": True, "key": "sk-launch-a"}

    def test_unknown_profile_rejected_before_handler(self):
        # The dependency raises HTTPException(404) before the handler runs, so this is a
        # 404 — NOT the handler's folded ``{"ok": False}`` no-data contract.
        resp = self._probe("ghost")
        assert resp.status_code == 404


class TestDashboardPluginManifestExtensions:
    """Tests for the extended plugin manifest fields (tab.override,
    tab.hidden, slots) read by _discover_dashboard_plugins()."""

    def _write_plugin(self, tmp_path, name, manifest):
        import json
        plug_dir = tmp_path / "plugins" / name / "dashboard"
        plug_dir.mkdir(parents=True)
        (plug_dir / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
        return plug_dir

    def test_override_and_hidden_carried_through(self, tmp_path, monkeypatch):
        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
        self._write_plugin(tmp_path, "skin-home", {
            "name": "skin-home",
            "label": "Skin Home",
            "tab": {"path": "/skin-home", "override": "/", "hidden": True},
            "slots": ["sidebar", "header-left"],
            "entry": "dist/index.js",
        })
        from hermes_cli import web_server
        # Bust the process-level cache so the test plugin is picked up.
        web_server._dashboard_plugins_cache = None
        plugins = web_server._get_dashboard_plugins(force_rescan=True)
        entry = next(p for p in plugins if p["name"] == "skin-home")
        assert entry["tab"]["override"] == "/"
        assert entry["tab"]["hidden"] is True
        assert entry["slots"] == ["sidebar", "header-left"]

    def test_user_plugins_ignore_profile_home_override(self, tmp_path, monkeypatch):
        """Regression: user dashboard extensions are a dashboard-owned asset
        (like theme YAML), so they must stay visible after a context-local
        HERMES_HOME override scopes a request to another profile."""
        from hermes_constants import (
            reset_hermes_home_override,
            set_hermes_home_override,
        )
        launch_home = tmp_path / "launch"
        launch_home.mkdir()
        self._write_plugin(launch_home, "skin-home", {
            "name": "skin-home",
            "label": "Skin Home",
            "tab": {"path": "/skin-home"},
            "entry": "dist/index.js",
        })
        other = tmp_path / "other-profile"
        other.mkdir()

        monkeypatch.setenv("HERMES_HOME", str(launch_home))
        token = set_hermes_home_override(str(other))
        try:
            plugins = _web_server_dashboard._discover_dashboard_plugins()
        finally:
            reset_hermes_home_override(token)
        assert any(p["name"] == "skin-home" for p in plugins)

    def test_user_plugins_found_under_profile_scoped_process(self, tmp_path, monkeypatch):
        """Regression #87197: a profile-scoped process (``--profile <name>``
        sets HERMES_HOME=<root>/profiles/<name>) must still discover user
        plugins installed in the hermes root's plugins/ directory."""
        root = tmp_path / "hermes-root"
        profile_home = root / "profiles" / "presale"
        profile_home.mkdir(parents=True)
        self._write_plugin(root, "meeting-intelligence", {
            "name": "meeting-intelligence",
            "label": "Meeting Intelligence",
            "tab": {"path": "/meetings"},
            "entry": "dist/index.js",
        })

        monkeypatch.setenv("HERMES_HOME", str(profile_home))
        plugins = _web_server_dashboard._discover_dashboard_plugins()
        assert any(p["name"] == "meeting-intelligence" for p in plugins)

    def test_profile_local_plugin_wins_over_root_plugin(self, tmp_path, monkeypatch):
        """A same-named plugin in the profile home takes precedence over the
        root copy (seen_names dedupe, profile scanned first)."""
        root = tmp_path / "hermes-root"
        profile_home = root / "profiles" / "presale"
        profile_home.mkdir(parents=True)
        self._write_plugin(profile_home, "dupe", {
            "name": "dupe",
            "label": "Profile Copy",
            "tab": {"path": "/from-profile"},
            "entry": "dist/index.js",
        })
        self._write_plugin(root, "dupe", {
            "name": "dupe",
            "label": "Root Copy",
            "tab": {"path": "/from-root"},
            "entry": "dist/index.js",
        })

        monkeypatch.setenv("HERMES_HOME", str(profile_home))
        plugins = _web_server_dashboard._discover_dashboard_plugins()
        entries = [p for p in plugins if p["name"] == "dupe"]
        assert len(entries) == 1
        assert entries[0]["tab"]["path"] == "/from-profile"

    def test_unreadable_plugin_paths_do_not_block_discovery(self, tmp_path, monkeypatch):
        """A denied plugin directory or manifest must not prevent valid plugins loading."""
        from pathlib import Path

        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
        self._write_plugin(tmp_path, "valid", {
            "name": "valid",
            "label": "Valid Plugin",
            "entry": "dist/index.js",
        })
        denied_root = tmp_path / "denied-root"
        denied_root.mkdir()
        denied_plugin = tmp_path / "plugins" / "denied"
        (denied_plugin / "dashboard").mkdir(parents=True)
        (denied_plugin / "dashboard" / "manifest.json").write_text("{}", encoding="utf-8")

        from hermes_cli import web_server_dashboard
        original_search_dirs = web_server_dashboard._dashboard_plugin_search_dirs
        original_scandir = web_server_dashboard.os.scandir
        original_exists = Path.exists

        def search_dirs():
            return [(denied_root, "user"), *original_search_dirs()]

        def guarded_scandir(path):
            if Path(path) == denied_root:
                raise PermissionError("[WinError 5] Access is denied")
            return original_scandir(path)

        def guarded_exists(path):
            if path == denied_plugin / "dashboard" / "manifest.json":
                raise PermissionError("[WinError 5] Access is denied")
            return original_exists(path)

        monkeypatch.setattr(web_server_dashboard, "_dashboard_plugin_search_dirs", search_dirs)
        monkeypatch.setattr(web_server_dashboard.os, "scandir", guarded_scandir)
        monkeypatch.setattr(Path, "exists", guarded_exists)

        plugins = web_server_dashboard._discover_dashboard_plugins()

        assert "valid" in {plugin["name"] for plugin in plugins}
        assert "denied" not in {plugin["name"] for plugin in plugins}


# ---------------------------------------------------------------------------
# /api/pty WebSocket — terminal bridge for the dashboard "Chat" tab.
#
# These tests drive the endpoint with a tiny fake command (typically ``cat``
# or ``sh -c 'printf …'``) instead of the real ``hermes --tui`` binary.  The
# endpoint resolves its argv through ``_resolve_chat_argv``, so tests
# monkeypatch that hook.
# ---------------------------------------------------------------------------

from hermes_cli import main_tui_launch


skip_on_windows = pytest.mark.skipif(
    sys.platform.startswith("win"), reason="PTY bridge is POSIX-only"
)


@skip_on_windows
class TestPtyWebSocket:
    @pytest.fixture(autouse=True)
    def _setup(self, monkeypatch, _isolate_hermes_home):
        from starlette.testclient import TestClient

        import hermes_cli.web_server as ws

        # Avoid exec'ing the actual TUI in tests: every test below installs
        # its own fake argv via ``web_server_chat._resolve_chat_argv``.
        self.ws_module = ws
        monkeypatch.setattr(ws, "_DASHBOARD_EMBEDDED_CHAT_ENABLED", True)
        ws.app.state.pty_active_session_files = {}
        self.token = ws._SESSION_TOKEN
        self.client = TestClient(ws.app)

    def _url(self, token: str | None = None, **params: str) -> str:
        tok = token if token is not None else self.token
        # TestClient.websocket_connect takes the path; it reconstructs the
        # query string, so we pass it inline.
        from urllib.parse import urlencode

        q = {"token": tok, **params}
        return f"/api/pty?{urlencode(q)}"


    def test_tui_python_command_uses_child_path(self, tmp_path):
        """Bare Python commands are resolved from the TUI child's PATH."""

        command = f"hermes-review-python{Path(sys.executable).suffix}"
        bin_dir = tmp_path / "bin"
        bin_dir.mkdir()
        executable = bin_dir / command
        # copy2, not os.link: tmp_path may sit on a different filesystem than
        # the venv (tmpfs /tmp vs disk home) where hard links raise EXDEV.
        shutil.copy2(sys.executable, executable)
        env = {
            "HERMES_CWD": str(tmp_path),
            "HERMES_PYTHON": command,
            "PATH": str(bin_dir),
        }

        main_tui_launch._apply_tui_python_env(env)

        assert env["HERMES_PYTHON"] == command


    def test_unavailable_platform_closes_with_message(self, monkeypatch):
        from hermes_cli.pty_bridge import PtyUnavailableError

        def _raise(argv, **kwargs):
            raise PtyUnavailableError("pty missing for tests")

        monkeypatch.setattr(
            _web_server_chat,
            "_resolve_chat_argv",
            lambda resume=None, sidecar_url=None, profile=None: (["/bin/cat"], None, None),
        )
        # Patch PtyBridge.spawn at the web_server_chat module's binding.
        monkeypatch.setattr(_web_server_chat.PtyBridge, "spawn", classmethod(lambda cls, *a, **k: _raise(*a, **k)))

        with self.client.websocket_connect(self._url()) as conn:
            # Expect a final text frame with the error message, then close.
            msg = conn.receive_text()
            assert "pty missing" in msg or "unavailable" in msg.lower() or "pty" in msg.lower()


    def test_pub_broadcasts_to_events_subscribers(self):
        """A frame handed to _broadcast_event is sent verbatim to every
        subscriber registered on that channel — and not to subscribers on
        other channels.

        This drives the broadcast unit directly under asyncio rather than
        round-tripping through Starlette's TestClient WebSocket portal. The
        portal version was flaky under heavy parallel CI load: the broadcast
        had to traverse two nested threaded portals within a 10s wall-clock
        budget, and a starved ASGI thread occasionally blew that budget even
        though the server logic was correct. Testing _broadcast_event with
        fake subscribers removes the scheduling surface entirely while
        asserting the exact fan-out contract.
        """
        import asyncio
        from hermes_cli import web_server as ws_mod

        class _FakeSub:
            def __init__(self):
                self.sent: list[str] = []

            async def send_text(self, payload: str) -> None:
                self.sent.append(payload)

        app = ws_mod.app

        async def _run():
            sub_a1 = _FakeSub()
            sub_a2 = _FakeSub()
            sub_other = _FakeSub()
            frame = '{"type":"tool.start","payload":{"tool_id":"t1"}}'

            event_channels, event_lock = _rt_chat_ws._get_event_state(app)
            # Register two subscribers on the target channel and one on a
            # different channel, exactly as the /api/events handler does.
            async with event_lock:
                event_channels.setdefault("broadcast-test", set()).update(
                    {sub_a1, sub_a2}
                )
                event_channels.setdefault("other-channel", set()).add(sub_other)
            try:
                await _rt_chat_ws._broadcast_event(app, "broadcast-test", frame)
            finally:
                async with event_lock:
                    event_channels.pop("broadcast-test", None)
                    event_channels.pop("other-channel", None)

            return sub_a1, sub_a2, sub_other, frame

        sub_a1, sub_a2, sub_other, frame = asyncio.run(_run())

        # Every subscriber on the channel got the frame verbatim, exactly once.
        assert sub_a1.sent == [frame]
        assert sub_a2.sent == [frame]
        # A subscriber on a different channel got nothing.
        assert sub_other.sent == []


def test_resolve_chat_argv_injects_gateway_ws_url(monkeypatch):
    import hermes_cli.main_tui_launch as tui_launch
    import hermes_cli.web_server as ws

    monkeypatch.setenv("PATH", "/run/current-system/sw/bin:/usr/bin")
    monkeypatch.setattr(
        tui_launch,
        "_make_tui_argv",
        lambda *_args, **_kwargs: (["node", "fake-tui.js"], Path("/tmp")),
    )
    monkeypatch.setattr(ws.app.state, "bound_host", "127.0.0.1", raising=False)
    monkeypatch.setattr(ws.app.state, "bound_port", 9119, raising=False)

    _argv, _cwd, env = _web_server_chat._resolve_chat_argv()

    assert env is not None
    gateway_url = env.get("HERMES_TUI_GATEWAY_URL", "")
    assert gateway_url.startswith("ws://127.0.0.1:9119/api/ws?")
    assert "token=" in gateway_url


class TestDashboardPluginStaticAssetAllowlist:
    """``/dashboard-plugins/<name>/<path>`` is unauthenticated by design —
    the SPA loads plugin JS via ``<script src>`` and CSS via
    ``<link href>``, neither of which can attach a custom auth header.
    Instead the route restricts file types to the browser-asset
    allowlist (JS/CSS/JSON/images/fonts) so that user-installed
    plugins shipping a ``plugin_api.py`` backend module don't leak
    their Python source to anyone reachable on the loopback port.

    Regression test for the dashboard pentest finding filed alongside
    the ``web-pentest`` skill (PR #32265 / issue #32267).
    """

    @pytest.fixture(autouse=True)
    def _setup_test_client(self, monkeypatch, _isolate_hermes_home, _install_example_plugin):
        """Create a TestClient and install the example-dashboard fixture.

        The static-asset allowlist tests need a plugin to point at —
        they verify that ``/dashboard-plugins/example/manifest.json``
        is served while ``plugin_api.py`` and ``__pycache__/*.pyc``
        from the same directory are not. Since the example plugin is
        no longer bundled, ``_install_example_plugin`` lays it down in
        the per-test ``HERMES_HOME`` user-plugins dir.
        """
        try:
            from starlette.testclient import TestClient
        except ImportError:
            pytest.skip("fastapi/starlette not installed")

        from hermes_cli.web_server import app

        self.client = TestClient(app)

    def test_python_source_is_404(self):
        """The example plugin's ``plugin_api.py`` must NOT be served as
        a static asset, even though the file exists under the plugin's
        dashboard directory. Suffix not in the allowlist → 404."""
        resp = self.client.get("/dashboard-plugins/example/plugin_api.py")
        assert resp.status_code == 404


    def test_manifest_json_still_served(self):
        """JSON files remain browser-fetchable — manifests, localized
        data, source maps, etc. all sit in this bucket."""
        resp = self.client.get("/dashboard-plugins/example/manifest.json")
        assert resp.status_code == 200
        assert resp.headers["content-type"].startswith("application/json")
        # And the body is actually the manifest, not the SPA fallback.
        body = resp.json()
        assert body.get("name") == "example"


    def test_path_traversal_still_blocked(self):
        """The allowlist is on top of the existing ``.resolve()`` /
        ``is_relative_to()`` check — a ``.js`` named file at an
        out-of-base path is still rejected as traversal, not served."""
        resp = self.client.get(
            "/dashboard-plugins/example/..%2Fplugin_api.py"
        )
        # 403 traversal-blocked OR 404 (depending on URL decode order)
        # — never 200.
        assert resp.status_code in (403, 404)


def _fake_httpx_async_client(*, status: int | None = None, raise_exc: bool = False):
    """Build a drop-in for httpx.AsyncClient with a canned GET response."""

    class _Resp:
        def __init__(self, code):
            self.status_code = code

        @property
        def is_success(self):
            return 200 <= self.status_code < 300

    class _Client:
        def __init__(self, *a, **k):
            pass

        async def __aenter__(self):
            return self

        async def __aexit__(self, *a):
            return False

        async def get(self, *a, **k):
            if raise_exc:
                raise RuntimeError("connection refused")
            return _Resp(status)

    return _Client


class TestValidateProviderCredential:
    """Live-probe credential validation (/api/providers/validate)."""

    @pytest.fixture(autouse=True)
    def _setup_test_client(self, monkeypatch, _isolate_hermes_home):
        try:
            from starlette.testclient import TestClient
        except ImportError:
            pytest.skip("fastapi/starlette not installed")

        from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN

        self.client = TestClient(app)
        self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN

        class _BlockingClient:
            def __init__(self, *args, **kwargs):
                raise AssertionError(
                    "async validation route used blocking httpx.Client"
                )

        monkeypatch.setattr("httpx.Client", _BlockingClient)

    def _post(self, key, value):
        return self.client.post(
            "/api/providers/validate", json={"key": key, "value": value}
        )

    def test_rejected_key_blocks(self, monkeypatch):
        monkeypatch.setattr("httpx.AsyncClient", _fake_httpx_async_client(status=401))
        data = self._post("OPENROUTER_API_KEY", "sk-bogus").json()
        assert data["ok"] is False and data["reachable"] is True

    def test_valid_key_passes(self, monkeypatch):
        monkeypatch.setattr("httpx.AsyncClient", _fake_httpx_async_client(status=200))
        data = self._post("OPENAI_API_KEY", "sk-real").json()
        assert data["ok"] is True and data["reachable"] is True

    def test_rate_limited_counts_as_valid(self, monkeypatch):
        monkeypatch.setattr("httpx.AsyncClient", _fake_httpx_async_client(status=429))
        data = self._post("XAI_API_KEY", "xai-real").json()
        assert data["ok"] is True

    def test_network_error_is_unreachable_not_blocking(self, monkeypatch):
        monkeypatch.setattr(
            "httpx.AsyncClient", _fake_httpx_async_client(raise_exc=True)
        )
        data = self._post("OPENROUTER_API_KEY", "sk-real").json()
        assert data["ok"] is False and data["reachable"] is False


    def test_local_endpoint_forwards_api_key_as_bearer(self, monkeypatch):
        """A custom endpoint that gates /v1/models behind auth must still
        enumerate models: the optional api_key is sent as a Bearer header so the
        probe doesn't come back empty (the desktop loop's root cause)."""
        captured = {}

        class _Resp:
            status_code = 200
            is_success = True

            def json(self):
                return {"data": [{"id": "gpt-oss-120b"}]}

        class _Client:
            def __init__(self, *a, **k):
                pass

            async def __aenter__(self):
                return self

            async def __aexit__(self, *a):
                return False

            async def get(self, url, *a, headers=None, **k):
                captured["url"] = url
                captured["headers"] = headers
                return _Resp()

        monkeypatch.setattr("httpx.AsyncClient", _Client)

        resp = self.client.post(
            "/api/providers/validate",
            json={
                "key": "OPENAI_BASE_URL",
                "value": "https://text.example.com/v1",
                "api_key": "sk-secret",
            },
        )
        data = resp.json()
        assert data["ok"] is True and data["reachable"] is True
        assert data["models"] == ["gpt-oss-120b"]
        assert captured["url"] == "https://text.example.com/v1/models"
        assert captured["headers"] == {"Authorization": "Bearer sk-secret"}

    def test_local_endpoint_without_key_sends_no_auth_header(self, monkeypatch):
        """No key → no Authorization header (keyless local servers unaffected)."""
        captured = {}

        class _Resp:
            status_code = 200
            is_success = True

            def json(self):
                return {"data": []}

        class _Client:
            def __init__(self, *a, **k):
                pass

            async def __aenter__(self):
                return self

            async def __aexit__(self, *a):
                return False

            async def get(self, url, *a, headers=None, **k):
                captured["headers"] = headers
                return _Resp()

        monkeypatch.setattr("httpx.AsyncClient", _Client)

        self.client.post(
            "/api/providers/validate",
            json={"key": "OPENAI_BASE_URL", "value": "http://127.0.0.1:8000/v1"},
        )
        assert captured["headers"] is None

    def test_named_custom_endpoint_probe_is_async(self, monkeypatch):
        """Custom endpoint validation must not block the dashboard event loop."""
        captured = {}

        class _Resp:
            status_code = 200
            is_success = True

            def json(self):
                return {"data": [{"id": "local-model"}]}

        class _Client:
            def __init__(self, *args, **kwargs):
                pass

            async def __aenter__(self):
                return self

            async def __aexit__(self, *args):
                return False

            async def get(self, url, *args, headers=None, **kwargs):
                captured["url"] = url
                captured["headers"] = headers
                return _Resp()

            async def post(self, url, *args, json=None, headers=None, **kwargs):
                captured["posted"] = url
                return _Resp()

        monkeypatch.setattr("httpx.AsyncClient", _Client)

        response = self.client.post(
            "/api/providers/custom-endpoints/validate",
            json={
                "name": "Local",
                "base_url": "http://localhost:8000/v1",
                "model": "local-model",
                "api_key": "local-secret",
            },
        )

        body = response.json()
        assert body["ok"] is True and body["reachable"] is True
        assert body["models"] == ["local-model"]
        assert captured["url"] == "http://localhost:8000/v1/models"
        assert captured["headers"]["Authorization"] == "Bearer local-secret"


class TestDesktopCronTicker:
    """The dashboard backend fires cron jobs itself only when desktop-spawned."""

    def _client(self):
        try:
            from starlette.testclient import TestClient
        except ImportError:
            pytest.skip("fastapi/starlette not installed")
        from hermes_cli.web_server import app

        return TestClient(app)

    def test_ticker_runs_when_desktop(self, monkeypatch, _isolate_hermes_home):
        import cron.scheduler as sched

        called = threading.Event()
        monkeypatch.setattr(sched, "tick", lambda *a, **k: called.set())
        monkeypatch.setenv("HERMES_DESKTOP", "1")
        monkeypatch.setenv("HERMES_DASHBOARD_SESSION_TOKEN", "desktop-spawn-token")

        with self._client():
            assert called.wait(3.0), "expected cron tick under a Desktop-owned backend"


class TestServeIndexMissingIndex:
    """_serve_index must not raise per-request when index.html vanishes
    (partial build, wiped dist) after mount_spa saw an existing dist dir.
    It should return the same JSON 404 payload mount_spa emits for a
    fully-missing dist."""

    @staticmethod
    def _client_with_dist(tmp_path, monkeypatch, *, write_index: bool):
        from fastapi import FastAPI
        from starlette.testclient import TestClient
        import hermes_cli.web_server as ws

        dist = tmp_path / "web_dist"
        (dist / "assets").mkdir(parents=True)
        if write_index:
            (dist / "index.html").write_text(
                "<html><head></head><body>SPA</body></html>", encoding="utf-8"
            )
        monkeypatch.setattr(ws, "WEB_DIST", dist)
        monkeypatch.delenv("HERMES_SERVE_HEADLESS", raising=False)
        spa_app = FastAPI()
        _web_server_dashboard.mount_spa(spa_app)
        return TestClient(spa_app), dist

    def test_missing_index_inside_existing_dist_returns_json_404(
        self, tmp_path, monkeypatch
    ):
        client, _dist = self._client_with_dist(
            tmp_path, monkeypatch, write_index=False
        )
        for route in ("/", "/chat"):
            resp = client.get(route)
            assert resp.status_code == 404
            assert resp.json()["error"]

    def test_index_deleted_after_mount_returns_json_404(self, tmp_path, monkeypatch):
        client, dist = self._client_with_dist(tmp_path, monkeypatch, write_index=True)
        assert client.get("/chat").status_code == 200  # healthy first
        (dist / "index.html").unlink()
        resp = client.get("/chat")
        assert resp.status_code == 404
        assert resp.json()["error"]
        # And recovers once the index reappears (e.g. a rebuild finished).
        (dist / "index.html").write_text(
            "<html><head></head><body>SPA-rebuilt</body></html>", encoding="utf-8"
        )
        resp = client.get("/chat")
        assert resp.status_code == 200
        assert "SPA-rebuilt" in resp.text

    def test_index_uses_ssh_token_applied_after_spa_mount(
        self, tmp_path, monkeypatch
    ):
        import hermes_cli.web_server as ws

        monkeypatch.setattr(ws, "_SESSION_TOKEN", "before-mount")
        client, _dist = self._client_with_dist(
            tmp_path, monkeypatch, write_index=True
        )

        ws._apply_ssh_session_token("after-mount")
        resp = client.get("/chat")

        assert resp.status_code == 200
        assert 'window.__HERMES_SESSION_TOKEN__="after-mount"' in resp.text


class TestHeadlessServeTokenPage:
    """Headless `hermes serve` must serve the Desktop token handshake page
    at `/` when the dashboard auth gate is off (#94227).

    The Electron renderer boots by fetching `/` and extracting
    ``window.__HERMES_SESSION_TOKEN__`` for WebSocket auth. Headless serve
    used to 404 every path, so after an update replaced the backend (and
    the spawn-token env pin no longer matched the token the new backend
    generated) the renderer was stuck with a stale token, /api/ws rejected
    it, and the window white-screened (#95575).
    """

    @staticmethod
    def _headless_client(monkeypatch, *, gated: bool):
        from fastapi import FastAPI
        from starlette.testclient import TestClient
        import hermes_cli.web_server as ws

        monkeypatch.setenv("HERMES_SERVE_HEADLESS", "1")
        spa_app = FastAPI()
        spa_app.state.auth_required = gated
        _web_server_dashboard.mount_spa(spa_app)
        return TestClient(spa_app), ws

    def test_root_serves_token_page_when_not_gated(self, monkeypatch):
        import re

        client, ws = self._headless_client(monkeypatch, gated=False)
        resp = client.get("/")
        assert resp.status_code == 200
        assert resp.headers["content-type"].startswith("text/html")
        assert "no-store" in resp.headers.get("cache-control", "")
        # Must match the desktop's extraction regex exactly
        # (apps/desktop/electron/dashboard-token.ts).
        match = re.search(
            r'window\.__HERMES_SESSION_TOKEN__\s*=\s*("(?:\\.|[^"\\])*")',
            resp.text,
        )
        assert match, resp.text
        import json as _json

        assert _json.loads(match.group(1)) == ws._SESSION_TOKEN
        assert "window.__HERMES_AUTH_REQUIRED__=false" in resp.text

    def test_root_uses_ssh_token_applied_after_spa_mount(self, monkeypatch):
        import json
        import re

        import hermes_cli.web_server as ws

        monkeypatch.setattr(ws, "_SESSION_TOKEN", "before-mount")
        client, ws = self._headless_client(monkeypatch, gated=False)

        ws._apply_ssh_session_token("after-mount")
        resp = client.get("/")
        match = re.search(
            r'window\.__HERMES_SESSION_TOKEN__\s*=\s*("(?:\\.|[^"\\])*")',
            resp.text,
        )

        assert match, resp.text
        assert json.loads(match.group(1)) == "after-mount"

    def test_root_stays_404_json_when_auth_gated(self, monkeypatch):
        client, ws = self._headless_client(monkeypatch, gated=True)
        resp = client.get("/")
        assert resp.status_code == 404
        assert ws._SESSION_TOKEN not in resp.text

    def test_non_root_paths_stay_404_json(self, monkeypatch):
        client, ws = self._headless_client(monkeypatch, gated=False)
        for route in ("/chat", "/api/status-page", "/assets/index-abc.js"):
            resp = client.get(route)
            assert resp.status_code == 404
            assert ws._SESSION_TOKEN not in resp.text


class TestHashedAssetCacheHeaders:
    """Hashed /assets/* responses must be immutable-cacheable; index.html
    must stay no-store so it always references the current hashes
    (salvaged from PR #28543)."""

    _IMMUTABLE = "public, max-age=31536000, immutable"

    @staticmethod
    def _client(tmp_path, monkeypatch):
        from fastapi import FastAPI
        from starlette.testclient import TestClient
        import hermes_cli.web_server as ws

        dist = tmp_path / "web_dist"
        (dist / "assets").mkdir(parents=True)
        (dist / "index.html").write_text(
            "<html><head></head><body>SPA</body></html>", encoding="utf-8"
        )
        (dist / "assets" / "index-abc123.js").write_text(
            "console.log('bundle');", encoding="utf-8"
        )
        (dist / "assets" / "index-abc123.css").write_text(
            "body{background:url(/ds-assets/bg.png);"
            "font-family:url(/fonts-terminal/x.woff2)}",
            encoding="utf-8",
        )
        monkeypatch.setattr(ws, "WEB_DIST", dist)
        monkeypatch.delenv("HERMES_SERVE_HEADLESS", raising=False)
        spa_app = FastAPI()
        _web_server_dashboard.mount_spa(spa_app)
        return TestClient(spa_app)

    def test_hashed_js_asset_is_immutable(self, tmp_path, monkeypatch):
        client = self._client(tmp_path, monkeypatch)
        resp = client.get("/assets/index-abc123.js")
        assert resp.status_code == 200
        assert resp.headers["cache-control"] == self._IMMUTABLE

    def test_serve_css_is_immutable_and_keeps_prefix_rewrites(
        self, tmp_path, monkeypatch
    ):
        client = self._client(tmp_path, monkeypatch)
        resp = client.get("/assets/index-abc123.css")
        assert resp.status_code == 200
        assert resp.headers["cache-control"] == self._IMMUTABLE

        # The proxy-prefix rewrite path (main's ds-assets/fonts-terminal
        # handling) must survive the header change.
        prefixed = client.get(
            "/assets/index-abc123.css",
            headers={"X-Forwarded-Prefix": "/hermes"},
        )
        assert prefixed.status_code == 200
        assert prefixed.headers["cache-control"] == self._IMMUTABLE
        assert "url(/hermes/ds-assets/bg.png)" in prefixed.text
        assert "url(/hermes/fonts-terminal/x.woff2)" in prefixed.text

    def test_index_html_stays_no_store(self, tmp_path, monkeypatch):
        client = self._client(tmp_path, monkeypatch)
        for route in ("/", "/chat"):
            resp = client.get(route)
            assert resp.status_code == 200
            cache_control = resp.headers["cache-control"]
            assert "no-store" in cache_control
            assert "immutable" not in cache_control

    def test_missing_asset_is_not_marked_immutable(self, tmp_path, monkeypatch):
        """A 404 must never be cached for a year — a later rebuild can
        legitimately create the file."""
        client = self._client(tmp_path, monkeypatch)
        resp = client.get("/assets/nope-000000.js")
        assert resp.status_code == 404
        assert "immutable" not in resp.headers.get("cache-control", "")


class TestDashboardComponentHealth:
    """Component-health rollup: error middleware, /api/status components, self-test."""

    @pytest.fixture(autouse=True)
    def _setup(self, monkeypatch, _isolate_hermes_home):
        try:
            from starlette.testclient import TestClient
        except ImportError:
            pytest.skip("fastapi/starlette not installed")

        import hermes_state
        from hermes_constants import get_hermes_home
        import hermes_cli.web_server as ws

        monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db")
        # Fresh state holder per test so counters don't leak across tests.
        monkeypatch.setattr(ws, "DASHBOARD_HEALTH", ws.DashboardHealth())
        self.ws = ws
        self.client = TestClient(ws.app, raise_server_exceptions=False)
        self.client.headers[ws._SESSION_HEADER_NAME] = ws._SESSION_TOKEN

    # -- middleware -------------------------------------------------------


    # -- /api/status components ------------------------------------------


    def test_public_component_payload_carries_no_secret_bearing_fields(self):
        """PUBLIC_API_PATHS contract: counts/enums only — no paths/messages."""
        self.ws.DASHBOARD_HEALTH.record_error("RuntimeError", "/api/secret-route?token=abc")
        resp = self.client.get("/api/status")
        payload = json.dumps(resp.json()["components"])
        assert "secret-route" not in payload
        assert "token=abc" not in payload
        assert "last_error_path" not in payload
        assert "last_error_type" not in payload
        assert "kaboom" not in payload

    # -- self-test ---------------------------------------------------------

    def test_selftest_records_failure_on_500(self, monkeypatch):
        httpx = pytest.importorskip("httpx")

        class _FakeResponse:
            status_code = 500

        class _FakeClient:
            def __init__(self, *args, **kwargs):
                pass

            async def __aenter__(self):
                return self

            async def __aexit__(self, *exc):
                return False

            async def get(self, *args, **kwargs):
                return _FakeResponse()

        monkeypatch.setattr(httpx, "AsyncClient", _FakeClient)
        asyncio.run(self.ws._dashboard_selftest_once())
        assert self.ws.DASHBOARD_HEALTH.selftest_status == "failing"
        assert self.ws.DASHBOARD_HEALTH.selftest_http_status == 500
        assert self.ws.DASHBOARD_HEALTH.snapshot()["status"] == "degraded"


class TestSessionPatchUnread:
    """PATCH /api/sessions/{id} with {"unread": bool} marks the session
    read/unread, and GET /api/sessions surfaces the derived flag."""

    @pytest.fixture(autouse=True)
    def _setup_test_client(self, monkeypatch, _isolate_hermes_home):
        try:
            from starlette.testclient import TestClient
        except ImportError:
            pytest.skip("fastapi/starlette not installed")

        import hermes_state
        from hermes_constants import get_hermes_home
        from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN

        monkeypatch.setattr(
            hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db"
        )

        self.client = TestClient(app)
        self.auth_client = TestClient(app)
        self.auth_client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN

        from hermes_state import SessionDB

        db = SessionDB()
        try:
            db.create_session(session_id="s1", source="cli")
            db.append_message(session_id="s1", role="user", content="hi")
            db.set_session_read("s1")  # start read, like a conversation you opened
        finally:
            db.close()

    def test_patch_unread_true_marks_row_unread(self):
        resp = self.auth_client.patch("/api/sessions/s1", json={"unread": True})
        assert resp.status_code == 200
        assert resp.json()["unread"] is True

        rows = self.auth_client.get("/api/sessions?limit=100").json()["sessions"]
        assert next(s for s in rows if s["id"] == "s1")["unread"] is True

    def test_patch_unread_false_marks_row_read(self):
        self.auth_client.patch("/api/sessions/s1", json={"unread": True})
        resp = self.auth_client.patch("/api/sessions/s1", json={"unread": False})
        assert resp.status_code == 200
        assert resp.json()["unread"] is False

        rows = self.auth_client.get("/api/sessions?limit=100").json()["sessions"]
        assert next(s for s in rows if s["id"] == "s1")["unread"] is False


    def test_patch_unread_rejects_non_bool(self):
        # NB: pydantic v2 coerces "yes"/"no"/"1"/"0"/"on"/"off" to bool, so use
        # a string outside the accepted set to prove validation rejects it.
        resp = self.auth_client.patch("/api/sessions/s1", json={"unread": "maybe"})
        assert resp.status_code == 422  # pydantic validation

    def test_patch_hidden_updates_persisted_session_without_live_runtime(self):
        resp = self.auth_client.patch("/api/sessions/s1", json={"hidden": True})
        assert resp.status_code == 200
        assert resp.json()["hidden"] is True

        rows = self.auth_client.get("/api/sessions?limit=100").json()["sessions"]
        assert all(s["id"] != "s1" for s in rows)

        restored = self.auth_client.patch(
            "/api/sessions/s1", json={"hidden": False}
        )
        assert restored.status_code == 200
        rows = self.auth_client.get("/api/sessions?limit=100").json()["sessions"]
        assert bool(next(s for s in rows if s["id"] == "s1")["hidden"]) is False


def test_mount_spa_dynamic_web_dist_recheck(tmp_path, monkeypatch):
    from fastapi import FastAPI
    from fastapi.testclient import TestClient
    from hermes_cli import web_server

    app = FastAPI()
    dist = tmp_path / "web_dist"
    monkeypatch.setattr(web_server, "WEB_DIST", dist)

    _web_server_dashboard.mount_spa(app)
    client = TestClient(app)

    # 1. missing build -> 404
    res1 = client.get("/")
    assert res1.status_code == 404
    assert res1.json()["error"]

    # 2. build created dynamically -> 200
    dist.mkdir(parents=True, exist_ok=True)
    (dist / "index.html").write_text("<html><body>Test</body></html>")
    res2 = client.get("/")
    assert res2.status_code == 200
    assert "Test" in res2.text


class TestSubmittedCustomEndpointSurvivesAssignment:
    """#115661 follow-up: a bare-``custom`` main-slot pick carries the submitted endpoint as the
    current one (see ``_validated_main_model_selection``). Once the switch's credential step
    re-resolves that target, an env endpoint (``CUSTOM_BASE_URL`` / ``OPENROUTER_BASE_URL``) could
    replace what the user typed and had persisted."""

    def test_submitted_custom_endpoint_wins_over_an_env_endpoint(self, monkeypatch):
        from hermes_cli.web_server_config import _apply_main_model_assignment, _validated_main_model_selection

        monkeypatch.setenv("CUSTOM_BASE_URL", "http://127.0.0.1:9999/v1")
        monkeypatch.setattr(
            "hermes_cli.models_validate.validate_requested_model",
            lambda *a, **k: {"accepted": True, "persist": True, "recognized": True, "message": None})
        monkeypatch.setattr("hermes_cli.model_switch.get_model_info", lambda *a, **k: None)
        monkeypatch.setattr("hermes_cli.model_switch.get_model_capabilities", lambda *a, **k: None)

        cfg = {"model": {"provider": "openrouter", "default": "m"}}
        result = _validated_main_model_selection(
            cfg, "custom", "qwen3:8b", "https://api.anthropic.com", "submitted-key")

        assert result.base_url == "https://api.anthropic.com"
        # The wire protocol follows the endpoint that gets persisted, not the displaced env host.
        assert result.api_mode == "anthropic_messages"
        applied = _apply_main_model_assignment(cfg.get("model", {}), result, "submitted-key")
        assert applied["base_url"] == "https://api.anthropic.com"
        assert applied["api_mode"] == "anthropic_messages"
        assert applied["api_key"] == "submitted-key"