from __future__ import annotations

import asyncio
import logging
import time
import uuid
from typing import Any, Dict, List, Optional

from app.services.talkback.types import (
    ExecuteTalkback,
    GetAskmeService,
    NodeStatus,
    ResolveAuthorizedAgentIds,
    RunInThreadpool,
    TalkbackState,
)
from app.services.talkback.utils import (
    _clone_ask_request,
    _coerce_route_handler_result_to_dict,
    _parse_df_input_light,
    _safe_str,
)
from app.services.talkback.meta import build_talkback_answer_meta
from app.services.talkback.routing import (
    _derive_request_id_for_direct,
    _print_node,
    _resolve_talkback_effective_session_key,
)
from app.services.talkback.nodes import (
    BaseNode,
    _BroadScopeProcessingWarningNode,
    _ClarificationShortCircuitNode,
    _ContextRelevanceNode,
    _ContextRewriteForRoutingNode,
    _DispatchDecisionNode,
    _ExecuteNode,
    _FileCommandDetectNode,
    _IntentRoutingNode,
    _LoadSessionNode,
    _NormalizeDatasourceNode,
    _PERSIST_TURN_NODE_NAME,
    _PersistTurnNode,
    _PreparePriorQuestionsNode,
    _PriorDfConsentNode,
    _ResolveAgentsNode,
    _TalkbackStop,
    _UQL_CACHE_NODE_NAME,
    _UserQuestionLogCacheHitNode,
)

logger = logging.getLogger(__name__)


