"""Runs a DeerFlow agent for one workflow ``agent`` / ``skill`` node.

This is the single place where the workflow runtime touches LangGraph. We build
the lead agent with :func:`make_lead_agent` (so the node inherits the same
middleware chain, SOUL, skills and model resolution as a normal chat turn) and
stream ``stream_mode=["values", "messages"]`` so the node can forward token
deltas into the workflow event log.

Each node gets its own thread id (``wf-{run_id}-{node_id}``) unless the node
asks for ``shared_run_thread``, which keeps the whole run in one conversation.
"""

from __future__ import annotations

import json
import logging
from collections.abc import Awaitable, Callable
from typing import Any

from langchain_core.messages import AIMessage, ToolMessage

from deerflow.workflows.errors import WorkflowError

logger = logging.getLogger(__name__)


def _extract_text(content: Any) -> str:
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts: list[str] = []
        for block in content:
            if isinstance(block, str):
                parts.append(block)
            elif isinstance(block, dict) and block.get("type") == "text":
                parts.append(str(block.get("text") or ""))
        return "".join(parts)
    return ""


def _extract_reasoning(message: AIMessage) -> str:
    """Read provider reasoning that LangChain keeps outside ``content``.

    Patched OpenAI/vLLM providers preserve the streaming field under
    ``additional_kwargs.reasoning_content``.  Some compatible providers use
    the shorter ``reasoning`` spelling, so accept both without making either
    field part of the workflow's executable output.
    """

    additional_kwargs = getattr(message, "additional_kwargs", None)
    if not isinstance(additional_kwargs, dict):
        return ""
    for key in ("reasoning_content", "reasoning"):
        value = additional_kwargs.get(key)
        text = _extract_text(value)
        if text:
            return text
        if isinstance(value, str) and value:
            return value
    return ""


def _thread_id(run_id: str, node_id: str, thread_mode: str) -> str:
    if thread_mode == "shared_run_thread":
        return f"wf-{run_id}"
    return f"wf-{run_id}-{node_id}"


def _native_message_frame(chunk: Any) -> list[Any]:
    """Keep the DeerFlow/LangGraph ``messages`` tuple intact for the client.

    The workflow event envelope adds run/node identity, but the nested value is
    deliberately the same two-item shape consumed by DeerFlow's chat UI:
    ``[message, metadata]``.  Do not project tool args, result content, or
    ``additional_kwargs`` here; those fields are what power rich tool and user
    assistance cards after a live stream reconnects.
    """

    def encode(value: Any) -> Any:
        if hasattr(value, "model_dump"):
            return encode(value.model_dump(mode="json"))
        if isinstance(value, dict):
            return {str(key): encode(item) for key, item in value.items()}
        if isinstance(value, (list, tuple)):
            return [encode(item) for item in value]
        # Force a JSON-compatible copy without discarding or truncating any
        # field. LangGraph metadata is normally already JSON-native; ``str``
        # is only a serialization fallback for provider-specific objects.
        try:
            json.dumps(value, ensure_ascii=False)
        except (TypeError, ValueError):
            return str(value)
        return value

    return encode(chunk if isinstance(chunk, tuple) else (chunk, {}))


def _resume_message(payload: dict[str, Any]) -> str:
    """Represent a workflow resume as the next user turn in the same thread."""
    return (
        "用户已回复你刚才的协助请求。请基于已有对话继续当前任务，不要重复询问已经回答的信息。\n"
        "<workflow-user-assistance-response>\n"
        f"{json.dumps(payload, ensure_ascii=False, separators=(',', ':'), default=str)}\n"
        "</workflow-user-assistance-response>"
    )


def _clarification_pending(message: ToolMessage) -> dict[str, Any] | None:
    """Identify DeerFlow's native ``ask_clarification`` tool result.

    The complete card payload continues to live in the untouched ToolMessage
    sent through ``node.message``.  The pause descriptor only carries the
    stable tool-call identity required to reconnect it to this node run.
    """
    if str(getattr(message, "name", "") or "") != "ask_clarification":
        return None
    additional_kwargs = getattr(message, "additional_kwargs", None)
    clarification = (
        additional_kwargs.get("clarification")
        if isinstance(additional_kwargs, dict)
        else None
    )
    if not isinstance(clarification, dict):
        return None
    call_id = str(getattr(message, "tool_call_id", "") or getattr(message, "id", "") or "")
    question = str(clarification.get("question") or "")
    return {
        "toolCallId": call_id,
        "prompt": question or "智能体需要你补充信息后才能继续。",
        "actions": ["submit"],
    }


