"""Tests for the Runtime and session view (Phase 1, #641).

Covers clean and stale session rendering, contamination marker surfacing,
worktree binding display, sanctioned recovery links (no pkill), nav/live
status, and the JSON API export.
"""

from __future__ import annotations

import sys
import unittest
from pathlib import Path
from unittest import mock

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from tests.webui_testclient import TestClient

from webui.app import create_app
from webui.inventory import (
    AUTHORITY_CONTROL_PLANE_DB,
    AUTHORITY_FILESYSTEM,
    InventorySection,
    InventorySnapshot,
    STATUS_OK,
    STATUS_UNAVAILABLE,
)
from webui.nav import NAV_GROUPS, STUB_PAGES, iter_nav_items
from webui.runtime_health import FileHash, RuntimeSnapshot
from webui.session_loader import (
    ContaminationMarker,
    SessionRow,
    SessionViewSnapshot,
    _build_session_rows,
    load_session_view_snapshot,
    snapshot_to_dict,
)
from webui.session_views import render_sessions_page


def _runtime(
    *,
    stale: str | None = None,
    profile: str = "prgs-author",
    role: str = "author",
) -> RuntimeSnapshot:
    return RuntimeSnapshot(
        project_id="gitea-tools",
        repo_root="/tmp/repo",
        remote="prgs",
        host="gitea.prgs.cc",
        profile_name=profile,
        role_kind=role,
        config_model="v2-contexts",
        profile_mode="dynamic-profile",
        profile_source="config file profile",
        authenticated_username="jcwalker3",
        identity_error=None,
        repo_sha="a" * 40,
        remote_master_sha="a" * 40,
        commits_behind_master=0,
        stale_runtime_warning=stale,
        shell_health={"shell_use_allowed": True, "consecutive_spawn_failures": 0},
        workflow_hashes=(
            FileHash(label="SKILL.md", path="skills/llm-project-workflow/SKILL.md", sha256="abc"),
        ),
        schema_hashes=(),
        restart_guidance="docs/mcp-namespace-eof-recovery.md",
        fetch_error=None,
    )


def _inventory(
    *,
    sessions: tuple[dict, ...] = (),
    leases: tuple[dict, ...] = (),
    locks: tuple[dict, ...] = (),
    worktrees: tuple[dict, ...] = (),
    namespaces: tuple[dict, ...] = (),
) -> InventorySnapshot:
    sections = (
        InventorySection(
            name="sessions",
            authority=AUTHORITY_CONTROL_PLANE_DB,
            status=STATUS_OK,
            items=sessions,
        ),
        InventorySection(
            name="leases",
            authority=AUTHORITY_CONTROL_PLANE_DB,
            status=STATUS_OK,
            items=leases,
        ),
        InventorySection(
            name="locks",
            authority=AUTHORITY_FILESYSTEM,
            status=STATUS_OK,
            items=locks,
        ),
        InventorySection(
            name="worktrees",
            authority=AUTHORITY_FILESYSTEM,
            status=STATUS_OK,
            items=worktrees,
        ),
        InventorySection(
            name="namespaces",
            authority=AUTHORITY_FILESYSTEM,
            status=STATUS_OK,
            items=namespaces
            or (
                {
                    "profile_name": "prgs-author",
                    "role": "author",
                    "mcp_namespace": "gitea-author",
                    "capability_summary": {
                        "can_author": True,
                        "can_review": False,
                        "can_merge": False,
                    },
                    "active": True,
                },
            ),
            reason="only the profile serving this web process is observable",
        ),
    )
    index = {section.name: section for section in sections}
    return InventorySnapshot(
        generated_at="2026-07-25T00:00:00+00:00",
        sections=sections,
        collisions=(),
        correlations=(),
        scan_ms=1.0,
        _section_index=index,
    )


def _clean_session() -> dict:
    return {
        "session_id": "prgs-author-111-clean",
        "role": "author",
        "profile": "prgs-author",
        "namespace": "gitea-author",
        "pid": 1111,
        "pid_alive": True,
        "status": "active",
        "started_at": "2026-07-25T00:00:00Z",
        "last_heartbeat_at": "2026-07-25T01:00:00Z",
    }


def _stale_session() -> dict:
    return {
        "session_id": "prgs-author-222-stale",
        "role": "author",
        "profile": "prgs-author",
        "namespace": "gitea-author",
        "pid": 2222,
        "pid_alive": False,
        "status": "active",
        "started_at": "2026-07-24T00:00:00Z",
        "last_heartbeat_at": "2026-07-24T01:00:00Z",
    }


