# -*- coding: utf-8 -*-
"""Console APIs: push messages, chat, and file upload for chat."""
from __future__ import annotations

import asyncio
import json
import logging
import re
import time
import uuid
from dataclasses import dataclass
from pathlib import Path
from typing import Any, AsyncGenerator, Dict, Optional, Union

from fastapi import (
    APIRouter,
    File,
    HTTPException,
    Query,
    Request,
    UploadFile,
)
from pydantic import BaseModel
from starlette.responses import StreamingResponse

from qwenpaw.schemas import (
    AgentRequest,
    _coerce_content_item,
)
from ...utils.logging import LOG_FILE_PATH, sanitize_log_value
from ..agent_context import get_agent_for_request
from ..approvals.display import approval_display_fields
from ..chats.title_generator import generate_and_update_title
from ..utils import check_upload_size


logger = logging.getLogger(__name__)

router = APIRouter(prefix="/console", tags=["console"])


# ── Background task store ──


@dataclass
class _BackgroundTask:
    """In-memory state for a background chat task."""

    status: str = "submitted"
    started_at: Optional[float] = None
    finished_at: Optional[float] = None
    result: Optional[Dict[str, Any]] = None
    asyncio_task: Optional[asyncio.Task] = None


_bg_tasks: Dict[str, _BackgroundTask] = {}
_bg_lock = asyncio.Lock()


class MarkInboxReadRequest(BaseModel):
    event_ids: list[str] = []
    all: bool = False


MAX_DEBUG_LOG_LINES = 1000


def _safe_filename(name: str) -> str:
    """Safe basename, alphanumeric/./-/_, max 200 chars."""
    base = Path(name).name if name else "file"
    return re.sub(r"[^\w.\-]", "_", base)[:200] or "file"


def _extract_placeholder_name(content_parts: list) -> tuple[str, str]:
    """Return ``(placeholder_name, first_user_text)`` for a new chat.

    The placeholder name shows up in the session drawer immediately while a
    background task asks the model for a real title. Content shapes match
    ``channels/base.py::_extract_chat_name``: dict blocks like
    ``{"type": "text", "text": "..."}``, raw strings, and objects with a
    ``.text`` attribute. Anything else (audio/image/file blocks) is treated
    as media and gets the generic "Media Message" placeholder.
    """
    if not content_parts:
        return "New Chat", ""
    content = content_parts[0]
    if not content:
        return "Media Message", ""
    if isinstance(content, str):
        first_text = content
    elif isinstance(content, dict):
        text = content.get("text", "")
        first_text = text if isinstance(text, str) else ""
    elif hasattr(content, "text"):
        first_text = content.text or ""
    else:
        first_text = ""
    if not first_text:
        return "Media Message", ""
    return first_text[:10], first_text


def _extract_session_and_payload(request_data: Union[AgentRequest, dict]):
    """Extract run_key (ChatSpec.id), session_id, and native payload.

    run_key must be ChatSpec.id (chat_id) so it matches list_chats/get_chat.
    """
    if isinstance(request_data, AgentRequest):
        channel_id = getattr(request_data, "channel", None) or "console"
        sender_id = request_data.user_id or "default"
        session_id = request_data.session_id or "default"
        content_parts = (
            list(request_data.input[0].content) if request_data.input else []
        )
    else:
        channel_id = request_data.get("channel", "console")
        sender_id = request_data.get("user_id", "default")
        session_id = request_data.get("session_id", "default")
        input_data = request_data.get("input", [])
        content_parts = []
        for content_part in input_data:
            if hasattr(content_part, "content"):
                content_parts.extend(list(content_part.content or []))
            elif isinstance(content_part, dict) and "content" in content_part:
                # Coerce raw dicts to typed Content models so downstream
                # getattr checks (e.g. _content_has_text) see real attrs.
                content_parts.extend(
                    _coerce_content_item(c)
                    for c in (content_part["content"] or [])
                )

    meta: dict = {
        "session_id": session_id,
        "user_id": sender_id,
    }

    # Preserve request_context (e.g. session-level approval_level)
    if isinstance(request_data, AgentRequest):
        rc = getattr(request_data, "request_context", None)
    else:
        rc = request_data.get("request_context")
    if isinstance(rc, dict) and rc:
        meta["request_context"] = rc

    native_payload = {
        "channel_id": channel_id,
        "sender_id": sender_id,
        "content_parts": content_parts,
        "meta": meta,
    }

    if isinstance(request_data, AgentRequest):
        mso = getattr(request_data, "model_slot_override", None)
    else:
        mso = request_data.get("model_slot_override")
    if mso is not None:
        native_payload["model_slot_override"] = mso

    return native_payload


