"""Controlled DeerFlow runtime invocation client for MAF roundtable.

All invocations require an explicit ``ExecutionPrincipal`` — the server-generated
identity derived ONLY from the authenticated MAF session owner. No raw tokens,
cookies, or credentials ever pass into events, checkpoints, or prompts.

Two invocation modes (the adapter uses mode 2):
1. Direct checkpointer access for thread creation and message reads.
2. In-process ``start_run`` via a typed internal request that carries the
   principal's user identity (never ``user=None``).

Security invariants:
- ExecutionPrincipal is constructed ONLY from the MAF session's user_id.
- The internal request object explicitly carries the principal's user so
  downstream code sees a real identity, not a None placeholder.
- Agent access is enforced server-side (enforce_access=True).
"""

from __future__ import annotations

import logging
import uuid
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import Any

from langgraph.checkpoint.base import empty_checkpoint

from deerflow.runtime.user_context import reset_current_user, set_current_user

logger = logging.getLogger(__name__)


def _serialize_tool_result(result: Any) -> Any:
    """Convert a tool result to a JSON-safe dict/list/str/int/float/bool/None."""
    if result is None:
        return None
    if isinstance(result, (str, int, float, bool)):
        return result
    if isinstance(result, (list, tuple)):
        return [_serialize_tool_result(v) for v in result]
    if isinstance(result, dict):
        return {str(k): _serialize_tool_result(v) for k, v in result.items()}
    return str(result)[:1000]


@dataclass(frozen=True)
class ExecutionPrincipal:
    """Server-generated execution identity for a MAF seat run.

    Constructed ONLY from the authenticated MAF session owner. Never from
    request body, query params, or client-supplied data.
    """

    user_id: str
    session_id: str
    run_id: str

    def __post_init__(self) -> None:
        if not self.user_id or not self.user_id.strip():
            raise ValueError("ExecutionPrincipal requires a non-empty user_id")


class _PrincipalUser:
    """Minimal CurrentUser protocol implementation backed by ExecutionPrincipal."""

    def __init__(self, principal: ExecutionPrincipal) -> None:
        self.id = principal.user_id


class _InternalState:
    """Typed state container for the internal run request.

    Carries the principal's user identity explicitly — never None — so
    downstream access checks and run metadata attribution work correctly.
    """

    def __init__(self, principal: ExecutionPrincipal) -> None:
        self.user = _PrincipalUser(principal)
        self.auth = None
        # This marker is set only on the server-created request used by the
        # MAF dispatcher.  Gateway code uses it to distinguish trusted MAF
        # execution context from user-supplied run configuration.
        self.maf_internal = True


@dataclass
class _InternalRunRequest:
    """Typed internal request for in-process agent run invocation.

    Replaces the former ``SimpleNamespace(app=..., state=SimpleNamespace(user=None))``
    pattern. The state explicitly carries the principal's user identity.
    """

    app: Any
    state: _InternalState

    @classmethod
    def from_principal(cls, app: Any, principal: ExecutionPrincipal) -> _InternalRunRequest:
        return cls(app=app, state=_InternalState(principal))