class TestBuildSessionRows(unittest.TestCase):
    def test_clean_session_has_no_stale_or_contamination_flags(self):
        inventory = _inventory(
            sessions=(_clean_session(),),
            leases=(
                {
                    "lease_id": "lease-clean",
                    "session_id": "prgs-author-111-clean",
                    "status": "active",
                    "expired": False,
                    "work_kind": "issue",
                    "work_number": 641,
                },
            ),
            locks=(
                {
                    "issue_number": 641,
                    "branch_name": "feat/issue-641-runtime-session-view",
                    "worktree_path": "~/Development/Gitea-Tools/branches/feat-issue-641",
                    "live": True,
                },
            ),
        )
        rows = _build_session_rows(inventory, contamination=())
        self.assertEqual(len(rows), 1)
        row = rows[0]
        self.assertEqual(row.session_id, "prgs-author-111-clean")
        self.assertEqual(row.role, "author")
        self.assertEqual(row.namespace, "gitea-author")
        self.assertEqual(row.pid_alive, True)
        self.assertEqual(row.lease_ids, ("lease-clean",))
        self.assertEqual(row.work_refs, ("issue#641",))
        self.assertTrue(row.worktree_paths)
        self.assertEqual(row.stale_flags, ())
        self.assertEqual(row.contamination_flags, ())

    def test_stale_session_flags_dead_pid(self):
        inventory = _inventory(sessions=(_stale_session(),))
        rows = _build_session_rows(inventory, contamination=())
        self.assertEqual(rows[0].stale_flags, ("pid-dead",))

    def test_contamination_marker_binds_to_session(self):
        inventory = _inventory(sessions=(_clean_session(),))
        marker = ContaminationMarker(
            kind="runtime_recovery_contamination",
            on_disk=True,
            has_payload=True,
            summary="manual daemon kill",
            reason_class="manual_daemon_kill",
            session_id="prgs-author-111-clean",
            role="author",
            command_summary="pkill -f mcp_server.py",
            cleared=False,
        )
        rows = _build_session_rows(inventory, contamination=(marker,))
        self.assertIn("runtime_recovery_contamination", rows[0].contamination_flags)

    def test_process_wide_contamination_surfaces_on_all_sessions(self):
        inventory = _inventory(sessions=(_clean_session(), _stale_session()))
        marker = ContaminationMarker(
            kind="stable_branch_contamination",
            on_disk=True,
            has_payload=True,
            summary="direct master push attempt",
            reason_class="stable_branch_push",
            session_id=None,
            cleared=False,
        )
        rows = _build_session_rows(inventory, contamination=(marker,))
        self.assertEqual(len(rows), 2)
        for row in rows:
            self.assertTrue(
                any("stable_branch_contamination" in f for f in row.contamination_flags)
            )


class TestRenderSessionsPage(unittest.TestCase):
    def _snapshot(
        self,
        *,
        sessions: tuple[dict, ...],
        contamination: tuple[ContaminationMarker, ...] = (),
        stale_runtime: str | None = None,
    ) -> SessionViewSnapshot:
        inventory = _inventory(
            sessions=sessions,
            leases=(
                {
                    "lease_id": "lease-1",
                    "session_id": sessions[0]["session_id"] if sessions else "",
                    "status": "active",
                    "expired": False,
                    "work_kind": "issue",
                    "work_number": 641,
                },
            )
            if sessions
            else (),
            locks=(
                {
                    "issue_number": 641,
                    "worktree_path": "branches/feat-issue-641",
                },
            )
            if sessions
            else (),
            worktrees=(
                {
                    "rel_path": "branches/feat-issue-641",
                    "branch": "feat/issue-641-runtime-session-view",
                    "classification": "active_issue_work",
                    "registered_worktree": True,
                    "dirty": False,
                },
            ),
        )
        rows = _build_session_rows(inventory, contamination)
        return SessionViewSnapshot(
            runtime=_runtime(stale=stale_runtime),
            inventory=inventory,
            sessions=rows,
            contamination_markers=contamination,
        )

    def test_clean_session_render(self):
        html = render_sessions_page(self._snapshot(sessions=(_clean_session(),)))
        self.assertIn("Runtime and sessions", html)
        self.assertIn("prgs-author-111-clean", html)
        self.assertIn("gitea-author", html)
        self.assertIn("branches/feat-issue-641", html)
        self.assertIn("Sanctioned recovery", html)
        self.assertIn("docs/mcp-namespace-eof-recovery.md", html)
        # Recovery section must name reconnect and forbid manual kill.
        recovery_idx = html.lower().find("sanctioned recovery")
        self.assertGreaterEqual(recovery_idx, 0)
        recovery = html[recovery_idx:].lower()
        self.assertIn("reconnect", recovery)
        self.assertIn("contamination", recovery)
        self.assertIn("not recovery", recovery)
        self.assertNotIn("run pkill", recovery)
        self.assertNotIn("killall", recovery)

    def test_stale_session_render(self):
        html = render_sessions_page(self._snapshot(sessions=(_stale_session(),)))
        self.assertIn("prgs-author-222-stale", html)
        self.assertIn("pid-dead", html)
        self.assertIn("badge-stale", html)

    def test_contamination_render_is_not_silent(self):
        marker = ContaminationMarker(
            kind="runtime_recovery_contamination",
            on_disk=True,
            has_payload=True,
            summary="manual kill",
            reason_class="manual_daemon_kill",
            session_id="prgs-author-111-clean",
            command_summary="pkill -f mcp_server.py",
            cleared=False,
        )
        html = render_sessions_page(
            self._snapshot(sessions=(_clean_session(),), contamination=(marker,))
        )
        self.assertIn("Contamination markers", html)
        self.assertIn("runtime_recovery_contamination", html)
        self.assertIn("ACTIVE", html)
        self.assertIn("badge-blocked", html)

    def test_stale_runtime_banner(self):
        html = render_sessions_page(
            self._snapshot(
                sessions=(_clean_session(),),
                stale_runtime="server behind master by 3 commits",
            )
        )
        self.assertIn("Stale runtime", html)
        self.assertIn("server behind master", html)