def _tail_text_file(
    path: Path,
    *,
    lines: int = 200,
    max_bytes: int = 512 * 1024,
) -> str:
    """Read the last N lines from a text file with bounded memory."""
    path = Path(path)
    if not path.exists() or not path.is_file():
        return ""
    try:
        size = path.stat().st_size
        if size == 0:
            return ""
        with open(path, "rb") as f:
            if size <= max_bytes:
                data = f.read()
            else:
                f.seek(max(size - max_bytes, 0))
                data = f.read()
        text = data.decode("utf-8", errors="replace")
        return "\n".join(text.splitlines()[-lines:])
    except Exception:
        logger.exception("Failed to read backend debug log file")
        return ""


@router.post(
    "/chat",
    status_code=200,
    summary="Chat with console (streaming response)",
    description="Agent API Request Format. See runtime.agentscope.io. "
    "Use body.reconnect=true to attach to a running stream.",
)
async def post_console_chat(
    request_data: Union[AgentRequest, dict],
    request: Request,
) -> StreamingResponse:
    """Stream agent response. Run continues in background after disconnect.
    Stop via POST /console/chat/stop. Reconnect with body.reconnect=true.
    """
    workspace = await get_agent_for_request(request)
    console_channel = await workspace.channel_manager.get_channel("console")
    if console_channel is None:
        raise HTTPException(
            status_code=503,
            detail="Channel Console not found",
        )
    try:
        native_payload = _extract_session_and_payload(request_data)
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e)) from e
    session_id = console_channel.resolve_session_id(
        sender_id=native_payload["sender_id"],
        channel_meta=native_payload["meta"],
    )
    name, first_text = _extract_placeholder_name(
        native_payload["content_parts"],
    )
    chat = await workspace.chat_manager.get_or_create_chat(
        session_id,
        native_payload["sender_id"],
        native_payload["channel_id"],
        name=name,
    )
    tracker = workspace.task_tracker

    # Kick off an LLM-backed title generation in the background when the chat
    # was just created with the truncated placeholder. This runs detached so
    # the streaming response is never blocked by title generation latency.
    if first_text and chat.name == name:
        asyncio.create_task(
            generate_and_update_title(
                workspace=workspace,
                chat_id=chat.id,
                user_message=first_text,
                placeholder_name=name,
            ),
        )

    is_reconnect = False
    if isinstance(request_data, dict):
        is_reconnect = request_data.get("reconnect") is True

    if is_reconnect:
        queue = await tracker.attach(chat.id)
        if queue is None:
            return
    else:
        queue, _ = await tracker.attach_or_start(
            chat.id,
            native_payload,
            console_channel.stream_one,
        )

    async def event_generator() -> AsyncGenerator[str, None]:
        # Hold iterator so finally can aclose(); guarantees stream_from_queue's
        # finally (detach_subscriber) on client abort / generator teardown.
        stream_it = tracker.stream_from_queue(queue, chat.id)
        try:
            try:
                async for event_data in stream_it:
                    yield event_data
            except Exception as e:
                logger.exception("Console chat stream error")
                yield f"data: {json.dumps({'error': str(e)})}\n\n"
        finally:
            await stream_it.aclose()

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
        },
    )