class WorkflowAgentRunner:
    def __init__(self, app: Any) -> None:
        self._app = app

    async def run_agent(
        self,
        *,
        run_id: str,
        node_id: str,
        owner_id: str,
        agent_id: str,
        prompt: str,
        skill_names: list[str] | None = None,
        thread_mode: str = "isolated_per_node",
        cancel: Any = None,
        on_delta: Callable[[str], Awaitable[None]] | None = None,
        on_tool: Callable[[dict[str, Any]], Awaitable[None]] | None = None,
        on_message: Callable[[list[Any]], Awaitable[None]] | None = None,
        resume_payload: dict[str, Any] | None = None,
        model_name: str | None = None,
        disable_tools: bool = False,
        thinking_enabled: bool | None = None,
        force_disable_thinking: bool = False,
    ) -> dict[str, Any]:
        from deerflow.agents.lead_agent.agent import make_lead_agent
        from deerflow.config.app_config import get_app_config

        thread_id = _thread_id(run_id, node_id, thread_mode)
        configurable: dict[str, Any] = {
            "thread_id": thread_id,
            "agent_id": agent_id,
            "agent_name": agent_id,
            "user_id": owner_id,
            "model_name": model_name or "",
            # Workflow nodes are non-interactive: no plan mode, no clarification.
            "is_plan_mode": False,
            # Control-plane callers (such as candidate planning) may need a
            # model's reasoning but must not expose any callable capability.
            "disable_tools": disable_tools,
            "thinking_force_disabled": force_disable_thinking,
            "workflow_run_id": run_id,
            "workflow_node_id": node_id,
        }
        if thinking_enabled is not None:
            configurable["thinking_enabled"] = thinking_enabled
        if skill_names:
            configurable["skill_names"] = list(skill_names)
        # Tool-using research agents need more than the old chat-sized 60
        # super-steps (each model → tool exchange is roughly two steps).
        config = {
            "configurable": configurable,
            "recursion_limit": get_app_config().workflows.agent_recursion_limit,
        }

        try:
            agent = make_lead_agent(config)
        except Exception as exc:  # noqa: BLE001
            logger.warning("failed to build agent %s for workflow node %s", agent_id, node_id, exc_info=True)
            raise WorkflowError(
                "WORKFLOW_RESOURCE_MISSING",
                "引用的智能体不存在或配置无效",
                node_id=node_id,
                details={"agentId": agent_id},
            ) from exc

        checkpointer = getattr(self._app.state, "checkpointer", None)
        if checkpointer is not None:
            agent.checkpointer = checkpointer
        store = getattr(self._app.state, "store", None)
        if store is not None:
            agent.store = store

        seen_tool_calls: set[str] = set()
        finished_tool_calls: set[str] = set()
        tool_names: dict[str, str] = {}
        final_state: dict[str, Any] | None = None
        streamed: dict[str, list[str]] = {}
        last_message_id = ""
        open_reasoning_message_ids: set[str] = set()

        async def emit_display_delta(text: str) -> None:
            if on_delta is not None and text:
                await on_delta(text)

        async def close_reasoning(*, except_message_id: str | None = None) -> None:
            """Keep thinking blocks well-formed across tool/model turns."""

            for message_id in tuple(open_reasoning_message_ids):
                if message_id == except_message_id:
                    continue
                open_reasoning_message_ids.remove(message_id)
                await emit_display_delta("</think>")

        initial_messages = [
            {
                "role": "user",
                "content": _resume_message(resume_payload)
                if resume_payload is not None
                else prompt,
            }
        ]
        awaiting_input: dict[str, Any] | None = None

        async for mode, chunk in agent.astream(
            {"messages": initial_messages},
            config=config,
            stream_mode=["values", "messages"],
        ):
            if cancel is not None and getattr(cancel, "cancelled", False):
                raise WorkflowError("WORKFLOW_CANCELLED", "运行已被取消", node_id=node_id)
            if mode == "values":
                if isinstance(chunk, dict):
                    final_state = chunk
                continue
            if mode != "messages":
                continue
            if on_message is not None:
                await on_message(_native_message_frame(chunk))
            message = chunk[0] if isinstance(chunk, tuple) and chunk else chunk
            if isinstance(message, ToolMessage):
                # Tool result for an earlier call: close its step in the event log.
                await close_reasoning()
                call_id = str(getattr(message, "tool_call_id", "") or "")
                if call_id and call_id not in finished_tool_calls:
                    finished_tool_calls.add(call_id)
                    if on_tool is not None:
                        await on_tool(
                            {
                                "phase": "finished",
                                "id": call_id,
                                "name": tool_names.get(call_id) or getattr(message, "name", "") or "",
                                "ok": getattr(message, "status", None) != "error",
                                "result": getattr(message, "content", ""),
                            }
                        )
                awaiting_input = _clarification_pending(message)
                if awaiting_input is not None:
                    break
                continue
            if not isinstance(message, AIMessage):
                continue
            message_id = str(getattr(message, "id", "") or "__workflow_reasoning__")
            reasoning = _extract_reasoning(message)
            if reasoning:
                if message_id not in open_reasoning_message_ids:
                    open_reasoning_message_ids.add(message_id)
                    await emit_display_delta("<think>")
                await emit_display_delta(reasoning)
            text = _extract_text(getattr(message, "content", ""))
            if text:
                await close_reasoning()
                streamed.setdefault(message_id, []).append(text)
                last_message_id = message_id
                await emit_display_delta(text)
            for call in getattr(message, "tool_calls", None) or []:
                call_id = str(call.get("id") or "")
                if not call_id or call_id in seen_tool_calls:
                    continue
                seen_tool_calls.add(call_id)
                tool_names[call_id] = str(call.get("name") or "")
                if on_tool is not None:
                    await on_tool(
                        {
                            "phase": "started",
                            "id": call_id,
                            "name": tool_names[call_id],
                            "args": call.get("args"),
                        }
                    )

            # A reasoning-only turn that delegates to a tool must end before
            # the tool result and next model turn arrive.  Otherwise a later
            # answer could be swallowed by one long unclosed `<think>` block.
            if getattr(message, "tool_calls", None):
                await close_reasoning()

        await close_reasoning()

        if awaiting_input is not None:
            return {
                "awaiting_input": awaiting_input,
                "thread_id": thread_id,
                "messages": [],
                "artifacts": self._artifacts(final_state),
                "usage": {},
            }

        final_text = "".join(streamed.get(last_message_id, ())) or self._final_text(final_state)
        if not final_text.strip():
            raise WorkflowError(
                "WORKFLOW_AGENT_FAILED",
                "智能体没有返回任何内容",
                retryable=True,
                node_id=node_id,
            )
        return {
            "text": final_text,
            "thread_id": thread_id,
            "messages": [],
            "artifacts": self._artifacts(final_state),
            "usage": {},
        }

    def _final_text(self, state: dict[str, Any] | None) -> str:
        for message in reversed((state or {}).get("messages") or []):
            if isinstance(message, AIMessage):
                text = _extract_text(getattr(message, "content", "")).strip()
                if text:
                    return text
        return ""

    def _artifacts(self, state: dict[str, Any] | None) -> list[dict[str, Any]]:
        raw = (state or {}).get("artifacts")
        if not isinstance(raw, list):
            return []
        out: list[dict[str, Any]] = []
        for item in raw:
            if isinstance(item, dict) and item.get("name"):
                out.append(
                    {
                        "name": str(item.get("name")),
                        "path": str(item.get("path") or ""),
                        "mime_type": str(item.get("mime_type") or item.get("mimeType") or ""),
                        "size_bytes": item.get("size_bytes") or item.get("sizeBytes"),
                        "preview": item.get("preview"),
                    }
                )
        return out


__all__ = ["WorkflowAgentRunner"]