import asyncio
from pathlib import Path
from types import SimpleNamespace

import pytest
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from lib.db.base import Base
from server.agent_runtime.service import AssistantService
from server.agent_runtime.session_store import SessionMetaStore
from server.agent_runtime.stream_projector import AssistantStreamProjector
from tests.factories import make_session_meta


class _FakePM:
    def __init__(self, valid_project="demo"):
        self.valid_project = valid_project

    def get_project_path(self, project_name):
        if project_name != self.valid_project:
            raise FileNotFoundError(project_name)
        return Path("/tmp") / project_name


class _FakeMetaStore:
    def __init__(self, metas=None):
        self.metas = {m.id: m for m in (metas or [])}

    async def get(self, session_id):
        return self.metas.get(session_id)

    async def list(self, project_name=None, status=None, limit=50, offset=0):
        return list(self.metas.values())

    async def delete(self, session_id):
        return self.metas.pop(session_id, None) is not None


class _FakeSessionManager:
    def __init__(self):
        self.sessions = {}
        self.new_sessions = []
        self.sent = []
        self.answered = []
        self.interrupted = []
        self.closed = []
        self.unsubscribed = []
        self.status = "running"
        self.buffer = []
        self.pending = []
        self.close_error = None

    async def send_new_session(self, project_name, prompt, **kwargs):
        self.new_sessions.append((project_name, prompt))
        return "sdk-new-id"

    async def get_status(self, session_id):
        return self.status

    def get_buffered_messages(self, session_id):
        return list(self.buffer)

    async def get_pending_questions_snapshot(self, session_id):
        return list(self.pending)

    async def send_message(self, session_id, content, **kwargs):
        self.sent.append((session_id, content))

    async def answer_user_question(self, session_id, question_id, answers):
        self.answered.append((session_id, question_id, answers))

    async def interrupt_session(self, session_id):
        self.interrupted.append(session_id)
        return "interrupted"

    async def subscribe(self, session_id, replay_buffer=True):
        q = asyncio.Queue()
        for m in self.buffer:
            q.put_nowait(m)
        return q

    async def unsubscribe(self, session_id, queue):
        self.unsubscribed.append(session_id)

    async def close_session(self, session_id, *, reason="session closed"):
        self.closed.append((session_id, reason))
        if self.close_error is not None:
            raise self.close_error
        self.sessions.pop(session_id, None)

    async def shutdown_gracefully(self):
        return None


class _FakeTranscriptAdapter:
    def __init__(self, history=None):
        self.history = history or []

    async def read_raw_messages(self, sdk_session_id=None, project_cwd=None):
        return list(self.history)


