"""Reusable product E2E smoke drivers."""

from __future__ import annotations

import asyncio
import json
import os
import subprocess
import time
import uuid
import wave
from collections.abc import AsyncGenerator, Awaitable, Callable, Iterator, Sequence
from contextlib import contextmanager
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

import httpx
import pytest

from config.provider_ids import SUPPORTED_PROVIDER_IDS
from core.anthropic.stream_contracts import (
    SSEEvent,
    assert_anthropic_stream_contract,
    event_index,
    has_tool_use,
    parse_sse_lines,
    text_content,
)
from messaging.handler import ClaudeMessageHandler
from messaging.models import IncomingMessage
from messaging.platforms.base import MessagingPlatform
from messaging.session import SessionStore
from smoke.lib.config import ProviderModel, SmokeConfig, auth_headers
from smoke.lib.server import RunningServer, start_server
from smoke.lib.skips import fail_missing_env


@dataclass(slots=True)
class ConversationTurn:
    request: dict[str, Any]
    events: list[SSEEvent]

    @property
    def assistant_content(self) -> list[dict[str, Any]]:
        return assistant_content_from_events(self.events)

    @property
    def text(self) -> str:
        return text_content(self.events)


class SmokeServerDriver:
    """Start a local proxy server for a product scenario."""

    def __init__(
        self,
        config: SmokeConfig,
        *,
        name: str,
        env_overrides: dict[str, str] | None = None,
        command: list[str] | None = None,
    ) -> None:
        self.config = config
        self.name = name
        self.env_overrides = env_overrides
        self.command = command

    @contextmanager
    def run(self) -> Iterator[RunningServer]:
        with start_server(
            self.config,
            env_overrides=self.env_overrides,
            command=self.command,
            name=self.name,
        ) as server:
            yield server


class ConversationDriver:
    """Drive multi-turn Anthropic-compatible conversations through the server."""

    def __init__(self, server: RunningServer, config: SmokeConfig) -> None:
        self.server = server
        self.config = config
        self.messages: list[dict[str, Any]] = []
        self.turns: list[ConversationTurn] = []

    def ask(
        self,
        text: str,
        *,
        model: str = "fcc-smoke-default",
        max_tokens: int = 256,
        extra: dict[str, Any] | None = None,
        headers: dict[str, str] | None = None,
        append_assistant: bool = True,
    ) -> ConversationTurn:
        self.messages.append({"role": "user", "content": text})
        payload = {
            "model": model,
            "max_tokens": max_tokens,
            "messages": list(self.messages),
        }
        if extra:
            payload.update(extra)
        turn = self.stream(payload, headers=headers)
        if append_assistant:
            self.messages.append(
                {"role": "assistant", "content": turn.assistant_content or turn.text}
            )
        return turn

    def stream(
        self,
        payload: dict[str, Any],
        *,
        headers: dict[str, str] | None = None,
    ) -> ConversationTurn:
        request_headers = headers or auth_headers()
        with httpx.stream(
            "POST",
            f"{self.server.base_url}/v1/messages",
            headers=request_headers,
            json=payload,
            timeout=self.config.timeout_s,
        ) as response:
            if response.status_code != 200:
                body = response.read().decode("utf-8", errors="replace")
                raise AssertionError(
                    f"stream request failed: HTTP {response.status_code} {body[:1000]}"
                )
            events = parse_sse_lines(response.iter_lines())
        assert_anthropic_stream_contract(events)
        turn = ConversationTurn(payload, events)
        self.turns.append(turn)
        return turn

    def stream_expect_http_error(
        self,
        payload: dict[str, Any],
        *,
        expected_status: int,
    ) -> dict[str, Any]:
        response = httpx.post(
            f"{self.server.base_url}/v1/messages",
            headers=auth_headers(),
            json=payload,
            timeout=self.config.timeout_s,
        )
        assert response.status_code == expected_status, response.text
        return response.json()


