"""圆桌真网关（Phase 3b）—— 进程内驱动真实 leader / seat / report agent run。

实现 ``RoundtableGateway`` 协议（定义在 harness 的 roundtable_orchestrator.types），
但**进程内**跑真模型：复用 ``services.start_run``（scheduled task 同款）驱动一轮 agent，
跑完从 checkpointer 读最终 messages，用 ``multi_agent`` 的纯解析函数解析派活 / 澄清。
不走 HTTP、不需要 Request（用 ``SimpleNamespace(app=app)`` 当虚拟 request，
``set_current_user`` 由执行器设好）。

本网关位于 app 层（依赖 app.state + multi_agent helpers），实现 harness 定义的协议——
不违反 harness→app 边界（harness 永不 import app；app import harness 合法）。

⚠️ 需在能起完整后端栈 + 真模型的环境验证（见 scripts/probe_roundtable_job.py）。
"""

from __future__ import annotations

import logging
import uuid
from datetime import UTC, datetime
from types import SimpleNamespace
from typing import Any

from langgraph.checkpoint.base import empty_checkpoint

from app.gateway.roundtable_run_policy import roundtable_run_policy
from app.gateway.routers._roundtable_seed import (
    COORDINATOR_AGENT_ID,
    REPORT_AGENT_ID,
    ensure_roundtable_functional_agents,
)
from deerflow.agents.roundtable_orchestrator.types import (
    LeaderTurn,
    ReportResult,
    SeatRef,
    SeatTurn,
)
from deerflow.config.paths import get_paths
from deerflow.runtime.serialization import serialize_channel_values

logger = logging.getLogger(__name__)

_ROUNDTABLE_THREAD_METADATA: dict[str, Any] = {"thread_type": "roundtable", "system": True}
_REPORT_FILE = "方案总览.html"

# 各角色的工具裁剪 / thinking / reasoning 全部取自单一数据源 roundtable_run_policy
# （与 multi_agent 的 _leader_run / _special_run 同源），不再在本文件硬编码——历史上
# 这里漏禁了 web_search，导致后台席位 / 报告能联网搜索而前端不能，现由 policy 统一。


# ── 进程内底层操作 ───────────────────────────────────────────────────────────


async def _create_thread(app) -> str:
    """同进程建一个带圆桌 metadata 的 thread + 空 checkpoint（镜像 _create_thread_direct）。"""
    checkpointer = app.state.checkpointer
    thread_store = app.state.thread_store
    thread_id = str(uuid.uuid4())
    await thread_store.create(thread_id, assistant_id=None, metadata=dict(_ROUNDTABLE_THREAD_METADATA))
    config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
    await checkpointer.aput(
        config,
        empty_checkpoint(),
        {"step": -1, "source": "input", "writes": None, "parents": {}, "created_at": datetime.now(UTC).isoformat()},
        {},
    )
    return thread_id


async def _read_thread_messages(app, thread_id: str) -> list[dict[str, Any]]:
    """读 thread 最新 checkpoint 的 messages（已 serialize 成 dict 列表）。"""
    checkpointer = app.state.checkpointer
    config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
    ct = await checkpointer.aget_tuple(config)
    if ct is None:
        return []
    checkpoint = getattr(ct, "checkpoint", {}) or {}
    values = serialize_channel_values(checkpoint.get("channel_values", {}) or {})
    return values.get("messages") or []


def _latest_ai_message(messages: list[dict[str, Any]]) -> dict[str, Any] | None:
    for m in reversed(messages):
        if not isinstance(m, dict):
            continue
        if (m.get("type") or m.get("role")) in {"ai", "assistant"}:
            return m
    return None


def _message_text(message: dict[str, Any] | None) -> str:
    if not message:
        return ""
    content = message.get("content")
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts = [p.get("text", "") for p in content if isinstance(p, dict) and p.get("type") == "text"]
        return "\n".join(parts)
    return ""


def _latest_ai_text(messages: list[dict[str, Any]]) -> str:
    """取**最后一条有正文**的 AI 消息文本。

    席位常以「调一堆工具」的 AIMessage 收尾（正文为空、只有 tool_calls，甚至 dangling），
    直接取最后一条 AI 消息会拿到空串 → 报告就拿不到席位产物。这里向前找到真正有文字的那条。
    """
    for m in reversed(messages):
        if not isinstance(m, dict):
            continue
        if (m.get("type") or m.get("role")) not in {"ai", "assistant"}:
            continue
        text = _message_text(m)
        if text.strip():
            return text
    return ""


