"""Session lineage and continuity contracts for Hermes chat."""

from __future__ import annotations

import asyncio
import importlib.util
import json
import sqlite3
import textwrap
from pathlib import Path

import pytest

from testing.tests.test_hermes_chat_support import (
    HERMES,
    _documents,
)


def _containers(path: Path) -> list[dict]:
    """Return every container and init container in one workload document."""
    spec = _documents(path)[0]["spec"]["template"]["spec"]
    return [*spec.get("initContainers", []), *spec.get("containers", [])]


def _env(container: dict) -> dict[str, str]:
    return {
        entry["name"]: entry.get("value", "")
        for entry in container.get("env", [])
        if isinstance(entry, dict) and "name" in entry
    }


def test_chat_continuity_polls_the_route_its_own_backend_serves():
    """The injected fallback must speak the tenant WebUI session contract."""
    fallback = (HERMES / "router" / "session_continuity.go").read_text(encoding="utf-8")
    snapshot = (HERMES / "router" / "session_snapshot.go").read_text(encoding="utf-8")
    script = fallback.split("const sessionContinuityJS = `", 1)[1].rsplit("`", 1)[0]

    assert "'/api/session?session_id=' + encodeURIComponent(sessionId)" in script
    assert "&messages=1&msg_limit=24" in script
    assert "hermes_fallback=1" in script
    # A 409 is the only answer that means "not in this account's active scope".
    assert "session_profile_mismatch" in script
    assert 'sessionFallbackPath = "/api/session"' in fallback
    assert "request.URL.Path != sessionFallbackPath" in snapshot

    # /api/sessions/<id>[/messages] is the Hermes agent dashboard contract. The
    # chat tenants never route it, so polling it was a permanent 404 that
    # reported a false ownership failure after every full page load.
    assert "/api/sessions/" not in script


def test_dashboard_session_route_never_backs_the_chat_tenants():
    """Only the agent dashboard gains /api/sessions/{id}/messages."""
    activity_patch = "patch_web_session_activity.py"
    marker = '@app.get("/api/sessions/{session_id}/messages")'
    assert marker in (HERMES / "scripts" / activity_patch).read_text(encoding="utf-8")

    agent = _containers(HERMES / "agent-deployment.yaml")
    assert any(
        activity_patch in " ".join(map(str, container.get("command", []) + container.get("args", [])))
        for container in agent
    )

    tenants = _containers(HERMES / "chat-statefulset.yaml")
    assert not any(
        activity_patch in " ".join(map(str, container.get("command", []) + container.get("args", [])))
        for container in tenants
    )

    # The router proxies browser traffic to this WebUI container, and asserts
    # the tenant identity through the header the WebUI is told to trust.
    webui = next(container for container in tenants if container["name"] == "webui")
    router = (HERMES / "router" / "main.go").read_text(encoding="utf-8")
    header = _env(webui)["HERMES_WEBUI_TRUSTED_AUTH_HEADER"]
    assert f'trustedTenantHeader = "{header}"' in router