class ProviderMatrixDriver:
    """Resolve provider models and enforce matrix semantics for product smoke."""

    ALL_PROVIDERS: tuple[str, ...] = SUPPORTED_PROVIDER_IDS

    def __init__(self, config: SmokeConfig) -> None:
        self.config = config

    def configured_models(self) -> list[ProviderModel]:
        return self.config.provider_models()

    def provider_smoke_models(self) -> list[ProviderModel]:
        selected = self.config.provider_matrix
        missing_selected = [
            provider
            for provider in selected
            if provider in self.ALL_PROVIDERS
            and not self.config.has_provider_configuration(provider)
        ]
        if missing_selected:
            fail_missing_env(
                "selected providers are not configured: "
                + ", ".join(sorted(missing_selected))
            )

        models = self.config.provider_smoke_models()
        if not models and os.getenv("FCC_ALLOW_NO_PROVIDER_SMOKE") != "1":
            fail_missing_env(
                "no configured provider smoke models; set FCC_ALLOW_NO_PROVIDER_SMOKE=1 "
                "only for no-provider smoke collection"
            )
        return models

    def first_model(self) -> ProviderModel:
        models = self.provider_smoke_models()
        if not models:
            pytest.skip("missing_env: no configured provider model")
        return models[0]


class ClientProtocolDriver:
    """Build recorded/representative client protocol requests."""

    @staticmethod
    def vscode_headers() -> dict[str, str]:
        headers = auth_headers()
        headers.update(
            {
                "anthropic-beta": "messages-2023-12-15",
                "user-agent": "Claude-Code-VSCode product smoke",
            }
        )
        return headers

    @staticmethod
    def jetbrains_headers(config: SmokeConfig) -> dict[str, str]:
        headers = auth_headers()
        token = config.settings.anthropic_auth_token
        if token:
            headers.pop("x-api-key", None)
            headers["authorization"] = f"Bearer {token}"
        headers["user-agent"] = "JetBrains-ACP product smoke"
        return headers

    @staticmethod
    def adaptive_thinking_payload() -> dict[str, Any]:
        return {
            "model": "claude-opus-4-7",
            "max_tokens": 256,
            "messages": [
                {"role": "user", "content": "hello"},
                {
                    "role": "assistant",
                    "content": [
                        {"type": "thinking", "thinking": "unsigned thought"},
                        {"type": "redacted_thinking", "data": "opaque"},
                        {"type": "text", "text": "Hello."},
                    ],
                },
                {"role": "user", "content": "Reply with exactly FCC_SMOKE_CLIENT"},
            ],
            "thinking": {"type": "adaptive", "budget_tokens": 1024},
        }

    @staticmethod
    def tool_result_payload() -> dict[str, Any]:
        return {
            "model": "claude-sonnet-4-5-20250929",
            "max_tokens": 256,
            "messages": [
                {"role": "user", "content": "Use echo_smoke once."},
                {
                    "role": "assistant",
                    "content": [
                        {
                            "type": "tool_use",
                            "id": "toolu_client_smoke",
                            "name": "echo_smoke",
                            "input": {"value": "FCC_SMOKE_CLIENT"},
                        }
                    ],
                },
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "tool_result",
                            "tool_use_id": "toolu_client_smoke",
                            "content": "FCC_SMOKE_CLIENT",
                        }
                    ],
                },
            ],
            "tools": [echo_tool_schema()],
            "thinking": {"type": "adaptive"},
        }

    @staticmethod
    def run_claude_prompt(
        *,
        claude_bin: str,
        server: RunningServer,
        config: SmokeConfig,
        cwd: Path,
        prompt: str,
    ) -> subprocess.CompletedProcess[str]:
        env = os.environ.copy()
        env["ANTHROPIC_BASE_URL"] = server.base_url
        env["ANTHROPIC_API_URL"] = f"{server.base_url}/v1"
        env.pop("ANTHROPIC_API_KEY", None)
        if config.settings.anthropic_auth_token:
            env["ANTHROPIC_AUTH_TOKEN"] = config.settings.anthropic_auth_token
        else:
            env.pop("ANTHROPIC_AUTH_TOKEN", None)
        return subprocess.run(
            [
                claude_bin,
                "--bare",
                "--tools",
                "",
                "--system-prompt",
                "Reply with exactly the requested smoke token and no other text.",
                "-p",
                prompt,
            ],
            cwd=cwd,
            env=env,
            capture_output=True,
            text=True,
            timeout=config.timeout_s,
            check=False,
        )