@router.post(
    "/chat/stop",
    status_code=200,
    summary="Stop running console chat",
)
async def post_console_chat_stop(
    request: Request,
    chat_id: str = Query(..., description="Chat id (ChatSpec.id) to stop"),
) -> dict:
    """Stop the running chat. Only stops when called."""
    logger.debug("[STOP API] Received stop request for chat_id=%s", chat_id)
    workspace = await get_agent_for_request(request)

    # Try to stop with the provided chat_id first
    logger.debug(
        "[STOP API] Got workspace, calling task_tracker.request_stop...",
    )
    stopped = await workspace.task_tracker.request_stop(chat_id)

    # If not found, the chat_id might be a session_id (timestamp)
    # Try to resolve it to the actual chat UUID
    if not stopped:
        logger.debug(
            "[STOP API] chat_id not found in tracker, trying to resolve "
            "from session_id...",
        )
        chat_manager = workspace.chat_manager
        if chat_manager:
            resolved_chat_id = await chat_manager.get_chat_id_by_session(
                session_id=chat_id,
                channel="console",
            )
            if resolved_chat_id:
                logger.debug(
                    "[STOP API] Resolved session_id=%s to chat_id=%s",
                    chat_id[:12] if len(chat_id) >= 12 else chat_id,
                    resolved_chat_id,
                )
                stopped = await workspace.task_tracker.request_stop(
                    resolved_chat_id,
                )

    logger.debug(
        "[STOP API] task_tracker.request_stop returned: stopped=%s",
        stopped,
    )
    return {"stopped": stopped}


@router.post("/upload", response_model=dict, summary="Upload file for chat")
async def post_console_upload(
    request: Request,
    file: UploadFile = File(..., description="File to attach"),
) -> dict:
    """Save to console channel media_dir."""

    workspace = await get_agent_for_request(request)
    console_channel = await workspace.channel_manager.get_channel("console")
    if console_channel is None:
        raise HTTPException(
            status_code=503,
            detail="Channel Console not found",
        )
    media_dir = console_channel.media_dir
    media_dir.mkdir(parents=True, exist_ok=True)
    data = await file.read()
    check_upload_size(data)
    safe_name = _safe_filename(file.filename or "file")
    stored_name = f"{uuid.uuid4().hex}_{safe_name}"

    path = (media_dir / stored_name).resolve()
    path.write_bytes(data)
    return {
        "url": path,
        "file_name": safe_name,
        "size": len(data),
    }


@router.get(
    "/debug/backend-logs",
    response_model=dict,
    summary="Read backend daemon logs for debug page",
)
async def get_backend_debug_logs(
    lines: int = Query(
        200,
        ge=20,
        le=MAX_DEBUG_LOG_LINES,
        description="Number of trailing log lines to return",
    ),
) -> dict:
    """Return the tail of the project log file for the debug UI."""
    log_path = LOG_FILE_PATH.resolve()
    try:
        st = log_path.stat()
        return {
            "path": str(log_path),
            "exists": True,
            "lines": lines,
            "updated_at": st.st_mtime,
            "size": st.st_size,
            "content": _tail_text_file(log_path, lines=lines),
        }
    except FileNotFoundError:
        return {
            "path": str(log_path),
            "exists": False,
            "lines": lines,
            "updated_at": None,
            "size": 0,
            "content": "",
        }


@router.get("/push-messages")
async def get_push_messages(
    session_id: str | None = Query(None, description="Optional session id"),
):
    """
    Return pending push messages and ALL approval requests.

    Messages:
    - With session_id: consumed messages for that session
    - Without session_id: recent messages (all sessions, last 60s)

    Approvals:
    - Always returns ALL pending approvals across all sessions
    - Frontend filters by current session_id for display
    - Includes session_id in each approval for filtering
    """
    from ..console_push_store import get_recent, take
    from ..approvals import get_approval_service

    # Get messages (session-specific or global)
    if session_id:
        messages = await take(session_id)
    else:
        messages = await get_recent()

    # Get ALL pending approvals (not filtered by session)
    approval_svc = get_approval_service()
    # pylint: disable=protected-access
    async with approval_svc._lock:
        all_pending = list(approval_svc._pending.values())

    # Serialize approval data with root_session_id for frontend filtering
    approvals_data = [
        {
            "request_id": p.request_id,
            "session_id": p.session_id,
            "root_session_id": p.root_session_id,
            "owner_agent_id": p.owner_agent_id,
            "agent_id": p.agent_id,
            "tool_name": p.tool_name,
            **approval_display_fields(p),
            "severity": p.severity,
            "findings_count": p.findings_count,
            "findings_summary": p.result_summary,
            "tool_params": p.extra.get("tool_call", {}).get("input", {}),
            "source_type": p.extra.get("source_type", "tool_guard"),
            "driver": p.extra.get("driver"),
            "created_at": p.created_at,
            "timeout_seconds": p.timeout_seconds,
        }
        for p in all_pending
    ]

    return {"messages": messages, "pending_approvals": approvals_data}