class TestSessionLoaderComposition(unittest.TestCase):
    def test_load_with_injected_sources(self):
        inventory = _inventory(sessions=(_clean_session(), _stale_session()))
        snap = load_session_view_snapshot(
            load_runtime=lambda: _runtime(),
            load_inventory=lambda: inventory,
            inspect_contamination=lambda **_k: {
                "on_disk": False,
                "has_payload": False,
                "summary": "absent",
            },
            load_contamination_payload=lambda **_k: None,
        )
        self.assertEqual(len(snap.sessions), 2)
        self.assertEqual(snap.stale_session_count, 1)
        self.assertEqual(snap.contaminated_session_count, 0)
        data = snapshot_to_dict(snap)
        self.assertEqual(data["view"], "runtime-sessions")
        self.assertEqual(data["issue"], 641)
        self.assertTrue(data["read_only"])
        self.assertEqual(data["session_counts"]["total"], 2)
        self.assertEqual(data["session_counts"]["stale"], 1)
        self.assertIn("recovery_docs", data)


class TestSessionsRoutes(unittest.TestCase):
    def setUp(self):
        self.client = TestClient(create_app())
        inventory = _inventory(
            sessions=(_clean_session(), _stale_session()),
            leases=(
                {
                    "lease_id": "lease-x",
                    "session_id": "prgs-author-111-clean",
                    "status": "active",
                    "expired": False,
                    "work_kind": "issue",
                    "work_number": 641,
                },
            ),
            locks=(
                {
                    "issue_number": 641,
                    "worktree_path": "branches/feat-issue-641",
                },
            ),
            worktrees=(
                {
                    "rel_path": "branches/feat-issue-641",
                    "branch": "feat/issue-641-runtime-session-view",
                    "classification": "active_issue_work",
                    "registered_worktree": True,
                    "dirty": False,
                },
            ),
        )
        rows = _build_session_rows(inventory, contamination=())
        self.snapshot = SessionViewSnapshot(
            runtime=_runtime(stale="stale for test"),
            inventory=inventory,
            sessions=rows,
            contamination_markers=(),
        )
        self._patch = mock.patch(
            "webui.app.load_session_view_snapshot",
            return_value=self.snapshot,
        )
        self._patch.start()

    def tearDown(self):
        self._patch.stop()

    def test_sessions_page_live(self):
        response = self.client.get("/sessions")
        self.assertEqual(response.status_code, 200)
        self.assertIn("Runtime and sessions", response.text)
        self.assertIn("prgs-author-111-clean", response.text)
        self.assertIn("prgs-author-222-stale", response.text)
        self.assertIn("pid-dead", response.text)
        self.assertIn("Sanctioned recovery", response.text)
        self.assertNotIn("Phase 1 shell placeholder", response.text)
        self.assertNotIn("child issue of #425", response.text.lower())

    def test_api_sessions_json(self):
        for path in ("/api/sessions", "/api/v1/sessions"):
            response = self.client.get(path)
            self.assertEqual(response.status_code, 200, path)
            data = response.json()
            self.assertEqual(data["view"], "runtime-sessions")
            self.assertEqual(data["session_counts"]["total"], 2)
            self.assertEqual(data["session_counts"]["stale"], 1)
            self.assertTrue(data["read_only"])
            self.assertEqual(data["mutations"], [])

    def test_nav_marks_sessions_live(self):
        sessions_items = [
            item for item in iter_nav_items() if item.href == "/sessions"
        ]
        self.assertEqual(len(sessions_items), 1)
        self.assertEqual(sessions_items[0].status, "live")
        self.assertNotIn("/sessions", STUB_PAGES)
        # Home page should not mark Sessions as stub.
        home = self.client.get("/")
        self.assertEqual(home.status_code, 200)
        self.assertIn('href="/sessions"', home.text)
        # Stub marker only appears next to remaining stub destinations.
        self.assertNotIn(
            'href="/sessions">Sessions</a> <span class="muted">(stub)</span>',
            home.text,
        )


if __name__ == "__main__":
    unittest.main()