class TestAssistantServiceMore:
    @pytest.mark.asyncio
    async def test_service_init_interrupts_stale_running_sessions(self, tmp_path):
        # Create an in-memory async store and seed data
        engine = create_async_engine("sqlite+aiosqlite:///:memory:")
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
        factory = async_sessionmaker(engine, expire_on_commit=False)
        store = SessionMetaStore(session_factory=factory)

        running = await store.create("demo", "sdk-running-1")
        completed = await store.create("demo", "sdk-completed-1")
        await store.update_status(running.id, "running")
        await store.update_status(completed.id, "completed")

        service = AssistantService(project_root=tmp_path)
        # Replace the service's meta_store with our test store
        service.meta_store = store
        service.session_manager.meta_store = store

        # Manually run the interrupt logic (normally done in startup())
        await service._interrupt_stale_running_sessions()

        refreshed_running = await service.meta_store.get(running.id)
        refreshed_completed = await service.meta_store.get(completed.id)
        assert refreshed_running is not None
        assert refreshed_running.status == "interrupted"
        assert refreshed_completed is not None
        assert refreshed_completed.status == "completed"

        await engine.dispose()

    @pytest.mark.asyncio
    async def test_startup_waits_cleanup_and_is_idempotent(self, tmp_path, monkeypatch):
        service = AssistantService(project_root=tmp_path)
        calls = 0
        entered = asyncio.Event()
        release = asyncio.Event()

        async def fake_interrupt():
            nonlocal calls
            calls += 1
            entered.set()
            await release.wait()

        monkeypatch.setattr(service, "_interrupt_stale_running_sessions", fake_interrupt)

        startup_task = asyncio.create_task(service.startup())
        await asyncio.wait_for(entered.wait(), timeout=0.2)
        assert not startup_task.done()

        release.set()
        await asyncio.wait_for(startup_task, timeout=0.2)
        assert calls == 1

        await service.startup()
        assert calls == 1

    @pytest.mark.asyncio
    async def test_crud_and_message_validation(self, tmp_path):
        service = AssistantService(project_root=tmp_path)
        meta = make_session_meta(id="s1", status="idle")

        sm = _FakeSessionManager()
        service.pm = _FakePM(valid_project="demo")
        service.session_manager = sm
        service.meta_store = _FakeMetaStore([meta])

        listed = await service.list_sessions()
        assert len(listed) == 1

        fetched = await service.get_session("s1")
        assert fetched.status == "idle"
        sm.sessions["s1"] = SimpleNamespace(status="running")
        fetched_live = await service.get_session("s1")
        assert fetched_live.status == "running"

        # send_or_create — new session (no session_id)
        new_result = await service.send_or_create("demo", "hello")
        assert new_result == {"status": "accepted", "session_id": "sdk-new-id"}
        assert len(sm.new_sessions) == 1
        assert sm.new_sessions[0][0] == "demo"

        # send_or_create — existing session
        existing_result = await service.send_or_create("demo", "world", session_id="s1")
        assert existing_result == {"status": "accepted", "session_id": "s1"}
        assert sm.sent == [("s1", "world")]

        # send_or_create — empty message raises ValueError
        with pytest.raises(ValueError):
            await service.send_or_create("demo", "   ")

        # send_or_create — missing session raises FileNotFoundError
        with pytest.raises(FileNotFoundError):
            await service.send_or_create("demo", "hello", session_id="missing")

        # send_or_create — project mismatch raises FileNotFoundError
        with pytest.raises(FileNotFoundError):
            await service.send_or_create("other_project", "hello", session_id="s1")

        with pytest.raises(FileNotFoundError):
            await service.answer_user_question("missing", "q1", {"a": "b"})
        await service.answer_user_question("s1", "q1", {"a": "b"})
        assert sm.answered == [("s1", "q1", {"a": "b"})]

        with pytest.raises(FileNotFoundError):
            await service.interrupt_session("missing")
        interrupted = await service.interrupt_session("s1")
        assert interrupted["session_status"] == "interrupted"

    @pytest.mark.asyncio
    async def test_delete_session_closes_active_session_before_delete(self, tmp_path):
        service = AssistantService(project_root=tmp_path)
        meta = make_session_meta(id="s1")
        service.meta_store = _FakeMetaStore([meta])
        sm = _FakeSessionManager()
        sm.sessions["s1"] = SimpleNamespace()
        service.session_manager = sm

        ok = await service.delete_session("s1")
        assert ok is True
        assert sm.closed == [("s1", "session deleted")]
        assert "s1" not in sm.sessions

        missing = await service.delete_session("missing")
        assert missing is False

    @pytest.mark.asyncio
    async def test_delete_session_propagates_close_error(self, tmp_path):
        service = AssistantService(project_root=tmp_path)
        meta = make_session_meta(id="s1")
        service.meta_store = _FakeMetaStore([meta])
        sm = _FakeSessionManager()
        sm.sessions["s1"] = SimpleNamespace()
        sm.close_error = RuntimeError("close failed")
        service.session_manager = sm

        with pytest.raises(RuntimeError, match="close failed"):
            await service.delete_session("s1")

        assert "s1" in sm.sessions

    @pytest.mark.asyncio
    async def test_snapshot_and_stream_helpers(self, tmp_path):
        service = AssistantService(project_root=tmp_path)
        meta = make_session_meta(id="s1", status="running")
        service.meta_store = _FakeMetaStore([meta])
        sm = _FakeSessionManager()
        sm.status = "running"
        sm.buffer = [{"type": "runtime_status", "status": "running"}]
        sm.pending = [{"type": "ask_user_question", "question_id": "aq-1"}]
        service.session_manager = sm
        service.transcript_adapter = _FakeTranscriptAdapter(history=[])

        with pytest.raises(FileNotFoundError):
            await service.get_snapshot("missing")

        snapshot = await service.get_snapshot("s1")
        assert snapshot["status"] == "running"
        assert snapshot["pending_questions"][0]["question_id"] == "aq-1"

        replayed, overflow = service._drain_replay(asyncio.Queue())
        assert replayed == []
        assert overflow is False
        q = asyncio.Queue()
        q.put_nowait({"type": "_queue_overflow"})
        replayed2, overflow2 = service._drain_replay(q)
        assert replayed2 == []
        assert overflow2 is True

        projector = AssistantStreamProjector(initial_messages=[])
        events, should_break = await service._dispatch_live_message(
            {"type": "_queue_overflow"},
            projector,
            "s1",
        )
        assert should_break is True
        assert events == []

        events2, stop2 = await service._dispatch_live_message(
            {"type": "system", "subtype": "compact_boundary"},
            projector,
            "s1",
        )
        assert stop2 is False
        assert any(event.event == "compact" for event in events2)

        events3, stop3 = await service._dispatch_live_message(
            {"type": "runtime_status", "status": "interrupted"},
            projector,
            "s1",
        )
        assert stop3 is True
        assert any(event.event == "status" for event in events3)

        events4, stop4 = await service._dispatch_live_message(
            {"type": "result", "subtype": "success", "is_error": False},
            projector,
            "s1",
        )
        assert stop4 is True
        assert any(event.event == "status" for event in events4)

        assert service._check_runtime_status_terminal({"status": "???."}, "s1") is None
        assert await service._handle_heartbeat_timeout("s1", "running", projector) is None
        sm.status = "completed"
        status_event = await service._handle_heartbeat_timeout("s1", "running", projector)
        assert status_event is not None
        assert status_event.event == "status"
        patch_event = service._sse_event("patch", {"x": 1})
        assert patch_event.event == "patch"
        assert patch_event.data == {"x": 1}

    def test_merge_and_dedup_helpers(self, tmp_path):
        service = AssistantService(project_root=tmp_path)

        # _fingerprint tests
        assert service._fingerprint({"type": "assistant", "content": [{"text": "A"}]}) == "fp:assistant:t:A"
        result_fp = service._fingerprint(
            {
                "type": "result",
                "subtype": "success",
                "is_error": False,
            }
        )
        assert result_fp == "fp:result:success:False"
        assert service._fingerprint({"type": "user", "content": "x"}) is None

        # _fingerprint_tail tests
        tail_fps = service._fingerprint_tail(
            [
                {"type": "user", "content": "hello", "uuid": "u1"},
                {"type": "assistant", "content": [{"text": "A"}], "uuid": "a1"},
            ]
        )
        assert "fp:assistant:t:A" in tail_fps

        # _is_buffer_duplicate tests
        assert service._is_buffer_duplicate({"uuid": "u1", "type": "user"}, "user", {"u1"}, set(), []) is True
        assert (
            service._is_buffer_duplicate(
                {"type": "assistant", "content": [{"text": "A"}]},
                "assistant",
                set(),
                {"fp:assistant:t:A"},
                [],
            )
            is True
        )

        assert service._parse_iso_datetime(None) is None
        assert service._parse_iso_datetime("bad") is None
        naive = service._parse_iso_datetime("2026-02-01T00:00:00")
        assert naive.tzinfo is not None
        assert service._parse_iso_datetime("2026-02-01T00:00:00Z") is not None

        # _echo_in_transcript tests — round-aware dedup
        # Case 1: in-progress round (user only, no result after) → dedup
        history_in_progress = [{"type": "user", "content": "hello", "timestamp": "2026-02-01T00:00:01Z"}]
        local_echo = {
            "type": "user",
            "content": "hello",
            "local_echo": True,
            "timestamp": "2026-02-01T00:00:00Z",
        }
        assert service._echo_in_transcript(local_echo, history_in_progress) is True

        # Case 2: same text from an older round → do NOT dedup
        history_complete = [
            {"type": "user", "content": "hello", "timestamp": "2026-01-31T23:59:59Z"},
            {"type": "assistant", "content": [{"type": "text", "text": "hi"}]},
        ]
        assert service._echo_in_transcript(local_echo, history_complete) is False

        # Case 3: non-user message → False
        assert service._echo_in_transcript({"type": "assistant"}, history_in_progress) is False

        assert service._extract_plain_user_content({"type": "assistant"}) is None
        assert (
            service._extract_plain_user_content({"type": "user", "content": [{"type": "text", "text": " ok "}]}) == "ok"
        )
        assert service._is_groupable_message("bad") is False  # type: ignore[arg-type]

        assert service._resolve_result_status({"session_status": "interrupted"}) == "interrupted"
        assert service._resolve_result_status({"subtype": "error_x", "is_error": True}) == "error"
        payload = service._build_status_event_payload("error", "s1", None)
        assert payload["status"] == "error"
        assert payload["subtype"] == "error"
        assert payload["is_error"] is True

    def test_skill_listing_and_metadata_parsing(self, tmp_path, monkeypatch):
        service = AssistantService(project_root=tmp_path)
        service.pm = _FakePM(valid_project="demo")

        agent_skill = tmp_path / "agent_runtime_profile" / ".claude" / "skills" / "s1"
        agent_skill.mkdir(parents=True)
        (agent_skill / "SKILL.md").write_text(
            "---\nname: project-skill\ndescription: from frontmatter\n---\n# body\n",
            encoding="utf-8",
        )

        # Create a fallback skill for metadata parsing test
        fallback_skill_dir = tmp_path / "agent_runtime_profile" / ".claude" / "skills" / "s2"
        fallback_skill_dir.mkdir(parents=True)
        (fallback_skill_dir / "SKILL.md").write_text(
            "first non heading line\n# heading\n",
            encoding="utf-8",
        )

        all_skills = service.list_available_skills()
        names = {item["name"] for item in all_skills}
        assert "project-skill" in names
        assert "s2" in names

        for_project = service.list_available_skills(project_name="demo")
        assert len(for_project) >= 1

        fallback = service._load_skill_metadata(fallback_skill_dir / "SKILL.md", "fallback")
        assert fallback["name"] == "fallback"
        assert fallback["description"] == "first non heading line"