class FakePlatform(MessagingPlatform):
    """In-memory platform that exercises the real message handler."""

    def __init__(self, name: str) -> None:
        self.name = name
        self.handler: Callable[[IncomingMessage], Awaitable[None]] | None = None
        self.sent: list[dict[str, Any]] = []
        self.edits: list[dict[str, Any]] = []
        self.deletes: list[dict[str, Any]] = []
        self._counter = 0
        self._tasks: list[asyncio.Future[Any]] = []
        self._pending_voice: dict[tuple[str, str], tuple[str, str]] = {}

    async def start(self) -> None:
        return None

    async def stop(self) -> None:
        for task in self._tasks:
            if not task.done():
                task.cancel()
        if self._tasks:
            await asyncio.gather(*self._tasks, return_exceptions=True)

    @property
    def is_connected(self) -> bool:
        return True

    def on_message(self, handler: Callable[[IncomingMessage], Awaitable[None]]) -> None:
        self.handler = handler

    async def emit(self, incoming: IncomingMessage) -> None:
        assert self.handler is not None
        await self.handler(incoming)

    def fire_and_forget(self, task: Awaitable[Any]) -> None:
        self._tasks.append(asyncio.ensure_future(task))

    async def send_message(
        self,
        chat_id: str,
        text: str,
        reply_to: str | None = None,
        parse_mode: str | None = None,
        message_thread_id: str | None = None,
    ) -> str:
        self._counter += 1
        message_id = f"{self.name}_msg_{self._counter}"
        self.sent.append(
            {
                "chat_id": chat_id,
                "message_id": message_id,
                "text": text,
                "reply_to": reply_to,
                "parse_mode": parse_mode,
                "message_thread_id": message_thread_id,
            }
        )
        return message_id

    async def edit_message(
        self,
        chat_id: str,
        message_id: str,
        text: str,
        parse_mode: str | None = None,
    ) -> None:
        self.edits.append(
            {
                "chat_id": chat_id,
                "message_id": message_id,
                "text": text,
                "parse_mode": parse_mode,
            }
        )

    async def delete_message(self, chat_id: str, message_id: str) -> None:
        self.deletes.append({"chat_id": chat_id, "message_id": message_id})

    async def queue_send_message(
        self,
        chat_id: str,
        text: str,
        reply_to: str | None = None,
        parse_mode: str | None = None,
        fire_and_forget: bool = True,
        message_thread_id: str | None = None,
    ) -> str | None:
        message_id = await self.send_message(
            chat_id,
            text,
            reply_to=reply_to,
            parse_mode=parse_mode,
            message_thread_id=message_thread_id,
        )
        return None if fire_and_forget else message_id

    async def queue_edit_message(
        self,
        chat_id: str,
        message_id: str,
        text: str,
        parse_mode: str | None = None,
        fire_and_forget: bool = True,
    ) -> None:
        await self.edit_message(chat_id, message_id, text, parse_mode=parse_mode)

    async def queue_delete_message(
        self,
        chat_id: str,
        message_id: str,
        fire_and_forget: bool = True,
    ) -> None:
        await self.delete_message(chat_id, message_id)

    async def queue_delete_messages(
        self,
        chat_id: str,
        message_ids: Sequence[str],
        fire_and_forget: bool = True,
    ) -> None:
        for message_id in message_ids:
            await self.queue_delete_message(chat_id, message_id, fire_and_forget=False)

    def register_pending_voice(
        self, chat_id: str, voice_message_id: str, status_message_id: str
    ) -> None:
        self._pending_voice[(chat_id, voice_message_id)] = (
            voice_message_id,
            status_message_id,
        )

    async def cancel_pending_voice(
        self, chat_id: str, voice_message_id: str
    ) -> tuple[str, str] | None:
        return self._pending_voice.pop((chat_id, voice_message_id), None)


