from __future__ import annotations

import asyncio
import json
import os
import re
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Dict, List, Mapping, Optional

import httpx

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", "progress"}
_TERMINAL_TYPES = {"done", "final", "message", "output", "response", "result", "proactive_result"}
_PREFETCH_TERMINAL_TYPES = {"prefetch_done", "prefetch_skipped", "prefetch_failed"}
_TEXT_KEYS = ("text", "response", "output", "message", "content")
_USER_TEXT_KEYS = ("text", "transcript", "utterance", "message", "content")
_OFERTA_INITIAL_MESSAGE_ENV = "REMOTE_AGENT_OFERTA_INITIAL_MESSAGE"
_OFERTA_INITIAL_MESSAGE_DEFAULT = "inicio_atendimento"
_OFERTA_SERVICE_STATUS_RESULT_TYPE = {
    "RESOLVED": "resolvido",
}
_OFERTA_NON_FINAL_SERVICE_STATUSES = {"UNRESOLVED", "RESOLVED_WITH_NEW_REQUEST"}
_FALSE_ENV_VALUES = {"0", "false", "no", "off", "disable", "disabled"}
_TRUE_ENV_VALUES = {"1", "true", "yes", "on", "enable", "enabled"}
_CONTA_USER_RESPONSE_MESSAGE_TYPES = {"ready", "final"}


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


