1373 lines
76 KiB
Python
1373 lines
76 KiB
Python
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 s2s_client.core import config as settings
|
|
from s2s_client.core import logging as app_logging
|
|
from s2s_client.core import urls as core_urls
|
|
from s2s_client.providers import apipilot as providers_apipilot
|
|
from s2s_client.providers import caller as providers_caller
|
|
from s2s_client.realtime import events as realtime_events
|
|
from s2s_client.services import context as services_context
|
|
from s2s_client.services import reasoning as services_reasoning
|
|
from s2s_client.services import tool_context as services_tool_context
|
|
|
|
|
|
async def realtime_proxy(ws: WebSocket) -> None:
|
|
"""Relay the browser's same-origin realtime WebSocket to the configured upstream."""
|
|
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 = ""
|
|
injected_apipilot_pass1 = ""
|
|
apipilot_tools_pass1: list = []
|
|
inject_phone = ""
|
|
inject_episode_id = ""
|
|
client_query = {}
|
|
login_token: str | None = None
|
|
login_response_obj: dict | None = None
|
|
profile_obj: dict | None = None
|
|
needs_apipilot = bool(settings.S2S_APIPILOT_TOOLS_URL) or bool(settings.S2S_APIPILOT_TOOLS_EXECUTE_URL)
|
|
try:
|
|
client_query = dict(parse_qsl(ws.url.query, keep_blank_values=True))
|
|
except Exception: # noqa: BLE001
|
|
client_query = {}
|
|
|
|
# ApiPilot works independently of profile injection. When enabled, extract
|
|
# phone/episode query parameters, resolve the bearer for authenticated tools,
|
|
# and load the profile for email defaults when its endpoint is configured.
|
|
if needs_apipilot:
|
|
inject_phone = (client_query.get("phoneNumber") or "").strip() or inject_phone
|
|
inject_episode_id = (client_query.get("episode_id") or "").strip() or inject_episode_id
|
|
_needs_login_apipilot = bool(settings.S2S_LOGIN_BY_PHONE.strip())
|
|
_needs_profile_apipilot = bool(settings.VOCERO_USER_PHONE_AUTH_URL.strip())
|
|
# Always log initial WS state to diagnose missing phone, token or login
|
|
# configuration that could leave the AquaPortal user anonymous.
|
|
_s_init = {
|
|
"needs_apipilot": needs_apipilot,
|
|
"phoneLen": len(inject_phone or ""),
|
|
"phoneHead": (
|
|
("+" + inject_phone.lstrip("+")[:3] + "...")
|
|
if (inject_phone and len(inject_phone) > 6)
|
|
else (inject_phone or "")
|
|
),
|
|
"loginTokenLen": (len(login_token) if isinstance(login_token, str) and login_token.strip() else 0),
|
|
"loginByPhone": ("YES" if _needs_login_apipilot else "NO"),
|
|
"profileUrl": ("YES" if _needs_profile_apipilot else "NO"),
|
|
"S2S_INJECT_PROFILE": settings.S2S_INJECT_PROFILE,
|
|
}
|
|
app_logging.logger.info(
|
|
"[INJECT-APIPILOT][WS connect] INITIAL STATE: " + ", ".join(f"{k}={v}" for k, v in sorted(_s_init.items())),
|
|
)
|
|
# The API-key bypass makes login_by_phone unnecessary.
|
|
if (
|
|
not settings.S2S_APIPILOT_API_KEY
|
|
and _needs_login_apipilot
|
|
and inject_phone
|
|
and (
|
|
not isinstance(login_token, str)
|
|
or not login_token.strip()
|
|
or (isinstance(login_token, str) and 0 < len(login_token.strip()) < 80)
|
|
)
|
|
):
|
|
try:
|
|
login_token, _lresp = await providers_caller._login_by_phone(inject_phone)
|
|
if isinstance(_lresp, dict) and _lresp:
|
|
login_response_obj = _lresp
|
|
app_logging.logger.info(
|
|
"[INJECT-APIPILOT][WS connect] login_by_phone for authenticated ApiPilot execution phoneLen=%d token=%s",
|
|
len(inject_phone or ""),
|
|
f"YES len={len(login_token or '')}"
|
|
if isinstance(login_token, str) and login_token
|
|
else "NONE (403!)",
|
|
)
|
|
except Exception as _exc_login: # noqa: BLE001
|
|
app_logging.logger.info(
|
|
"[INJECT-APIPILOT][WS connect] login_by_phone failed (anonymous ApiPilot user, likely 403) phoneLen=%d err=%r",
|
|
len(inject_phone or ""),
|
|
_exc_login,
|
|
)
|
|
if _needs_profile_apipilot and inject_phone and (not isinstance(profile_obj, dict) or not profile_obj):
|
|
try:
|
|
profile_obj = await providers_caller._fetch_user_profile_for_phone(inject_phone, auth_token=login_token)
|
|
app_logging.logger.info(
|
|
"[INJECT-APIPILOT][WS connect] profile fetch for email defaults phoneLen=%d profile=%s",
|
|
len(inject_phone or ""),
|
|
("FOUND user=%s" % (((profile_obj or {}).get("user") or {}).get("name") or "?"))
|
|
if isinstance(profile_obj, dict)
|
|
else "NONE",
|
|
)
|
|
except Exception as _exc_p: # noqa: BLE001
|
|
app_logging.logger.info(
|
|
"[INJECT-APIPILOT][WS connect] profile fetch failed phoneLen=%d err=%r",
|
|
len(inject_phone or ""),
|
|
_exc_p,
|
|
)
|
|
|
|
if not needs_apipilot:
|
|
app_logging.logger.info(
|
|
"[INJECT-APIPILOT] ⏭️ WS connect: APIPILOT tools injection DISABLED (S2S_APIPILOT_TOOLS_URL='%s' S2S_APIPILOT_TOOLS_EXECUTE_URL='%s'). "
|
|
"No ApiPilot fetch or session.update patch in this tunnel.",
|
|
settings.S2S_APIPILOT_TOOLS_URL,
|
|
settings.S2S_APIPILOT_TOOLS_EXECUTE_URL,
|
|
)
|
|
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 injects EXCEPT apipilot."
|
|
)
|
|
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, login_response_obj = 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)."
|
|
)
|
|
# 3) Fetch the ApiPilot pass-1 tool list independently of S2S_INJECT_PROFILE,
|
|
# even without login/phone. Use the resolved token or anonymous fingerprint,
|
|
# matching the reasoning endpoint's server cache.
|
|
if needs_apipilot:
|
|
apipilot_auth_header = f"Bearer {login_token}" if isinstance(login_token, str) and login_token.strip() else None
|
|
# Phone login resolves the AquaPortal JWT for execution. The configured
|
|
# S2S_APIPILOT_SERVICE_TOKEN overrides it.
|
|
if settings.S2S_APIPILOT_SERVICE_TOKEN:
|
|
_svc = settings.S2S_APIPILOT_SERVICE_TOKEN.strip()
|
|
if _svc:
|
|
apipilot_auth_header = f"Bearer {_svc}" if not _svc.lower().startswith("bearer ") else _svc
|
|
demo_session_key = client_query.get("x-demo-session") if isinstance(client_query, dict) else None
|
|
if not demo_session_key:
|
|
demo_session_key = f"anon-ws-{client_ip}-{id(ws)}"
|
|
try:
|
|
# Always log discovery, matching reasoning pass 1.
|
|
app_logging.logger.info(
|
|
"[INJECT-APIPILOT] WS connect: fetching tools (auth=%s demoSession=%.40s...) upstream=%s...",
|
|
"yes" if apipilot_auth_header else "anon",
|
|
(demo_session_key[:40] if isinstance(demo_session_key, str) else "?"),
|
|
settings.S2S_APIPILOT_TOOLS_URL or "<using override execute>",
|
|
)
|
|
if settings.S2S_DEBUG_PROMPT:
|
|
app_logging.logger.info(
|
|
"[DEBUG-APIPILOT][WS connect] tools fetch detail toolsURL=%s executeURL=%s overrideExec=%s",
|
|
settings.S2S_APIPILOT_TOOLS_URL,
|
|
settings.S2S_APIPILOT_TOOLS_EXECUTE_URL,
|
|
providers_apipilot._apipilot_resolve_execute_url(),
|
|
)
|
|
_t0 = providers_apipilot._monotonic_now()
|
|
apipilot_tools_pass1 = (
|
|
await providers_apipilot._apipilot_fetch_tools(
|
|
auth_header=apipilot_auth_header,
|
|
demo_session=demo_session_key if isinstance(demo_session_key, str) else None,
|
|
refresh=False,
|
|
)
|
|
or []
|
|
)
|
|
_el = providers_apipilot._monotonic_now() - _t0
|
|
# Supply known phone/episode defaults to pass 1 so the model does
|
|
# not leave required arguments empty (observed ARGUMENTS=>>> bug).
|
|
_apipilot_defaults_p1: dict[str, str] = {}
|
|
if isinstance(inject_phone, str) and inject_phone.strip():
|
|
_apipilot_defaults_p1["phone"] = inject_phone.strip()
|
|
if isinstance(inject_episode_id, str) and inject_episode_id.strip():
|
|
_apipilot_defaults_p1["episode_id"] = inject_episode_id.strip()
|
|
# Also supply email when the resolved profile has a valid value.
|
|
if isinstance(profile_obj, dict) and profile_obj:
|
|
try:
|
|
u = profile_obj.get("user") if isinstance(profile_obj.get("user"), dict) else None
|
|
if isinstance(u, dict):
|
|
email = cast(str, u.get("email")) if isinstance(u.get("email"), str) else ""
|
|
if email.strip():
|
|
_apipilot_defaults_p1["email"] = email.strip()
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
session_ctx_apipilot_defaults = dict(_apipilot_defaults_p1)
|
|
injected_apipilot_pass1 = (
|
|
services_tool_context._resolve_apipilot_reasoning_context_block(
|
|
tools_list=apipilot_tools_pass1,
|
|
context_defaults=_apipilot_defaults_p1,
|
|
)
|
|
or ""
|
|
)
|
|
if injected_apipilot_pass1.strip():
|
|
app_logging.logger.info(
|
|
"[APIPILOT] REASONING PASS-1 (WS realtime): tool list ready N=%d t=%.0fms auth=%s defaults=%s",
|
|
len(apipilot_tools_pass1),
|
|
_el,
|
|
"Y" if isinstance(apipilot_auth_header, str) else "anon",
|
|
list(_apipilot_defaults_p1.keys()),
|
|
)
|
|
app_logging.logger.info(
|
|
"[INJECT-APIPILOT] ✅ WS connect: APIPILOT TOOLS list PASS-1 READY Ntools=%d wrappedLen=%d defaultsPhone=%s -> will patch session.update frames in this tunnel.",
|
|
len(apipilot_tools_pass1),
|
|
len(injected_apipilot_pass1),
|
|
("yes" if _apipilot_defaults_p1.get("phone") else "no"),
|
|
)
|
|
else:
|
|
app_logging.logger.info(
|
|
"[INJECT-APIPILOT] ⏭️ WS connect: apipilot fetch returned empty tools list; PASS-1 injection skipped for this tunnel.",
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
app_logging.logger.info(
|
|
"[INJECT-APIPILOT] ⛔ WS connect: apipilot tools fetch failed. Skipping PASS-1 injection for tunnel. err=%r",
|
|
exc,
|
|
)
|
|
injected_apipilot_pass1 = ""
|
|
apipilot_tools_pass1 = []
|
|
session_ctx_apipilot_defaults = {}
|
|
# Preserve the exact discovery authentication and demo-session values.
|
|
# AquaPortal binds authorized tools to GET /tools' x-demo-session; a new
|
|
# ws-pass2 value would appear anonymous and produce a 403 response.
|
|
_ws_apipilot_auth_for_execute = apipilot_auth_header
|
|
_ws_apipilot_demo_for_execute = demo_session_key if isinstance(demo_session_key, str) else None
|
|
# Preserve API-key identity in every propagation variable, matching the
|
|
# discovery cache fingerprint and HTTP headers even during pass-2 fallback.
|
|
if settings.S2S_APIPILOT_API_KEY:
|
|
_ws_apipilot_auth_for_execute = f"Bearer {settings.S2S_APIPILOT_API_KEY}"
|
|
_ws_apipilot_demo_for_execute = settings.S2S_APIPILOT_API_KEY
|
|
else:
|
|
# With ApiPilot unconfigured, keep defaults as an empty dict rather than None.
|
|
session_ctx_apipilot_defaults = {}
|
|
injected_apipilot_pass1 = ""
|
|
apipilot_tools_pass1 = []
|
|
_ws_apipilot_auth_for_execute = None
|
|
_ws_apipilot_demo_for_execute = None
|
|
app_logging.logger.info(
|
|
"[INJECT-PROFILE] WS connect FINAL: injectEpisode=%s injectProfile=%s injectReports=%s injectSummary=%s injectApipilotPass1=%s(Ntools=%d) login=%s episodeIdLen=%d phoneLen=%d episodeCtxLen=%d profileCtxLen=%d reportsCtxLen=%d summaryCtxLen=%d apipilotCtxLen=%d target=%s",
|
|
bool(injected_episode_context),
|
|
bool(injected_profile_context),
|
|
bool(injected_reports_context),
|
|
bool(injected_summary_context),
|
|
bool(injected_apipilot_pass1.strip()),
|
|
len(apipilot_tools_pass1),
|
|
("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),
|
|
len(injected_apipilot_pass1),
|
|
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 = {
|
|
"last_user_transcript": "",
|
|
"last_item_id": None,
|
|
"injected_episode_context": injected_episode_context,
|
|
"injected_profile_context": injected_profile_context,
|
|
"injected_reports_context": injected_reports_context,
|
|
"injected_summary_context": injected_summary_context,
|
|
"injected_apipilot_pass1": injected_apipilot_pass1,
|
|
"inject_phone": inject_phone,
|
|
"inject_episode_id": inject_episode_id,
|
|
"login_token": login_token,
|
|
"login_response": login_response_obj,
|
|
# 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,
|
|
"apipilot_tools_pass1": apipilot_tools_pass1,
|
|
# Accumulate transcripts per response_id and prevent pass-2
|
|
# injection loops after ApiPilot execution.
|
|
"_ws_apipilot_asst_transcript_acc": {}, # response_id -> str full transcript
|
|
"_ws_apipilot_pass2_active": False, # True during the second response; prevents infinite marker loops.
|
|
"_ws_apipilot_pass2_count": 0, # Session execution count, limited to three attempts.
|
|
# Defaults prepared by pass 1 fill missing pass-2 arguments.
|
|
"apipilot_defaults": (session_ctx_apipilot_defaults if needs_apipilot else {}),
|
|
# Pass-2 phone fallback also works when defaults are empty but
|
|
# the query string supplied phoneNumber.
|
|
"inject_phone": (inject_phone if isinstance(inject_phone, str) else ""), # noqa: F601
|
|
# Bearer for authenticated execution, resolved by phone login
|
|
# independently of S2S_INJECT_PROFILE.
|
|
"login_token": (login_token if isinstance(login_token, str) else None), # noqa: F601
|
|
# Execute with the exact auth/demo values used for WS discovery;
|
|
# AquaPortal binds the authorized tool list to that demo session.
|
|
"apipilot_exec_auth": _ws_apipilot_auth_for_execute,
|
|
"apipilot_exec_demo": _ws_apipilot_demo_for_execute,
|
|
}
|
|
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_apipilot_pass1 = session_ctx.get("injected_apipilot_pass1") 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,
|
|
apipilot_context_text_pass1=ctx_apipilot_pass1 or "",
|
|
)
|
|
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):
|
|
# Drop audio only for the current response_id marked as containing
|
|
# a CALL marker: Qwen3 may already have spoken it. Matching any ID
|
|
# would also discard valid pass-2 audio.
|
|
drop_now = False
|
|
matched_rid = ""
|
|
drop_ids = session_ctx.get("_ws_drop_audio_for_response_ids")
|
|
if isinstance(drop_ids, dict) and drop_ids:
|
|
rid_ctx = session_ctx.get("_ws_current_response_id")
|
|
if isinstance(rid_ctx, str) and rid_ctx and drop_ids.get(rid_ctx):
|
|
drop_now = True
|
|
matched_rid = rid_ctx
|
|
elif (not isinstance(rid_ctx, str) or not rid_ctx) and len(drop_ids) == 1:
|
|
# Fallback when no current ID exists but exactly one response
|
|
# remains in the dictionary.
|
|
only_rid = next(iter(drop_ids.keys()))
|
|
if drop_ids.get(only_rid):
|
|
drop_now = True
|
|
matched_rid = only_rid
|
|
if drop_now:
|
|
app_logging.logger.info(
|
|
"[APIPILOT-AUDIO-DROP] FRAME DROP bytes=%dB matchedResponse=%s (CALL marker in this response → leaked Qwen3 audio DISCARDED)",
|
|
len(message),
|
|
(matched_rid or session_ctx.get("_ws_current_response_id") or "-"),
|
|
)
|
|
continue
|
|
else:
|
|
app_logging.logger.debug(
|
|
"[APIPILOT-AUDIO-ALLOW] FRAME PASS bytes=%dB responseCtx=%s dropIds=%s",
|
|
len(message),
|
|
(session_ctx.get("_ws_current_response_id") or "-"),
|
|
(sorted(list(drop_ids.keys())) if isinstance(drop_ids, dict) else "none"),
|
|
)
|
|
# #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:
|
|
# Extract the original message before sanitation removes pass-2 markers.
|
|
pre_sanitize_summary = realtime_events._realtime_event_summary(message)
|
|
pre_event_type = pre_sanitize_summary.get("type") if isinstance(pre_sanitize_summary, dict) else None
|
|
pre_response_id: str | None = None
|
|
if isinstance(pre_sanitize_summary, dict):
|
|
if isinstance(pre_sanitize_summary.get("response_id"), str):
|
|
pre_response_id = pre_sanitize_summary["response_id"]
|
|
elif isinstance(pre_sanitize_summary.get("id"), str):
|
|
pre_response_id = pre_sanitize_summary["id"]
|
|
# Track the current response_id for the binary-audio gate.
|
|
if isinstance(pre_response_id, str) and pre_response_id:
|
|
session_ctx["_ws_current_response_id"] = pre_response_id
|
|
# Detect CALL markers in every text event, including before response.done.
|
|
# Mark this response_id to drop its audio and sanitize later events.
|
|
if (
|
|
isinstance(pre_sanitize_summary, dict)
|
|
and isinstance(pre_event_type, str)
|
|
and pre_event_type not in {"response.done"}
|
|
):
|
|
quick_text_parts: list[str] = []
|
|
for _k in (
|
|
"text",
|
|
"transcript",
|
|
"output_text",
|
|
"audio_text",
|
|
"output_transcript",
|
|
"final_transcript",
|
|
"output_audio_text",
|
|
"content",
|
|
"message",
|
|
"utterance",
|
|
"delta",
|
|
"_raw_preview",
|
|
):
|
|
_v = pre_sanitize_summary.get(_k)
|
|
if isinstance(_v, str) and _v.strip():
|
|
quick_text_parts.append(_v)
|
|
for _cname in ("item", "part", "output", "audio"):
|
|
_sub = pre_sanitize_summary.get(_cname)
|
|
if isinstance(_sub, dict):
|
|
for _k2 in ("text", "transcript", "output_text"):
|
|
_v2 = _sub.get(_k2)
|
|
if isinstance(_v2, str) and _v2.strip():
|
|
quick_text_parts.append(_v2)
|
|
quick_text = (" ".join(quick_text_parts)).strip()
|
|
if quick_text:
|
|
_detected_name, _detected_args = services_tool_context._apipilot_parse_call_marker(quick_text)
|
|
if isinstance(_detected_name, str) and _detected_name:
|
|
drop_ids2 = session_ctx.get("_ws_drop_audio_for_response_ids")
|
|
if not isinstance(drop_ids2, dict):
|
|
drop_ids2 = {}
|
|
session_ctx["_ws_drop_audio_for_response_ids"] = drop_ids2
|
|
_flag_rid = (
|
|
pre_response_id
|
|
if isinstance(pre_response_id, str) and pre_response_id
|
|
else f"auto-{id(session_ctx)}"
|
|
)
|
|
drop_ids2[_flag_rid] = True
|
|
# Remove older IDs when more than ten remain.
|
|
if len(drop_ids2) > 10:
|
|
try:
|
|
for _old_k in list(drop_ids2.keys())[:-10]:
|
|
drop_ids2.pop(_old_k, None)
|
|
except Exception:
|
|
pass
|
|
app_logging.logger.info(
|
|
"[APIPILOT-SANITIZE] CALL marker in event=%s response=%s name=%s → AUDIO DROP GATE enabled (discard potential leaked Qwen3 audio)",
|
|
pre_event_type,
|
|
_flag_rid,
|
|
_detected_name,
|
|
)
|
|
# 1a) Accumulate original TTS transcripts for pass-2 marker parsing.
|
|
if (
|
|
isinstance(pre_sanitize_summary, dict)
|
|
and isinstance(pre_event_type, str)
|
|
and pre_event_type in {"response.audio_transcript.done", "response.output_audio_transcript.done"}
|
|
):
|
|
pre_agent_text = (
|
|
cast(str, pre_sanitize_summary.get("text"))
|
|
if isinstance(pre_sanitize_summary.get("text"), str)
|
|
else ""
|
|
).strip()
|
|
rid_key = (
|
|
pre_response_id
|
|
if isinstance(pre_response_id, str) and pre_response_id
|
|
else f"auto-{id(session_ctx)}"
|
|
)
|
|
if rid_key and pre_agent_text:
|
|
acc = session_ctx.get("_ws_apipilot_asst_transcript_acc")
|
|
if not isinstance(acc, dict):
|
|
acc = {}
|
|
session_ctx["_ws_apipilot_asst_transcript_acc"] = acc
|
|
prev = acc.get(rid_key, "")
|
|
if prev:
|
|
acc[rid_key] = f"{prev} {pre_agent_text}".strip()
|
|
else:
|
|
acc[rid_key] = pre_agent_text
|
|
# 1b) Also accumulate assistant conversation.item.created text,
|
|
# which can contain markers outside TTS transcripts.
|
|
elif (
|
|
isinstance(pre_sanitize_summary, dict)
|
|
and isinstance(pre_event_type, str)
|
|
and pre_event_type == "conversation.item.created"
|
|
and pre_sanitize_summary.get("role") == "assistant"
|
|
):
|
|
pre_conv_text = (
|
|
cast(str, pre_sanitize_summary.get("text"))
|
|
if isinstance(pre_sanitize_summary.get("text"), str)
|
|
else ""
|
|
).strip()
|
|
rid_key2 = (
|
|
pre_response_id
|
|
if isinstance(pre_response_id, str) and pre_response_id
|
|
else f"auto-{id(session_ctx)}"
|
|
)
|
|
if rid_key2 and pre_conv_text:
|
|
acc2 = session_ctx.get("_ws_apipilot_asst_transcript_acc")
|
|
if not isinstance(acc2, dict):
|
|
acc2 = {}
|
|
session_ctx["_ws_apipilot_asst_transcript_acc"] = acc2
|
|
prev2 = acc2.get(rid_key2, "")
|
|
if prev2:
|
|
acc2[rid_key2] = f"{prev2} {pre_conv_text}".strip()
|
|
else:
|
|
acc2[rid_key2] = pre_conv_text
|
|
# 2) Parse the original marker at response.done to trigger pass 2.
|
|
if (
|
|
isinstance(pre_sanitize_summary, dict)
|
|
and isinstance(pre_event_type, str)
|
|
and pre_event_type == "response.done"
|
|
):
|
|
rid_val: str = (
|
|
pre_response_id
|
|
if isinstance(pre_response_id, str)
|
|
else (
|
|
cast(str, pre_sanitize_summary.get("id"))
|
|
if isinstance(pre_sanitize_summary.get("id"), str)
|
|
else f"auto-{id(session_ctx)}"
|
|
)
|
|
)
|
|
full_text = ""
|
|
acc3 = session_ctx.get("_ws_apipilot_asst_transcript_acc")
|
|
if rid_val and isinstance(acc3, dict):
|
|
full_text = (acc3.pop(rid_val, "") if isinstance(acc3, dict) else "") or ""
|
|
# Use the sole remaining accumulator when an upstream changes
|
|
# response_id between events.
|
|
if (
|
|
(not isinstance(full_text, str) or not full_text.strip())
|
|
and isinstance(acc3, dict)
|
|
and len(acc3) == 1
|
|
):
|
|
_only_key = next(iter(acc3.keys()))
|
|
full_text = (acc3.pop(_only_key, "") if isinstance(acc3, dict) else "") or ""
|
|
marker_name, marker_args = None, None
|
|
if isinstance(full_text, str) and full_text.strip():
|
|
_p1_defaults = (
|
|
session_ctx.get("apipilot_defaults")
|
|
if isinstance(session_ctx.get("apipilot_defaults"), dict)
|
|
else {}
|
|
)
|
|
marker_name, marker_args = services_tool_context._apipilot_parse_call_marker(
|
|
full_text, defaults=_p1_defaults
|
|
)
|
|
app_logging.logger.info(
|
|
"[APIPILOT-CALL-NAME] USE=WS_RESPONSE_DONE_DETECTION markerFound=%s name=%r argsKeys=%s",
|
|
("Y" if isinstance(marker_name, str) and marker_name else "N"),
|
|
(marker_name or ""),
|
|
list((marker_args or {}).keys()),
|
|
)
|
|
# Always log pass-2 marker detection diagnostics.
|
|
try:
|
|
_preview = full_text.replace("\n", " ")
|
|
if len(_preview) > 500:
|
|
_preview = _preview[:500] + "…"
|
|
except Exception:
|
|
_preview = str(full_text)[:500]
|
|
app_logging.logger.info(
|
|
"[APIPILOT] REASONING PASS-2 (WS realtime): response.done DIAGNOSTIC fullLen=%d markerFound=%s markerName=%s argsKeys=%s defaultsAutoFill=%s acc3Remaining=%d fullTextPreview=%r",
|
|
len(full_text or ""),
|
|
("Y" if isinstance(marker_name, str) and marker_name else "N"),
|
|
(marker_name if isinstance(marker_name, str) and marker_name else ""),
|
|
(list(marker_args.keys()) if isinstance(marker_args, dict) else []),
|
|
(list(_p1_defaults.keys()) if isinstance(_p1_defaults, dict) and _p1_defaults else "none"),
|
|
(len(acc3) if isinstance(acc3, dict) else -1),
|
|
_preview,
|
|
)
|
|
elif isinstance(full_text, str):
|
|
app_logging.logger.info(
|
|
"[APIPILOT] REASONING PASS-2 (WS realtime): response.done DIAGNOSTIC fullLen=0 EMPTY markerFound=N response_id=%r acc3Remaining=%d",
|
|
rid_val,
|
|
(len(acc3) if isinstance(acc3, dict) else -1),
|
|
)
|
|
else:
|
|
app_logging.logger.info(
|
|
"[APIPILOT] REASONING PASS-2 (WS realtime): response.done DIAGNOSTIC fullText is not str type=%r acc3=%r",
|
|
type(full_text).__name__,
|
|
(list(acc3.keys()) if isinstance(acc3, dict) else "?"),
|
|
)
|
|
needs_pass2 = (
|
|
isinstance(marker_name, str)
|
|
and bool(marker_name)
|
|
and not bool(session_ctx.get("_ws_apipilot_pass2_active"))
|
|
and (int(session_ctx.get("_ws_apipilot_pass2_count") or 0) < 3)
|
|
)
|
|
if needs_pass2:
|
|
# Clear active audio-drop gates before pass 2: its response
|
|
# contains actual user data and must be played normally.
|
|
drop_ids_clean = session_ctx.get("_ws_drop_audio_for_response_ids")
|
|
if isinstance(drop_ids_clean, dict) and drop_ids_clean:
|
|
drop_ids_clean.clear()
|
|
app_logging.logger.info(
|
|
"[APIPILOT-AUDIO-DROP] PASS-2 start: clear audio-drop gate → forward pass-2 briefing audio normally"
|
|
)
|
|
args = marker_args if isinstance(marker_args, dict) else {}
|
|
# Fill missing arguments or replace fictional prompt example
|
|
# values with actual defaults. When defaults are empty, use
|
|
# session_ctx["inject_phone"] from the query string.
|
|
_p1_defaults_use = (
|
|
session_ctx.get("apipilot_defaults")
|
|
if isinstance(session_ctx.get("apipilot_defaults"), dict)
|
|
else {}
|
|
)
|
|
_fallback_inject_phone = (
|
|
session_ctx.get("inject_phone") if isinstance(session_ctx.get("inject_phone"), str) else ""
|
|
)
|
|
# Known fictional example values appearing in prompts.
|
|
_FAKE_PHONE_PLACEHOLDERS: set[str] = {
|
|
"+56912345678", # Prompt example from rule 7.
|
|
"56912345678", # Variant without +.
|
|
"+5612345678",
|
|
"+1234567890",
|
|
"+12345678",
|
|
"+56XXXXXXX",
|
|
"+56xxxxxxx",
|
|
"",
|
|
}
|
|
|
|
def _is_phone_fake(v: str) -> bool:
|
|
c = str(v or "").strip()
|
|
if c in _FAKE_PHONE_PLACEHOLDERS: # noqa: B023
|
|
return True
|
|
if len(c) < 7:
|
|
return True
|
|
if c.endswith("00000"):
|
|
return True
|
|
if "123456" in c or "xxxxxx" in c.lower() or "xxxx" in c.lower():
|
|
return True
|
|
if c.startswith("+56") and len(c) <= 9:
|
|
return True
|
|
return False
|
|
|
|
# Apply defaults and fallback to all available keys.
|
|
if isinstance(_p1_defaults_use, dict) and _p1_defaults_use:
|
|
for k, default_v in _p1_defaults_use.items():
|
|
if default_v in (None, "", []):
|
|
continue
|
|
if k not in args or args[k] in (None, "", []):
|
|
args[k] = default_v
|
|
elif k == "phone" and isinstance(args.get("phone"), str):
|
|
if _is_phone_fake(str(args.get("phone"))):
|
|
args[k] = default_v
|
|
# Replace fictional phone values even when defaults lacks phone.
|
|
if (
|
|
isinstance(args.get("phone"), str)
|
|
and _is_phone_fake(str(args.get("phone")))
|
|
and isinstance(_fallback_inject_phone, str)
|
|
and _fallback_inject_phone.strip()
|
|
):
|
|
args["phone"] = _fallback_inject_phone.strip()
|
|
# Refresh _p1_defaults for subsequent logging.
|
|
_p1_defaults = _p1_defaults_use
|
|
session_ctx["_ws_apipilot_pass2_count"] = int(session_ctx.get("_ws_apipilot_pass2_count") or 0) + 1
|
|
session_ctx["_ws_apipilot_pass2_active"] = True
|
|
# Reuse exact discovery auth/demo values: AquaPortal binds
|
|
# authorized tools to the x-demo-session used for GET /tools.
|
|
_exec_auth = (
|
|
session_ctx.get("apipilot_exec_auth")
|
|
if isinstance(session_ctx.get("apipilot_exec_auth"), str)
|
|
else None
|
|
)
|
|
_exec_demo = (
|
|
session_ctx.get("apipilot_exec_demo")
|
|
if isinstance(session_ctx.get("apipilot_exec_demo"), str)
|
|
else None
|
|
)
|
|
# Compatibility fallback for missing values:
|
|
if not isinstance(_exec_auth, str) or not _exec_auth.strip():
|
|
login_tok = session_ctx.get("login_token")
|
|
_exec_auth = f"Bearer {login_tok}" if isinstance(login_tok, str) and login_tok.strip() else None
|
|
if not isinstance(_exec_demo, str) or not _exec_demo.strip():
|
|
_exec_demo = f"ws-pass2-{client_ip}-{id(session_ctx)}"
|
|
# The SUPER API KEY overrides even SERVICE_TOKEN. Force bearer
|
|
# and demo-session values to match the discovery bypass identity.
|
|
if settings.S2S_APIPILOT_API_KEY:
|
|
_exec_auth = f"Bearer {settings.S2S_APIPILOT_API_KEY}"
|
|
_exec_demo = settings.S2S_APIPILOT_API_KEY
|
|
# Apply the service-token override only without a SUPER API KEY.
|
|
elif settings.S2S_APIPILOT_SERVICE_TOKEN:
|
|
_svc = settings.S2S_APIPILOT_SERVICE_TOKEN.strip()
|
|
if _svc:
|
|
_exec_auth = f"Bearer {_svc}" if not _svc.lower().startswith("bearer ") else _svc
|
|
# For authenticated tools with an empty/short bearer, force
|
|
# phone login again so AquaPortal receives a valid user token.
|
|
# The length heuristic treats fewer than 80 characters as short.
|
|
_phone_lfb = (
|
|
session_ctx.get("inject_phone") if isinstance(session_ctx.get("inject_phone"), str) else ""
|
|
) or (inject_phone if isinstance(inject_phone, str) else "") # type: ignore[name-defined] # noqa: F821 - preserve the legacy fallback failure.
|
|
_phone_lfb = _phone_lfb.strip() if isinstance(_phone_lfb, str) else ""
|
|
_current_bearer_token = ""
|
|
if isinstance(_exec_auth, str) and _exec_auth.lower().startswith("bearer "):
|
|
_current_bearer_token = _exec_auth[len("bearer ") :].strip()
|
|
_is_jwt_like = (
|
|
bool(_current_bearer_token)
|
|
and len(_current_bearer_token) >= 120
|
|
and _current_bearer_token.count(".") >= 2
|
|
)
|
|
# Resolve requires_auth from recent session/cache tools; default true.
|
|
_tools_recent: list[dict] = (
|
|
cast(list[dict], session_ctx.get("apipilot_tools_pass1"))
|
|
if isinstance(session_ctx.get("apipilot_tools_pass1"), list)
|
|
else []
|
|
)
|
|
if not _tools_recent:
|
|
try:
|
|
_cached = (
|
|
providers_apipilot._APIPILOT_CACHE.get("tools")
|
|
if providers_apipilot._APIPILOT_CACHE
|
|
else None
|
|
)
|
|
if isinstance(_cached, tuple) and _cached and isinstance(_cached[0], list):
|
|
_tools_recent = _cached[0]
|
|
except Exception: # noqa: BLE001
|
|
_tools_recent = []
|
|
_mtool = next(
|
|
(t for t in _tools_recent if isinstance(t, dict) and t.get("name") == marker_name), None
|
|
)
|
|
_requires_auth = True if (_mtool is None) else bool(_mtool.get("requires_auth"))
|
|
if (
|
|
(not settings.S2S_APIPILOT_API_KEY)
|
|
and settings.S2S_LOGIN_BY_PHONE
|
|
and _phone_lfb
|
|
and _requires_auth
|
|
and (not _is_jwt_like)
|
|
):
|
|
try:
|
|
_l_tok, _l_obj = await providers_caller._login_by_phone(_phone_lfb)
|
|
if isinstance(_l_tok, str) and _l_tok.strip():
|
|
_exec_auth = f"Bearer {_l_tok.strip()}"
|
|
_tok_len = len(_l_tok.strip())
|
|
_is_new_jwt = _tok_len >= 120 and _l_tok.strip().count(".") >= 2
|
|
app_logging.logger.info(
|
|
"[APIPILOT] REASONING PASS-2 (WS realtime): FORCED PASS-2 LOGIN (tool requires_auth) phoneLen=%d token=%s%s",
|
|
len(_phone_lfb),
|
|
f"YES len={_tok_len}",
|
|
(" JWT?" if _is_new_jwt else (" SHORT?" if _tok_len < 80 else " OK")),
|
|
)
|
|
else:
|
|
app_logging.logger.info(
|
|
"[APIPILOT] REASONING PASS-2 (WS realtime): FORCED PASS-2 LOGIN token EXTRACTION FAILED phoneLen=%d (tool requires_auth → likely 403)",
|
|
len(_phone_lfb),
|
|
)
|
|
except Exception as _lfb_exc: # noqa: BLE001
|
|
app_logging.logger.info(
|
|
"[APIPILOT] REASONING PASS-2 (WS realtime): FORCED PASS-2 LOGIN failed phoneLen=%d err=%r",
|
|
len(_phone_lfb),
|
|
_lfb_exc,
|
|
)
|
|
bearer = _exec_auth
|
|
demo_key = _exec_demo
|
|
try:
|
|
app_logging.logger.info(
|
|
"[APIPILOT-CALL-NAME] USE=WS_REALTIME_PASS2_TRIGGER name=%r argsKeys=%s argsPhone=%r",
|
|
marker_name,
|
|
list(args.keys()) if args else [],
|
|
(args.get("phone") if isinstance(args.get("phone"), str) else ""),
|
|
)
|
|
app_logging.logger.info(
|
|
"[APIPILOT] REASONING PASS-2 (WS realtime): trigger marker name=%s argsKeys=%s argsPhone=%s auth=%s defaultsAutoFill=%s demoSessionKey=%.40s",
|
|
marker_name,
|
|
list(args.keys()) if args else [],
|
|
(args.get("phone") if isinstance(args.get("phone"), str) else ""),
|
|
"Y" if bearer else "anon",
|
|
list(_p1_defaults.keys()) if isinstance(_p1_defaults, dict) and _p1_defaults else "none",
|
|
demo_key if isinstance(demo_key, str) else "?",
|
|
)
|
|
t0 = providers_apipilot._monotonic_now()
|
|
app_logging.logger.info(
|
|
"[APIPILOT-CALL-NAME] USE=WS_REALTIME_PASS2_EXEC_HELPER name=%r → forward a _apipilot_execute_tool",
|
|
marker_name,
|
|
)
|
|
env_exec = providers_apipilot._apipilot_execute_tool(
|
|
auth_header=bearer,
|
|
demo_session=demo_key,
|
|
name=cast(str, marker_name),
|
|
arguments=args,
|
|
)
|
|
if hasattr(asyncio, "create_task"):
|
|
envelope = await asyncio.wait_for(env_exec, timeout=60.0)
|
|
else:
|
|
envelope = await env_exec
|
|
t1 = providers_apipilot._monotonic_now() - t0
|
|
ok = bool(isinstance(envelope, dict) and envelope.get("success"))
|
|
data = envelope.get("data") if isinstance(envelope, dict) else None
|
|
err_obj = (
|
|
envelope.get("error")
|
|
if isinstance(envelope, dict) and isinstance(envelope.get("error"), dict)
|
|
else None
|
|
)
|
|
if isinstance(data, str):
|
|
tool_text = data
|
|
elif isinstance(data, (dict, list)):
|
|
try:
|
|
tool_text = json.dumps(data, indent=2, ensure_ascii=False)
|
|
except Exception: # noqa: BLE001
|
|
tool_text = str(data)
|
|
else:
|
|
tool_text = ""
|
|
if not ok:
|
|
code = (
|
|
err_obj.get("code")
|
|
if isinstance(err_obj, dict) and err_obj.get("code") is not None
|
|
else (envelope.get("code") if isinstance(envelope, dict) else None)
|
|
)
|
|
msg = (
|
|
err_obj.get("message")
|
|
if isinstance(err_obj, dict) and isinstance(err_obj.get("message"), str)
|
|
else ""
|
|
)
|
|
prefix = ""
|
|
if code is not None or msg:
|
|
prefix = f"[{code if code is not None else 'ERROR'}] {msg}".strip() + "\n"
|
|
tool_text = f"{prefix}{tool_text}".strip()
|
|
app_logging.logger.info(
|
|
"[APIPILOT] REASONING PASS-2 (WS realtime): execute DONE name=%s success=%s len=%d t=%.0fms",
|
|
marker_name,
|
|
"ok" if ok else "error",
|
|
len(tool_text),
|
|
t1,
|
|
)
|
|
call_result_dict: dict[str, Any] = {
|
|
"name": marker_name,
|
|
"text": tool_text or "",
|
|
"error": (not ok),
|
|
}
|
|
# Use the light internal sanitizer for pass 2. The user-output
|
|
# sanitizer would remove the RESULT wrapper and prevent the
|
|
# upstream model from receiving the briefing result.
|
|
len_raw_tool_text = len(call_result_dict.get("text") or "")
|
|
if isinstance(call_result_dict.get("text"), str):
|
|
call_result_dict["text"] = services_tool_context._sanitize_pass2_internal_context(
|
|
call_result_dict["text"]
|
|
)
|
|
len_sanitized_tool_text = len(call_result_dict.get("text") or "")
|
|
blk_pass2 = (
|
|
services_tool_context._resolve_apipilot_reasoning_context_block(
|
|
None,
|
|
call_result=call_result_dict,
|
|
)
|
|
or ""
|
|
)
|
|
if not blk_pass2.strip():
|
|
blk_pass2 = (tool_text or "").strip()
|
|
len_raw_wrapper = len(blk_pass2 or "")
|
|
# Light sanitation preserves the pass-2 RESULT wrapper.
|
|
if blk_pass2.strip():
|
|
blk_pass2 = services_tool_context._sanitize_pass2_internal_context(blk_pass2)
|
|
len_sanitized_wrapper = len(blk_pass2 or "")
|
|
app_logging.logger.info(
|
|
"[APIPILOT] REASONING PASS-2 (WS realtime): block SIZES len_raw_tool_text=%d len_sanitized_tool_text=%d len_raw_wrapper=%d len_sanitized_wrapper=%d success=%s",
|
|
len_raw_tool_text,
|
|
len_sanitized_tool_text,
|
|
len_raw_wrapper,
|
|
len_sanitized_wrapper,
|
|
("ok" if ok else "error"),
|
|
)
|
|
if blk_pass2.strip():
|
|
# Reset _ws_current_response_id before pass-2 injection so
|
|
# valid briefing audio cannot match the previous CALL response.
|
|
session_ctx["_ws_current_response_id"] = None
|
|
app_logging.logger.info(
|
|
"[APIPILOT] REASONING PASS-2 (WS realtime): inject PASS-2 block len=%d -> send conversation.item.create + response.create upstream",
|
|
len(blk_pass2),
|
|
)
|
|
try:
|
|
await upstream.send(
|
|
json.dumps(
|
|
{
|
|
"type": "conversation.item.create",
|
|
"item": {
|
|
"type": "message",
|
|
"role": "user",
|
|
"content": [{"type": "input_text", "text": blk_pass2}],
|
|
},
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
)
|
|
await upstream.send(json.dumps({"type": "response.create"}, ensure_ascii=False))
|
|
app_logging.logger.info(
|
|
"[APIPILOT] REASONING PASS-2 (WS realtime): injection OK. Awaiting model response (pass2_active=Y until next response.done)",
|
|
)
|
|
except Exception as exc2: # noqa: BLE001
|
|
session_ctx["_ws_apipilot_pass2_active"] = False
|
|
app_logging.logger.info(
|
|
"[APIPILOT] REASONING PASS-2 (WS realtime): inject FAIL upstream send err=%r. pass2_active=N.",
|
|
exc2,
|
|
)
|
|
else:
|
|
session_ctx["_ws_apipilot_pass2_active"] = False
|
|
app_logging.logger.info(
|
|
"[APIPILOT] REASONING PASS-2 (WS realtime): EMPTY result block -> skip injection. pass2_active=N.",
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
session_ctx["_ws_apipilot_pass2_active"] = False
|
|
app_logging.logger.info(
|
|
"[APIPILOT] REASONING PASS-2 (WS realtime): ABORT marker=%s err=%r. pass2_active=N (safety).",
|
|
marker_name,
|
|
exc,
|
|
)
|
|
elif isinstance(session_ctx.get("_ws_apipilot_pass2_active"), bool) and bool(
|
|
session_ctx.get("_ws_apipilot_pass2_active")
|
|
):
|
|
session_ctx["_ws_apipilot_pass2_active"] = False
|
|
app_logging.logger.info(
|
|
"[APIPILOT] REASONING PASS-2 (WS realtime): response.done PASS-2 FINALE -> pass2_active=N safety counter=%s",
|
|
int(session_ctx.get("_ws_apipilot_pass2_count") or 0),
|
|
)
|
|
# Layer 7 post-processing: after marker handling, safely remove
|
|
# internal 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()
|
|
item_id = summary.get("item_id")
|
|
if user_text:
|
|
app_logging.logger.info(
|
|
"[USER INPUT] client=%s stt=%s",
|
|
client_ip,
|
|
user_text,
|
|
)
|
|
session_ctx["last_user_transcript"] = user_text
|
|
if item_id:
|
|
session_ctx["last_item_id"] = item_id
|
|
elif event_type == "conversation.item.created" and summary.get("role") == "user":
|
|
user_text = (summary.get("text") or "").strip()
|
|
item_id = summary.get("item_id") or summary.get("id")
|
|
if user_text:
|
|
app_logging.logger.info(
|
|
"[USER INPUT] client=%s text=%s",
|
|
client_ip,
|
|
user_text,
|
|
)
|
|
session_ctx["last_user_transcript"] = user_text
|
|
if item_id:
|
|
session_ctx["last_item_id"] = item_id
|
|
# Agent output, including all event variants.
|
|
agent_text = ""
|
|
response_id = summary.get("response_id") or summary.get("id")
|
|
item_id = summary.get("item_id")
|
|
if event_type in {"response.audio_transcript.done", "response.output_audio_transcript.done"}:
|
|
# Pass-2 accumulation already retained the original text.
|
|
# This sanitized agent_text is only 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,
|
|
)
|
|
asyncio.create_task(
|
|
services_reasoning._forward_agent_text_to_nodered(
|
|
{
|
|
"tts": agent_text,
|
|
"response_id": response_id,
|
|
"item_id": item_id,
|
|
"client_ip": client_ip,
|
|
"role": "assistant",
|
|
"source_event": event_type,
|
|
"user_transcript": session_ctx.get("last_user_transcript") or "",
|
|
}
|
|
)
|
|
)
|
|
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,
|
|
)
|
|
asyncio.create_task(
|
|
services_reasoning._forward_agent_text_to_nodered(
|
|
{
|
|
"tts": agent_text,
|
|
"response_id": response_id,
|
|
"item_id": item_id,
|
|
"client_ip": client_ip,
|
|
"role": "assistant",
|
|
"source_event": event_type,
|
|
"user_transcript": session_ctx.get("last_user_transcript") or "",
|
|
}
|
|
)
|
|
)
|
|
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,
|
|
)
|
|
asyncio.create_task(
|
|
services_reasoning._forward_agent_text_to_nodered(
|
|
{
|
|
"tts": agent_text,
|
|
"response_id": response_id,
|
|
"item_id": item_id,
|
|
"client_ip": client_ip,
|
|
"role": "assistant",
|
|
"source_event": event_type,
|
|
"user_transcript": session_ctx.get("last_user_transcript") or "",
|
|
}
|
|
)
|
|
)
|
|
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,
|
|
)
|
|
asyncio.create_task(
|
|
services_reasoning._forward_agent_text_to_nodered(
|
|
{
|
|
"tts": agent_text,
|
|
"response_id": response_id,
|
|
"item_id": item_id,
|
|
"client_ip": client_ip,
|
|
"role": "assistant",
|
|
"source_event": event_type,
|
|
"user_transcript": session_ctx.get("last_user_transcript") or "",
|
|
}
|
|
)
|
|
)
|
|
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)
|