class FakeCLISession:
    def __init__(self, events: list[dict[str, Any]]) -> None:
        self.events = events
        self.calls: list[dict[str, Any]] = []
        self.is_busy = False

    async def start_task(
        self, prompt: str, session_id: str | None = None, fork_session: bool = False
    ) -> AsyncGenerator[dict[str, Any]]:
        self.calls.append(
            {"prompt": prompt, "session_id": session_id, "fork_session": fork_session}
        )
        self.is_busy = True
        try:
            for event in self.events:
                await asyncio.sleep(0)
                yield event
        finally:
            self.is_busy = False


class FakeCLIManager:
    def __init__(self, event_batches: list[list[dict[str, Any]]] | None = None) -> None:
        self.event_batches = event_batches or [default_cli_events("fake_session_1")]
        self.sessions: list[FakeCLISession] = []
        self.registered: list[tuple[str, str]] = []
        self.removed: list[str] = []
        self.stopped = False

    async def get_or_create_session(
        self, session_id: str | None = None
    ) -> tuple[FakeCLISession, str, bool]:
        index = len(self.sessions)
        events = self.event_batches[min(index, len(self.event_batches) - 1)]
        session = FakeCLISession(events)
        self.sessions.append(session)
        return session, session_id or f"pending_{index}", session_id is None

    async def register_real_session_id(
        self, temp_id: str, real_session_id: str
    ) -> bool:
        self.registered.append((temp_id, real_session_id))
        return True

    async def stop_all(self) -> None:
        self.stopped = True

    async def remove_session(self, session_id: str) -> bool:
        self.removed.append(session_id)
        return True

    def get_stats(self) -> dict[str, int]:
        return {"active_sessions": len(self.sessions), "pending_sessions": 0}


@dataclass(slots=True)
class FakePlatformDriver:
    platform_name: str
    tmp_path: Path
    event_batches: list[list[dict[str, Any]]] | None = None
    platform: FakePlatform = field(init=False)
    cli_manager: FakeCLIManager = field(init=False)
    session_store: SessionStore = field(init=False)
    handler: ClaudeMessageHandler = field(init=False)

    def __post_init__(self) -> None:
        self.platform = FakePlatform(self.platform_name)
        self.cli_manager = FakeCLIManager(self.event_batches)
        self.session_store = SessionStore(
            storage_path=str(self.tmp_path / f"{self.platform_name}-sessions.json")
        )
        self.handler = ClaudeMessageHandler(
            self.platform, self.cli_manager, self.session_store
        )
        self.platform.on_message(self.handler.handle_message)

    async def send(
        self,
        text: str,
        *,
        message_id: str | None = None,
        reply_to: str | None = None,
    ) -> IncomingMessage:
        incoming = IncomingMessage(
            text=text,
            chat_id="chat_1",
            user_id="user_1",
            message_id=message_id or f"in_{uuid.uuid4().hex[:8]}",
            platform=self.platform_name,
            reply_to_message_id=reply_to,
        )
        await self.platform.emit(incoming)
        await self.wait_for_idle()
        return incoming

    async def wait_for_idle(self, *, timeout_s: float = 5.0) -> None:
        deadline = time.monotonic() + timeout_s
        while time.monotonic() < deadline:
            pending = [task for task in self.platform._tasks if not task.done()]
            if not pending and self._all_tree_nodes_terminal():
                self.session_store.flush_pending_save()
                return
            await asyncio.sleep(0.02)
        raise AssertionError("fake platform did not become idle")

    def _all_tree_nodes_terminal(self) -> bool:
        data = self.handler.tree_queue.to_dict()
        for tree in data.get("trees", {}).values():
            nodes = tree.get("nodes", {}) if isinstance(tree, dict) else {}
            for node in nodes.values():
                if not isinstance(node, dict):
                    continue
                if node.get("state") in {"pending", "in_progress"}:
                    return False
        return True