def _normalize_tool_calls(tool_calls: Any) -> list[dict[str, Any]]:
    """把 checkpoint 里的 tool_calls 归一化成 {name, args, id}（丢掉无名项）。"""
    out: list[dict[str, Any]] = []
    for tc in tool_calls or []:
        if not isinstance(tc, dict):
            continue
        name = tc.get("name") or ""
        if not name:
            continue
        out.append({"name": name, "args": tc.get("args") or {}, "id": tc.get("id") or ""})
    return out


def _tool_calls_since_last_human(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """收集「最后一条 human 消息之后」各 AI 消息的 tool_calls（即**本轮**调用），保序。

    席位 thread 跨轮累积历史，直接扫全量会把往轮的工具调用也算进来；以最后一条 human
    （本轮 run 的输入）为界，只取其后的 AI 工具调用，对应前端「这一条气泡的步骤卡」。
    """
    last_human = -1
    for i, m in enumerate(messages):
        if isinstance(m, dict) and (m.get("type") or m.get("role")) in {"human", "user"}:
            last_human = i
    out: list[dict[str, Any]] = []
    for m in messages[last_human + 1 :]:
        if not isinstance(m, dict):
            continue
        if (m.get("type") or m.get("role")) not in {"ai", "assistant"}:
            continue
        out.extend(_normalize_tool_calls(m.get("tool_calls")))
    return out


async def _run_agent_turn(
    app,
    *,
    thread_id: str,
    agent_name: str,
    message: str,
    model: str | None,
    excluded_tools: list[str],
    skill_stop_names: list[str],
    thinking_enabled: bool,
    reasoning_effort: str,
) -> list[dict[str, Any]]:
    """进程内驱动 agent_name 在 thread_id 上跑一轮（发 message），阻塞到完成，返回最终 messages。

    镜像 multi_agent ``_run_payload`` 的 body 形状：agent 通过 ``context.agent_name`` 选，
    assistant_id 固定 lead_agent；关闭 memory recall / subagent / plan mode（临时研讨任务）。
    """
    from app.gateway.routers.thread_runs import RunCreateRequest
    from app.gateway.services import start_run

    body = RunCreateRequest(
        assistant_id="lead_agent",
        input={"messages": [{"type": "human", "content": [{"type": "text", "text": message}], "additional_kwargs": {}}]},
        config={"recursion_limit": 1000},
        context={
            "agent_name": agent_name,
            "model_name": model or "",
            "mode": "pro",
            "reasoning_effort": reasoning_effort,
            "thinking_enabled": thinking_enabled,
            "is_plan_mode": False,
            "subagent_enabled": False,
            "thread_id": thread_id,
            "memory_recall_disabled": True,
            # 与 multi_agent._run_payload 对齐:关 builtin 记忆注入,避免单例总控/席位
            # 被上轮会话 memory.json 带偏(臆造席位 / 直接干活)。详见该处注释。
            "memory_injection_enabled": False,
        },
        excluded_tools=excluded_tools,
        skill_stop_names=skill_stop_names,
        on_disconnect="continue",
        multitask_strategy="reject",
    )
    request = SimpleNamespace(app=app)
    record = await start_run(body, thread_id, request)  # type: ignore[arg-type]
    if getattr(record, "task", None) is not None:
        await record.task  # 阻塞到这轮 agent 跑完，checkpoint 已落盘
    return await _read_thread_messages(app, thread_id)


def _read_report_html(app, thread_id: str, user_id: str) -> str | None:
    """从 report thread 的 outputs 目录读出报告 HTML。

    优先 ``方案总览.html``；找不到就扫目录里**任意 .html**（取体积最大的那个），
    容忍报告 agent 用了别的文件名。
    """
    try:
        outputs_dir = get_paths().sandbox_outputs_dir(thread_id, user_id=user_id)
        preferred = outputs_dir / _REPORT_FILE
        if preferred.exists():
            return preferred.read_text(encoding="utf-8")
        if outputs_dir.exists():
            htmls = sorted(
                outputs_dir.glob("*.html"),
                key=lambda p: p.stat().st_size,
                reverse=True,
            )
            if htmls:
                logger.info("report html: 方案总览.html not found, using %s", htmls[0].name)
                return htmls[0].read_text(encoding="utf-8")
        logger.warning("report html: no .html found in %s", outputs_dir)
    except Exception:  # noqa: BLE001
        logger.warning("read report html failed for thread %s", thread_id, exc_info=True)
    return None


def _build_seat_intro(agents: list[SeatRef]) -> str:
    """镜像 init 注入 coordinator 的「可调度席位清单」。"""
    lines = [f"- agent_name: {a.agent_id}（{a.name}）" for a in agents]
    return (
        "【圆桌系统初始化】本次研讨可调度的席位清单如下，请严格按 agent_name 通过 "
        "agent_orchestration 技能派活，不要派给不在清单内的 agent：\n" + "\n".join(lines)
    )


# ── 网关实现 ─────────────────────────────────────────────────────────────────


class InProcessRoundtableGateway:
    """进程内真网关。executor 先调 ``init_threads`` 建好线程，再跑编排引擎。"""

    def __init__(self, app, *, user_id: str, agents: list[SeatRef], model: str | None = None) -> None:
        self._app = app
        self._user_id = user_id
        self._agents = list(agents)
        self._model = model
        self._seat_intro = ""
        self._first_leader = True

    async def init_threads(self) -> tuple[dict[str, str], str]:
        """建 coordinator + 各席位 thread，返回 (thread_ids, coordinator_name)。

        席位清单通过「首轮 leader 消息前缀」注入（coordinator thread 跨轮累积历史，
        首轮带上即长期可见），无需单独写 checkpoint。
        """
        ensure_roundtable_functional_agents()
        thread_ids: dict[str, str] = {}
        for a in self._agents:
            thread_ids[a.agent_id] = await _create_thread(self._app)
        thread_ids[COORDINATOR_AGENT_ID] = await _create_thread(self._app)
        self._seat_intro = _build_seat_intro(self._agents)
        return thread_ids, COORDINATOR_AGENT_ID

    async def run_leader(self, *, thread_ids, coordinator_name, message, model) -> LeaderTurn:
        msg = message
        if self._first_leader and self._seat_intro:
            msg = self._seat_intro + "\n\n" + message
            self._first_leader = False
        policy = roundtable_run_policy("leader")
        messages = await _run_agent_turn(
            self._app,
            thread_id=thread_ids[coordinator_name],
            agent_name=coordinator_name,
            message=msg,
            model=model or self._model,
            excluded_tools=policy.excluded_tools,
            skill_stop_names=policy.skill_stop_names,
            thinking_enabled=policy.thinking_enabled,
            reasoning_effort=policy.reasoning_effort,
        )
        ai = _latest_ai_message(messages)
        # tool_calls 取最后一条 AI 消息（派活就在这条）；正文取最后一条有文字的 AI 消息。
        final_text = _latest_ai_text(messages)
        tool_calls = (ai.get("tool_calls") if ai else None) or []
        # 本轮工具调用（派活 / 澄清），带进 LeaderTurn 供草稿 run 重建步骤卡。
        turn_tool_calls = _normalize_tool_calls(tool_calls)

        # 澄清优先
        clar = next((tc for tc in tool_calls if (tc.get("name") or "").lower() == "ask_clarification"), None)
        if clar is not None:
            cargs = clar.get("args") or {}
            question = str(cargs.get("question") or "").strip() or "总控需要补充信息，请补充后继续。"
            return LeaderTurn(final_text=final_text, clarification=question, tool_calls=turn_tool_calls)

        # 派活解析（复用 multi_agent 纯函数）
        from app.gateway.routers.multi_agent import _extract_dispatch

        dispatched: list[tuple[str, str]] = []
        for tc in tool_calls:
            try:
                name, task = _extract_dispatch(tc)
                if name and task:
                    dispatched.append((name, task))
            except Exception:  # noqa: BLE001
                logger.warning("dispatch parse failed for tool_call", exc_info=True)
        return LeaderTurn(final_text=final_text, dispatched=dispatched, tool_calls=turn_tool_calls)

    async def run_seat(self, *, thread_ids, agent_id, task, model) -> SeatTurn:
        policy = roundtable_run_policy("seat")
        messages = await _run_agent_turn(
            self._app,
            thread_id=thread_ids[agent_id],
            agent_name=agent_id,
            message=task,
            model=model or self._model,
            excluded_tools=policy.excluded_tools,
            skill_stop_names=policy.skill_stop_names,
            thinking_enabled=policy.thinking_enabled,
            reasoning_effort=policy.reasoning_effort,
        )
        return SeatTurn(final_text=_latest_ai_text(messages), tool_calls=_tool_calls_since_last_human(messages))

    async def create_report_thread(self) -> str:
        ensure_roundtable_functional_agents()
        return await _create_thread(self._app)

    async def run_report(self, *, report_thread_id, prompt, model) -> ReportResult:
        policy = roundtable_run_policy("report")
        messages = await _run_agent_turn(
            self._app,
            thread_id=report_thread_id,
            agent_name=REPORT_AGENT_ID,
            message=prompt,
            model=model or self._model,
            excluded_tools=policy.excluded_tools,
            skill_stop_names=policy.skill_stop_names,
            thinking_enabled=policy.thinking_enabled,
            reasoning_effort=policy.reasoning_effort,
        )
        summary = _latest_ai_text(messages)
        html = _read_report_html(self._app, report_thread_id, self._user_id) or summary
        return ReportResult(html=html, summary=summary, model=model or self._model)