async def _run_talkback_prior_df_search_web_direct(
    *,
    request: Any,
    current_user: Any,
    current_user_id: str,
    resolve_authorized_agent_ids: ResolveAuthorizedAgentIds,
    get_askme_service: GetAskmeService,
    run_in_threadpool: RunInThreadpool,
    execute: ExecuteTalkback,
) -> Optional[Dict[str, Any]]:
    """
    When `search_web` is true and `df` is present, skip the full talkback pipeline (intent routing,
    relevance, rewrite, etc.) and run @askme / @web answer generation only via LLM — no SQL/API
    agent routing. Agent id is used only to obtain AskmeService / API credentials (first resolved id).

    The question must include ``@web`` or ``@askme``; otherwise this handler returns None and the
    normal talkback pipeline runs (no defaulting to web mode).
    """
    if not bool(getattr(request, "search_web", False)):
        return None
    q_raw = (getattr(request, "question", None) or getattr(request, "my_question", None) or "").strip()
    if not q_raw:
        return None
    rows = _parse_df_input_light(getattr(request, "df", None))
    if not isinstance(rows, list) or len(rows) == 0:
        return None

    effective_sk = _resolve_talkback_effective_session_key(request, current_user_id)

    resolved = resolve_authorized_agent_ids(getattr(request, "agent_id", None), current_user_id)
    if not resolved:
        return {
            "success": False,
            "message": "No authorized agents found for this user or requested agent(s)",
            "details": {},
            "trace_id": str(uuid.uuid4()),
            "session_key": effective_sk,
        }

    from app.core.config import settings

    service_key = resolved if len(resolved) > 1 else resolved[0]
    service = get_askme_service(agent_id=service_key)
    if not service:
        return {
            "success": False,
            "message": "Could not initialize service for prior-result analysis",
            "details": {},
            "trace_id": str(uuid.uuid4()),
            "session_key": effective_sk,
        }

    session_key = await run_in_threadpool(
        lambda: service.derive_talkback_session_key(
            current_user_id, resolved, client_session_key=effective_sk
        )
    )

    ql = q_raw.lower()
    if "@web" in ql:
        prior_mode = "web_enrichment"
    elif "@askme" in ql and "@web" not in ql:
        prior_mode = "askme_analysis"
    else:
        # Require explicit @web or @askme; do not infer from search_web alone.
        return None

    if prior_mode == "askme_analysis" and not settings.IS_ASKME_ACTIVE:
        return {
            "success": False,
            "message": "@askme analysis is disabled for this deployment.",
            "details": {"feature": "IS_ASKME_ACTIVE"},
            "trace_id": str(uuid.uuid4()),
            "session_key": effective_sk,
        }
    if prior_mode == "web_enrichment" and not settings.IS_WEB_ACTIVE:
        return {
            "success": False,
            "message": "@web search is disabled for this deployment.",
            "details": {"feature": "IS_WEB_ACTIVE"},
            "trace_id": str(uuid.uuid4()),
            "session_key": effective_sk,
        }

    new_q = q_raw.strip()

    # Single-agent LLM path only: no multi-agent SQL routing.
    credential_agent_id = str(resolved[0])
    exec_req = _clone_ask_request(
        request,
        {
            "question": new_q,
            "agent_id": [credential_agent_id],
            "df": rows,
            "inContext": False,
            "previous_questions": None,
            "search_web": True,
            "session_key": effective_sk,
        },
    )

    trace_id = str(uuid.uuid4())
    t0 = time.perf_counter()
    try:
        raw = await execute(exec_req, current_user)
    except Exception as e:
        logger.exception("PriorDf direct LLM bypass failed: %s", e)
        return {
            "success": False,
            "message": str(e),
            "details": {},
            "trace_id": trace_id,
            "session_key": effective_sk,
        }
    duration_ms = int((time.perf_counter() - t0) * 1000)

    result = _coerce_route_handler_result_to_dict(raw)

    result["trace_id"] = trace_id
    result["intent_routing"] = {
        "intent": "other",
        "prior_df_followup": prior_mode,
        "needs_clarification": False,
        "suggested_agent_ids": [],
        "misrouted_web": False,
    }
    result["direct_prior_df_llm_bypass"] = True
    result["context_cleared"] = False

    # Persist talkback turn on success (same scope as full pipeline).
    if result.get("success") is True:
        direct_scope = [str(resolved[0])] if resolved else []
        try:
            await run_in_threadpool(service.clear_talkback_pending_clarification_only, session_key)
        except Exception as e:
            logger.warning("Talkback direct bypass clear pending failed: %s", e)

        answer_text = str(result.get("message", "") or "")
        try:
            rid = _derive_request_id_for_direct(request, current_user_id, resolved, q_raw)
            direct_meta = build_talkback_answer_meta(
                result,
                dispatch_agent_ids=[str(resolved[0])] if resolved else None,
            )
            await run_in_threadpool(
                service.append_turn_to_talkback_session,
                session_key,
                current_user_id,
                resolved,
                q_raw,
                answer_text,
                10,
                rid,
                direct_meta,
                direct_scope,
            )
        except Exception as e:
            logger.warning("Talkback direct bypass persist failed: %s", e)

        try:
            agent = service._ensure_agent()
            last_id = getattr(agent, "_last_session_response_id", None)
            if last_id:
                await run_in_threadpool(
                    service.upsert_talkback_session_state,
                    session_key,
                    current_user_id,
                    resolved,
                    None,
                    last_id,
                    False,
                    None,
                    {},
                    direct_scope,
                )
        except Exception:
            pass

    result["trace_steps"] = [
        {
            "name": "PriorDfDirectLLM",
            "status": "success",
            "duration_ms": duration_ms,
            "error": None,
        }
    ]
    result["session_key"] = effective_sk
    return result