def test_codex_native_health_overrides_historical_router_errors(
    tmp_path: Path, monkeypatch
):
    """Fresh first-party health is authoritative over old Switchyard probes."""
    plugin_path = HERMES / "plugins" / "auto-router" / "provider_status.py"
    spec = importlib.util.spec_from_file_location("hermes_provider_status", plugin_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    health_path = tmp_path / "codex.json"
    health_path.write_text(
        json.dumps(
            {
                "state": "available",
                "authenticated": True,
                "transport": "codex-chatgpt-subscription",
            }
        )
    )
    monkeypatch.setattr(module, "CODEX_HEALTH_PATH", health_path)
    monkeypatch.setattr(module, "CLAUDE_HEALTH_PATH", tmp_path / "missing.json")
    monkeypatch.setattr(
        module,
        "_get_json",
        lambda url: {"status": "ok"}
        if url.endswith("/health")
        else {
            "models": {
                "route/codex/terra/medium": {
                    "calls": 1,
                    "errors": 99,
                    "total_tokens": 12,
                }
            }
        },
    )
    monkeypatch.setattr(module, "_codex_account", lambda: {})
    monkeypatch.setattr(module, "_claude_account", lambda: {})

    codex = module.provider_status_payload()["providers"]["codex"]

    assert codex["errors"] == 99
    assert codex["state"] == "available"
    assert codex["native_health"]["transport"] == "codex-chatgpt-subscription"


def test_api_session_patch_accepts_parent_lineage(tmp_path: Path):
    """API-created workers must persist the originating Hermes session."""
    module_path = HERMES / "scripts" / "patch_api_server_sessions.py"
    spec = importlib.util.spec_from_file_location("patch_api_sessions", module_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    source = tmp_path / "api_server.py"
    destination = tmp_path / "patched.py"
    source.write_text(
        "prefix\n"
        + module.BEFORE
        + "middle\n"
        + module.RUNS_BEFORE
        + "run body\n"
        + module.RUN_CLOSE_BEFORE
        + module.RESPONSES_SESSION_BEFORE
        + module.EVENT_CALLBACK_SIGNATURE_BEFORE
        + "callback docstring and push helper\n"
        + module.EVENT_CALLBACK_BODY_BEFORE
        + "tool start body\n"
        + module.EVENT_CALLBACK_END_BEFORE
        + module.EVENT_CALLBACK_CALL_BEFORE
        + module.RUN_SWEEP_BEFORE
        + "suffix\n",
        encoding="utf-8",
    )

    module.patch(source, destination)
    patched = destination.read_text(encoding="utf-8")

    assert "X-Hermes-Parent-Session-Id" in patched
    assert "parent_session_id=parent_session_id" in patched
    assert "Parent session not found" in patched
    assert "HERMES_API_DEFAULT_PARENT_MATCH_PREFIXES" in patched
    assert "user_message.startswith(default_prefixes)" in patched
    assert "session_parent_conflict" in patched
    assert "X-Hermes-Conversation-Platform" in patched
    assert "X-Hermes-Conversation-Title" in patched
    assert 'conversation_platform != "telegram"' in patched
    assert "db.record_gateway_session_peer(" in patched
    assert 'display_name="Telegram"' in patched
    assert "db.reopen_session(session_id)" in patched
    assert 'db.end_session(session_id, f"api_run_{terminal_status}")' in patched
    assert "def _record_run_activity(" in patched
    assert '"_thinking": "Hermes is reasoning"' in patched
    assert '"run.started": "Worker started"' in patched
    assert '"run.completed": "Worker completed"' in patched
    assert '"reasoning.available": "Hermes finished a reasoning step"' in patched
    assert '"subagent.progress": "Nested worker progress"' in patched
    assert "redact_sensitive_text" in patched
    assert 'getattr(os, "O_NOFOLLOW", 0)' in patched
    assert "os.fchmod(fd, 0o600)" in patched
    assert "session_id=session_id" in patched
    assert 'self._record_run_activity(session_id, "run.started")' in patched
    assert "detail = tool_name if event_type in {" in patched
    assert 'if event_type == "subagent.tool"' in patched
    assert "_RUN_ACTIVITY_HEARTBEAT_SECONDS = 15.0" in patched
    assert "heartbeats.get(session_id, 0.0)" in patched
    assert '"subagent.thinking",' in patched
    assert "Stream retention and run lifetime are separate" in patched
    assert 'terminal_status in {"completed", "failed", "cancelled"}' in patched
    assert patched.index("terminal_status = self._run_statuses") < patched.index(
        "self._active_run_tasks.pop(run_id, None)"
    )


def test_telegram_api_session_migration_is_bounded_and_idempotent(tmp_path: Path):
    """Only named Telegram conversations receive presentation metadata."""
    module_path = HERMES / "scripts" / "migrate_telegram_api_sessions.py"
    spec = importlib.util.spec_from_file_location(
        "migrate_telegram_sessions", module_path
    )
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    state = tmp_path / "state.db"
    responses = tmp_path / "response_store.db"
    with sqlite3.connect(state) as connection:
        connection.execute(
            """CREATE TABLE sessions (
                id TEXT PRIMARY KEY, source TEXT, session_key TEXT, chat_type TEXT,
                display_name TEXT, origin_json TEXT, title TEXT
            )"""
        )
        connection.executemany(
            "INSERT INTO sessions (id, source) VALUES (?, ?)",
            (("telegram-session", "api_server"), ("other-session", "api_server")),
        )
    with sqlite3.connect(responses) as connection:
        connection.execute(
            "CREATE TABLE conversations (name TEXT PRIMARY KEY, response_id TEXT NOT NULL)"
        )
        connection.execute(
            "CREATE TABLE responses (response_id TEXT PRIMARY KEY, data TEXT NOT NULL, accessed_at REAL NOT NULL)"
        )
        connection.executemany(
            "INSERT INTO responses VALUES (?, ?, 0)",
            (
                ("telegram-response", json.dumps({"session_id": "telegram-session"})),
                ("other-response", json.dumps({"session_id": "other-session"})),
            ),
        )
        connection.executemany(
            "INSERT INTO conversations VALUES (?, ?)",
            (("telegram", "telegram-response"), ("unrelated", "other-response")),
        )

    assert module.migrate(state, responses) == 1
    assert module.migrate(state, responses) == 0
    with sqlite3.connect(state) as connection:
        telegram = connection.execute(
            "SELECT session_key, chat_type, display_name, origin_json, title "
            "FROM sessions WHERE id = 'telegram-session'"
        ).fetchone()
        other = connection.execute(
            "SELECT session_key, title FROM sessions WHERE id = 'other-session'"
        ).fetchone()
    assert telegram[:3] == ("telegram", "private", "Telegram")
    assert json.loads(telegram[3]) == {
        "platform": "telegram",
        "session_key": "telegram",
    }
    assert telegram[4] == "Telegram · General"
    assert other == (None, None)


def test_web_session_activity_patch_projects_bounded_events(tmp_path: Path):
    """The DOM transcript includes events without polluting agent history."""
    module_path = HERMES / "scripts" / "patch_web_session_activity.py"
    spec = importlib.util.spec_from_file_location("patch_web_activity", module_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    source = tmp_path / "web_server.py"
    destination = tmp_path / "patched.py"
    source.write_text(
        "prefix\n"
        + module.LATEST_ROWS_BEFORE
        + module.LATEST_SELECTION_BEFORE
        + module.HELPER_MARKER
        + "    db = object()\n"
        + module.MESSAGES_BEFORE
        + module.SPA_INDEX_BEFORE
        + "suffix\n",
        encoding="utf-8",
    )

    module.patch(source, destination)
    patched = destination.read_text(encoding="utf-8")

    assert "def _run_activity_messages(" in patched
    assert 'root = get_hermes_home() / "run-activity"' in patched
    assert "path.stat().st_size > 600_000" in patched
    assert "entries[-1_000:]" in patched
    assert "*_run_activity_messages(sid)" in patched
    assert "messages.sort(" in patched
    assert "limit: Optional[int] = None" in patched
    assert "total_messages = len(messages)" in patched
    assert "min(int(limit), 10_000)" in patched
    assert '"total_messages": total_messages' in patched
    assert "SELECT id, parent_session_id, started_at, ended_at" in patched
    assert "newest still-open member of the lineage" in patched
    assert "immediate objective parent" in patched
    assert "mixes unrelated workstreams" in patched
    assert "orphaned children" in patched
    assert 'item[0].get("ended_at") is None' in patched


def test_activity_projection_preserves_user_text_and_labels_tool_only_turns():
    """A saved tool invocation cannot become a blank resumed-chat card."""
    module_path = HERMES / "scripts" / "patch_web_session_activity.py"
    spec = importlib.util.spec_from_file_location("patch_web_activity_projection", module_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)

    helper = module.HELPER_REPLACEMENT.split('@app.get("/api/sessions/', 1)[0]
    namespace: dict[str, object] = {}
    exec("from typing import Any, Dict, List\n" + helper, namespace)
    present = namespace["_present_activity_message"]

    first_user = {"role": "user", "content": "Please inspect the saved work."}
    second_user = {"role": "user", "content": "Continue from the recorded findings."}
    assert present(first_user) == first_user
    assert present(second_user) == second_user

    tool_only = {
        "role": "assistant",
        "content": None,
        # This is the parsed API shape for the serialized database column.
        "tool_calls": [{"function": {"name": "terminal"}}],
    }
    projected = present(tool_only)
    assert projected["content"] == "Calling terminal."
    assert projected["observed"] is True
    assert tool_only["content"] is None

    class FakeDB:
        def get_messages(self, session_id):
            assert session_id == "resume-fixture"
            return [first_user, tool_only, second_user]

    response_source = "async def project_messages(db, sid, limit=None):\n" + textwrap.indent(
        textwrap.dedent(module.MESSAGES_AFTER), "    "
    )
    exec(response_source, namespace)
    namespace["_run_activity_messages"] = lambda _session_id: []
    payload = asyncio.run(namespace["project_messages"](FakeDB(), "resume-fixture"))
    assert [item["content"] for item in payload["messages"]] == [
        "Please inspect the saved work.",
        "Calling terminal.",
        "Continue from the recorded findings.",
    ]
    assert payload["total_messages"] == 3

def test_web_session_lineage_returns_to_resumed_parent():
    """An ended reviewer must not strand the live view away from its parent."""
    module_path = HERMES / "scripts" / "patch_web_session_activity.py"
    spec = importlib.util.spec_from_file_location("patch_web_lineage", module_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)

    namespace: dict[str, object] = {}
    exec(
        "def select_active(sid, db, rows):\n" + module.LATEST_SELECTION_AFTER,
        namespace,
    )

    class FakeDB:
        def get_session(self, session_id):
            return next((row for row in rows if row["id"] == session_id), None)

    rows = [
        {
            "id": "umbrella",
            "parent_session_id": None,
            "started_at": 0.0,
            "ended_at": None,
        },
        {
            "id": "root",
            "parent_session_id": "umbrella",
            "started_at": 1.0,
            "ended_at": None,
        },
        {
            "id": "review-1",
            "parent_session_id": "root",
            "started_at": 2.0,
            "ended_at": 3.0,
        },
        {
            "id": "review-2",
            "parent_session_id": "root",
            "started_at": 4.0,
            "ended_at": None,
        },
    ]
    select_active = namespace["select_active"]

    assert select_active("root", FakeDB(), rows) == (
        "review-2",
        ["root", "review-2"],
    )
    assert select_active("review-1", FakeDB(), rows) == (
        "review-2",
        ["root", "review-2"],
    )
    rows[2]["ended_at"] = 5.0
    rows[3]["ended_at"] = 5.0
    assert select_active("root", FakeDB(), rows) == ("root", ["root"])

    rows[1]["ended_at"] = 9.0
    rows.append(
        {
            "id": "orphaned-review",
            "parent_session_id": "root",
            "started_at": 6.0,
            "ended_at": None,
        }
    )
    rows.append(
        {
            "id": "accepted-review",
            "parent_session_id": "root",
            "started_at": 7.0,
            "ended_at": 8.0,
        }
    )
    assert select_active("review-1", FakeDB(), rows) == (
        "accepted-review",
        ["root", "accepted-review"],
    )


def test_api_activity_patch_coalesces_streaming_heartbeats(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
):
    """Token callbacks stay bounded while every tool transition is retained."""
    module_path = HERMES / "scripts" / "patch_api_server_sessions.py"
    spec = importlib.util.spec_from_file_location("patch_api_activity", module_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    activity_body = module.EVENT_CALLBACK_SIGNATURE_AFTER.split(
        "    def _make_run_event_callback(", 1
    )[0]
    namespace: dict[str, object] = {}
    exec(
        "import hashlib, json, logging, os, time\n"
        "from pathlib import Path\n"
        "logger = logging.getLogger(__name__)\n"
        "def redact_sensitive_text(value): return value\n"
        "class ActivityRecorder:\n" + activity_body,
        namespace,
    )
    recorder = namespace["ActivityRecorder"]()
    monkeypatch.setenv("HERMES_HOME", str(tmp_path))

    recorder._record_run_activity("session", "_thinking")
    recorder._record_run_activity("session", "reasoning.available")
    recorder._record_run_activity("session", "subagent.thinking")
    recorder._record_run_activity("session", "tool.started", tool_name="terminal")
    recorder._record_run_activity("session", "tool.completed", tool_name="terminal")

    journals = list((tmp_path / "run-activity").glob("*.jsonl"))
    assert len(journals) == 1
    entries = [json.loads(line) for line in journals[0].read_text().splitlines()]
    assert [entry["activity_event"] for entry in entries] == [
        "_thinking",
        "tool.started",
        "tool.completed",
    ]


def test_legacy_api_sessions_are_nested_idempotently(tmp_path: Path):
    """Known standalone API workers move under Cassandra without data loss."""
    module_path = HERMES / "scripts" / "migrate_api_session_lineage.py"
    spec = importlib.util.spec_from_file_location("migrate_api_sessions", module_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    database = tmp_path / "state.db"
    with sqlite3.connect(database) as connection:
        connection.execute(
            "CREATE TABLE sessions (id TEXT PRIMARY KEY, source TEXT, "
            "parent_session_id TEXT, title TEXT, transcript TEXT, archived INTEGER DEFAULT 0)"
        )
        connection.execute(
            "INSERT INTO sessions (id, source, parent_session_id, title, transcript) "
            "VALUES (?, 'tui', NULL, 'Cassandra', 'parent-data')",
            (module.LEGACY_CASSANDRA_PARENT,),
        )
        worker_id = next(iter(module.LEGACY_CASSANDRA_WORKERS))
        connection.execute(
            "INSERT INTO sessions (id, source, parent_session_id, title, transcript) "
            "VALUES (?, 'api_server', NULL, 'old', 'keep-me')",
            (worker_id,),
        )
        connection.executemany(
            "INSERT INTO sessions (id, source, parent_session_id, title, transcript) "
            "VALUES (?, 'api_server', NULL, 'old smoke', 'keep-smoke')",
            ((orphan_id,) for orphan_id in module.LEGACY_ORPHANED_SMOKE_SESSIONS),
        )

    assert module.migrate(database) == 1 + len(module.LEGACY_ORPHANED_SMOKE_SESSIONS)
    assert module.migrate(database) == 0
    with sqlite3.connect(database) as connection:
        row = connection.execute(
            "SELECT parent_session_id, title, transcript FROM sessions WHERE id = ?",
            (worker_id,),
        ).fetchone()
    assert row == (
        module.LEGACY_CASSANDRA_PARENT,
        module.LEGACY_CASSANDRA_WORKERS[worker_id],
        "keep-me",
    )
    with sqlite3.connect(database) as connection:
        orphans = connection.execute(
            "SELECT id, archived, title, transcript FROM sessions "
            "WHERE id IN ({}) ORDER BY id".format(
                ",".join("?" for _ in module.LEGACY_ORPHANED_SMOKE_SESSIONS)
            ),
            tuple(module.LEGACY_ORPHANED_SMOKE_SESSIONS),
        ).fetchall()
    assert orphans == sorted(
        (
            orphan_id,
            1,
            module.LEGACY_ORPHANED_SMOKE_SESSIONS[orphan_id],
            "keep-smoke",
        )
        for orphan_id in module.LEGACY_ORPHANED_SMOKE_SESSIONS
    )


def test_automated_triage_sessions_are_grouped_without_touching_interactive_runs(
    tmp_path: Path,
):
    """Only the stable Ariadne contract moves below the triage parent."""
    module_path = HERMES / "scripts" / "migrate_api_session_lineage.py"
    spec = importlib.util.spec_from_file_location(
        "migrate_triage_sessions", module_path
    )
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    database = tmp_path / "state.db"
    with sqlite3.connect(database) as connection:
        connection.execute(
            "CREATE TABLE sessions (id TEXT PRIMARY KEY, source TEXT, "
            "parent_session_id TEXT, title TEXT, started_at REAL, archived INTEGER DEFAULT 0)"
        )
        connection.execute(
            "CREATE TABLE messages (session_id TEXT, role TEXT, content TEXT)"
        )
        connection.executemany(
            "INSERT INTO sessions (id, source, started_at) VALUES (?, 'api_server', ?)",
            (("triage-run", 1.0), ("jenkins-run", 1.5), ("interactive-run", 2.0)),
        )
        connection.executemany(
            "INSERT INTO messages (session_id, role, content) VALUES (?, 'user', ?)",
            (
                (
                    "triage-run",
                    module.TRIAGE_MESSAGE_PREFIXES[0]
                    + " Fix for incident sonar/bstein_home/python:S2208/finding-key.",
                ),
                (
                    "jenkins-run",
                    module.TRIAGE_MESSAGE_PREFIXES[1]
                    + "\nAnalyze incident soteria/291 for the Jenkins job soteria.",
                ),
                ("interactive-run", "Please explain this alert to me."),
            ),
        )

    assert module.migrate(database, group_triage=True) == 2
    assert module.migrate(database, group_triage=True) == 0
    with sqlite3.connect(database) as connection:
        parent = connection.execute(
            "SELECT title FROM sessions WHERE id = ?", (module.TRIAGE_PARENT,)
        ).fetchone()
        triage = connection.execute(
            "SELECT parent_session_id, title FROM sessions WHERE id = 'triage-run'"
        ).fetchone()
        interactive = connection.execute(
            "SELECT parent_session_id FROM sessions WHERE id = 'interactive-run'"
        ).fetchone()
        jenkins = connection.execute(
            "SELECT parent_session_id, title FROM sessions WHERE id = 'jenkins-run'"
        ).fetchone()
    assert parent == (module.TRIAGE_PARENT_TITLE,)
    assert triage == (
        module.TRIAGE_PARENT,
        "Sonar · bstein_home · python:S2208 · iage-run",
    )
    assert interactive == (None,)
    assert jenkins == (
        module.TRIAGE_PARENT,
        "Sonar · soteria/291 · kins-run",
    )