from __future__ import annotations

import asyncio
import importlib
import json
import logging
import os
import re
import socket
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Dict, List, Mapping, Optional
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit

from app.livekit.adapters.agent_backend import BackendReply
from app.livekit.policies.agent_finalization import final_stop_from_agent_result
from app.utils.logging import setup_minimal_logging
from app.utils.turn_ids import next_turn_message_id

logger = setup_minimal_logging()

_STREAM_TYPES = {"chunk", "delta", "partial", "token"}
_TERMINAL_TYPES = {"done", "final", "message", "output", "response", "result"}
_TEXT_KEYS = ("text", "response", "output", "message", "content")
_USER_TEXT_KEYS = ("text", "transcript", "utterance", "message", "content")
_CONTA_USER_RESPONSE_MESSAGE_TYPES = {"ready", "final"}


@dataclass(slots=True)
class _RemoteAgentReply:
    stage: str
    text: str
    done: bool
    result: Any
    metadata: Any = None


class RemoteAgentWSAdapter:
    def __init__(
        self,
        *,
        intro: str,
        url: str,
        backend_label: str = "remote_ws",
        url_by_agent: Optional[Dict[str, str]] = None,
        request_context: Optional[Dict[str, Any]] = None,
        timeline: Any | None = None,
        streaming: bool = False,
        default_stage: str = "PRESENTATION",
        open_timeout_s: float = 10.0,
        read_timeout_s: float = 45.0,
        write_timeout_s: float = 10.0,
        close_timeout_s: float = 10.0,
        max_message_bytes: int = 1024 * 1024,
    ) -> None:
        self._intro = intro
        self._url = url
        self._backend_label = str(backend_label or "remote_ws").strip().lower()
        self._url_by_agent = {
            self._normalize_agent_name(key): value.strip()
            for key, value in (url_by_agent or {}).items()
            if value and value.strip()
        }
        self._request_context = dict(request_context or {})
        self._timeline = timeline
        self._streaming = streaming
        self._default_stage = (default_stage or "PRESENTATION").strip().upper()
        self._open_timeout_s = open_timeout_s
        self._read_timeout_s = read_timeout_s
        self._write_timeout_s = write_timeout_s
        self._close_timeout_s = close_timeout_s
        self._max_message_bytes = max_message_bytes
        self._instance_name = (os.getenv("HOSTNAME") or socket.gethostname() or "").strip() or "-"

        self._elegibility = True
        self._protocol = ""
        self._last_stage = "INTRO"
        self._pending_interrupt: Optional[Dict[str, Any]] = None
        self._pending_events: List[Dict[str, Any]] = []

        self._end_lock = asyncio.Lock()
        self._ended = False
        self._end_reply: Optional[BackendReply] = None
        self._session_lock = asyncio.Lock()
        self._session_cm: Any | None = None
        self._session_websocket: Any | None = None
        self._pending_ready_reply: Optional[_RemoteAgentReply] = None
        self._reader_task: asyncio.Task[Any] | None = None
        self._response_waiter: asyncio.Future[_RemoteAgentReply] | None = None
        self._push_queue: asyncio.Queue[_RemoteAgentReply] = asyncio.Queue()
        self._reader_error: BaseException | None = None
        self._terminal_reply: Optional[_RemoteAgentReply] = None
        self._session_actions: set[str] | None = None
        self._query_message_seq = 0
        self._connection_message_id = ""
        self._active_request_message_id = ""

    @staticmethod
    def _normalize_agent_name(value: Any) -> str:
        raw = str(value or "").strip().lower()
        aliases = {
            "conta": "conta",
            "contas": "conta",
            "ofert": "oferta",
            "oferta": "oferta",
            "ofertas": "oferta",
            "cobra": "cobranca",
            "cobranca": "cobranca",
            "cobrança": "cobranca",
            "cobrancas": "cobranca",
            "cobranças": "cobranca",
        }
        return aliases.get(raw, raw)

    @staticmethod
    def _string_or_empty(value: Any) -> str:
        return str(value or "").strip()

    @classmethod
    def _message_id_from_mapping(cls, payload: Mapping[str, Any] | None) -> str:
        if not isinstance(payload, Mapping):
            return ""

        for key in ("message_id", "messageId"):
            value = payload.get(key)
            if value not in (None, ""):
                return cls._string_or_empty(value)

        for container_key in ("metadata", "data", "payload", "context"):
            metadata = payload.get(container_key)
            if not isinstance(metadata, Mapping):
                continue
            for key in ("message_id", "messageId"):
                value = metadata.get(key)
                if value not in (None, ""):
                    return cls._string_or_empty(value)

        return ""

    @classmethod
    def _metadata_with_message_id(cls, metadata: Any, message_id: Any) -> Dict[str, Any] | None:
        normalized_message_id = cls._string_or_empty(message_id)
        if not normalized_message_id:
            return dict(metadata) if isinstance(metadata, Mapping) else metadata

        merged: Dict[str, Any] = dict(metadata) if isinstance(metadata, Mapping) else {}
        merged.setdefault("message_id", normalized_message_id)
        return merged

    @classmethod
    def _reply_with_message_id(
        cls,
        reply: _RemoteAgentReply,
        message_id: Any,
    ) -> _RemoteAgentReply:
        metadata = cls._metadata_with_message_id(reply.metadata, message_id)
        if metadata is not reply.metadata:
            reply.metadata = metadata
        return reply

    def _import_websockets(self):
        module = importlib.import_module("websockets")
        connect = getattr(module, "connect", None)
        if callable(connect):
            return connect

        client_module = getattr(module, "client", None)
        connect = getattr(client_module, "connect", None)
        if callable(connect):
            return connect

        raise RuntimeError("websockets.connect is not available")

    def _serialize_message(self, payload: Dict[str, Any]) -> str:
        return json.dumps(payload, ensure_ascii=False)

    def _decode_message(self, raw_message: Any) -> Any:
        if isinstance(raw_message, bytes):
            raw_message = raw_message.decode("utf-8")
        if not isinstance(raw_message, str):
            raise RuntimeError(f"Unsupported websocket payload type: {type(raw_message)!r}")

        message = raw_message.strip()
        if not message:
            return {}

        try:
            return json.loads(message)
        except json.JSONDecodeError:
            return message

    def _extract_text_from_value(self, value: Any) -> str:
        if isinstance(value, str):
            return value.strip()
        if isinstance(value, Mapping):
            for key in ("result", * _TEXT_KEYS):
                text = self._extract_text_from_value(value.get(key))
                if text:
                    return text
            return ""
        if isinstance(value, list):
            for item in value:
                text = self._extract_text_from_value(item)
                if text:
                    return text
        return ""

    def _extract_user_text(self, user_input: Any) -> str:
        if isinstance(user_input, str):
            return user_input.strip()

        if isinstance(user_input, Mapping):
            for key in _USER_TEXT_KEYS:
                text = self._extract_text_from_value(user_input.get(key))
                if text:
                    return text

            for value in user_input.values():
                text = self._extract_text_from_value(value)
                if text:
                    return text

        if isinstance(user_input, list):
            for item in user_input:
                text = self._extract_user_text(item)
                if text:
                    return text

        return str(user_input).strip()

    def _copy_context(self) -> Dict[str, Any]:
        return {
            "intro": self._intro,
            "elegibility": self._elegibility,
            "streaming": self._streaming,
        }

    def _current_timestamp(self) -> str:
        return datetime.now(timezone.utc).isoformat()

    def _request_field(self, *keys: str) -> str:
        for key in keys:
            value = self._request_context.get(key)
            if value not in (None, ""):
                return self._string_or_empty(value)

        lowered = {str(key).lower(): value for key, value in self._request_context.items()}
        for key in keys:
            value = lowered.get(str(key).lower())
            if value not in (None, ""):
                return self._string_or_empty(value)

        return ""

    def _current_agent_name(self) -> str:
        return self._normalize_agent_name(
            self._request_field("agent", "Agent", "agente")
        )

    def _resolve_url(self) -> str:
        agent_name = self._current_agent_name()
        if agent_name:
            routed_url = self._url_by_agent.get(agent_name, "")
            if routed_url:
                return routed_url
        return self._url

    def _current_invoice_number(self) -> str:
        return self._request_field(
            "current_invoice_number",
            "currentInvoiceNumber",
            "ID_FATURA",
            "idFatura",
            "id_fatura",
            "IdFatura",
        )

    def _current_msisdn(self) -> str:
        return self._request_field("msisdn", "GSM", "gsm", "NUM_TELEFONE")

    def _current_ani(self) -> str:
        return self._request_field("ANI", "ani")

    def _current_protocol_number(self) -> str:
        return self._request_field("protocol_id", "protocolId", "protocolo", "protocolNumber", "protocol")

    def _current_session_id(self) -> str:
        return self._request_field("session_id", "sessionId")

    def _next_query_message_id(self) -> str:
        source = dict(self._request_context)
        if self._protocol:
            source.setdefault("protocol", self._protocol)
        return next_turn_message_id(source)

    def _current_connection_message_id(self) -> str:
        explicit_message_id = self._request_field("message_id", "messageId")
        if explicit_message_id:
            return explicit_message_id

        if not self._connection_message_id:
            self._connection_message_id = self._next_query_message_id()
        return self._connection_message_id

    def _current_channel_id(self) -> str:
        channel_id = self._request_field("channelId", "channel_id")
        if channel_id:
            return channel_id
        return "ura"

    def _current_ura_call_id(self) -> str:
        return self._request_field("uraCallId", "ura_call_id", "callIdGed")

    def _current_channel(self) -> str:
        channel = self._request_field("channel", "Channel", "canal")
        if channel:
            return channel

        configured = self._string_or_empty(
            os.getenv("REMOTE_AGENT_CHANNEL")
            or os.getenv("REMOTE_AGENT_DEFAULT_CHANNEL")
        )
        return configured or "SUPERVISOR"

    def _connection_query_params(self) -> Dict[str, str]:
        if self._current_agent_name() != "conta":
            return {}

        params: Dict[str, str] = {}
        msisdn = self._current_msisdn()
        if msisdn:
            params["msisdn"] = msisdn

        current_invoice_number = self._current_invoice_number()
        if current_invoice_number:
            params["current_invoice_number"] = current_invoice_number

        ani = self._current_ani()
        if ani:
            params["ani"] = ani

        protocol_id = self._current_protocol_number()
        if protocol_id:
            params["protocol_id"] = protocol_id

        session_id = self._current_session_id()
        if session_id:
            params["session_id"] = session_id

        message_id = self._current_connection_message_id()
        if message_id:
            params["message_id"] = message_id

        channel_id = self._current_channel_id()
        if channel_id:
            params["channelId"] = channel_id

        ura_call_id = self._current_ura_call_id()
        if ura_call_id:
            params["uraCallId"] = ura_call_id

        return params

    @staticmethod
    def _append_query_params(url: str, params: Mapping[str, Any]) -> str:
        filtered_params = {
            str(key): str(value).strip()
            for key, value in params.items()
            if value not in (None, "") and str(value).strip()
        }
        if not filtered_params:
            return url

        parsed = urlsplit(url)
        query_params = dict(parse_qsl(parsed.query, keep_blank_values=True))
        query_params.update(filtered_params)
        return urlunsplit(
            (
                parsed.scheme,
                parsed.netloc,
                parsed.path,
                urlencode(query_params),
                parsed.fragment,
            )
        )

    def _uses_persistent_session(self) -> bool:
        return self._current_agent_name() == "conta"

    def _connection_url(self) -> str:
        return self._append_query_params(
            self._resolve_url(),
            self._connection_query_params(),
        )

    def _connection_kwargs(self) -> Dict[str, Any]:
        return {
            "open_timeout": self._open_timeout_s,
            "close_timeout": self._close_timeout_s,
            "max_size": self._max_message_bytes,
        }

    @staticmethod
    def _compact_log_value(value: Any) -> str:
        text = str(value or "").strip()
        if not text:
            return "-"
        if len(text) > 240:
            return f"{text[:237]}..."
        return text.replace("\n", "\\n")

    def _log_ws_event(self, event: str, *, level: int = logging.INFO, **fields: Any) -> None:
        base_fields: Dict[str, Any] = {
            "backend": self._backend_label,
            "agent": self._current_agent_name() or "-",
            "persistent": int(self._uses_persistent_session()),
            "instance": self._instance_name,
        }
        base_fields.update(fields)

        ordered_keys = (
            "backend",
            "agent",
            "persistent",
            "instance",
            "url",
            "host",
            "action",
            "stage",
            "protocol",
            "duration_ms",
            "text_len",
            "text_preview",
            "payload_bytes",
            "done",
            "result_type",
            "error_type",
            "error",
        )
        parts = []
        used_keys: set[str] = set()
        for key in ordered_keys:
            if key not in base_fields:
                continue
            value = base_fields[key]
            if value in (None, ""):
                continue
            parts.append(f"{key}={self._compact_log_value(value)}")
            used_keys.add(key)

        for key, value in base_fields.items():
            if key in used_keys or value in (None, ""):
                continue
            parts.append(f"{key}={self._compact_log_value(value)}")

        logger.log(level, "REMOTE_AGENT_WS_%s | %s", event, " | ".join(parts))

    def _log_ws_failure(self, event: str, error: BaseException, **fields: Any) -> None:
        self._log_ws_event(
            event,
            level=logging.ERROR,
            error_type=type(error).__name__,
            error=str(error),
            **fields,
        )

    def _payload_metadata(self, payload: Mapping[str, Any]) -> Dict[str, Any]:
        body = payload.get("payload")
        if not isinstance(body, Mapping):
            body = payload

        action = str(payload.get("action") or body.get("action") or "").strip().lower()
        stage = str(body.get("stage") or payload.get("stage") or self._fallback_stage()).strip().upper()
        protocol = str(body.get("protocol") or payload.get("protocol") or self._protocol or "").strip()
        if "message" in body:
            user_input = body.get("message") or ""
        elif "text" in body:
            user_input = body.get("text") or ""
        else:
            user_input = ""
        user_text = self._extract_user_text(user_input)
        return {
            "action": action or "chat",
            "stage": stage or self._fallback_stage(),
            "protocol": protocol or "-",
            "text_len": len(user_text),
        }

    @staticmethod
    def _url_host(url: str) -> str:
        parsed = urlsplit(str(url or ""))
        return (parsed.hostname or "").strip()

    def _update_session_actions(self, payload: Mapping[str, Any]) -> None:
        if "actions" not in payload:
            return

        raw_actions = payload.get("actions")
        parsed_actions: set[str] = set()
        if isinstance(raw_actions, list):
            for item in raw_actions:
                action = str(item or "").strip().lower()
                if action:
                    parsed_actions.add(action)
        elif isinstance(raw_actions, str):
            action = raw_actions.strip().lower()
            if action:
                parsed_actions.add(action)

        self._session_actions = parsed_actions

    def _persistent_session_supports_action(self, action: str) -> bool:
        normalized_action = str(action or "").strip().lower()
        if not normalized_action:
            return False

        if not self._uses_persistent_session():
            return True

        if self._session_actions is None:
            return normalized_action != "end"

        return normalized_action in self._session_actions

    @staticmethod
    def _is_unsupported_action_error(error: BaseException, action: str) -> bool:
        action_name = str(action or "").strip().lower()
        if not action_name:
            return False
        message = str(error or "").strip().lower()
        return "não suportada" in message and action_name in message

    def _emit_request(self, *, url: str, payload: Dict[str, Any]) -> None:
        if self._timeline is not None:
            self._timeline.emit(
                "remote_agent_request",
                backend=self._backend_label,
                backend_family="remote_ws",
                agent=self._current_agent_name(),
                url=url,
                action=payload.get("action", ""),
                payload=payload,
            )
        try:
            payload_bytes = len(self._serialize_message(payload).encode("utf-8"))
        except Exception:
            payload_bytes = None
        self._log_ws_event(
            "REQUEST",
            url=url,
            host=self._url_host(url),
            payload_bytes=payload_bytes,
            **self._payload_metadata(payload),
        )

    async def _open_persistent_session_locked(self, *, capture_ready: bool) -> None:
        if self._session_websocket is not None:
            return

        connect = self._import_websockets()
        url = self._connection_url()
        self._log_ws_event(
            "CONNECT_OPEN",
            url=url,
            host=self._url_host(url),
            protocol=self._protocol or "-",
        )
        session_cm = connect(url, **self._connection_kwargs())
        try:
            websocket = await session_cm.__aenter__()
        except Exception as exc:
            self._log_ws_failure(
                "CONNECT_FAIL",
                exc,
                url=url,
                host=self._url_host(url),
                protocol=self._protocol or "-",
            )
            raise
        self._session_cm = session_cm
        self._session_websocket = websocket
        self._reader_error = None
        self._terminal_reply = None
        self._clear_push_queue()
        self._log_ws_event(
            "CONNECT_OK",
            url=url,
            host=self._url_host(url),
            protocol=self._protocol or "-",
        )

        ready_waiter: asyncio.Future[_RemoteAgentReply] | None = None
        if capture_ready:
            ready_waiter = self._create_response_waiter_locked()

        self._start_reader_task_locked()

        if ready_waiter is None:
            return

        try:
            self._pending_ready_reply = await asyncio.wait_for(
                ready_waiter,
                timeout=self._read_timeout_s,
            )
        except asyncio.TimeoutError:
            self._log_ws_event(
                "READY_TIMEOUT",
                url=url,
                host=self._url_host(url),
                protocol=self._protocol or "-",
            )
            if self._response_waiter is ready_waiter:
                self._response_waiter = None
            if ready_waiter is not None and not ready_waiter.done():
                ready_waiter.cancel()
            await self._close_persistent_session_locked()
            raise
        except Exception as exc:
            self._log_ws_failure(
                "READY_FAIL",
                exc,
                url=url,
                host=self._url_host(url),
                protocol=self._protocol or "-",
            )
            await self._close_persistent_session_locked()
            raise

    async def _close_persistent_session_locked(self) -> None:
        reader_task = self._reader_task
        self._reader_task = None
        waiter = self._response_waiter
        self._response_waiter = None
        session_cm = self._session_cm
        self._session_cm = None
        self._session_websocket = None
        self._pending_ready_reply = None
        self._reader_error = None
        self._terminal_reply = None
        self._session_actions = None
        self._query_message_seq = 0
        self._connection_message_id = ""
        self._active_request_message_id = ""
        self._clear_push_queue()

        if waiter is not None and not waiter.done():
            waiter.cancel()

        if reader_task is not None and not reader_task.done():
            reader_task.cancel()
            await asyncio.gather(reader_task, return_exceptions=True)

        if session_cm is None:
            return

        await session_cm.__aexit__(None, None, None)

    async def _send_persistent_payload_locked(self, payload: Dict[str, Any]) -> _RemoteAgentReply:
        websocket = self._session_websocket
        if websocket is None:
            raise RuntimeError("Persistent websocket session is not connected")

        waiter = self._create_response_waiter_locked()
        url = self._connection_url()
        self._emit_request(url=url, payload=payload)
        request_started = time.perf_counter()
        request_message_id = self._message_id_from_mapping(payload)
        previous_request_message_id = self._active_request_message_id
        self._active_request_message_id = request_message_id
        try:
            await asyncio.wait_for(
                websocket.send(self._serialize_message(payload)),
                timeout=self._write_timeout_s,
            )
        except Exception as exc:
            self._log_ws_failure(
                "SEND_FAIL",
                exc,
                url=url,
                host=self._url_host(url),
                **self._payload_metadata(payload),
            )
            if self._response_waiter is waiter:
                self._response_waiter = None
            if not waiter.done():
                waiter.cancel()
            self._active_request_message_id = previous_request_message_id
            raise
        try:
            reply = await asyncio.wait_for(waiter, timeout=self._read_timeout_s)
            if request_message_id:
                self._reply_with_message_id(reply, request_message_id)
            done_fields = {
                **self._payload_metadata(payload),
                "duration_ms": round((time.perf_counter() - request_started) * 1000),
                "stage": reply.stage,
                "text_len": len(reply.text),
                "text_preview": reply.text,
                "done": int(reply.done),
                "result_type": type(reply.result).__name__ if reply.result is not None else "-",
            }
            self._log_ws_event(
                "TURN_DONE",
                url=url,
                host=self._url_host(url),
                **done_fields,
            )
            return reply
        except asyncio.TimeoutError:
            self._log_ws_event(
                "RESPONSE_TIMEOUT",
                url=url,
                host=self._url_host(url),
                duration_ms=round((time.perf_counter() - request_started) * 1000),
                **self._payload_metadata(payload),
            )
            if self._response_waiter is waiter:
                self._response_waiter = None
            if not waiter.done():
                waiter.cancel()
            raise
        finally:
            self._active_request_message_id = previous_request_message_id

    async def _run_with_persistent_session(self, user_input: Any) -> tuple[_RemoteAgentReply, bool]:
        user_text = self._extract_user_text(user_input)

        async with self._session_lock:
            await self._open_persistent_session_locked(capture_ready=self._session_websocket is None)

            if self._pending_ready_reply is not None:
                if not user_text:
                    reply = self._pending_ready_reply
                    self._pending_ready_reply = None
                    return reply, False
                self._pending_ready_reply = None

            try:
                reply = await self._send_persistent_payload_locked(self._build_turn_payload(user_input))
            except Exception:
                await self._close_persistent_session_locked()
                raise

        return reply, True

    def _clear_push_queue(self) -> None:
        while True:
            try:
                self._push_queue.get_nowait()
            except asyncio.QueueEmpty:
                return

    def _create_response_waiter_locked(self) -> asyncio.Future[_RemoteAgentReply]:
        current = self._response_waiter
        if current is not None and not current.done():
            raise RuntimeError("Another persistent websocket response is already pending")

        waiter: asyncio.Future[_RemoteAgentReply] = asyncio.get_running_loop().create_future()
        self._response_waiter = waiter
        return waiter

    def _start_reader_task_locked(self) -> None:
        if self._reader_task is not None and not self._reader_task.done():
            return
        websocket = self._session_websocket
        if websocket is None:
            raise RuntimeError("Persistent websocket session is not connected")
        self._reader_task = asyncio.create_task(
            self._reader_loop(websocket),
            name="remote_agent_ws_reader",
        )

    async def _reader_loop(self, websocket: Any) -> None:
        try:
            while True:
                reply = await self._recv_reply(websocket, timeout_s=None)
                waiter = self._response_waiter
                if waiter is not None and not waiter.done():
                    self._response_waiter = None
                    waiter.set_result(reply)
                else:
                    await self._push_queue.put(reply)

                if reply.done:
                    self._terminal_reply = reply
                    return
        except asyncio.CancelledError:
            raise
        except Exception as exc:
            self._reader_error = exc
            self._log_ws_failure(
                "READER_FAIL",
                exc,
                url=self._connection_url(),
                host=self._url_host(self._connection_url()),
                protocol=self._protocol or "-",
            )
            waiter = self._response_waiter
            if waiter is not None and not waiter.done():
                self._response_waiter = None
                waiter.set_exception(exc)

    def supports_server_push(self) -> bool:
        return self._uses_persistent_session()

    def supports_inflight_backend_push(self) -> bool:
        return self._current_agent_name() == "conta"

    async def wait_for_server_push(self) -> BackendReply | None:
        if not self._uses_persistent_session():
            return None

        while True:
            try:
                reply = self._push_queue.get_nowait()
            except asyncio.QueueEmpty:
                reply = None

            if reply is not None:
                self._last_stage = reply.stage or self._last_stage or self._default_stage
                return BackendReply(
                    stage=self._last_stage,
                    text=reply.text,
                    done=bool(reply.done) or self._last_stage == "DONE",
                    export_payload=reply.result,
                    metadata=reply.metadata,
                )

            if self._reader_error is not None:
                raise RuntimeError("Persistent websocket reader failed") from self._reader_error

            reader_task = self._reader_task
            if reader_task is None or reader_task.done():
                return None

            get_task = asyncio.create_task(self._push_queue.get(), name="remote_agent_ws_push_wait")
            try:
                done, pending = await asyncio.wait(
                    {get_task, reader_task},
                    return_when=asyncio.FIRST_COMPLETED,
                )
            except asyncio.CancelledError:
                get_task.cancel()
                await asyncio.gather(get_task, return_exceptions=True)
                raise

            if get_task in done:
                reply = get_task.result()
                self._last_stage = reply.stage or self._last_stage or self._default_stage
                return BackendReply(
                    stage=self._last_stage,
                    text=reply.text,
                    done=bool(reply.done) or self._last_stage == "DONE",
                    export_payload=reply.result,
                    metadata=reply.metadata,
                )

            get_task.cancel()
            await asyncio.gather(get_task, return_exceptions=True)

            if self._reader_error is not None:
                raise RuntimeError("Persistent websocket reader failed") from self._reader_error

            if reader_task.done() and self._push_queue.empty():
                return None

    def _base_payload(self) -> Dict[str, Any]:
        payload = {
            "timestamp": self._current_timestamp(),
            "agent": self._current_agent_name(),
            "RouterCallKeyDay": self._request_field("RouterCallKeyDay", "router_call_key_day", "routerCallKeyDay"),
            "RouterCallKey": self._request_field("RouterCallKey", "router_call_key", "routerCallKey"),
            "ANI": self._request_field("ANI", "ani"),
            "GSM": self._request_field("GSM", "gsm", "NUM_TELEFONE"),
            "callIdGed": self._request_field("callIdGed"),
        }

        if payload["agent"] == "conta":
            id_fatura = self._request_field("ID_FATURA", "id_fatura", "IdFatura")
            if id_fatura:
                payload["ID_FATURA"] = id_fatura

        return payload

    def _build_conta_turn_payload(self, user_input: Any) -> Dict[str, Any]:
        payload = {
            "message": self._extract_user_text(user_input),
            "channel": self._current_channel(),
            "msisdn": self._current_msisdn(),
        }
        message_id = self._message_id_from_mapping(user_input)
        if message_id:
            payload["message_id"] = message_id

        current_invoice_number = self._current_invoice_number()
        if current_invoice_number:
            payload["current_invoice_number"] = current_invoice_number

        self._add_pending_speech_interruption(payload)
        if self._pending_events:
            payload["events"] = list(self._pending_events)

        return {
            "action": "chat",
            "payload": payload,
        }

    def _build_turn_payload(self, user_input: Any) -> Dict[str, Any]:
        if self._current_agent_name() == "conta":
            return self._build_conta_turn_payload(user_input)

        payload = self._base_payload()
        payload["text"] = self._extract_user_text(user_input)
        payload["protocol"] = self._protocol
        payload["stage"] = self._last_stage
        message_id = self._message_id_from_mapping(user_input)
        if message_id:
            payload["message_id"] = message_id

        self._add_pending_speech_interruption(payload)
        if self._pending_events:
            payload["events"] = list(self._pending_events)

        return payload

    def _build_end_payload(self) -> Dict[str, Any]:
        if self._current_agent_name() == "conta":
            payload = self._base_payload()
            payload["protocol"] = self._protocol
            payload["stage"] = self._last_stage
            payload["msisdn"] = self._current_msisdn()
            payload["channel"] = self._current_channel()

            current_invoice_number = self._current_invoice_number()
            if current_invoice_number:
                payload["current_invoice_number"] = current_invoice_number

            self._add_pending_speech_interruption(payload)
            if self._pending_events:
                payload["events"] = list(self._pending_events)

            return {
                "action": "end",
                "payload": payload,
            }

        payload = self._base_payload()
        payload["type"] = "end"
        payload["protocol"] = self._protocol
        payload["stage"] = self._last_stage

        self._add_pending_speech_interruption(payload)
        if self._pending_events:
            payload["events"] = list(self._pending_events)

        return payload

    def _clear_pending_state(self) -> None:
        self._pending_interrupt = None
        self._pending_events.clear()

    def _extract_stage(self, payload: Mapping[str, Any]) -> str:
        stage = payload.get("stage")
        if isinstance(stage, str) and stage.strip():
            return stage.strip().upper()
        return ""

    def _extract_delta(self, payload: Mapping[str, Any]) -> str:
        for key in ("delta", "chunk", "token"):
            value = payload.get(key)
            if isinstance(value, str) and value.strip():
                return value
        return ""

    def _extract_reply_text(self, payload: Mapping[str, Any]) -> str:
        message_type = str(payload.get("type") or "").strip().lower()
        if message_type == "final":
            text = self._extract_text_from_value(payload.get("content"))
            if text:
                return text

        for key in ("result", *_TEXT_KEYS):
            value = payload.get(key)
            text = self._extract_text_from_value(value)
            if text:
                return text
        return ""

    @staticmethod
    def _bool_or_none(value: Any) -> bool | None:
        if isinstance(value, bool):
            return value
        if isinstance(value, str):
            normalized = value.strip().lower()
            if normalized in {"1", "true", "yes", "on"}:
                return True
            if normalized in {"0", "false", "no", "off"}:
                return False
        return None

    def _reply_metadata(self, payload: Mapping[str, Any], result: Any = None) -> Dict[str, Any] | None:
        raw_metadata = payload.get("metadata")
        metadata: Dict[str, Any] = dict(raw_metadata) if isinstance(raw_metadata, Mapping) else {}

        sources: list[Mapping[str, Any]] = []
        if isinstance(result, Mapping):
            sources.append(result)
        sources.append(payload)

        for source in sources:
            message_id = self._message_id_from_mapping(source)
            if message_id:
                metadata["message_id"] = message_id
                break

        for source in sources:
            speech_id = self._string_or_empty(source.get("speech_id"))
            if speech_id:
                metadata["speech_id"] = speech_id
                break

        for source in sources:
            interruptible = self._bool_or_none(source.get("is_interruptible"))
            if interruptible is not None:
                metadata["is_interruptible"] = interruptible
                break

        if self._current_agent_name() == "conta":
            message_type = str(payload.get("type") or "").strip().lower()
            result_type = ""
            if isinstance(result, Mapping):
                result_type = str(result.get("type") or "").strip().lower()

            expects_user_response = (
                result_type == "final"
                or (not result_type and message_type in _CONTA_USER_RESPONSE_MESSAGE_TYPES)
            )
            terminal_result = final_stop_from_agent_result(result) is not None
            protected_speech = (
                message_type in {"ready", "feedback"}
                or terminal_result
                or (bool(result_type) and result_type != "final")
            )

            if message_type:
                metadata["agent_message_type"] = message_type
            if result_type:
                metadata["agent_result_type"] = result_type
            metadata["expects_user_response"] = expects_user_response
            metadata["drop_user_input_while_speaking"] = protected_speech
            if protected_speech:
                metadata["is_interruptible"] = False

        return metadata or None

    def _is_conta_feedback(self, message_type: str) -> bool:
        return self._current_agent_name() == "conta" and message_type == "feedback"

    def _is_conta_result_feedback(self, result: Any) -> bool:
        return (
            self._current_agent_name() == "conta"
            and isinstance(result, Mapping)
            and str(result.get("type") or "").strip().lower() == "feedback"
        )

    def _feedback_metadata(
        self,
        payload: Mapping[str, Any],
        base_metadata: Any = None,
    ) -> Dict[str, Any]:
        metadata: Dict[str, Any] = {}
        if isinstance(base_metadata, Mapping):
            metadata.update(base_metadata)
        metadata["event"] = "feedback"
        metadata["payload"] = dict(payload)
        return metadata

    def _queue_feedback_reply(
        self,
        *,
        payload: Mapping[str, Any],
        stage: str,
        text: str,
        metadata: Any = None,
    ) -> None:
        feedback_text = (text or "").strip()
        if not feedback_text:
            return

        if self._current_agent_name() == "conta":
            metadata = self._metadata_with_message_id(
                metadata,
                self._active_request_message_id
                or self._connection_message_id
                or self._request_field("message_id", "messageId"),
            )
        reply = _RemoteAgentReply(
            stage=stage or self._fallback_stage(),
            text=feedback_text,
            done=False,
            result=None,
            metadata=self._feedback_metadata(payload, metadata),
        )
        self._push_queue.put_nowait(reply)

        if self._timeline is not None:
            self._timeline.emit(
                "remote_agent_response",
                backend=self._backend_label,
                backend_family="remote_ws",
                agent=self._current_agent_name(),
                message_type="feedback",
                stage=reply.stage,
                done=False,
                text_len=len(reply.text),
                has_result=False,
            )

    def _add_pending_speech_interruption(self, payload: Dict[str, Any]) -> None:
        if self._pending_interrupt is not None:
            interruption = dict(self._pending_interrupt)
            interruption_key = str(
                interruption.pop("_interruption_field", "speech_interruption")
            )
            payload[interruption_key] = interruption

    def _fallback_stage(self) -> str:
        if self._last_stage and self._last_stage != "INTRO":
            return self._last_stage
        return self._default_stage

    async def _recv_reply(self, websocket: Any, timeout_s: float | None = None) -> _RemoteAgentReply:
        chunks: List[str] = []
        final_text = ""
        final_stage = ""
        done = False
        result: Any = None
        metadata: Any = None
        saw_message = False

        while True:
            try:
                recv_coro = websocket.recv()
                if timeout_s is None:
                    raw_message = await recv_coro
                else:
                    raw_message = await asyncio.wait_for(recv_coro, timeout=timeout_s)
            except asyncio.TimeoutError:
                if saw_message or chunks or final_text or result is not None:
                    break
                raise
            except asyncio.CancelledError:
                raise
            except Exception:
                if saw_message or chunks or final_text or result is not None:
                    break
                raise

            saw_message = True
            message = self._decode_message(raw_message)

            if isinstance(message, str):
                final_text = message
                break

            if not isinstance(message, Mapping):
                continue

            self._update_session_actions(message)

            error = message.get("error")
            if error:
                raise RuntimeError(str(error))

            message_type = str(message.get("type") or "").strip().lower()
            stage = self._extract_stage(message)
            if stage and not self._is_conta_feedback(message_type):
                final_stage = stage

            previous_result = result
            if "result" in message:
                result = message.get("result")
            message_metadata = self._reply_metadata(message, result)

            delta = self._extract_delta(message)
            if delta:
                chunks.append(delta)
                if self._timeline is not None:
                    self._timeline.emit(
                        "remote_agent_stream_chunk",
                        agent=self._current_agent_name(),
                        chunk_len=len(delta),
                    )

            reply_text = self._extract_reply_text(message)

            if self._is_conta_feedback(message_type) or self._is_conta_result_feedback(result):
                self._queue_feedback_reply(
                    payload=message,
                    stage=stage or final_stage or self._fallback_stage(),
                    text=reply_text,
                    metadata=message_metadata,
                )
                result = previous_result
                continue

            if message_metadata is not None:
                metadata = message_metadata

            if reply_text:
                final_text = reply_text

            if final_stop_from_agent_result(result) is not None:
                done = True
            if message_type == "done":
                done = True
            if done or bool(message.get("done")) or final_stage == "DONE":
                done = True
                break

            if message_type in _STREAM_TYPES:
                continue

            if bool(message.get("final")) or message_type in _TERMINAL_TYPES or reply_text:
                break

        text = (final_text or "".join(chunks)).strip()
        stage = final_stage or self._fallback_stage()
        if done:
            stage = "DONE"

        if not text and result is None:
            raise RuntimeError("Remote agent websocket closed without a usable response")

        if self._timeline is not None:
            self._timeline.emit(
                "remote_agent_response",
                backend=self._backend_label,
                backend_family="remote_ws",
                agent=self._current_agent_name(),
                stage=stage,
                done=done,
                text_len=len(text),
                has_result=result is not None,
            )
        self._log_ws_event(
            "RESPONSE",
            url=self._connection_url(),
            host=self._url_host(self._connection_url()),
            stage=stage,
            protocol=self._protocol or "-",
            text_len=len(text),
            text_preview=text,
            done=int(done),
            result_type=type(result).__name__ if result is not None else "-",
        )

        reply = _RemoteAgentReply(
            stage=stage,
            text=text,
            done=done,
            result=result,
            metadata=metadata,
        )
        if self._current_agent_name() == "conta":
            self._reply_with_message_id(
                reply,
                self._active_request_message_id
                or self._connection_message_id
                or self._request_field("message_id", "messageId"),
            )
        return reply

    async def _request(self, payload: Dict[str, Any]) -> _RemoteAgentReply:
        connect = self._import_websockets()
        url = self._connection_url()
        self._log_ws_event(
            "CONNECT_OPEN",
            url=url,
            host=self._url_host(url),
            protocol=self._protocol or "-",
        )
        request_message_id = self._message_id_from_mapping(payload)
        previous_request_message_id = self._active_request_message_id
        self._active_request_message_id = request_message_id
        try:
            async with connect(url, **self._connection_kwargs()) as websocket:
                self._log_ws_event(
                    "CONNECT_OK",
                    url=url,
                    host=self._url_host(url),
                    protocol=self._protocol or "-",
                )
                self._emit_request(url=url, payload=payload)
                request_started = time.perf_counter()
                try:
                    await asyncio.wait_for(
                        websocket.send(self._serialize_message(payload)),
                        timeout=self._write_timeout_s,
                    )
                except Exception as exc:
                    self._log_ws_failure(
                        "SEND_FAIL",
                        exc,
                        url=url,
                        host=self._url_host(url),
                        **self._payload_metadata(payload),
                    )
                    raise
                reply = await self._recv_reply(websocket, timeout_s=self._read_timeout_s)
                if request_message_id:
                    self._reply_with_message_id(reply, request_message_id)
                done_fields = {
                    **self._payload_metadata(payload),
                    "duration_ms": round((time.perf_counter() - request_started) * 1000),
                    "stage": reply.stage,
                    "text_len": len(reply.text),
                    "text_preview": reply.text,
                    "done": int(reply.done),
                    "result_type": type(reply.result).__name__ if reply.result is not None else "-",
                }
                self._log_ws_event(
                    "TURN_DONE",
                    url=url,
                    host=self._url_host(url),
                    **done_fields,
                )
                self._active_request_message_id = previous_request_message_id
                return reply
        except asyncio.TimeoutError:
            self._active_request_message_id = previous_request_message_id
            self._log_ws_event(
                "REQUEST_TIMEOUT",
                url=url,
                host=self._url_host(url),
                **self._payload_metadata(payload),
            )
            raise
        except Exception as exc:
            self._active_request_message_id = previous_request_message_id
            self._log_ws_failure(
                "REQUEST_FAIL",
                exc,
                url=url,
                host=self._url_host(url),
                **self._payload_metadata(payload),
            )
            raise

    async def prepare(
        self,
        elegibility: bool,
        protocol: str,
    ) -> None:
        if self._timeline is not None:
            self._timeline.emit(
                "backend_prepare_started",
                backend=self._backend_label,
                backend_family="remote_ws",
                agent=self._current_agent_name(),
            )
        self._elegibility = bool(elegibility)
        self._protocol = str(protocol or "")
        self._last_stage = "INTRO"
        self._pending_interrupt = None
        self._pending_events.clear()
        self._pending_ready_reply = None
        if self._uses_persistent_session():
            async with self._session_lock:
                await self._close_persistent_session_locked()
                await self._open_persistent_session_locked(capture_ready=True)
        if self._timeline is not None:
            self._timeline.emit(
                "backend_prepare_completed",
                backend=self._backend_label,
                backend_family="remote_ws",
                agent=self._current_agent_name(),
                protocol=self._protocol,
            )

    async def run(self, user_input: Any) -> BackendReply:
        sent_payload = True
        if self._uses_persistent_session():
            reply, sent_payload = await self._run_with_persistent_session(user_input)
        else:
            reply = await self._request(self._build_turn_payload(user_input))

        if sent_payload:
            self._clear_pending_state()
        self._last_stage = reply.stage or self._last_stage or self._default_stage
        return BackendReply(
            stage=self._last_stage,
            text=reply.text,
            done=bool(reply.done) or self._last_stage == "DONE",
            export_payload=reply.result,
            metadata=reply.metadata,
        )

    async def set_interruption(
        self,
        interrupted: bool,
        listened_text: str = "",
        skipped: bool = False,
        speech_id: str = "",
    ) -> None:
        if not interrupted:
            self._pending_interrupt = None
            if self._timeline is not None:
                self._timeline.emit(
                    "backend_set_interruption",
                    backend=self._backend_label,
                    backend_family="remote_ws",
                    agent=self._current_agent_name(),
                    interrupted=False,
                )
            return

        speech_id = self._string_or_empty(speech_id)
        if not speech_id:
            self._pending_interrupt = None
            if self._timeline is not None:
                self._timeline.emit(
                    "backend_set_interruption",
                    backend=self._backend_label,
                    backend_family="remote_ws",
                    agent=self._current_agent_name(),
                    interrupted=True,
                    skipped=True,
                    listened_text=listened_text,
                    speech_id="",
                    reason="missing_speech_id",
                )
            return
        self._pending_interrupt = {
            "speech_id": speech_id,
            "heard_text": listened_text or "",
        }
        if self._timeline is not None:
            self._timeline.emit(
                "backend_set_interruption",
                backend=self._backend_label,
                backend_family="remote_ws",
                agent=self._current_agent_name(),
                interrupted=True,
                skipped=bool(skipped),
                listened_text=listened_text,
                speech_id=speech_id,
            )

    async def set_processing_interruption(
        self,
        listened_text: str = "",
        skipped: bool = False,
        speech_id: str = "",
    ) -> None:
        await self.set_interruption(
            True,
            listened_text=listened_text,
            skipped=skipped,
            speech_id=speech_id,
        )
        if self._pending_interrupt is not None:
            self._pending_interrupt["_interruption_field"] = "processing_interruption"

    async def inject_idle_nudge(self, nudge_text: str) -> None:
        text = (nudge_text or "").strip()
        if not text:
            return

        event = {
            "type": "idle_nudge",
            "text": text,
        }
        # Cada frase de inatividade substitui a anterior: o cliente responde ao
        # que ouviu por ultimo, e as intermediarias so empilham falas do agente
        # no historico remoto -- inclusive o aviso de encerramento, que passa a
        # constar como dito logo antes de a conversa seguir normalmente.
        if self._pending_events and self._pending_events[-1].get("type") == "idle_nudge":
            self._pending_events[-1] = event
        else:
            self._pending_events.append(event)
        if self._timeline is not None:
            self._timeline.emit(
                "backend_idle_nudge_buffered",
                backend=self._backend_label,
                backend_family="remote_ws",
                agent=self._current_agent_name(),
                text=text,
            )

    async def end_service_once(self) -> BackendReply:
        async with self._end_lock:
            if self._ended:
                return self._end_reply or BackendReply(stage="DONE", done=True, export_payload=[])

            self._ended = True
            if self._timeline is not None:
                self._timeline.emit(
                    "backend_end_started",
                    backend=self._backend_label,
                    backend_family="remote_ws",
                    agent=self._current_agent_name(),
                )
            try:
                if self._uses_persistent_session():
                    async with self._session_lock:
                        await self._open_persistent_session_locked(
                            capture_ready=self._session_websocket is None
                        )
                        self._pending_ready_reply = None
                        if self._reader_task is not None and self._reader_task.done():
                            if self._reader_error is not None:
                                raise RuntimeError("Persistent websocket reader failed") from self._reader_error
                            reply = self._terminal_reply or _RemoteAgentReply(
                                stage="DONE",
                                text="",
                                done=True,
                                result=[],
                            )
                        elif not self._persistent_session_supports_action("end"):
                            reply = self._terminal_reply or _RemoteAgentReply(
                                stage="DONE",
                                text="",
                                done=True,
                                result=[],
                            )
                        else:
                            try:
                                reply = await self._send_persistent_payload_locked(self._build_end_payload())
                            except RuntimeError as exc:
                                if not self._is_unsupported_action_error(exc, "end"):
                                    raise
                                reply = self._terminal_reply or _RemoteAgentReply(
                                    stage="DONE",
                                    text="",
                                    done=True,
                                    result=[],
                                )
                        await self._close_persistent_session_locked()
                else:
                    reply = await self._request(self._build_end_payload())
            except Exception:
                logger.exception("[remote-agent] end_service websocket falhou")
                if self._uses_persistent_session():
                    async with self._session_lock:
                        await self._close_persistent_session_locked()
                if self._timeline is not None:
                    self._timeline.emit(
                        "backend_end_failed",
                        backend=self._backend_label,
                        backend_family="remote_ws",
                        agent=self._current_agent_name(),
                    )
                self._end_reply = BackendReply(stage="DONE", done=True, export_payload=[])
                return self._end_reply

            self._clear_pending_state()
            stage = (reply.stage or self._last_stage or self._default_stage).upper()
            self._last_stage = stage
            self._end_reply = BackendReply(
                stage=stage,
                text=reply.text,
                done=bool(reply.done) or stage == "DONE",
                export_payload=reply.result if reply.result is not None else [],
                metadata=reply.metadata,
            )
            if self._timeline is not None:
                self._timeline.emit(
                    "backend_end_completed",
                    backend=self._backend_label,
                    backend_family="remote_ws",
                    agent=self._current_agent_name(),
                    result_type=type(self._end_reply.export_payload).__name__,
                )
            return self._end_reply