class TalkbackPipeline:
    """
    Pipeline-style orchestrator mirroring `askme_open_agent/askme_base/sql_nodes.py`.
    Nodes update a shared TalkbackState; this runner controls ordering + fallback behavior.
    """

    def __init__(
        self,
        *,
        resolve_authorized_agent_ids: ResolveAuthorizedAgentIds,
        get_askme_service: GetAskmeService,
        run_in_threadpool: RunInThreadpool,
        execute: ExecuteTalkback,
    ):
        self._resolve_authorized_agent_ids = resolve_authorized_agent_ids
        self._get_askme_service = get_askme_service
        self._run_in_threadpool = run_in_threadpool
        self._execute = execute

        self._nodes: List[BaseNode] = [
            _UserQuestionLogCacheHitNode(
                self._resolve_authorized_agent_ids,
                self._run_in_threadpool,
                self._execute,
            ),
            _ResolveAgentsNode(self._resolve_authorized_agent_ids),
            _NormalizeDatasourceNode(),
            _FileCommandDetectNode(self._run_in_threadpool),
            _LoadSessionNode(self._get_askme_service, self._run_in_threadpool),
            _PreparePriorQuestionsNode(),
            _ContextRewriteForRoutingNode(self._run_in_threadpool),
            _IntentRoutingNode(self._run_in_threadpool),
            _ContextRelevanceNode(self._run_in_threadpool),
            _DispatchDecisionNode(self._get_askme_service, self._run_in_threadpool),
            _PriorDfConsentNode(self._get_askme_service, self._run_in_threadpool),
            _BroadScopeProcessingWarningNode(self._get_askme_service, self._run_in_threadpool),
            _ClarificationShortCircuitNode(self._get_askme_service, self._run_in_threadpool),
            _ExecuteNode(self._execute),
            _PersistTurnNode(self._get_askme_service, self._run_in_threadpool),
        ]

    async def _run_node(self, node: BaseNode, state: TalkbackState) -> TalkbackState:
        start = time.perf_counter()
        _print_node("start", node.name, state.trace.trace_id, state)
        logger.info(
            "TalkbackOrchestrator trace_id=%s node=%s status=%s start",
            state.trace.trace_id,
            node.name,
            "IN_PROGRESS",
        )
        state.trace.start(node.name)
        try:
            timeout_seconds: Optional[float] = None
            # Only apply timeouts to the steps that typically do LLM/network work.
            if node.name in {"ContextRelevanceNode", "ContextRewriteForRoutingNode", "IntentRoutingNode"}:
                timeout_seconds = 20.0

            if timeout_seconds is not None:
                state = await asyncio.wait_for(node.process(state), timeout=timeout_seconds)
            else:
                state = await node.process(state)
            duration_ms = int((time.perf_counter() - start) * 1000)
            state.trace.finish(node.name, NodeStatus.SUCCESS, duration_ms)
            _print_node("success", node.name, state.trace.trace_id, state, duration_ms=duration_ms)
            # Add node-specific debug context (without changing response shape).
            if node.name == "ResolveAgentsNode":
                logger.info(
                    "TalkbackOrchestrator trace_id=%s node=%s resolved_agents=%s",
                    state.trace.trace_id,
                    node.name,
                    state.resolved_agent_ids,
                )
            elif node.name == "ContextRelevanceNode":
                logger.info(
                    "TalkbackOrchestrator trace_id=%s node=%s context_cleared=%s reason=%s",
                    state.trace.trace_id,
                    node.name,
                    state.context_cleared,
                    state.context_clear_reason,
                )
            elif node.name == "IntentRoutingNode":
                if isinstance(state.intent_routing, dict):
                    logger.info(
                        "TalkbackOrchestrator trace_id=%s node=%s intent=%s suggested_agents=%s needs_clarification=%s error=%s",
                        state.trace.trace_id,
                        node.name,
                        state.intent_routing.get("intent"),
                        state.intent_routing.get("suggested_agent_ids"),
                        state.intent_routing.get("needs_clarification"),
                        state.intent_routing.get("error"),
                    )
            elif node.name == "DispatchDecisionNode":
                logger.info(
                    "TalkbackOrchestrator trace_id=%s node=%s dispatch_agents=%s",
                    state.trace.trace_id,
                    node.name,
                    state.dispatch_agent_ids,
                )
            return state
        except asyncio.TimeoutError:
            duration_ms = int((time.perf_counter() - start) * 1000)
            state.trace.finish(node.name, NodeStatus.FAILED, duration_ms, error="timeout")
            _print_node("timeout", node.name, state.trace.trace_id, state, duration_ms=duration_ms)
            logger.warning(
                "TalkbackOrchestrator trace_id=%s node=%s status=%s error=timeout duration_ms=%s",
                state.trace.trace_id,
                node.name,
                "FAILED",
                duration_ms,
            )

            # Provide safe fallbacks so the orchestrator can still return something.
            if node.name == "IntentRoutingNode":
                state.intent_routing = {"error": "timeout"}
            elif node.name == "ContextRewriteForRoutingNode":
                state.routing_question_text = state.question_text
            # For ContextRelevanceNode timeout: keep defaults (no context clearing).
            return state
        except _TalkbackStop:
            # Node already set response
            duration_ms = int((time.perf_counter() - start) * 1000)
            state.trace.finish(node.name, NodeStatus.SUCCESS, duration_ms)
            _print_node("early_exit", node.name, state.trace.trace_id, state, duration_ms=duration_ms)
            logger.info(
                "TalkbackOrchestrator trace_id=%s node=%s status=%s duration_ms=%s early_exit=true",
                state.trace.trace_id,
                node.name,
                "SUCCESS",
                duration_ms,
            )
            return state
        except Exception as e:
            duration_ms = int((time.perf_counter() - start) * 1000)
            state.trace.finish(node.name, NodeStatus.FAILED, duration_ms, error=str(e))
            _print_node("failed", node.name, state.trace.trace_id, state, duration_ms=duration_ms, error=_safe_str(e, 400))
            logger.exception(
                "TalkbackOrchestrator trace_id=%s node=%s status=%s error=%s duration_ms=%s",
                state.trace.trace_id,
                node.name,
                "FAILED",
                str(e),
                duration_ms,
            )
            raise

    async def run(self, state: TalkbackState) -> TalkbackState:
        for node in self._nodes:
            if state.response is not None:
                break
            if getattr(state, "_uql_cache_hit_short_circuit", False) and node.name != _PERSIST_TURN_NODE_NAME:
                continue
            state = await self._run_node(node, state)
        return state