class VoiceFixtureDriver:
    @staticmethod
    def write_tone_wav(path: Path) -> None:
        import math

        sample_rate = 16000
        duration_s = 0.25
        amplitude = 8000
        frames = bytearray()
        for i in range(int(sample_rate * duration_s)):
            sample = int(amplitude * math.sin(2 * math.pi * 440 * i / sample_rate))
            frames.extend(sample.to_bytes(2, byteorder="little", signed=True))
        with wave.open(str(path), "wb") as wav:
            wav.setnchannels(1)
            wav.setsampwidth(2)
            wav.setframerate(sample_rate)
            wav.writeframes(bytes(frames))


def echo_tool_schema() -> dict[str, Any]:
    return {
        "name": "echo_smoke",
        "description": "Echo a test value.",
        "input_schema": {
            "type": "object",
            "properties": {"value": {"type": "string"}},
            "required": ["value"],
        },
    }


def assistant_content_from_events(events: list[SSEEvent]) -> list[dict[str, Any]]:
    blocks: dict[int, dict[str, Any]] = {}
    block_order: list[int] = []
    for event in events:
        if event.event == "content_block_start":
            index = event_index(event)
            block = event.data.get("content_block", {})
            if isinstance(block, dict):
                blocks[index] = dict(block)
                block_order.append(index)
            continue
        if event.event == "content_block_delta":
            index = event_index(event)
            block = blocks.get(index)
            delta = event.data.get("delta", {})
            if not isinstance(block, dict) or not isinstance(delta, dict):
                continue
            delta_type = delta.get("type")
            if delta_type == "text_delta":
                block["text"] = str(block.get("text", "")) + str(delta.get("text", ""))
            elif delta_type == "thinking_delta":
                block["thinking"] = str(block.get("thinking", "")) + str(
                    delta.get("thinking", "")
                )
            elif delta_type == "input_json_delta":
                block["_partial_json"] = str(block.get("_partial_json", "")) + str(
                    delta.get("partial_json", "")
                )

    content: list[dict[str, Any]] = []
    for index in block_order:
        block = blocks[index]
        if block.get("type") == "tool_use":
            partial = str(block.pop("_partial_json", ""))
            if partial:
                try:
                    block["input"] = json.loads(partial)
                except json.JSONDecodeError:
                    block["input"] = {}
        content.append(block)
    return content


def tool_use_blocks(events: list[SSEEvent]) -> list[dict[str, Any]]:
    return [
        block
        for block in assistant_content_from_events(events)
        if block.get("type") == "tool_use"
    ]


def default_cli_events(session_id: str) -> list[dict[str, Any]]:
    return [
        {"type": "session_info", "session_id": session_id},
        {
            "type": "assistant",
            "message": {
                "content": [
                    {"type": "thinking", "thinking": "Inspect the request."},
                    {
                        "type": "tool_use",
                        "id": "toolu_fake",
                        "name": "Read",
                        "input": {"file_path": "README.md"},
                    },
                    {
                        "type": "tool_result",
                        "tool_use_id": "toolu_fake",
                        "content": "Free Claude Code",
                    },
                    {"type": "text", "text": "Fake platform answer."},
                ]
            },
        },
        {"type": "exit", "code": 0, "stderr": None},
    ]


def assert_product_stream(events: list[SSEEvent]) -> None:
    assert_anthropic_stream_contract(events)
    assert text_content(events).strip() or has_tool_use(events), (
        "product stream emitted neither text nor tool_use"
    )