@router.get("/inbox/events")
async def get_inbox_events(
    limit: int = Query(50, ge=1, le=500),
    offset: int = Query(0, ge=0),
    source_type: str | None = Query(None),
    status: str | None = Query(None),
    agent_id: str | None = Query(None),
    unread_only: bool = Query(False),
):
    from ..inbox_store import list_events

    events = await list_events(
        limit=limit,
        offset=offset,
        source_type=source_type,
        status=status,
        agent_id=agent_id,
        unread_only=unread_only,
    )
    return {"events": events}


@router.post("/inbox/read")
async def post_mark_inbox_read(payload: MarkInboxReadRequest):
    from ..inbox_store import mark_all_read, mark_read

    if payload.all:
        updated = await mark_all_read()
    else:
        updated = await mark_read(payload.event_ids)
    return {"updated": updated}


@router.delete("/inbox/events/{event_id}")
async def delete_inbox_event(event_id: str):
    from ..inbox_store import delete_event
    from ..inbox_trace_store import delete_trace

    deleted, run_id, run_id_still_referenced = await delete_event(event_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="event not found")
    trace_deleted = False
    if run_id and not run_id_still_referenced:
        trace_deleted = await delete_trace(run_id)
    return {
        "deleted": True,
        "trace_deleted": trace_deleted,
        "run_id": run_id,
    }


@router.get("/inbox/traces/{run_id}")
async def get_inbox_trace(run_id: str):
    from ..inbox_trace_store import get_trace

    trace = await get_trace(run_id)
    if trace is None:
        raise HTTPException(
            status_code=404,
            detail="trace not found",
        )
    return trace


# ── Background chat task endpoints ──


def _parse_sse_payload(line: str) -> Optional[Dict[str, Any]]:
    """Parse a single SSE data line into a dict."""
    stripped = line.strip()
    if stripped.startswith("data: "):
        try:
            return json.loads(stripped[6:])
        except (json.JSONDecodeError, ValueError):
            return None
    return None