async def run_talkback_pipeline(
    *,
    request: Any,
    current_user: Any,
    resolve_authorized_agent_ids: ResolveAuthorizedAgentIds,
    get_askme_service: GetAskmeService,
    run_in_threadpool: RunInThreadpool,
    execute: ExecuteTalkback,
) -> Dict[str, Any]:
    """
    Public entrypoint used by the talkback route. Kept generic so it can be called from
    other transports later (background jobs) by injecting the same dependencies.
    """
    # Normally enforce authenticated user identity as the root of all session keys
    # (do not trust client-provided ``request.user_id``). However when BYPASS_AUTH is on,
    # ``current_user`` is always the admin id=1, which makes ``get_user_agents`` return
    # every agent (admin special case in ``auth.get_user_agents``). In that mode, prefer
    # the form-provided ``user_id`` so resolved agents are filtered by ``AccessManage``.
    from app.core.config import settings as _settings

    inferred_user_id = getattr(current_user, "id", None)
    req_user_id_raw = getattr(request, "user_id", None)
    req_user_id = str(req_user_id_raw).strip() if req_user_id_raw is not None else ""
    if _settings.BYPASS_AUTH and req_user_id:
        inferred_user_id = req_user_id
    elif inferred_user_id is None:
        inferred_user_id = req_user_id_raw
    current_user_id = str(inferred_user_id)
    try:
        setattr(request, "user_id", current_user_id)
    except Exception:
        pass

    direct = await _run_talkback_prior_df_search_web_direct(
        request=request,
        current_user=current_user,
        current_user_id=current_user_id,
        resolve_authorized_agent_ids=resolve_authorized_agent_ids,
        get_askme_service=get_askme_service,
        run_in_threadpool=run_in_threadpool,
        execute=execute,
    )
    if direct is not None:
        return direct

    state = TalkbackState(
        request=request,
        current_user=current_user,
        current_user_id=current_user_id,
    )
    state.talkback_effective_session_key = _resolve_talkback_effective_session_key(
        request, current_user_id
    )
    # Propagate minted thread id so ``multi_agent_ask_question`` does not mint a different key per call.
    try:
        setattr(request, "session_key", state.talkback_effective_session_key)
    except Exception:
        pass

    pipeline = TalkbackPipeline(
        resolve_authorized_agent_ids=resolve_authorized_agent_ids,
        get_askme_service=get_askme_service,
        run_in_threadpool=run_in_threadpool,
        execute=execute,
    )
    state = await pipeline.run(state)

    if state.response is None:
        logger.error(
            "TalkbackOrchestrator trace_id=%s finished without state.response (check PersistTurnNode / early exits)",
            state.trace.trace_id,
        )
        out: Dict[str, Any] = {
            "success": False,
            "message": "Unknown talkback orchestration error",
            "details": {},
            "trace_id": state.trace.trace_id,
        }
    else:
        out = state.response

    if isinstance(out, dict):
        out.pop("_log_meta", None)
        # Expose structured timing/status info for debugging without changing main response shape.
        out["trace_steps"] = [
            {
                "name": step.name,
                "status": step.status.value,
                "duration_ms": step.duration_ms,
                "error": step.error,
            }
            for step in state.trace.steps.values()
        ]
        eff = getattr(state, "talkback_effective_session_key", None) or ""
        if eff.strip():
            out["session_key"] = eff.strip()

    return out
