from __future__ import annotations import asyncio import json from typing import Any, cast from urllib.parse import parse_qsl import websockets from fastapi import WebSocket, WebSocketDisconnect from websockets.asyncio.client import ClientConnection from websockets.typing import Subprotocol from vocero_api.core import config as settings from vocero_api.core import logging as app_logging from vocero_api.core import urls as core_urls from vocero_api.providers import caller as providers_caller from vocero_api.realtime import events as realtime_events from vocero_api.realtime.mcp_proxy import realtime_mcp_proxy from vocero_api.services import context as services_context async def realtime_proxy(ws: WebSocket) -> None: """Relay the browser's same-origin realtime WebSocket to the configured upstream.""" if settings.S2S_MCP_URL: await realtime_mcp_proxy(ws) return client_ip = ws.client.host if ws.client else "?" if not settings.S2S_PROXY_TARGET: app_logging.logger.debug("[EXT][S2S-PROXY] client=%s rejected: S2S_PROXY_TARGET not configured", client_ip) await ws.accept() await ws.close(code=1013, reason="El proxy de tiempo real no está configurado.") return await ws.accept() upstream_url = core_urls._proxy_upstream_url(ws) subprotocols = _subprotocols(ws) # ── S2S_INJECT_PROFILE: resolve caller profile + podcast episode + latest summary ── # Query params (browser URL carries ?phoneNumber=...&episode_id=...) are forwarded by the # client WS library: we read ?phoneNumber= and ?episode_id= from the incoming WS query. # When S2S_INJECT_PROFILE=false or envs missing -> ctx fields stay empty # and the tunnel stays 100% pure passthrough (original behaviour). injected_episode_context = "" injected_profile_context = "" injected_reports_context = "" injected_summary_context = "" inject_phone = "" inject_episode_id = "" client_query = {} login_token: str | None = None profile_obj: dict | None = None try: client_query = dict(parse_qsl(ws.url.query, keep_blank_values=True)) except Exception: # noqa: BLE001 client_query = {} if not settings.S2S_INJECT_PROFILE: app_logging.logger.info( "[INJECT-PROFILE] ⏭️ WS connect: SKIPPED (S2S_INJECT_PROFILE=false). " "No profile/episode fetch; tunnel stays pure passthrough for context injection." ) else: needs_phone_authed = bool(settings.VOCERO_USER_PHONE_AUTH_URL.strip()) # 2026-09-11 user fix: remove mass injection of scheduled reports. # SAME gate as the REASONING endpoint above. "needs_reports" becomes True # only when BOTH the URL env is configured AND the new opt-in S2S_INJECT_REPORTS # flag is true. With the default (S2S_INJECT_REPORTS=false) this entire 1.5) # branch of the WS connect path is skipped — reports-list fetch never runs. needs_reports = bool(settings.S2S_REPORT_LIST.strip()) and settings.S2S_INJECT_REPORTS needs_summary = bool(settings.S2S_LATEST_SUMMARY.strip()) or bool(settings.S2S_LATEST_SUMMARY_TEXT) needs_login = bool(settings.S2S_LOGIN_BY_PHONE.strip()) inject_phone = (client_query.get("phoneNumber") or "").strip() inject_episode_id = (client_query.get("episode_id") or "").strip() can_inject = needs_phone_authed or needs_reports or needs_summary or needs_login or bool(inject_episode_id) if not can_inject: app_logging.logger.info( "[INJECT-PROFILE] ⏭️ WS connect: SKIPPED (VOCERO_USER_PHONE_AUTH_URL, S2S_REPORT_LIST, S2S_LATEST_SUMMARY/S2S_LATEST_SUMMARY_TEXT and S2S_LOGIN_BY_PHONE all empty, AND no episode_id in query; OR S2S_INJECT_REPORTS is false so scheduled report injection is globally disabled). " "Set at least one AND set S2S_INJECT_REPORTS=true (if you want scheduled reports) to enable caller context / reports / login / episode." ) elif ( (not inject_phone) and (not needs_reports) and (not inject_episode_id) and (not settings.S2S_LATEST_SUMMARY_TEXT) ): # reports list can be fetched without phone (generic list); episode can be fetched without phone; # S2S_LATEST_SUMMARY_TEXT also works without phone (static text override) app_logging.logger.info( "[INJECT-PROFILE] ⏭️ WS connect: SKIPPED (incoming WS query has NO ?phoneNumber= AND no ?episode_id=, AND S2S_REPORT_LIST+S2S_INJECT_REPORTS not enabled). " "Make sure the browser opens the WS with the phone or episode_id query param, or set S2S_INJECT_REPORTS=true if you want scheduled reports without phone." ) else: # 0) Login-by-phone (optional — only if S2S_LOGIN_BY_PHONE env is set). # Obtains a bearer token used on subsequent profile/reports/summary fetches. if needs_login and inject_phone: login_token, _ = await providers_caller._login_by_phone(inject_phone) # 0.5) Podcast episode fetch (independent of phone — uses only episode_id from query) if inject_episode_id: if settings.S2S_DEBUG_PROMPT: app_logging.logger.info( "[INJECT-EPISODE] WS connect: starting podcast episode fetch episodeId=%.80s...", inject_episode_id, ) episode_raw = await providers_caller._fetch_podcast_episode(inject_episode_id) injected_episode_context = services_context._resolve_episode_context(episode_raw) if injected_episode_context.strip(): app_logging.logger.info( "[INJECT-EPISODE] ✅ WS connect: PODCAST EPISODE READY episodeId=%.80s wrappedLen=%d -> will patch session.update frames in this tunnel.", inject_episode_id, len(injected_episode_context), ) else: app_logging.logger.info( "[INJECT-EPISODE] ⏭️ WS connect: episode skipped (context empty / upstream returned empty / fetch failed)." ) # 1) Profile fetch (optional — only if env is set) if needs_phone_authed and inject_phone: profile_obj = await providers_caller._fetch_user_profile_for_phone(inject_phone, auth_token=login_token) injected_profile_context = services_context._resolve_user_context_text( inject_phone, profile_obj, loaded=True ) if injected_profile_context.strip(): app_logging.logger.info( "[INJECT-PROFILE] ✅ WS connect: profile READY phoneLen=%d contextLen=%d -> will patch all client session.update frames in this tunnel.", len(inject_phone), len(injected_profile_context), ) else: app_logging.logger.info( "[INJECT-PROFILE] ⏭️ WS connect: profile fetch OK but resolved empty. Tunnel stays pure passthrough for profile." ) # 1.5) Reports list fetch (optional — only if env is set) if needs_reports: if settings.S2S_DEBUG_PROMPT: app_logging.logger.info( "[INJECT-REPORTS] WS connect: starting S2S_REPORT_LIST fetch (phoneLen=%s bearer=%s)...", len(inject_phone) if inject_phone else "none", "yes" if login_token else "no", ) reports_raw = await providers_caller._fetch_reports_list( phone_number=inject_phone if inject_phone else None, auth_token=login_token, ) injected_reports_context = services_context._resolve_reports_context(reports_raw) if injected_reports_context.strip(): app_logging.logger.info( "[INJECT-REPORTS] ✅ WS connect: reports list READY wrappedLen=%d -> will patch session.update frames in this tunnel.", len(injected_reports_context), ) else: app_logging.logger.info( "[INJECT-REPORTS] ⏭️ WS connect: reports list skipped (context empty / upstream returned empty list / fetch failed)." ) # 2) Latest summary fetch (optional — only if env is set) if needs_summary and (inject_phone or settings.S2S_LATEST_SUMMARY_TEXT): if settings.S2S_DEBUG_PROMPT: app_logging.logger.info( "[INJECT-SUMMARY] WS connect: starting summary load (S2S_LATEST_SUMMARY=%s S2S_LATEST_SUMMARY_TEXT=%s phoneLen=%s)...", bool(settings.S2S_LATEST_SUMMARY), bool(settings.S2S_LATEST_SUMMARY_TEXT), len(inject_phone) if inject_phone else "none", ) latest_summary_raw = await providers_caller._fetch_latest_summary_for_phone( inject_phone, auth_token=login_token ) injected_summary_context = services_context._resolve_latest_summary_context(latest_summary_raw) if injected_summary_context.strip(): app_logging.logger.info( "[INJECT-SUMMARY] ✅ WS connect: latest SUMMARY READY phoneLen=%s summaryTextOverride=%s wrappedLen=%d -> will patch session.update frames in this tunnel.", len(inject_phone) if inject_phone else "none", bool(settings.S2S_LATEST_SUMMARY_TEXT), len(injected_summary_context), ) else: app_logging.logger.info( "[INJECT-SUMMARY] ⏭️ WS connect: latest summary skipped (context empty / upstream returned empty summary / fetch failed)." ) app_logging.logger.info( "[INJECT-PROFILE] WS connect FINAL: injectEpisode=%s injectProfile=%s injectReports=%s injectSummary=%s login=%s episodeIdLen=%d phoneLen=%d episodeCtxLen=%d profileCtxLen=%d reportsCtxLen=%d summaryCtxLen=%d target=%s", bool(injected_episode_context), bool(injected_profile_context), bool(injected_reports_context), bool(injected_summary_context), ("token-present" if login_token else "skipped"), len(inject_episode_id), len(inject_phone), len(injected_episode_context), len(injected_profile_context), len(injected_reports_context), len(injected_summary_context), upstream_url, ) if settings.S2S_DEBUG_PROMPT and injected_episode_context.strip(): app_logging.logger.info( "[DEBUG-EPISODE][WS connect] FULL podcast episode block loaded (len=%d):\n%s\n", len(injected_episode_context), injected_episode_context, ) if settings.S2S_DEBUG_PROMPT and injected_reports_context.strip(): app_logging.logger.info( "[DEBUG-REPORTS][WS connect] FULL reports list block loaded (len=%d):\n%s\n", len(injected_reports_context), injected_reports_context, ) if settings.S2S_DEBUG_PROMPT and injected_summary_context.strip(): app_logging.logger.info( "[DEBUG-SUMMARY][WS connect] FULL latest summary block loaded (len=%d):\n%s\n", len(injected_summary_context), injected_summary_context, ) try: async with websockets.connect( upstream_url, subprotocols=cast(list[Subprotocol], subprotocols) or None, open_timeout=15, max_size=None, ) as upstream: app_logging.logger.info("[INJECT-PROFILE] WS tunnel ESTABLISHED client=%s", client_ip) session_ctx = { "injected_episode_context": injected_episode_context, "injected_profile_context": injected_profile_context, "injected_reports_context": injected_reports_context, "injected_summary_context": injected_summary_context, # 2026-09-11 proper-name anglicization guard: keep the full # upstream profile object so: # (a) _build_voice_rules_head_lock_block can emit rule #5 # "Exact name: Alessandro Valenti — NEVER Alexander" # (b) layer 7 post-processing _sanitize_agent_output can # deterministically replace Alexander/Alexandre back to # Alessandro with word-boundary regex even when the # prompt lock fails on GPTQ. "profile_obj": profile_obj, } client_to_upstream = asyncio.create_task(_client_to_upstream(ws, upstream, client_ip, session_ctx)) upstream_to_client = asyncio.create_task(_upstream_to_client(ws, upstream, client_ip, session_ctx)) done, pending = await asyncio.wait( {client_to_upstream, upstream_to_client}, return_when=asyncio.FIRST_COMPLETED, ) for task in pending: task.cancel() await asyncio.gather(*pending, return_exceptions=True) for task in done: task_exc = task.exception() if task_exc: raise task_exc except WebSocketDisconnect: app_logging.logger.debug("[EXT][S2S-PROXY] client=%s disconnected (WSDisconnect)", client_ip) return except Exception as exc: app_logging.logger.warning("Realtime proxy failed: %r", exc) app_logging.logger.debug("[EXT][S2S-PROXY] client=%s ERROR %s", client_ip, type(exc).__name__) try: await ws.close(code=1011, reason="El servicio de tiempo real no está disponible.") except Exception: pass finally: app_logging.logger.debug("[EXT][S2S-PROXY] client=%s TUNNEL CLOSED", client_ip) def _subprotocols(ws: WebSocket) -> list[str]: raw = ws.headers.get("sec-websocket-protocol", "") return [item.strip() for item in raw.split(",") if item.strip()] async def _client_to_upstream( ws: WebSocket, upstream: ClientConnection, client_ip: str, session_ctx: dict[str, Any] ) -> None: while True: message = await ws.receive() if message["type"] == "websocket.disconnect": # #region debug-point B:client-disconnect await realtime_events._report_server_debug( "B", "server.py:_client_to_upstream", "client websocket disconnected", {"code": message.get("code")}, ) # #endregion app_logging.logger.debug( "[EXT][S2S-PROXY] client=%s client→upstream DISCONNECT code=%s", client_ip, message.get("code"), ) await upstream.close() return if message.get("text") is not None: # #region debug-point B:client-text-forward summary = realtime_events._realtime_event_summary(message["text"]) if summary: await realtime_events._report_server_debug( "B", "server.py:_client_to_upstream", "forwarding client event to upstream", summary, ) event_type = summary.get("type") # Client-side user input: typed text and inserted items. user_text = "" if event_type == "conversation.item.created" and summary.get("role") == "user": user_text = (summary.get("text") or "").strip() if user_text: app_logging.logger.info( "[USER INPUT] client=%s text=%s", client_ip, user_text, ) if ( event_type not in realtime_events._NOISY_CLIENT_EVENT_TYPES and event_type not in realtime_events._NOISY_REALTIME_EVENT_TYPES ): app_logging.logger.debug( "[EXT][S2S-PROXY] client=%s →UPSTREAM text event=%s summary=%s", client_ip, event_type, json.dumps(summary, ensure_ascii=True)[:300], ) # #endregion # --- S2S_INJECT_PROFILE: if this is a session.update client message # and we have a resolved caller profile / podcast episode / latest caller summary, # patch instructions to include them before forwarding upstream. # Pure no-op when S2S_INJECT_PROFILE=false (original pure # passthrough behaviour). ctx_episode_context = session_ctx.get("injected_episode_context") if isinstance(session_ctx, dict) else "" ctx_profile_context = session_ctx.get("injected_profile_context") if isinstance(session_ctx, dict) else "" ctx_reports_context = session_ctx.get("injected_reports_context") if isinstance(session_ctx, dict) else "" ctx_summary_context = session_ctx.get("injected_summary_context") if isinstance(session_ctx, dict) else "" ctx_profile_obj = session_ctx.get("profile_obj") if isinstance(session_ctx, dict) else None forward_text = services_context._patch_client_session_update_with_profile( message["text"], ctx_profile_context or "", ctx_reports_context or "", ctx_summary_context or "", ctx_episode_context or "", profile_obj=ctx_profile_obj, ) await upstream.send(forward_text) elif message.get("bytes") is not None: # #region debug-point B:client-bytes-forward await realtime_events._report_server_debug( "B", "server.py:_client_to_upstream", "forwarding client binary frame to upstream", {"bytes": len(message["bytes"])}, ) app_logging.logger.debug( "[EXT][S2S-PROXY] client=%s →UPSTREAM bytes=%dB", client_ip, len(message["bytes"]), ) # #endregion await upstream.send(message["bytes"]) async def _upstream_to_client( ws: WebSocket, upstream: ClientConnection, client_ip: str, session_ctx: dict[str, Any] ) -> None: async for message in upstream: if isinstance(message, bytes): # #region debug-point B:upstream-bytes-forward await realtime_events._report_server_debug( "B", "server.py:_upstream_to_client", "forwarding upstream binary frame to client", {"bytes": len(message)}, ) app_logging.logger.debug( "[EXT][S2S-PROXY] client=%s UPSTREAM→ bytes=%dB", client_ip, len(message), ) # #endregion await ws.send_bytes(message) else: # Remove internal context markers before forwarding text to the client. message, was_sanitized = realtime_events._sanitize_realtime_event_raw(message, session_ctx) # #region debug-point B:upstream-text-forward summary = realtime_events._realtime_event_summary(message) if summary: await realtime_events._report_server_debug( "B", "server.py:_upstream_to_client", "forwarding upstream event to client", summary, ) event_type = summary.get("type") # User input from upstream server-side STT. user_text = "" if event_type == "conversation.item.input_audio_transcription.completed": user_text = (summary.get("transcript") or "").strip() if user_text: app_logging.logger.info( "[USER INPUT] client=%s stt=%s", client_ip, user_text, ) elif event_type == "conversation.item.created" and summary.get("role") == "user": user_text = (summary.get("text") or "").strip() if user_text: app_logging.logger.info( "[USER INPUT] client=%s text=%s", client_ip, user_text, ) # Agent output, including all event variants. agent_text = "" if event_type in {"response.audio_transcript.done", "response.output_audio_transcript.done"}: # Use the sanitized agent text for UI and logs. agent_text = (summary.get("text") or "").strip() if agent_text: app_logging.logger.info( "[AGENT OUTPUT] client=%s tts=%s", client_ip, agent_text, ) else: raw_preview = summary.get("_raw_preview") or json.dumps(summary, ensure_ascii=True) app_logging.logger.debug( "[AGENT OUTPUT][INSPECTION] client=%s %s has summary=%s raw=%s", client_ip, event_type, json.dumps(summary, ensure_ascii=True)[:400], raw_preview[:800], ) elif event_type == "response.content_part.done": agent_text = (summary.get("text") or "").strip() if agent_text: app_logging.logger.info( "[AGENT OUTPUT] client=%s content_part=%s", client_ip, agent_text, ) elif event_type == "response.output_item.done" and summary.get("role") == "assistant": agent_text = (summary.get("text") or "").strip() if agent_text: app_logging.logger.info( "[AGENT OUTPUT] client=%s output_item=%s", client_ip, agent_text, ) elif event_type == "conversation.item.created" and summary.get("role") == "assistant": agent_text = (summary.get("text") or "").strip() if agent_text: app_logging.logger.info( "[AGENT OUTPUT] client=%s created=%s", client_ip, agent_text, ) if event_type not in realtime_events._NOISY_REALTIME_EVENT_TYPES: app_logging.logger.debug( "[EXT][S2S-PROXY] client=%s UPSTREAM→ event=%s summary=%s", client_ip, event_type, json.dumps(summary, ensure_ascii=True)[:300], ) # Suppress internal tools[] injection errors from the browser. # Strategy B currently fails because ai-server does not support tools[]. # Inspect both message and nested type in these upstream formats: # (A) Standard wrapper observed in logs: # {"type":"error","error":{"message":"Unknown or invalid event: session.update","type":"unknown_or_invalid_event","code":null,...}} # (B) Flat fallback from other compatible upstreams: # {"type":"error","message":"...","code":"..."} if isinstance(message, str): try: _evt = json.loads(message) except Exception: _evt = None if isinstance(_evt, dict): _evt_type = _evt.get("type") if _evt_type == "error": # Prefer the nested wrapper for message and error_type. _err_msg = "" _err_type = "" _inner = _evt.get("error") if isinstance(_inner, dict): for _k in ("message", "error", "msg"): _v = _inner.get(_k) if isinstance(_v, str) and _v.strip(): _err_msg = _v.strip() break _t = _inner.get("type") if isinstance(_t, str): _err_type = _t # Fall back to flat fields when the wrapper has no message. if not _err_msg: for _k in ("message", "error", "msg"): _v = _evt.get(_k) if isinstance(_v, str) and _v.strip(): _err_msg = _v.strip() break if not _err_type: _t = _evt.get("code") if isinstance(_t, str): _err_type = _t # Decide whether to suppress the error. _suppress = False _suppress_reason = "" if "Unknown or invalid event: session.update" in _err_msg: _suppress = True _suppress_reason = ( "injected session.update with tools[] rejected — ai-server currently does not support " "tools[]. Log only on the server and do not forward to the browser." ) elif "session.update" in _err_msg and ( "tool" in _err_msg.lower() or "invalid" in _err_msg.lower() ): _suppress = True _suppress_reason = f"upstream rejected session.update (message): {_err_msg[:160]}. Do not forward to the browser." elif _err_type == "unknown_or_invalid_event" and "session.update" in _err_msg: _suppress = True _suppress_reason = f"upstream rejected session.update (type=unknown_or_invalid_event): {_err_msg[:160]}. Do not forward to the browser." if _suppress: app_logging.logger.info( "[WS SUPPRESS] suppressed upstream error (not sent to browser): %s", _suppress_reason, ) app_logging.logger.debug( "[WS SUPPRESS] full upstream suppressed error payload=%s", json.dumps(_evt, ensure_ascii=True)[:600], ) return # Log unsuppressed errors with message/type to assess new rules. app_logging.logger.debug( "[WS SUPPRESS] upstream error NOT suppressed (passthrough to browser) — " "evt.error.message=%s evt.error.type=%s evtTopKeys=%s", _err_msg[:200] if _err_msg else "(empty)", _err_type or "(none)", sorted(_evt.keys()), ) # #endregion await ws.send_text(message)