class DeerFlowRuntimeClient:
    """Controlled entry point for invoking the DeerFlow Agent Runtime from MAF.

    All calls require an ExecutionPrincipal. The client:
    - Sets the user ContextVar before each invocation.
    - Creates threads with proper user attribution.
    - Invokes start_run with server-generated identity.
    - Never passes raw tokens or credentials downstream.
    """

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

    @property
    def app(self) -> Any:
        return self._app

    async def create_thread(self, principal: ExecutionPrincipal) -> str:
        """Create a DeerFlow thread owned by the principal's user.

        The user ContextVar is set so thread_store.create records the correct
        user_id in thread_meta.
        """
        token = set_current_user(_PrincipalUser(principal))
        try:
            checkpointer = self._app.state.checkpointer
            thread_store = self._app.state.thread_store
            thread_id = str(uuid.uuid4())
            await thread_store.create(
                thread_id,
                assistant_id=None,
                metadata={
                    "thread_type": "maf_roundtable",
                    "system": True,
                    "maf_session_id": principal.session_id,
                    "maf_run_id": principal.run_id,
                },
            )
            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
        finally:
            reset_current_user(token)

    async def start_agent_run(
        self,
        principal: ExecutionPrincipal,
        *,
        thread_id: str,
        agent_id: str,
        message: str,
        model_name: str | None = None,
        reasoning_effort: str | None = None,
        thinking_enabled: bool = True,
        subagent_enabled: bool = False,
        excluded_tools: list[str] | None = None,
        memory_recall_disabled: bool | None = None,
        memory_injection_enabled: bool | None = None,
        turn_index: int = 0,
        capability_snapshot: dict[str, Any] | None = None,
        write_policy: str = "confirm_high_risk",
        approved_tool: str | None = None,
        approved_action_id: str | None = None,
    ) -> Any:
        """Start a DeerFlow agent run under the principal's identity.

        Returns the RunRecord (with .task and .run_id). The caller must await
        the task and handle exceptions.

        WriteGuard activation: when ``write_policy != "auto"`` the run context
        carries ``maf_run_id`` (the activation flag the WriteGuard middleware
        checks) plus the per-run data the interceptor needs to record a
        pending action. ``excluded_tools`` here MUST only contain tools the
        user explicitly disabled — write tools are no longer excluded (the
        middleware intercepts them instead, keeping them visible to the model).
        """
        from app.gateway.routers.thread_runs import RunCreateRequest
        from app.gateway.services import RUN_SCOPE_BACKGROUND_CHILD, start_run

        context: dict[str, Any] = {
            "agent_name": agent_id,
            "model_name": model_name or "",
            "mode": "pro",
            "reasoning_effort": reasoning_effort or "medium",
            "thinking_enabled": thinking_enabled,
            "is_plan_mode": False,
            "subagent_enabled": subagent_enabled,
            "thread_id": thread_id,
        }
        if memory_recall_disabled is not None:
            context["memory_recall_disabled"] = memory_recall_disabled
        if memory_injection_enabled is not None:
            context["memory_injection_enabled"] = memory_injection_enabled

        # WriteGuard activation + per-run data (server-generated identity only).
        # ``maf_run_id`` is the flag the WriteGuard middleware checks to decide
        # whether to intercept; the rest is the data the interceptor reads.
        context["maf_run_id"] = principal.run_id
        context["maf_session_id"] = principal.session_id
        context["maf_user_id"] = principal.user_id
        context["agent_id"] = agent_id
        context["turn_index"] = turn_index
        context["write_policy"] = write_policy
        if capability_snapshot is not None:
            context["capability_snapshot"] = capability_snapshot
        # Post-approval deterministic execution: when set, the WriteGuard
        # middleware passes this single tool through without interception.
        # ``maf_approved_action_id`` binds the passthrough to a specific
        # approved action — the middleware validates both tool name AND
        # action_id so a model calling another tool with the same name is
        # still intercepted (defense-in-depth when combined with direct
        # tool execution that bypasses the LLM).
        if approved_tool:
            context["maf_approved_tool"] = approved_tool
        if approved_action_id:
            context["maf_approved_action_id"] = approved_action_id

        body = RunCreateRequest(
            assistant_id="lead_agent",
            input={"messages": [{"type": "human", "content": [{"type": "text", "text": message}], "additional_kwargs": {}}]},
            metadata={"run_scope": RUN_SCOPE_BACKGROUND_CHILD, "maf_run_id": principal.run_id},
            config={"recursion_limit": 1000},
            context=context,
            excluded_tools=excluded_tools or [],
            on_disconnect="continue",
            multitask_strategy="reject",
        )
        # Typed internal request carrying the principal's identity — never user=None.
        request = _InternalRunRequest.from_principal(self._app, principal)

        token = set_current_user(_PrincipalUser(principal))
        try:
            record = await start_run(
                body,
                thread_id,
                request,
                run_scope=RUN_SCOPE_BACKGROUND_CHILD,
            )
            return record
        finally:
            reset_current_user(token)

    async def read_thread_messages(self, principal: ExecutionPrincipal, thread_id: str) -> list[dict[str, Any]]:
        """Read the latest checkpoint messages from a thread."""
        from deerflow.runtime.serialization import serialize_channel_values

        token = set_current_user(_PrincipalUser(principal))
        try:
            checkpointer = self._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 []
        finally:
            reset_current_user(token)

    async def execute_tool_directly(
        self,
        *,
        thread_id: str,
        agent_id: str,
        tool_name: str,
        params: dict[str, Any],
        principal: ExecutionPrincipal,
        turn_index: int,
        capability_snapshot: dict[str, Any] | None,
        write_policy: str,
        approved_action_id: str,
        action_record: dict[str, Any] | None,
        model_name: str | None = None,
        excluded_tools: list[str] | None = None,
    ) -> dict[str, Any]:
        """Execute a DeerFlow tool directly without LLM involvement.

        For post-approval deterministic execution this delegates to the harness
        ``execute_approved_tool``, which runs the call through the **real**
        LangGraph ``ToolNode`` / ``ToolRuntime`` / middleware chain (state
        recovered from the thread checkpoint), not a faked runtime. No raw
        params enter any prompt, checkpoint, or model context.

        Returns ``{"status": "succeeded", "result": ...}`` or
        ``{"status": "failed", "error": "..."}``. A security rejection
        (unauthorized / disabled / nonexistent tool, context mismatch) is a
        ``failed`` outcome and the tool executes 0 times.
        """
        from deerflow.config import get_app_config
        from deerflow.tools.approved_tool_executor import execute_approved_tool

        app_config = get_app_config()
        checkpointer = getattr(self._app.state, "checkpointer", None)
        store = getattr(self._app.state, "store", None)

        return await execute_approved_tool(
            app_config=app_config,
            checkpointer=checkpointer,
            store=store,
            thread_id=thread_id,
            user_id=principal.user_id,
            session_id=principal.session_id,
            run_id=principal.run_id,
            agent_id=agent_id,
            turn_index=turn_index,
            model_name=model_name,
            excluded_tools=excluded_tools,
            capability_snapshot=capability_snapshot,
            write_policy=write_policy,
            tool_name=tool_name,
            params=params,
            approved_action_id=approved_action_id,
            action_record=action_record,
        )

    async def cancel_run(self, deerflow_run_id: str) -> bool:
        """Cancel a DeerFlow run by its run_id."""
        run_manager = getattr(self._app.state, "run_manager", None)
        if run_manager is None:
            return False
        try:
            await run_manager.cancel(deerflow_run_id, action="interrupt")
            return True
        except Exception as exc:  # noqa: BLE001
            # Log only the exception type — never the raw message/traceback,
            # which may carry params or credentials.
            logger.warning(
                "cancel DeerFlow run %s failed (%s)", deerflow_run_id, type(exc).__name__
            )
            return False