@router.post(
    "/chat/task",
    status_code=200,
    summary="Submit a background chat task",
)
async def post_console_chat_task(  # pylint: disable=too-many-statements
    request_data: Union[AgentRequest, dict],
    request: Request,
) -> dict:
    """Run an agent chat as a background task.

    Returns a ``task_id`` immediately. Poll status via
    ``GET /console/chat/task/{task_id}``.
    """
    workspace = await get_agent_for_request(request)
    console_channel = await workspace.channel_manager.get_channel("console")
    if console_channel is None:
        raise HTTPException(
            status_code=503,
            detail="Channel Console not found",
        )

    task_id = f"task-{uuid.uuid4().hex[:12]}"
    native_payload = _extract_session_and_payload(request_data)
    session_id = console_channel.resolve_session_id(
        sender_id=native_payload["sender_id"],
        channel_meta=native_payload["meta"],
    )
    name, _ = _extract_placeholder_name(native_payload["content_parts"])
    await workspace.chat_manager.get_or_create_chat(
        session_id,
        native_payload["sender_id"],
        native_payload["channel_id"],
        name=name,
    )

    task_timeout: Optional[float] = None
    fork_project_dir = ""
    fork_worktree_branch = ""
    fork_scope_id = ""
    if isinstance(request_data, dict):
        task_timeout = request_data.get("timeout")
        rc = request_data.get("request_context")
        if isinstance(rc, dict):
            fork_project_dir = str(rc.get("fork_project_dir") or "")
            fork_worktree_branch = str(
                rc.get("fork_worktree_branch") or "",
            )
            fork_scope_id = str(rc.get("fork_scope_id") or "")
    elif hasattr(request_data, "timeout"):
        task_timeout = getattr(request_data, "timeout", None)
        rc = getattr(request_data, "request_context", None)
        if isinstance(rc, dict):
            fork_project_dir = str(rc.get("fork_project_dir") or "")
            fork_worktree_branch = str(
                rc.get("fork_worktree_branch") or "",
            )
            fork_scope_id = str(rc.get("fork_scope_id") or "")

    bg = _BackgroundTask(
        status="running",
        started_at=time.time(),
    )

    async def _run() -> None:
        last_response: Optional[Dict[str, Any]] = None
        try:
            async for sse_line in console_channel.stream_one(
                native_payload,
            ):
                parsed = _parse_sse_payload(sse_line)
                if parsed and parsed.get("type") != "turn_usage":
                    last_response = parsed
        except asyncio.CancelledError:
            bg.status = "finished"
            bg.finished_at = time.time()
            bg.result = {
                "status": "failed",
                "error": {"message": "Task cancelled"},
            }
            if fork_project_dir and fork_worktree_branch:
                try:
                    from qwenpaw.agents.fork_project import mark_fork_failed

                    await asyncio.to_thread(
                        mark_fork_failed,
                        fork_project_dir,
                        fork_worktree_branch,
                        reason="Task cancelled",
                        expected_scope=fork_scope_id or None,
                    )
                except Exception:
                    logger.warning(
                        "mark_fork_failed on cancel failed for %s",
                        sanitize_log_value(fork_worktree_branch),
                        exc_info=True,
                    )
            return
        except Exception as exc:
            bg.status = "finished"
            bg.finished_at = time.time()
            bg.result = {
                "status": "failed",
                "error": {"message": str(exc)},
            }
            if fork_project_dir and fork_worktree_branch:
                try:
                    from qwenpaw.agents.fork_project import mark_fork_failed

                    await asyncio.to_thread(
                        mark_fork_failed,
                        fork_project_dir,
                        fork_worktree_branch,
                        reason=str(exc),
                        expected_scope=fork_scope_id or None,
                    )
                except Exception:
                    logger.warning(
                        "mark_fork_failed on error failed for %s",
                        sanitize_log_value(fork_worktree_branch),
                        exc_info=True,
                    )
            return

        bg.status = "finished"
        bg.finished_at = time.time()
        if last_response is not None:
            bg.result = {
                "status": "completed",
                "session_id": session_id,
                **last_response,
            }
        else:
            bg.result = {
                "status": "completed",
                "session_id": session_id,
                "output": [],
            }
        # Fork subagents: commit dirty worktree so branch tips are mergeable.
        if fork_project_dir and fork_worktree_branch:
            try:
                from qwenpaw.agents.fork_project import (
                    finalize_fork_worktree_or_fail,
                )

                await asyncio.to_thread(
                    finalize_fork_worktree_or_fail,
                    fork_project_dir,
                    fork_worktree_branch,
                    message=f"fork worker {fork_worktree_branch}",
                    expected_scope=fork_scope_id or None,
                )
            except Exception:
                logger.warning(
                    "Background fork finalize failed for %s (%s)",
                    sanitize_log_value(fork_worktree_branch),
                    sanitize_log_value(fork_project_dir),
                    exc_info=True,
                )

    atask = asyncio.create_task(_run())
    bg.asyncio_task = atask

    if task_timeout is not None and task_timeout > 0:

        async def _timeout_guard() -> None:
            await asyncio.sleep(task_timeout)
            if not atask.done():
                atask.cancel()

        asyncio.create_task(_timeout_guard())

    async with _bg_lock:
        _bg_tasks[task_id] = bg

    return {"task_id": task_id}


@router.get(
    "/chat/task/{task_id}",
    status_code=200,
    summary="Check background chat task status",
)
async def get_console_chat_task(task_id: str) -> dict:
    """Return the current status of a background chat task."""
    async with _bg_lock:
        bg = _bg_tasks.get(task_id)
    if bg is None:
        raise HTTPException(
            status_code=404,
            detail=f"Task not found: {task_id}",
        )
    response: Dict[str, Any] = {"status": bg.status}
    if bg.started_at is not None:
        response["started_at"] = bg.started_at
    if bg.status == "finished" and bg.result is not None:
        response["result"] = bg.result
    return response