class RemoteAgentSSEAdapter:
    def __init__(
        self,
        *,
        intro: str,
        url: str,
        backend_label: str = "remote_sse",
        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",
        connect_timeout_s: float = 10.0,
        read_timeout_s: float = 45.0,
        write_timeout_s: float = 10.0,
    ) -> None:
        self._intro = intro
        self._url = url
        self._backend_label = str(backend_label or "remote_sse").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._connect_timeout_s = connect_timeout_s
        self._read_timeout_s = read_timeout_s
        self._write_timeout_s = write_timeout_s

        self._elegibility = True
        self._remote_session_id = ""
        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._pending_ready_reply: Optional[_RemoteAgentReply] = None
        self._push_queue: asyncio.Queue[_RemoteAgentReply] = asyncio.Queue()
        self._session_actions: set[str] | None = None
        self._protocol = ""
        self._turn_seq = 0
        self._offer_stream_task: asyncio.Task[None] | None = None
        self._conta_feedback_streams = 0
        self._conta_feedback_event = asyncio.Event()
        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

    @staticmethod
    def _decode_message(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 SSE 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 _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 _is_oferta_agent(self) -> bool:
        return self._current_agent_name() == "oferta"

    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(
            "invoice_id",
            "invoiceId",
            "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_session_id(self) -> str:
        return self._request_field("session_id", "sessionId")

    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_protocol_number(self) -> str:
        return self._request_field("protocol_id", "protocolId", "protocolo", "protocolNumber", "protocol")

    def _message_id_from_user_input(self, user_input: Any) -> str:
        if not isinstance(user_input, Mapping):
            return ""

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

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

        return ""

    def _query_message_id(self, user_input: Any = None) -> str:
        return (
            self._message_id_from_user_input(user_input)
            or self._request_field("message_id", "messageId")
            or self._next_message_id()
        )

    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 _stream_open_params(self) -> Dict[str, str]:
        params: Dict[str, str] = {}

        msisdn = self._current_msisdn()
        if msisdn:
            params["msisdn"] = msisdn

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

        if self._current_agent_name() != "conta":
            return params

        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._query_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

    def _action_params(
        self,
        user_input: Any = None,
        *,
        message_id: str = "",
    ) -> Dict[str, str]:
        params: Dict[str, str] = {}
        if self._remote_session_id:
            params["session_id"] = self._remote_session_id

        if self._current_agent_name() != "conta":
            return params

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

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

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

        request_message_id = message_id or self._query_message_id(user_input)
        if request_message_id:
            params["message_id"] = request_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

    def _stream_headers(self) -> Dict[str, str]:
        return {
            "Accept": "text/event-stream",
            "Cache-Control": "no-cache",
        }

    def _oferta_headers(self) -> Dict[str, str]:
        headers = self._stream_headers()
        headers["Content-Type"] = "application/json"
        headers["Channel-id"] = self._current_channel_id() or "ura"
        return headers

    def _client_timeout(self) -> httpx.Timeout:
        return httpx.Timeout(
            connect=self._connect_timeout_s,
            read=self._read_timeout_s,
            write=self._write_timeout_s,
            pool=self._connect_timeout_s,
        )

    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 _update_session_id(self, payload: Mapping[str, Any]) -> None:
        for key in ("session_id", "sessionId"):
            value = payload.get(key)
            if value not in (None, ""):
                self._remote_session_id = self._string_or_empty(value)
                return

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

        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,
        *,
        method: str,
        url: str,
        action: str,
        params: Mapping[str, Any],
        payload: Dict[str, Any] | None = None,
    ) -> None:
        if self._timeline is not None:
            self._timeline.emit(
                "remote_agent_request",
                backend=self._backend_label,
                backend_family="remote_sse",
                agent=self._current_agent_name(),
                method=str(method or "").upper(),
                url=url,
                action=action,
                params=dict(params),
                payload=payload,
            )

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

    def supports_server_push(self) -> bool:
        if not self._push_queue.empty():
            return True
        if not self._is_oferta_agent() or self._offer_stream_task is None:
            return False
        if not self._offer_stream_task.done():
            return True
        try:
            return self._offer_stream_task.exception() is not None
        except asyncio.CancelledError:
            return False

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

    async def wait_for_server_push(self) -> BackendReply | None:
        reply = await self._next_pushed_reply()
        if reply is None:
            return 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,
        )

    async def _next_pushed_reply(self) -> _RemoteAgentReply | None:
        try:
            return self._push_queue.get_nowait()
        except asyncio.QueueEmpty:
            pass

        if self._current_agent_name() == "conta":
            return await self._next_conta_feedback_reply()

        if not self._is_oferta_agent():
            return None

        stream_task = self._offer_stream_task
        if stream_task is None:
            return None

        if stream_task.done():
            return self._reply_from_finished_offer_stream(stream_task)

        get_task = asyncio.create_task(self._push_queue.get(), name="remote_sse_oferta_push_get")
        try:
            done, _pending = await asyncio.wait(
                {get_task, stream_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:
            return get_task.result()

        get_task.cancel()
        await asyncio.gather(get_task, return_exceptions=True)
        try:
            return self._push_queue.get_nowait()
        except asyncio.QueueEmpty:
            return self._reply_from_finished_offer_stream(stream_task)

    def _reply_from_finished_offer_stream(self, stream_task: asyncio.Task[None]) -> _RemoteAgentReply | None:
        try:
            exc = stream_task.exception()
        except asyncio.CancelledError:
            return None
        if exc is not None:
            raise RuntimeError("Remote oferta SSE stream failed") from exc
        return None

    async def _next_conta_feedback_reply(self) -> _RemoteAgentReply | None:
        saw_stream = False
        while True:
            try:
                return self._push_queue.get_nowait()
            except asyncio.QueueEmpty:
                pass

            if self._conta_feedback_streams <= 0:
                if saw_stream:
                    return None
                self._conta_feedback_event.clear()
                try:
                    await asyncio.wait_for(self._conta_feedback_event.wait(), timeout=1.0)
                except asyncio.TimeoutError:
                    return None
                continue

            saw_stream = True
            get_task = asyncio.create_task(self._push_queue.get(), name="remote_sse_conta_feedback_get")
            state_task = asyncio.create_task(
                self._conta_feedback_event.wait(),
                name="remote_sse_conta_feedback_state",
            )
            try:
                done, _pending = await asyncio.wait(
                    {get_task, state_task},
                    return_when=asyncio.FIRST_COMPLETED,
                )
            except asyncio.CancelledError:
                get_task.cancel()
                state_task.cancel()
                await asyncio.gather(get_task, state_task, return_exceptions=True)
                raise

            if get_task in done:
                state_task.cancel()
                await asyncio.gather(state_task, return_exceptions=True)
                return get_task.result()

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

    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"),
        }

        invoice_number = self._current_invoice_number()
        if invoice_number:
            payload["invoice_id"] = invoice_number
            if self._current_agent_name() == "conta":
                payload["ID_FATURA"] = invoice_number

        return payload

    def _next_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_oferta_initial_message(self) -> str:
        return (
            self._string_or_empty(os.getenv(_OFERTA_INITIAL_MESSAGE_ENV))
            or _OFERTA_INITIAL_MESSAGE_DEFAULT
        )

    def _env_first(self, *names: str) -> str:
        for name in names:
            value = self._string_or_empty(os.getenv(name))
            if value:
                return value
        return ""

    def _verify_tls(self) -> bool:
        agent_name = re.sub(r"[^A-Z0-9]+", "_", self._current_agent_name().upper()).strip("_")
        candidate_names = []
        if agent_name:
            candidate_names.extend(
                [
                    f"REMOTE_AGENT_SSE_TLS_VERIFY_{agent_name}",
                    f"REMOTE_AGENT_TLS_VERIFY_{agent_name}",
                    f"REMOTE_AGENT_SSE_VERIFY_TLS_{agent_name}",
                    f"REMOTE_AGENT_VERIFY_TLS_{agent_name}",
                ]
            )
        candidate_names.extend(
            [
                "REMOTE_AGENT_SSE_TLS_VERIFY",
                "REMOTE_AGENT_TLS_VERIFY",
                "REMOTE_AGENT_SSE_VERIFY_TLS",
                "REMOTE_AGENT_VERIFY_TLS",
            ]
        )

        raw_value = self._env_first(*candidate_names)
        if not raw_value:
            return True

        normalized = raw_value.strip().lower()
        if normalized in _FALSE_ENV_VALUES:
            return False
        if normalized in _TRUE_ENV_VALUES:
            return True
        return True

    def _build_oferta_context(self) -> Dict[str, Any]:
        context: Dict[str, Any] = {}

        explicit_session_id = self._request_field("session_id", "sessionId")
        if explicit_session_id:
            context["sessionId"] = explicit_session_id

        protocol_number = self._current_protocol_number()
        if protocol_number:
            context["protocolNumber"] = protocol_number
            context["protocolo"] = protocol_number

        gsm = self._current_msisdn()
        if gsm:
            context["gsm"] = gsm

        call_id_ged = self._request_field("callIdGed")
        if call_id_ged:
            context["uraId"] = call_id_ged
            context["callIdGed"] = call_id_ged

        optional_fields = (
            ("ani", ("ANI", "ani")),
            ("routerCallKey", ("RouterCallKey", "routerCallKey")),
            ("routerCallKeyDay", ("RouterCallKeyDay", "routerCallKeyDay")),
            ("agent", ("agent",)),
            ("assetId", ("assetId", "asset_id")),
        )
        for output_key, input_keys in optional_fields:
            value = self._request_field(*input_keys)
            if value:
                context[output_key] = value

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

        return context

    def _build_oferta_execute_payload(self, user_input: Any) -> Dict[str, Any]:
        user_text = self._extract_user_text(user_input)
        message = user_text or self._current_oferta_initial_message()
        message_id = self._message_id_from_user_input(user_input) or self._next_message_id()
        return {
            "messageId": message_id,
            "message": message,
            "context": self._build_oferta_context(),
        }

    def _build_conta_payload(
        self,
        user_input: Any,
        *,
        message_id: str = "",
    ) -> Dict[str, Any]:
        payload = {
            "message": self._extract_user_text(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_turn_payload(
        self,
        user_input: Any,
        *,
        message_id: str = "",
    ) -> Dict[str, Any]:
        if self._current_agent_name() == "conta":
            request_message_id = message_id or self._query_message_id(user_input)
            return {
                "action": "chat",
                "payload": self._build_conta_payload(
                    user_input,
                    message_id=request_message_id,
                ),
            }

        payload = self._base_payload()
        payload["message"] = self._extract_user_text(user_input)
        payload["msisdn"] = self._current_msisdn()
        payload["stage"] = self._last_stage

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

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

    def _build_end_payload(self, *, message_id: str = "") -> Dict[str, Any]:
        if self._current_agent_name() == "conta":
            request_message_id = message_id or self._query_message_id()
            return {
                "action": "end",
                "payload": self._build_conta_payload("", message_id=request_message_id),
            }

        payload = self._base_payload()
        payload["msisdn"] = self._current_msisdn()
        payload["stage"] = self._last_stage

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

        return {
            "action": "end",
            "payload": 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

        metadata = self._metadata_with_message_id(metadata, self._active_request_message_id)
        reply = _RemoteAgentReply(
            stage=stage or self._fallback_stage(),
            text=feedback_text,
            done=False,
            result=None,
            event_type="feedback",
            metadata=self._feedback_metadata(payload, metadata),
        )
        self._push_queue.put_nowait(reply)
        self._conta_feedback_event.set()

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

    def _is_conta_feedback_stream(
        self,
        *,
        method: str,
        action: str,
        payload: Mapping[str, Any] | None,
    ) -> bool:
        return (
            self._current_agent_name() == "conta"
            and str(method or "").upper() == "POST"
            and str(action or "").strip().lower() == "chat"
            and payload is not None
        )

    def _begin_conta_feedback_stream(self) -> None:
        self._conta_feedback_streams += 1
        self._conta_feedback_event.set()

    def _end_conta_feedback_stream(self) -> None:
        self._conta_feedback_streams = max(0, self._conta_feedback_streams - 1)
        self._conta_feedback_event.set()

    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 _next_sse_event(self, line_iterator: Any) -> tuple[str, Any] | None:
        event_name = ""
        data_lines: List[str] = []

        while True:
            try:
                raw_line = await line_iterator.__anext__()
            except StopAsyncIteration:
                if event_name or data_lines:
                    break
                return None

            line = str(raw_line or "")
            if not line:
                if event_name or data_lines:
                    break
                continue
            if line.startswith(":"):
                continue

            field, sep, value = line.partition(":")
            if not sep:
                field = line
                value = ""
            elif value.startswith(" "):
                value = value[1:]

            if field == "event":
                event_name = value.strip().lower()
            elif field == "data":
                data_lines.append(value)

        data = "\n".join(data_lines).strip()
        payload = self._decode_message(data) if data else {}
        return event_name, payload

    def _coerce_event_payload(self, event_name: str, payload: Any) -> Dict[str, Any]:
        if isinstance(payload, str):
            coerced = {"type": event_name or "message", "text": payload}
        elif not isinstance(payload, Mapping):
            coerced = {"type": event_name or "message"}
        else:
            coerced = dict(payload)

        if "type" not in coerced and event_name:
            coerced["type"] = event_name

        return coerced

    def _payload_to_reply(
        self,
        *,
        event_name: str,
        payload: Mapping[str, Any],
        chunks: List[str],
    ) -> tuple[_RemoteAgentReply | None, str, bool]:
        self._update_session_id(payload)
        self._update_session_actions(payload)

        message_type = str(payload.get("type") or event_name or "").strip().lower()
        prefetch_terminal = message_type in _PREFETCH_TERMINAL_TYPES or event_name in _PREFETCH_TERMINAL_TYPES

        error = payload.get("error")
        if event_name == "error" or (error and not prefetch_terminal):
            raise RuntimeError(str(error or self._extract_reply_text(payload) or "remote agent SSE error"))

        stage = self._extract_stage(payload)
        delta = self._extract_delta(payload)
        if message_type == "progress" and not delta:
            delta = self._extract_reply_text(payload)
        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),
                )

        if prefetch_terminal or message_type in _STREAM_TYPES:
            return None, message_type, prefetch_terminal

        result: Any = None
        if "result" in payload:
            result = payload.get("result")
        elif message_type in {"result", "proactive_result"}:
            result = dict(payload)
        metadata = self._reply_metadata(payload, result)

        reply_text = self._extract_reply_text(payload)
        if self._is_conta_feedback(message_type) or self._is_conta_result_feedback(result):
            self._queue_feedback_reply(
                payload=payload,
                stage=stage or self._fallback_stage(),
                text=reply_text,
                metadata=metadata,
            )
            return None, message_type, prefetch_terminal

        done = (
            final_stop_from_agent_result(result) is not None
            or message_type == "done"
            or bool(payload.get("done"))
            or stage == "DONE"
        )
        terminal = (
            bool(payload.get("final"))
            or message_type in _TERMINAL_TYPES
            or bool(reply_text)
            or result is not None
            or done
        )
        if not terminal:
            return None, message_type, prefetch_terminal

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

        reply = _RemoteAgentReply(
            stage=final_stage,
            text=text,
            done=done,
            result=result,
            event_type=message_type,
            metadata=metadata,
        )

        if self._timeline is not None:
            self._timeline.emit(
                "remote_agent_response",
                backend=self._backend_label,
                backend_family="remote_sse",
                agent=self._current_agent_name(),
                stage=reply.stage,
                done=reply.done,
                text_len=len(reply.text),
                has_result=reply.result is not None,
            )
            logger.info(
                "REMOTE_AGENT_SSE_RESPONSE | backend=%s | agent=%s | stage=%s | done=%s | text=%r",
                self._backend_label,
                self._current_agent_name(),
                reply.stage,
                reply.done,
                reply.text,
            )

        chunks.clear()
        return reply, message_type, prefetch_terminal

    def _oferta_done_result(self, payload: Mapping[str, Any]) -> tuple[Dict[str, Any], bool]:
        status = str(payload.get("status") or "").strip().lower()
        additional = payload.get("additionalInformations")
        if not isinstance(additional, Mapping):
            additional = {}

        service_status = str(additional.get("service_status") or "").strip().upper()
        if status == "transferred":
            return (
                {
                    "type": "transferred",
                    "content": "",
                    "status": str(payload.get("status") or "").strip() or "completed",
                    "additionalInformations": dict(additional),
                },
                False,
            )

        if service_status in _OFERTA_NON_FINAL_SERVICE_STATUSES:
            return (
                {
                    "type": service_status.lower(),
                    "content": "",
                    "status": str(payload.get("status") or "").strip() or "completed",
                    "additionalInformations": dict(additional),
                },
                False,
            )

        result_type = _OFERTA_SERVICE_STATUS_RESULT_TYPE.get(service_status)
        if not result_type:
            return (
                {
                    "type": service_status.lower() if service_status else "completed",
                    "content": "",
                    "status": str(payload.get("status") or "").strip() or "completed",
                    "additionalInformations": dict(additional),
                },
                False,
            )

        return (
            {
                "type": result_type,
                "content": "",
                "status": str(payload.get("status") or "").strip() or "completed",
                "additionalInformations": dict(additional),
            },
            True,
        )

    def _oferta_payload_to_reply(
        self,
        *,
        event_name: str,
        payload: Mapping[str, Any],
    ) -> _RemoteAgentReply | None:
        normalized_event = str(event_name or payload.get("type") or "").strip().lower()
        self._update_session_id(payload)

        if normalized_event == "schedule_message":
            scheduled = payload.get("scheduledMessage")
            if not isinstance(scheduled, Mapping):
                scheduled = {}
            text = self._extract_text_from_value(scheduled.get("message"))
            if not text:
                return None
            return _RemoteAgentReply(
                stage=self._fallback_stage(),
                text=text,
                done=False,
                result=None,
                event_type="schedule_message",
                metadata={
                    "event": "schedule_message",
                    "scheduledMessage": dict(scheduled),
                    "payload": dict(payload),
                    **(self._reply_metadata(payload) or {}),
                },
            )

        if normalized_event == "message":
            text = (
                self._extract_text_from_value(payload.get("response"))
                or self._extract_reply_text(payload)
            )
            if not text:
                return None
            metadata: Dict[str, Any] = {
                "event": "message",
                "payload": dict(payload),
            }
            speech_metadata = self._reply_metadata(payload)
            if speech_metadata:
                metadata.update(speech_metadata)
            if payload.get("sessionId") not in (None, ""):
                metadata["sessionId"] = self._string_or_empty(payload.get("sessionId"))
            additional = payload.get("additionalInformations")
            if isinstance(additional, Mapping):
                metadata["additionalInformations"] = dict(additional)
            return _RemoteAgentReply(
                stage=self._fallback_stage(),
                text=text,
                done=False,
                result=None,
                event_type="message",
                metadata=metadata,
            )

        if normalized_event == "done":
            additional = payload.get("additionalInformations")
            if not isinstance(additional, Mapping):
                additional = {}
            result, finalizes_conversation = self._oferta_done_result(payload)
            metadata = {
                "event": "done",
                "payload": dict(payload),
                "additionalInformations": dict(additional),
            }
            return _RemoteAgentReply(
                stage="DONE" if finalizes_conversation else self._fallback_stage(),
                text="",
                done=finalizes_conversation,
                result=result,
                event_type="done",
                metadata=metadata,
            )

        error = payload.get("error")
        if normalized_event == "error" or error:
            raise RuntimeError(str(error or self._extract_reply_text(payload) or "remote oferta SSE error"))

        return None

    async def _consume_oferta_execute_stream(
        self,
        *,
        url: str,
        payload: Dict[str, Any],
    ) -> None:
        self._emit_request(method="POST", url=url, action="execute", params={}, payload=payload)
        request_started = time.perf_counter()
        saw_event = False
        saw_done = False

        async with httpx.AsyncClient(
            follow_redirects=True,
            timeout=self._client_timeout(),
            verify=self._verify_tls(),
        ) as client:
            async with client.stream(
                "POST",
                url,
                params={},
                headers=self._oferta_headers(),
                json=payload,
            ) as response:
                response.raise_for_status()
                line_iterator = response.aiter_lines()
                while True:
                    event = await self._next_sse_event(line_iterator)
                    if event is None:
                        break

                    saw_event = True
                    event_name, raw_payload = event
                    event_payload = self._coerce_event_payload(event_name, raw_payload)
                    if self._timeline is not None:
                        self._timeline.emit(
                            "remote_agent_response_raw",
                            backend=self._backend_label,
                            backend_family="remote_sse",
                            agent=self._current_agent_name(),
                            sse_event=event_name or event_payload.get("type", ""),
                            response_payload=dict(event_payload),
                        )
                    reply = self._oferta_payload_to_reply(
                        event_name=event_name,
                        payload=event_payload,
                    )
                    if reply is not None:
                        self._reply_with_message_id(reply, payload.get("messageId"))
                        if self._timeline is not None:
                            self._timeline.emit(
                                "remote_agent_response",
                                backend=self._backend_label,
                                backend_family="remote_sse",
                                agent=self._current_agent_name(),
                                sse_event=reply.event_type,
                                stage=reply.stage,
                                done=reply.done,
                                text_len=len(reply.text),
                                has_result=reply.result is not None,
                            )
                        logger.info(
                            "REMOTE_AGENT_SSE_RESPONSE | backend=%s | agent=%s | event=%s | stage=%s | done=%s | duration_ms=%s | text=%r",
                            self._backend_label,
                            self._current_agent_name(),
                            reply.event_type,
                            reply.stage,
                            reply.done,
                            round((time.perf_counter() - request_started) * 1000),
                            reply.text,
                        )
                        await self._push_queue.put(reply)
                        if reply.event_type == "done":
                            saw_done = True
                            break

                if not saw_event:
                    raise RuntimeError("Remote oferta SSE stream closed without a usable response")
                if not saw_done:
                    raise RuntimeError("Remote oferta SSE stream closed before done")

    async def _start_oferta_execute_stream_locked(self, payload: Dict[str, Any]) -> _RemoteAgentReply:
        await self._cancel_oferta_stream_locked()
        self._clear_push_queue()
        url = self._resolve_url()
        self._offer_stream_task = asyncio.create_task(
            self._consume_oferta_execute_stream(url=url, payload=payload),
            name="remote_sse_oferta_stream",
        )
        reply = await self._next_pushed_reply()
        if reply is None:
            raise RuntimeError("Remote oferta SSE stream closed without a usable response")
        return reply

    async def _cancel_oferta_stream_locked(self) -> None:
        task = self._offer_stream_task
        if task is None:
            return
        if not task.done():
            task.cancel()
            await asyncio.gather(task, return_exceptions=True)
        else:
            try:
                task.exception()
            except asyncio.CancelledError:
                pass
        self._offer_stream_task = None

    async def _consume_sse_response(
        self,
        line_iterator: Any,
        *,
        stop_on_first_result: bool,
        stop_on_ready: bool,
        stop_on_prefetch_terminal: bool,
    ) -> list[_RemoteAgentReply]:
        chunks: List[str] = []
        replies: list[_RemoteAgentReply] = []
        saw_event = False

        while True:
            event = await self._next_sse_event(line_iterator)
            if event is None:
                break

            saw_event = True
            event_name, raw_payload = event
            payload = self._coerce_event_payload(event_name, raw_payload)
            reply, message_type, prefetch_terminal = self._payload_to_reply(
                event_name=event_name,
                payload=payload,
                chunks=chunks,
            )
            if reply is not None:
                replies.append(reply)
                if stop_on_ready and message_type == "ready":
                    break
                if stop_on_first_result and message_type in {"result", "proactive_result", "done", "final"}:
                    break

            if stop_on_prefetch_terminal and prefetch_terminal:
                break

        if chunks:
            replies.append(
                _RemoteAgentReply(
                    stage=self._fallback_stage(),
                    text="".join(chunks).strip(),
                    done=False,
                    result=None,
                    event_type="progress",
                )
            )

        if not saw_event:
            raise RuntimeError("Remote agent SSE stream closed without a usable response")

        return replies

    async def _stream_request(
        self,
        *,
        method: str,
        url: str,
        params: Mapping[str, Any],
        action: str,
        payload: Dict[str, Any] | None = None,
        stop_on_first_result: bool = False,
        stop_on_ready: bool = False,
        stop_on_prefetch_terminal: bool = False,
    ) -> list[_RemoteAgentReply]:
        self._emit_request(method=method, url=url, action=action, params=params, payload=payload)
        request_started = time.perf_counter()
        feedback_stream = self._is_conta_feedback_stream(
            method=method,
            action=action,
            payload=payload,
        )
        request_message_id = self._message_id_from_mapping(params)
        previous_request_message_id = self._active_request_message_id
        self._active_request_message_id = request_message_id
        if feedback_stream:
            self._begin_conta_feedback_stream()
        try:
            async with httpx.AsyncClient(
                follow_redirects=True,
                timeout=self._client_timeout(),
                verify=self._verify_tls(),
            ) as client:
                kwargs: Dict[str, Any] = {
                    "params": dict(params),
                    "headers": self._stream_headers(),
                }
                if payload is not None:
                    kwargs["json"] = payload

                stream_cm = client.stream(method, url, **kwargs)
                async with stream_cm as response:
                    response.raise_for_status()
                    replies = await self._consume_sse_response(
                        response.aiter_lines(),
                        stop_on_first_result=stop_on_first_result,
                        stop_on_ready=stop_on_ready,
                        stop_on_prefetch_terminal=stop_on_prefetch_terminal,
                    )
                    logger.info(
                        "REMOTE_AGENT_SSE_REQUEST_DONE | backend=%s | agent=%s | method=%s | action=%s | duration_ms=%s | replies=%s",
                        self._backend_label,
                        self._current_agent_name(),
                        str(method or "").upper(),
                        action,
                        round((time.perf_counter() - request_started) * 1000),
                        len(replies),
                    )
                    if request_message_id:
                        for reply in replies:
                            self._reply_with_message_id(reply, request_message_id)
                    return replies
        finally:
            self._active_request_message_id = previous_request_message_id
            if feedback_stream:
                self._end_conta_feedback_stream()

    @staticmethod
    def _select_ready_reply(replies: list[_RemoteAgentReply]) -> _RemoteAgentReply | None:
        for reply in replies:
            if reply.event_type == "ready":
                return reply
        return replies[0] if replies else None

    @staticmethod
    def _select_action_reply(replies: list[_RemoteAgentReply]) -> _RemoteAgentReply:
        for reply in replies:
            if reply.event_type in {"result", "proactive_result", "done", "final"}:
                return reply
        for reply in replies:
            if reply.event_type != "ready":
                return reply
        if replies:
            return replies[-1]
        raise RuntimeError("Remote agent SSE stream closed without a usable response")

    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_sse",
                agent=self._current_agent_name(),
            )
        self._elegibility = bool(elegibility)
        self._protocol = str(protocol or "")
        self._remote_session_id = ""
        self._turn_seq = 0
        self._last_stage = "INTRO"
        self._pending_interrupt = None
        self._pending_events.clear()
        self._pending_ready_reply = None
        self._session_actions = None
        self._conta_feedback_streams = 0
        self._conta_feedback_event.clear()
        async with self._session_lock:
            await self._cancel_oferta_stream_locked()
        self._clear_push_queue()

        if self._is_oferta_agent():
            if self._timeline is not None:
                self._timeline.emit(
                    "backend_prepare_completed",
                    backend=self._backend_label,
                    backend_family="remote_sse",
                    agent=self._current_agent_name(),
                    remote_session_id="",
                )
            return

        async with self._session_lock:
            url = self._resolve_url()
            replies = await self._stream_request(
                method="GET",
                url=url,
                params=self._stream_open_params(),
                action="connect",
                stop_on_first_result=False,
                stop_on_ready=True,
                stop_on_prefetch_terminal=True,
            )
            self._pending_ready_reply = self._select_ready_reply(replies)
            for reply in replies:
                if reply is not self._pending_ready_reply and reply.text:
                    await self._push_queue.put(reply)

        if self._timeline is not None:
            self._timeline.emit(
                "backend_prepare_completed",
                backend=self._backend_label,
                backend_family="remote_sse",
                agent=self._current_agent_name(),
                remote_session_id=self._remote_session_id,
            )

    async def run(self, user_input: Any) -> BackendReply:
        user_text = self._extract_user_text(user_input)
        sent_payload = False

        async with self._session_lock:
            if self._is_oferta_agent():
                payload = self._build_oferta_execute_payload(user_input)
                reply = await self._start_oferta_execute_stream_locked(payload)
                sent_payload = True
            elif self._pending_ready_reply is not None and not user_text:
                reply = self._pending_ready_reply
                self._pending_ready_reply = None
            elif not user_text:
                self._pending_ready_reply = None
                reply = _RemoteAgentReply(
                    stage=self._fallback_stage(),
                    text="",
                    done=False,
                    result=None,
                    event_type="noop",
                )
            else:
                self._pending_ready_reply = None
                message_id = (
                    self._query_message_id(user_input)
                    if self._current_agent_name() == "conta"
                    else ""
                )
                payload = self._build_turn_payload(user_input, message_id=message_id)
                replies = await self._stream_request(
                    method="POST",
                    url=self._resolve_url(),
                    params=self._action_params(user_input, message_id=message_id),
                    action=str(payload.get("action") or "chat"),
                    payload=payload,
                    stop_on_first_result=True,
                    stop_on_prefetch_terminal=False,
                )
                reply = self._select_action_reply(replies)
                sent_payload = True

        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_sse",
                    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_sse",
                    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_sse",
                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_sse",
                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._is_oferta_agent():
                async with self._session_lock:
                    await self._cancel_oferta_stream_locked()
                self._end_reply = BackendReply(stage="DONE", done=True, export_payload=[])
                return self._end_reply

            if self._timeline is not None:
                self._timeline.emit(
                    "backend_end_started",
                    backend=self._backend_label,
                    backend_family="remote_sse",
                    agent=self._current_agent_name(),
                )
            try:
                async with self._session_lock:
                    if not self._session_supports_action("end"):
                        reply = _RemoteAgentReply(
                            stage="DONE",
                            text="",
                            done=True,
                            result=[],
                            event_type="done",
                        )
                    else:
                        message_id = (
                            self._query_message_id()
                            if self._current_agent_name() == "conta"
                            else ""
                        )
                        payload = self._build_end_payload(message_id=message_id)
                        try:
                            replies = await self._stream_request(
                                method="POST",
                                url=self._resolve_url(),
                                params=self._action_params(message_id=message_id),
                                action="end",
                                payload=payload,
                                stop_on_first_result=True,
                                stop_on_prefetch_terminal=False,
                            )
                            reply = self._select_action_reply(replies)
                        except RuntimeError as exc:
                            if not self._is_unsupported_action_error(exc, "end"):
                                raise
                            reply = _RemoteAgentReply(
                                stage="DONE",
                                text="",
                                done=True,
                                result=[],
                                event_type="done",
                            )
            except Exception:
                logger.exception("[remote-agent] end_service SSE falhou")
                if self._timeline is not None:
                    self._timeline.emit(
                        "backend_end_failed",
                        backend=self._backend_label,
                        backend_family="remote_sse",
                        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_sse",
                    agent=self._current_agent_name(),
                    result_type=type(self._end_reply.export_payload).__name__,
                )
            return self._end_reply
