valenti/vocero-client
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
f3bae8ae0ccfac57220d4837594516a44427bf1b
vocero-client/vocero-api/vocero_api/realtime/proxy.py
T
alvaro abd03ed07d refactor: rename API and frontend projects to Vocero 
Rename the project directories, Python distribution and import package, and npm package. Update launcher, Docker, tooling, tests, lockfiles, and documentation while preserving existing API contracts.

Make AGENTS.md self-contained and remove external skill and workflow references.
2026-10-06 19:26:55 -03:00

1308 lines
73 KiB
Python
Raw Blame History

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 apipilot as providers_apipilot
from vocero_api.providers import caller as providers_caller
from vocero_api.realtime import events as realtime_events
from vocero_api.services import context as services_context
from vocero_api.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 = {
"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()
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"}:
# 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,
)
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)
Reference in New Issue View Git Blame Copy Permalink