valenti/vocero-client
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d58be9613ff62811b2c76a1827cc5844b997e910
vocero-client/s2s-client/server.py
T

2543 lines
116 KiB
Python
Raw Blame History

"""
Tiny server for the speech-to-speech demo.
The demo used to ship as a `sdk: static` Space, but the web-search tool needs a
search key the browser must NOT see. A static Space has no runtime process, so it
can't hold a secret the front-end uses. This server fixes that: it serves the
unchanged front-end AND exposes a same-origin `/api/search` proxy that holds the
Serper key server-side (see docs/adr/0001).
Everything lives in one container; the speech-to-speech backend stays a separate,
load-balanced service the browser talks to over WebSocket as before. The load
balancer's address is a secret too (like the Serper key): the browser never sees
it. `/api/session` proxies the session handshake server-side so only the
per-session compute URL the LB hands back (which the browser must dial) is exposed.
On the deployed Space the server also meters conversation time by HF login tier
(anonymous / signed-in / PRO) — see `limiter.py` and `auth.py`. That whole feature
is off unless BOTH `LOAD_BALANCER_URL` and `SPACE_ID` are set, so it runs only on
the live Space, never locally (even with the LB exported for testing).
`SPEECH_TO_SPEECH_URL` overrides everything: when set, the LB logic above is
disabled entirely (no session proxy, no queue, no metering, no sign-in) and the
browser connects directly to that URL, shown read-only in Settings.
Endpoints:
GET /api/config -> { search, lb, allowDirect, s2sUrl, auth }
GET /api/me -> login + tier + remaining budget (LB mode only)
POST /api/search -> { results, answer } Google via Serper.dev
POST /api/session -> proxies <LB>/session: a grant, or a queue ticket
GET /api/queue/{id} -> proxies <LB>/queue/{id}: position, or a grant on claim
DELETE /api/queue/{id} -> leave the queue (explicit "Leave queue" button)
POST /api/queue/end -> leave the queue (sendBeacon on teardown)
POST /api/session/heartbeat-> extend the reservation; { expired }
POST /api/session/end -> reconcile + refund (sendBeacon on teardown)
/* -> static files (index.html, main.js, ...)
When every compute slot is busy the load balancer hands back a queue ticket
instead of a grant; the browser polls /api/queue/{id} until it reaches the front
and a slot frees. Waiting reserves nothing — the daily budget is only reserved at
the moment a slot is actually claimed (a grant), never while queued.
"""
import asyncio
from datetime import datetime, timezone
import json
import logging
import os
import re
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
import httpx
import websockets
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException, Request, WebSocket, WebSocketDisconnect
from fastapi.responses import JSONResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
import auth
import limiter
logger = logging.getLogger("s2s")
_logging_configured = False
if not _logging_configured:
_handler = logging.StreamHandler()
_handler.setFormatter(logging.Formatter(
"%(asctime)s %(levelname)-7s [%(name)s] %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
))
logging.basicConfig(level=logging.DEBUG, handlers=[_handler], force=True)
_lvl_env = (os.environ.get("LOG_LEVEL") or os.environ.get("S2S_LOG_LEVEL") or "INFO").strip().upper()
_is_debug = _lvl_env in {"DEBUG", "TRACE", "10"}
# Logger rumorosi: a meno che LOG_LEVEL=debug esplicito, tienili quiet.
logging.getLogger("httpx").setLevel(logging.DEBUG if _is_debug else logging.WARNING)
logging.getLogger("websockets").setLevel(logging.DEBUG if _is_debug else logging.INFO)
logging.getLogger("uvicorn").setLevel(logging.DEBUG if _is_debug else logging.INFO)
logging.getLogger("uvicorn.access").setLevel(logging.DEBUG if _is_debug else logging.INFO)
logging.getLogger("httpcore").setLevel(logging.DEBUG if _is_debug else logging.WARNING)
logger.setLevel(logging.DEBUG if _is_debug else logging.DEBUG) # s2s sempre almeno DEBUG in dev
_logging_configured = True
logger.debug(
"LOG_LEVEL env=%s — noisy-logger levels: httpx=%s uvicorn=%s websockets=%s",
_lvl_env,
logging.getLevelName(logging.getLogger("httpx").level),
logging.getLevelName(logging.getLogger("uvicorn").level),
logging.getLevelName(logging.getLogger("websockets").level),
)
ROOT_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
load_dotenv(os.path.join(ROOT_DIR, ".env"), override=False)
logger.debug("Loaded .env from %s", ROOT_DIR)
def _normalize_reasoning_type(value: str) -> str:
normalized = (value or "").strip().strip('"').strip("'").replace("-", "_").upper()
if normalized == "OPENAI":
return "OPENAI"
if normalized in {"NODE_RED", "NODERED"}:
return "NODE_RED"
return "PROXY"
SERPER_KEY = os.environ.get("SERPER_API_KEY", "").strip()
S2S_PRIMARY_PROMPT = os.environ.get("S2S_PRIMARY_PROMPT", "").strip()
if S2S_PRIMARY_PROMPT:
S2S_PRIMARY_PROMPT = S2S_PRIMARY_PROMPT.replace("\\n", "\n")
# External user-by-phone info endpoint. When set, GET /api/user/phone proxies
# to this URL (passing ?phoneNumber=) and forwards the JSON payload to the
# browser so the UI can render caller identity / caller context.
VOCERO_USER_PHONE_AUTH_URL = os.environ.get("VOCERO_USER_PHONE_AUTH_URL", "").strip().rstrip("/")
# Feature flag: when "true" (case-insensitive) => injects the resolved caller
# profile into the realtime session (patch session.update client->upstream,
# append profile to the LLM system instructions prompt instructions, and prepend it to the
# reasoning user_context for the /api/reasoning HTTP body.
# Default = disabled (generic deploy; pure passthrough; all characterization lives
# in the backend RAG per original architectural decision.)
S2S_INJECT_PROFILE = (
(os.environ.get("S2S_INJECT_PROFILE", "") or "").strip().lower()
in {"1", "true", "yes", "on"}
)
# Endpoint URL for fetching the latest per-caller conversation summary, used when
# both S2S_INJECT_PROFILE=true AND S2S_LATEST_SUMMARY is set. Completely
# generic: the proxy performs a plain GET S2S_LATEST_SUMMARY?phoneNumber=<x>
# and expects JSON back (optionally wrapped as {"result": <value>}). Empty
# string means the feature is off (no network call, nothing injected).
S2S_LATEST_SUMMARY = (os.environ.get("S2S_LATEST_SUMMARY", "") or "").strip()
# Optional: login-by-phone endpoint. When set (and S2S_INJECT_PROFILE=true and
# phoneNumber is available), before fetching user profile / latest summary the proxy
# first performs a POST S2S_LOGIN_BY_PHONE with {"phoneNumber": "..."} JSON body,
# extracts an auth token from the response (if any), and forwards it as
# Authorization: Bearer <token> for subsequent profile / summary / reports HTTP calls.
S2S_LOGIN_BY_PHONE = (os.environ.get("S2S_LOGIN_BY_PHONE", "") or "").strip().rstrip("/")
# Optional: list of scheduled reports / podcasts endpoint. When set (and S2S_INJECT_PROFILE=true)
# the proxy performs a plain GET S2S_REPORT_LIST, unwraps the JSON array (or {"result": [...]}),
# filters per-item fields (sourceScheduleLabel, date, podcastScript), sorts by date DESC,
# and injects the compiled block into the LLM instructions / user_context between
# the caller profile block and the latest conversation summary block.
S2S_REPORT_LIST = (os.environ.get("S2S_REPORT_LIST", "") or "").strip().rstrip("/")
# Feature flag: when "true" the proxy logs the FULL PROMPT TEXT (not just
# lengths) going to the upstream LLM/realtime backends. Includes:
# - WS session.update session.instructions (before/after patch)
# - /api/reasoning NODE_RED instructions + user_context full text
# - /api/reasonING OPENAI full system message + user transcript
# Default = off; only length summaries are logged (current behaviour).
# WARNING: setting this on will log potentially PII / sensitive caller
# profile data / user transcripts. Use only in development / debugging.
S2S_DEBUG_PROMPT = (
(os.environ.get("S2S_DEBUG_PROMPT", "") or "").strip().lower()
in {"1", "true", "yes", "on"}
)
S2S_REASONING_TYPE = _normalize_reasoning_type(os.environ.get("S2S_REASONING_TYPE", "PROXY"))
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "").strip()
OPENAI_BASE_URL = os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1").strip().rstrip("/")
OPENAI_REASONING_MODEL = os.environ.get("OPENAI_REASONING_MODEL", "gpt-4.1-mini").strip()
OPENAI_ORGANIZATION = os.environ.get("OPENAI_ORGANIZATION", "").strip()
OPENAI_PROJECT = os.environ.get("OPENAI_PROJECT", "").strip()
# Node-RED reasoning engine endpoint (modalità NODE_RED).
# Es.: http://localhost:1881 oppure http://s2s-reasoning:1880 (rete docker)
NODE_RED_REASONING_URL = os.environ.get("S2S_REASONING_NODE_RED_URL", "").strip().rstrip("/")
# Speech-to-speech load balancer URL. When set, the browser POSTs /api/session
# (which proxies <lb>/session here, server-side) and connects to the URL the LB
# returns (the original flow). The LB address itself is never sent to the browser.
# When empty, the user may instead set a direct s2s server URL in Settings and the
# browser connects to it straight (no load balancer).
LOAD_BALANCER_URL = os.environ.get("LOAD_BALANCER_URL", "").strip()
# Direct s2s server URL pinned by the deploy. Takes priority over the load
# balancer: when set, ALL LB logic is disabled (no /api/session proxy, no queue,
# no limiter, no sign-in) and the browser connects to this URL directly. Unlike
# the LB address it is NOT a secret — /api/config sends it to the client, which
# shows it read-only in Settings.
SPEECH_TO_SPEECH_URL = os.environ.get("SPEECH_TO_SPEECH_URL", "").strip()
# Optional same-origin WebSocket relay target. When set, the browser dials this
# app over ws(s)://<this-app>/v1/realtime and the server relays frames to the
# upstream speech-to-speech endpoint, keeping the internal host private.
S2S_PROXY_TARGET = os.environ.get("S2S_PROXY_TARGET", "").strip()
if SPEECH_TO_SPEECH_URL or S2S_PROXY_TARGET:
LOAD_BALANCER_URL = ""
# HF injects SPACE_ID ("owner/space") into every Space runtime; it's absent
# locally and on a plain `docker run`. We meter conversation time ONLY on the
# deployed Space — i.e. when BOTH the LB is configured AND we're on a Space.
# Off-Space (local dev, even with the LB exported) the app still proxies the LB,
# but nothing is metered: no budget, no reservations, no sign-in gating.
SPACE_ID = os.environ.get("SPACE_ID", "").strip()
LIMITER_ENABLED = bool(LOAD_BALANCER_URL) and bool(SPACE_ID)
SERPER_URL = "https://google.serper.dev/search"
# Cap results so the tool output stays small enough to feed back to the model.
MAX_RESULTS = 5
HERE = os.path.dirname(os.path.abspath(__file__))
app = FastAPI(title="s2s-demo")
# Wire HF OAuth before the app serves (no-op unless the OAuth env is present).
# Sign-in only matters when we're metering (prod Space), so gate it on that.
AUTH_ENABLED = LIMITER_ENABLED and auth.attach(app)
logger.info(
"[STARTUP] Other externals — reasoning=%s nodeRed=%s serper=%s lb=%s s2sProxy=%s s2sDirect=%s userAuth=%s reportsApi=%s summaryApi=%s loginApi=%s injectProfile=%s debugPrompt=%s auth=%s",
S2S_REASONING_TYPE,
NODE_RED_REASONING_URL or "-",
bool(SERPER_KEY),
bool(LOAD_BALANCER_URL),
bool(S2S_PROXY_TARGET),
bool(SPEECH_TO_SPEECH_URL),
bool(VOCERO_USER_PHONE_AUTH_URL),
bool(S2S_REPORT_LIST),
bool(S2S_LATEST_SUMMARY),
bool(S2S_LOGIN_BY_PHONE),
S2S_INJECT_PROFILE,
S2S_DEBUG_PROMPT,
AUTH_ENABLED,
)
logger.debug(
"[STARTUP] Runtime config — SPACE_ID=%s LIMITER_ENABLED=%s MAX_RESULTS=%d",
bool(SPACE_ID), LIMITER_ENABLED, MAX_RESULTS,
)
async def _report_server_debug(hypothesis_id: str, location: str, msg: str, data: dict | None = None) -> None:
return
def _realtime_event_summary(raw: str) -> dict | None:
try:
payload = json.loads(raw)
except Exception:
return None
if not isinstance(payload, dict):
return None
event_type = payload.get("type")
if not isinstance(event_type, str):
return None
summary = {"type": event_type}
if event_type == "session.update":
session = payload.get("session") if isinstance(payload.get("session"), dict) else {}
tools = session.get("tools") if isinstance(session.get("tools"), list) else []
summary["toolCount"] = len(tools)
summary["toolNames"] = [tool.get("name") for tool in tools if isinstance(tool, dict)]
summary["toolChoice"] = session.get("tool_choice")
summary["instructionLength"] = len(session.get("instructions") or "")
elif event_type in {"response.function_call_arguments.done", "error", "session.updated"}:
for key in ("name", "call_id", "code", "message", "response_id"):
if key in payload:
summary[key] = payload.get(key)
# --- INPUT (user speech → STT transcript) ---
# Can arrive from upstream (server-side STT) — match either direction.
elif event_type == "conversation.item.input_audio_transcription.completed":
summary["transcript"] = payload.get("transcript", "")
summary["item_id"] = payload.get("item_id")
elif event_type == "conversation.item.created":
item = payload.get("item") if isinstance(payload.get("item"), dict) else None
if isinstance(item, dict):
summary["item_id"] = item.get("id")
summary["role"] = item.get("role")
summary["item_type"] = item.get("type")
summary["status"] = item.get("status")
content_parts = []
for c in (item.get("content") or []):
if isinstance(c, dict):
if c.get("type") in {"input_text", "text"} and isinstance(c.get("text"), str):
content_parts.append(c["text"])
elif c.get("type") == "transcript" and isinstance(c.get("transcript"), str):
content_parts.append(c["transcript"])
if content_parts:
summary["text"] = " ".join(content_parts)
# --- OUTPUT (agent text/speech generation) ---
elif event_type == "response.output_item.done":
item = payload.get("item") if isinstance(payload.get("item"), dict) else None
if isinstance(item, dict):
summary["item_id"] = item.get("id")
summary["role"] = item.get("role")
summary["status"] = item.get("status")
content_parts = []
for c in (item.get("content") or []):
if isinstance(c, dict):
if c.get("type") == "text" and isinstance(c.get("text"), str) and c["text"]:
content_parts.append(c["text"])
elif c.get("type") == "audio" and isinstance(c.get("transcript"), str) and c["transcript"]:
content_parts.append(c["transcript"])
if content_parts:
summary["text"] = " ".join(content_parts)
elif event_type == "response.content_part.done":
part = payload.get("part") if isinstance(payload.get("part"), dict) else None
if isinstance(part, dict):
summary["response_id"] = payload.get("response_id")
summary["item_id"] = payload.get("item_id")
summary["output_index"] = payload.get("output_index")
summary["content_index"] = payload.get("content_index")
if part.get("type") == "text":
summary["text"] = part.get("text", "")
elif part.get("type") == "audio":
if isinstance(part.get("transcript"), str) and part["transcript"]:
summary["text"] = part["transcript"]
# Variante standard OpenAI:
elif event_type == "response.audio_transcript.done":
summary["response_id"] = payload.get("response_id")
summary["item_id"] = payload.get("item_id")
summary["text"] = payload.get("transcript", "")
# Variante upstream custom (osservato nel log dell'utente):
elif event_type == "response.output_audio_transcript.done":
summary["response_id"] = payload.get("response_id")
summary["item_id"] = payload.get("item_id")
summary["output_index"] = payload.get("output_index")
# Try field names in order of likelihood
text = ""
for key in ("transcript", "text", "output_text", "audio_text",
"output_transcript", "final_transcript", "output_audio_text",
"content", "message", "utterance"):
v = payload.get(key)
if isinstance(v, str) and v.strip():
text = v.strip()
break
# Se text ancora vuoto, cerca all'interno di eventuali sotto-oggetti noti
if not text:
for container in ("item", "part", "content", "output", "delta", "audio"):
sub = payload.get(container)
if isinstance(sub, dict):
for key in ("transcript", "text", "output_text"):
v = sub.get(key)
if isinstance(v, str) and v.strip():
text = v.strip()
break
if text:
break
if not text:
# Ultima risorsa: content[] array
arr = payload.get("content")
if isinstance(arr, list):
parts = []
for c in arr:
if isinstance(c, dict):
for key in ("transcript", "text"):
v = c.get(key)
if isinstance(v, str) and v.strip():
parts.append(v.strip())
if parts:
text = " ".join(parts)
summary["text"] = text
# Se non abbiamo ancora trovato testo, salva un'anteprima del raw
# per capire dove lo sta emettendo l'upstream custom.
if not text:
preview = raw[:800]
try:
pretty = json.dumps(payload, ensure_ascii=True)[:800]
except Exception:
pretty = preview
summary["_raw_preview"] = pretty
elif event_type == "response.function_call_arguments.done":
summary["arguments"] = payload.get("arguments")
# Questo evento è il KEY CHECK della Strategia B: se lo vediamo =
# ai-server ha accettato tools[] nativi (native toolcall detected).
_tool_name = payload.get("name") if isinstance(payload.get("name"), str) else ""
_call_id = payload.get("call_id") if isinstance(payload.get("call_id"), str) else ""
_args_json = payload.get("arguments") if isinstance(payload.get("arguments"), str) else "{}"
logger.info(
"[TOOLCALL] ✓ UPSTREAM ha emesso response.function_call_arguments.done! "
"tool=%s callId=%s argsLen=%d argsPreview=%s",
_tool_name or "(empty)",
_call_id or "-",
len(_args_json),
_args_json[:200],
)
try:
_args_parsed = json.loads(_args_json or "{}")
logger.info(
"[TOOLCALL] ✓ Tool call parsed argsKeys=%s",
sorted(_args_parsed.keys()) if isinstance(_args_parsed, dict) else "N/A",
)
except Exception:
pass
return summary
# Eventi realtime ad alta frequenza da silenziare nel log dettagliato (rumore).
# Essi vengono comunque tracciati dal report_server_debug interno se abilitato.
_NOISY_REALTIME_EVENT_TYPES = frozenset({
"response.output_audio.delta",
"response.audio.delta",
"conversation.item.input_audio_transcription.delta",
"response.text.delta",
"response.output_text.delta",
"rate_limits.updated",
})
# Eventi realtime silenziati anche quando arrivano dal lato cliente (input_audio_buffer.append ecc.).
_NOISY_CLIENT_EVENT_TYPES = frozenset({
"input_audio_buffer.append",
"input_audio_buffer.commit",
"input_audio_buffer.clear",
"input_audio_buffer.speech_started",
"input_audio_buffer.speech_stopped",
})
@app.on_event("startup")
async def _startup():
"""Stand up the usage DB and a periodic sweeper — metered (prod Space) only."""
if not LIMITER_ENABLED:
return
limiter.init()
asyncio.create_task(_sweeper())
async def _sweeper():
while True:
await asyncio.sleep(limiter.REAP_AFTER_SEC)
try:
await asyncio.to_thread(limiter.sweep)
except Exception as exc: # pragma: no cover - defensive
logger.warning("usage sweep failed: %r", exc)
class SearchRequest(BaseModel):
query: str
# Optional user-supplied key (fallback when the deploy has no server key).
# Used for this request only; never stored.
key: str | None = None
VOCERO_USER_CONTEXT_HINT = (
" Caller profile resolved from the page phoneNumber: "
)
REPORTS_LIST_HINT = (
" Scheduled reports / podcast scripts available for this caller before the current session. "
"Each entry is a scheduled report; use the podcastScript contents as trusted reference material when the user asks about scheduled topics, reports, or prior content."
)
LATEST_SUMMARY_HINT = (
" Latest conversation summary for this caller (most recent before the current session): "
)
def _resolve_user_context_text(phone_number: str, profile_obj: dict | None, loaded: bool = True) -> str:
"""Given a phone+profile object -> the short context string appended to
LLM system prompt / reasoning user_context when S2S_INJECT_PROFILE is on.
Generic (deploy-agnostic): profile_obj is the JSON returned by
VOCERO_USER_PHONE_AUTH_URL (optionally nested under {"result": ...}).
"""
if not loaded or not profile_obj or not phone_number:
return ""
try:
serialized = json.dumps(profile_obj, ensure_ascii=False)
except Exception:
serialized = str(profile_obj)
if len(serialized) > 1600:
serialized = serialized[:1600] + "..."
return f"{VOCERO_USER_CONTEXT_HINT}{serialized}. Use this as trusted caller context when relevant."
def _resolve_latest_summary_context(summary_value: str | dict | None) -> str:
"""Normalize the payload returned by S2S_LATEST_SUMMARY endpoint into a
single context string to be appended next to the profile block.
Accepts plain string / object (dumped JSON) / None. Capped at 1600 chars.
Wrapped with LATEST_SUMMARY_BEGIN/END markers so log inspection is easy.
Returns "" when no summary is available.
"""
if summary_value is None:
return ""
if isinstance(summary_value, dict) or isinstance(summary_value, list):
try:
serialized = json.dumps(summary_value, ensure_ascii=False)
except Exception:
serialized = str(summary_value)
else:
serialized = str(summary_value)
trimmed = serialized.strip()
if not trimmed:
return ""
if len(trimmed) > 1600:
trimmed = trimmed[:1600] + "..."
return f"{LATEST_SUMMARY_HINT}<<<LATEST_SUMMARY_BEGIN>>>\n{trimmed}\n<<<LATEST_SUMMARY_END>>>. Use this as trusted history when the user references prior calls."
def _parse_date_key(value: object) -> tuple:
"""Return a sortable tuple for an arbitrary date-like value.
Sort order is DESC (newest first). Non-dates / empty sort to the end (lowest priority).
Accepts ISO strings ("YYYY-MM-DD", "YYYY-MM-DDTHH:MM:SS"), Python datetime objects,
int/float unix timestamps, numbers.
"""
import datetime as _dt
if value is None:
return (1, 0)
if isinstance(value, _dt.datetime):
try:
return (0, -int(value.timestamp()))
except Exception:
return (1, 0)
if isinstance(value, _dt.date):
try:
ts = int(_dt.datetime.combine(value, _dt.time.min).timestamp())
return (0, -ts)
except Exception:
return (1, 0)
if isinstance(value, (int, float)):
try:
return (0, -int(value))
except Exception:
return (1, 0)
if isinstance(value, str):
s = value.strip()
if not s:
return (1, 0)
# Try ISO parse first (handles most formats)
try:
dt = _dt.datetime.fromisoformat(s.replace("Z", "+00:00"))
return (0, -int(dt.timestamp()))
except Exception:
pass
# Fallback: date-only YYYY-MM-DD
try:
dt = _dt.datetime.strptime(s[:10], "%Y-%m-%d")
return (0, -int(dt.timestamp()))
except Exception:
pass
# Last resort: lexicographic sort DESC (newest first still works for ISO-ish)
return (0, tuple(-ord(c) for c in s[:32]))
return (1, 0)
def _resolve_reports_context(raw_value: object) -> str:
"""Normalize / filter the S2S_REPORT_LIST response into a prompt block.
Expects an array/list (or dict {"result": [...]} unwrapped by the fetcher).
For each dict item keeps only fields: sourceScheduleLabel, date, podcastScript.
Sorts by date DESC (newest first) via _parse_date_key. Capped at 12000 chars
(podcastScript can be long). Wrapped with <<<REPORTS_LIST_BEGIN/END>>> markers.
Returns "" if no valid items / empty list.
"""
if raw_value is None:
return ""
items_list: list
if isinstance(raw_value, list):
items_list = raw_value
elif isinstance(raw_value, dict) and isinstance(raw_value.get("result"), list):
items_list = raw_value["result"]
else:
return ""
# Keep only dict items; for each project ONLY the three required fields (in order).
filtered: list[dict] = []
for item in items_list:
if not isinstance(item, dict):
continue
cleaned: dict = {}
if "sourceScheduleLabel" in item and item["sourceScheduleLabel"] is not None:
cleaned["sourceScheduleLabel"] = item["sourceScheduleLabel"]
if "date" in item and item["date"] is not None:
cleaned["date"] = item["date"]
if "podcastScript" in item and item["podcastScript"] is not None:
cleaned["podcastScript"] = item["podcastScript"]
if cleaned:
filtered.append(cleaned)
if not filtered:
return ""
# Sort by date DESC (newest first)
filtered.sort(key=lambda r: _parse_date_key(r.get("date")))
# Serialize: compact JSON (pretty-ish but one line per report saves space)
entries_lines = []
for idx, r in enumerate(filtered, start=1):
try:
entries_lines.append(f" [{idx}] {json.dumps(r, ensure_ascii=False)}")
except Exception:
entries_lines.append(f" [{idx}] {str(r)[:500]}")
body = "\n".join(entries_lines)
wrapped = f"{REPORTS_LIST_HINT}\n<<<REPORTS_LIST_BEGIN>>>\n{body}\n<<<REPORTS_LIST_END>>>. Use these scheduled reports as trusted prior content relevant to the caller's company topics."
if len(wrapped) > 12000:
# Truncate body preserving markers + hint
budget = 12000 - len(REPORTS_LIST_HINT) - len("\n<<<REPORTS_LIST_BEGIN>>>\n") - len("\n<<<REPORTS_LIST_END>>>...truncated.")
if budget > 0:
body = body[:budget] + "..."
wrapped = f"{REPORTS_LIST_HINT}\n<<<REPORTS_LIST_BEGIN>>>\n{body}\n<<<REPORTS_LIST_END>>>...truncated (cap 12000 chars). Use these scheduled reports as trusted prior content relevant to the caller's company topics."
return wrapped
async def _fetch_reports_list(phone_number: str | None = None, auth_token: str | None = None) -> list | dict | None:
"""Fetch scheduled reports list from S2S_REPORT_LIST.
Completely generic / deploy-agnostic: performs GET ${S2S_REPORT_LIST} and appends
?phoneNumber=<x> if a phone is provided. Accepts JSON (optional {"result": [...]}
wrapper) and returns the list or unwrapped list. On any error logs and returns None
(never fails the proxy / reasoning tunnel).
"""
if not S2S_REPORT_LIST:
return None
from urllib.parse import urlencode
if phone_number:
sep = "&" if "?" in S2S_REPORT_LIST else "?"
target_url = f"{S2S_REPORT_LIST}{sep}{urlencode({'phoneNumber': phone_number})}"
else:
target_url = S2S_REPORT_LIST
logger.info(
"[INJECT-REPORTS] fetch upstream URL=%s phoneLen=%s bearer=%s",
S2S_REPORT_LIST,
len(phone_number) if phone_number else "none",
"yes" if auth_token else "no",
)
try:
headers = {}
if auth_token:
headers["Authorization"] = f"Bearer {auth_token}"
async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as http:
resp = await http.get(target_url, headers=headers or None)
if resp.status_code != 200:
logger.warning(
"[INJECT-REPORTS] ⛔ upstream HTTP %s response=%.250s",
resp.status_code, resp.text,
)
return None
text = (resp.text or "").strip()
if not text:
logger.info("[INJECT-REPORTS] ⏭️ upstream returned empty body.")
return None
try:
payload = resp.json()
except Exception:
logger.warning(
"[INJECT-REPORTS] ⛔ upstream returned non-JSON body len=%d; cannot parse list. response=%.250s",
len(text), text,
)
return None
except Exception as exc:
logger.warning(
"[INJECT-REPORTS] ⛔ fetch unreachable error=%r",
exc,
)
return None
# Unwrap {"result": [...]} or other common list-wrapping dict keys
# (e.g. {"data": [...], "reports": [...], "items": [...], "content": [...], "value": [...]}).
unwrapped: object = payload
if isinstance(payload, dict):
# Log full structure for debugging if enabled
if S2S_DEBUG_PROMPT:
try:
keys_preview = ",".join(sorted(str(k) for k in payload.keys()))
preview = json.dumps(payload, ensure_ascii=False)
preview = preview[:4000]
logger.info(
"[DEBUG-REPORTS][fetch] upstream response dict keys=[%s] full PREVIEW (len=%d):\n%s\n",
keys_preview, len(preview), preview,
)
except Exception:
pass
# Preferred explicit "result" key first
if "result" in payload and isinstance(payload["result"], list):
unwrapped = payload["result"]
else:
# Scan all common list-wrapper keys (case-insensitive match on the key)
wrapper_keys = {k.lower() for k in (
"data", "reports", "reportList", "report_list", "report",
"items", "itemList", "item_list", "list", "content",
"scheduled", "scheduledReports", "scheduled_reports",
"podcastList", "podcast_list", "podcasts", "entries",
"records", "rows", "pageContent", "page_content", "value",
)}
for k, v in payload.items():
if isinstance(k, str) and isinstance(v, list) and k.lower() in wrapper_keys:
unwrapped = v
break
# Fallback: if it's still a dict with only ONE list-valued key,
# assume that's the list (common pattern like {"data": [...]} with
# unusual key name we don't hardcode).
if not isinstance(unwrapped, list):
list_valued_keys = [
k for k, v in payload.items() if isinstance(v, list) and len(v) > 0
]
if len(list_valued_keys) == 1:
unwrapped = payload[list_valued_keys[0]]
# Validate array shape
if isinstance(unwrapped, list):
n_items = len(unwrapped)
n_dict = sum(1 for x in unwrapped if isinstance(x, dict))
logger.info(
"[INJECT-REPORTS] ✅ reports list loaded OK (list len=%d dictItems=%d)",
n_items, n_dict,
)
return unwrapped
logger.info(
"[INJECT-REPORTS] ⏭️ upstream returned type=%s (not list); nothing to inject.",
type(unwrapped).__name__,
)
if S2S_DEBUG_PROMPT and isinstance(payload, dict):
# Log final top-level keys to help diagnose wrapper shape
logger.info(
"[DEBUG-REPORTS][fetch] wrapper not detected — top-level keys of dict response: [%s] "
"(list values under keys: %s)",
",".join(sorted(str(k) for k in payload.keys())),
",".join(
sorted(str(k) + "=list[" + str(len(v)) + "]" for k, v in payload.items() if isinstance(v, list))
) or "none",
)
return None
def _extract_login_token(login_payload: dict | None) -> str | None:
"""Given the parsed response from the S2S_LOGIN_BY_PHONE endpoint (already
{"result":…}-unwrapped), try to extract an auth / session token.
Looks under common names (case-insensitive). Returns None if nothing found.
"""
if not login_payload or not isinstance(login_payload, dict):
return None
# Unwrap one level of nesting if result is still an dict: handled by caller already.
candidates = {c.lower() for c in (
"token", "jwt", "access_token", "accessToken", "auth_token",
"authToken", "session_token", "sessionToken", "id_token", "idToken",
"bearer", "bearer_token", "bearerToken", "authentication_token",
)}
for k, v in login_payload.items():
if isinstance(k, str) and k.lower() in candidates and isinstance(v, str) and v.strip():
return v.strip()
# Second pass: nested dict {"data": {"token": …}} or {"auth": {"jwt": …}}
for k, v in login_payload.items():
if isinstance(v, dict):
nested = _extract_login_token(v)
if nested:
return nested
return None
async def _login_by_phone(phone_number: str) -> tuple[str | None, dict | None]:
"""Perform login-by-phone: POST S2S_LOGIN_BY_PHONE with JSON
{"phoneNumber": phone_number}. Returns (bearer_token_or_None, response_obj_or_None).
Never fails: on any HTTP/parse error logs a warning and returns (None, None)
so subsequent profile/summary calls simply proceed without a Bearer token.
"""
if not S2S_LOGIN_BY_PHONE or not phone_number:
return None, None
body = {"phoneNumber": phone_number}
logger.info(
"[INJECT-LOGIN] POST upstream URL=%s phoneLen=%d",
S2S_LOGIN_BY_PHONE,
len(phone_number),
)
try:
async with httpx.AsyncClient(timeout=10.0, follow_redirects=True) as http:
resp = await http.post(
S2S_LOGIN_BY_PHONE,
json=body,
headers={"Content-Type": "application/json"},
)
if resp.status_code != 200:
logger.warning(
"[INJECT-LOGIN] ⛔ upstream HTTP %s phoneLen=%d response=%.250s",
resp.status_code, len(phone_number), resp.text,
)
return None, None
text = (resp.text or "").strip()
if not text:
logger.info("[INJECT-LOGIN] ⏭️ upstream returned empty body; no login token available.")
return None, None
try:
payload = resp.json()
except Exception:
logger.warning(
"[INJECT-LOGIN] ⛔ upstream returned non-JSON body len=%d; cannot extract token. response=%.250s",
len(text), text,
)
return None, None
except Exception as exc:
logger.warning(
"[INJECT-LOGIN] ⛔ fetch unreachable phoneLen=%d error=%r",
len(phone_number), exc,
)
return None, None
# Unwrap {"result": …} if present (common wrapper convention).
if isinstance(payload, dict) and isinstance(payload.get("result"), dict):
unwrapped = payload["result"]
else:
unwrapped = payload if isinstance(payload, dict) else None
token = _extract_login_token(unwrapped) if isinstance(unwrapped, dict) else _extract_login_token(
payload if isinstance(payload, dict) else None
)
if isinstance(unwrapped, dict):
logger.info(
"[INJECT-LOGIN] ✅ login response OK keys=%s token=%s",
",".join(sorted(str(k) for k in unwrapped.keys())),
"present" if token else "missing",
)
if S2S_DEBUG_PROMPT:
logger.info(
"[DEBUG-LOGIN][response] FULL login payload (len=%d):\n%s\n",
len(json.dumps(unwrapped, ensure_ascii=False)),
json.dumps(unwrapped, ensure_ascii=False, indent=2),
)
return token, unwrapped
# Fallback: non-dict payload (string/list/number) — still log it, but no token expected.
logger.info(
"[INJECT-LOGIN] ⏭️ login response parsed but type=%s (not dict); no token extracted.",
type(payload).__name__,
)
if S2S_DEBUG_PROMPT:
logger.info(
"[DEBUG-LOGIN][response] FULL login payload (len=%d):\n%s\n",
len(str(payload)), payload,
)
return None, None
async def _fetch_latest_summary_for_phone(phone_number: str, auth_token: str | None = None) -> str | dict | None:
"""Fetch the latest caller summary via S2S_LATEST_SUMMARY URL.
Generic / completely deploy-agnostic: plain GET
S2S_LATEST_SUMMARY?phoneNumber=<phone>; accepts JSON (optional {"result": …}
wrapper) or plain text. Never fails the proxy / reasoning; returns None on
any error and logs it. If `auth_token` is provided it is sent as
Authorization: Bearer <auth_token>.
"""
if not S2S_LATEST_SUMMARY or not phone_number:
return None
from urllib.parse import urlencode
target_url = f"{S2S_LATEST_SUMMARY}?{urlencode({'phoneNumber': phone_number})}"
logger.info(
"[INJECT-SUMMARY] fetch upstream URL=%s phoneLen=%d bearer=%s",
S2S_LATEST_SUMMARY,
len(phone_number),
"yes" if auth_token else "no",
)
try:
headers = {}
if auth_token:
headers["Authorization"] = f"Bearer {auth_token}"
async with httpx.AsyncClient(timeout=10.0, follow_redirects=True) as http:
resp = await http.get(target_url, headers=headers or None)
if resp.status_code != 200:
logger.warning(
"[INJECT-SUMMARY] ⛔ upstream HTTP %s phoneLen=%d response=%.200s",
resp.status_code, len(phone_number), resp.text,
)
return None
# Accept JSON first, then fall back to plain text (some summary APIs
# return text/plain).
text = (resp.text or "").strip()
if not text:
return ""
try:
payload = resp.json()
except Exception:
logger.info(
"[INJECT-SUMMARY] upstream returned non-JSON response (len=%d); treating as plain string summary.",
len(text),
)
return text
except Exception as exc:
logger.warning(
"[INJECT-SUMMARY] ⛔ fetch unreachable phoneLen=%d error=%r",
len(phone_number), exc,
)
return None
# Unwrap {"result": …} just like profile fetch (common wrapper pattern).
if isinstance(payload, dict) and "result" in payload:
payload = payload["result"]
# Filter out empty dict/list / empty string after unwrap.
empty = (
payload is None
or (isinstance(payload, str) and not payload.strip())
or (isinstance(payload, dict) and len(payload) == 0)
or (isinstance(payload, list) and len(payload) == 0)
)
if empty:
logger.info("[INJECT-SUMMARY] ⏭️ upstream returned empty summary; nothing to inject.")
return None
if isinstance(payload, dict):
logger.info(
"[INJECT-SUMMARY] ✅ summary loaded OK (object) keys=%s",
",".join(sorted(str(k) for k in payload.keys())),
)
elif isinstance(payload, list):
logger.info(
"[INJECT-SUMMARY] ✅ summary loaded OK (list) len=%d",
len(payload),
)
else:
logger.info(
"[INJECT-SUMMARY] ✅ summary loaded OK (string) len=%d",
len(str(payload)),
)
return payload
async def _fetch_user_profile_for_phone(phone_number: str, auth_token: str | None = None) -> dict | None:
"""Fetch the profile JSON from VOCERO_USER_PHONE_AUTH_URL for a phone.
Returns the raw {"result": ...}-unwrapped object or None. Errors are logged
then suppressed (never fail the proxy / reasoning because of this).
When `auth_token` is provided it is forwarded as Authorization: Bearer <auth_token>
to the upstream profile endpoint.
"""
if not VOCERO_USER_PHONE_AUTH_URL or not phone_number:
return None
from urllib.parse import urlencode
target_url = f"{VOCERO_USER_PHONE_AUTH_URL}?{urlencode({'phoneNumber': phone_number})}"
logger.info(
"[INJECT-PROFILE] fetch upstream URL=%s phoneLen=%d bearer=%s",
VOCERO_USER_PHONE_AUTH_URL,
len(phone_number),
"yes" if auth_token else "no",
)
try:
headers = {}
if auth_token:
headers["Authorization"] = f"Bearer {auth_token}"
async with httpx.AsyncClient(timeout=10.0, follow_redirects=True) as http:
resp = await http.get(target_url, headers=headers or None)
if resp.status_code != 200:
logger.info(
"[INJECT-PROFILE] ⛔ upstream HTTP %s phoneLen=%d response=%.200s",
resp.status_code, len(phone_number), resp.text,
)
return None
payload = resp.json()
except Exception as exc:
logger.warning(
"[INJECT-PROFILE] ⛔ fetch unreachable phoneLen=%d error=%r",
len(phone_number), exc,
)
return None
if (isinstance(payload, dict)
and "result" in payload
and isinstance(payload["result"], dict)):
profile_obj = payload["result"]
elif isinstance(payload, dict) and not isinstance(payload, list):
profile_obj = payload
else:
logger.info(
"[INJECT-PROFILE] ⛔ upstream returned non-object JSON type=%s keys=?",
type(payload).__name__,
)
return None
logger.info(
"[INJECT-PROFILE] ✅ profile loaded OK keys=%s",
",".join(sorted(profile_obj.keys())),
)
return profile_obj
def _augment_instructions_with_profile(base_instructions: str, user_context_text: str, reports_context_text: str | None = None, summary_context_text: str | None = None) -> str:
"""When S2S_INJECT_PROFILE is true: appends the profile + (optional) reports list
+ (optional) latest summary context strings to the LLM system/instructions prompt.
Append order (most generic → most specific / chronologically oldest → newest):
1. base instructions (primary prompt / user settings / default)
2. profile_ctx (caller identity block)
3. reports_ctx (scheduled reports / podcast scripts list, sorted by date DESC)
4. summary_ctx (latest conversation summary)
Preserves the existing instructions text.
"""
base = (base_instructions or "").strip()
profile_ctx = (user_context_text or "").strip()
reports_ctx = (reports_context_text or "").strip()
summary_ctx = (summary_context_text or "").strip()
contexts = [c for c in [profile_ctx, reports_ctx, summary_ctx] if c]
if not contexts:
return base_instructions or ""
augmented = base
for ctx in contexts:
augmented = (augmented + " " + ctx).strip() if augmented else ctx
changed = augmented != base
logger.info(
"[INJECT-PROFILE] ✏️ appended profile/reports/summary to LLM instructions. "
"baseLen=%d profileCtxLen=%d reportsCtxLen=%d summaryCtxLen=%d newLen=%d changed=%s",
len(base), len(profile_ctx), len(reports_ctx), len(summary_ctx), len(augmented), changed,
)
if S2S_DEBUG_PROMPT and reports_ctx:
logger.info(
"[DEBUG-REPORTS][instructions augment] FULL reports list block (len=%d):\n%s\n",
len(reports_ctx), reports_ctx,
)
if S2S_DEBUG_PROMPT and summary_ctx:
logger.info(
"[DEBUG-SUMMARY][instructions augment] FULL latest summary block (len=%d):\n%s\n",
len(summary_ctx), summary_ctx,
)
return augmented
def _patch_client_session_update_with_profile(raw_text: str, user_context_text: str, reports_context_text: str | None = None, summary_context_text: str | None = None) -> str:
"""If raw_text is a client session.update JSON and S2S_INJECT_PROFILE is
true, append user_context_text + (optional) reports_context_text +
(optional) summary_context_text to session.instructions.
Returns the modified JSON string, or the original string unchanged if:
- it is not session.update JSON (the VAST majority of frames -> silent passthrough, NO LOG)
- user_context_text AND reports_context_text AND summary_context_text are all empty
- S2S_INJECT_PROFILE is False
- parse fails
"""
# Step 1: only consider session.update frames. All other frame types
# (response.create, conversation.item.*, audio_* etc.) pass through
# SILENTLY — we never emit logs for them, which avoids spamming the
# console with 6-20 identical "SKIPPED" lines per session.
try:
payload = json.loads(raw_text)
except Exception:
return raw_text
if not isinstance(payload, dict) or payload.get("type") != "session.update":
return raw_text
# From here on: we know this is a real session.update (typically emitted
# just once or twice per realtime session). We can log safely — 1 line,
# not 20 repeated lines.
if not S2S_INJECT_PROFILE:
logger.info(
"[INJECT-PROFILE] ⏭️ WS patch SKIPPED on session.update (S2S_INJECT_PROFILE=false). "
"WS stays pure passthrough."
)
return raw_text
has_profile = bool((user_context_text or "").strip())
has_reports = bool((reports_context_text or "").strip())
has_summary = bool((summary_context_text or "").strip())
if not (has_profile or has_reports or has_summary):
logger.info(
"[INJECT-PROFILE] ⏭️ WS patch SKIPPED on session.update (no profile context, no reports list, no latest summary). "
"Probably phoneNumber query param missing / fetch failed upstream / "
"endpoints or S2S_LATEST_SUMMARY / S2S_REPORT_LIST returned empty result."
)
return raw_text
if has_summary and not has_profile:
logger.info(
"[INJECT-SUMMARY] profile missing, but LATEST SUMMARY block loaded. "
"Will inject summary alone into session.update."
)
if has_reports and not has_profile:
logger.info(
"[INJECT-REPORTS] profile missing, but REPORTS LIST block loaded. "
"Will inject reports alone into session.update."
)
top_session = payload.get("session")
if not isinstance(top_session, dict):
return raw_text
current_instructions = (
top_session.get("instructions")
if isinstance(top_session.get("instructions"), str)
else ""
)
augmented = _augment_instructions_with_profile(
current_instructions, user_context_text, reports_context_text, summary_context_text
)
if S2S_DEBUG_PROMPT:
logger.info(
"[DEBUG-PROMPT][WS session.update] BEFORE instructions (len=%d):\n%s\n",
len(current_instructions), current_instructions,
)
if augmented == current_instructions:
logger.info(
"[INJECT-PROFILE] ⏭️ WS patch SKIPPED on session.update (augmented == current; "
"profile already injected or context empty)"
)
if S2S_DEBUG_PROMPT and augmented.strip():
logger.info(
"[DEBUG-PROMPT][WS session.update] FINAL unchanged (len=%d):\n%s\n",
len(augmented), augmented,
)
return raw_text
top_session["instructions"] = augmented
logger.info(
"[INJECT-PROFILE] ✅ WS session.update PATCHED -> upstream instructions. "
"origLen=%d newLen=%d injectedLen=%d profile=%s reports=%s summary=%s",
len(current_instructions),
len(augmented),
(len(augmented) - len(current_instructions)),
has_profile, has_reports, has_summary,
)
if S2S_DEBUG_PROMPT:
logger.info(
"[DEBUG-PROMPT][WS session.update] AFTER patch (len=%d):\n%s\n",
len(augmented), augmented,
)
if S2S_DEBUG_PROMPT and reports_context_text and reports_context_text.strip():
logger.info(
"[DEBUG-REPORTS][WS session.update] FULL reports list block (len=%d):\n%s\n",
len(reports_context_text.strip()), reports_context_text.strip(),
)
# Also log the FINAL upstream instructions when S2S_DEBUG_PROMPT is on and
# this is a session.update (so we see the EXACT text reaching the model even
# when S2S_INJECT_PROFILE is off and no patch happened.)
if (not S2S_INJECT_PROFILE) and S2S_DEBUG_PROMPT and current_instructions.strip():
logger.info(
"[DEBUG-PROMPT][WS session.update] FINAL upstream instructions (injectProfile=false, len=%d):\n%s\n",
len(current_instructions), current_instructions,
)
try:
return json.dumps(payload)
except Exception:
logger.info(
"[INJECT-PROFILE] ⛔ JSON re-serialize failed on session.update. "
"Forwarding unpatched."
)
return raw_text
class ReasoningRequest(BaseModel):
transcript: str
instructions: str = ""
language: str = ""
user_context: str = ""
# Optional phone number. When S2S_INJECT_PROFILE is true the proxy fetches
# the caller profile from VOCERO_USER_PHONE_AUTH_URL and:
# 1. prepends the serialized profile to user_context before feeding the
# engine (NODE_RED or OPENAI);
# 2. appends the profile context to the LLM instructions prompt.
phoneNumber: str | None = None
def _url_with_query(url: str, **params: str) -> str:
target = urlsplit(url)
merged_query = dict(parse_qsl(target.query, keep_blank_values=True))
for key, value in params.items():
if value:
merged_query[key] = value
return urlunsplit(
(
target.scheme,
target.netloc,
target.path,
urlencode(merged_query, doseq=True),
target.fragment,
)
)
def _openai_headers() -> dict[str, str]:
if not OPENAI_API_KEY:
raise RuntimeError("OPENAI_API_KEY is not configured.")
headers = {
"Authorization": f"Bearer {OPENAI_API_KEY}",
"Content-Type": "application/json",
}
if OPENAI_ORGANIZATION:
headers["OpenAI-Organization"] = OPENAI_ORGANIZATION
if OPENAI_PROJECT:
headers["OpenAI-Project"] = OPENAI_PROJECT
return headers
def _effective_s2s_url(request: Request) -> str:
return SPEECH_TO_SPEECH_URL or _proxy_public_url(request)
@app.get("/api/config")
async def config(request: Request):
"""Client bootstrap: whether web search is available, whether the deploy runs
behind a load balancer (so the browser uses the /api/session proxy + limiter),
whether HF sign-in is available, and whether the user may instead set a direct
s2s server URL. The LB address itself is intentionally NOT included."""
logger.debug("[INTERNAL] /api/config requested client=%s", request.client.host if request.client else "?")
primary_prompt = S2S_PRIMARY_PROMPT
out = {
"search": bool(SERPER_KEY),
"lb": bool(LOAD_BALANCER_URL),
"allowDirect": not LOAD_BALANCER_URL,
"s2sUrl": _effective_s2s_url(request),
"primaryPrompt": primary_prompt,
"reasoningType": S2S_REASONING_TYPE,
"auth": AUTH_ENABLED,
"injectProfile": S2S_INJECT_PROFILE,
"debugPrompt": S2S_DEBUG_PROMPT,
}
logger.debug(
"[INTERNAL] /api/config returned promptLen=%d reasoning=%s",
len(primary_prompt),
S2S_REASONING_TYPE,
)
return out
@app.post("/api/reasoning")
async def reasoning(req: ReasoningRequest):
if S2S_REASONING_TYPE == "PROXY":
raise HTTPException(status_code=409, detail="Reasoning endpoint is disabled in PROXY mode.")
transcript = (req.transcript or "").strip()
if not transcript:
raise HTTPException(status_code=400, detail="transcript is required.")
# ── S2S_INJECT_PROFILE: resolve caller profile + reports list + latest summary ──
# (When S2S_INJECT_PROFILE=false this entire block is a no-op; user_context
# and instructions are taken verbatim from the request body.)
inject_profile_text = ""
inject_reports_text = ""
inject_summary_text = ""
phone = (req.phoneNumber or "").strip()
if not S2S_INJECT_PROFILE:
logger.info(
"[INJECT-PROFILE] ⏭️ REASONING: skipped (S2S_INJECT_PROFILE=false). "
"Using verbatim request body; no fetch, no augmentation."
)
else:
needs_phone_authed = bool(VOCERO_USER_PHONE_AUTH_URL.strip())
needs_reports = bool(S2S_REPORT_LIST.strip())
needs_summary = bool(S2S_LATEST_SUMMARY.strip())
needs_login = bool(S2S_LOGIN_BY_PHONE.strip())
can_profile_ok = needs_phone_authed or needs_reports or needs_summary or needs_login
if (not needs_phone_authed) and (not needs_reports) and (not needs_summary) and (not needs_login):
logger.info(
"[INJECT-PROFILE] ⏭️ REASONING: skipped (VOCERO_USER_PHONE_AUTH_URL, S2S_REPORT_LIST, S2S_LATEST_SUMMARY and S2S_LOGIN_BY_PHONE all empty). "
"Set at least one to enable login / fetch profile/reports/summary for REASONING endpoint."
)
elif not phone:
logger.info(
"[INJECT-PROFILE] ⏭️ REASONING: skipped (phoneNumber missing from request body). "
"Browser-side issue: check main.js passes voceroUserPhoneNumber in JSON body. (Note: S2S_REPORT_LIST fetch works even without phone.)"
)
elif not can_profile_ok:
logger.info(
"[INJECT-PROFILE] ⏭️ REASONING: skipped (env for phone empty and no reports/summary/login API)."
)
else:
# ###################################################################
# 0) Login-by-phone (optional — only if S2S_LOGIN_BY_PHONE env set)
# Obtains a bearer token that will be forwarded for profile/reports/summary fetches.
# ###################################################################
login_token = None
if S2S_LOGIN_BY_PHONE and phone:
login_token, _login_resp = await _login_by_phone(phone)
# ###################################################################
# 1) Profile fetch (optional — only if env is configured)
# ###################################################################
if needs_phone_authed and phone:
profile_obj = await _fetch_user_profile_for_phone(phone, auth_token=login_token)
inject_profile_text = _resolve_user_context_text(phone, profile_obj, loaded=True)
if inject_profile_text.strip():
logger.info(
"[INJECT-PROFILE] ✅ REASONING: profile READY phoneLen=%d contextLen=%d -> augmenting instructions + user_context before sending upstream.",
len(phone), len(inject_profile_text),
)
else:
logger.info(
"[INJECT-PROFILE] ⏭️ REASONING: profile fetch OK but resolved context empty."
)
# ###################################################################
# 1.5) Scheduled reports / podcast scripts list (optional — S2S_REPORT_LIST)
# ###################################################################
if needs_reports:
if S2S_DEBUG_PROMPT:
logger.info(
"[INJECT-REPORTS] starting REASONING fetch S2S_REPORT_LIST (phoneLen=%s, bearer=%s)...",
len(phone) if phone else "none",
"yes" if login_token else "no",
)
reports_raw = await _fetch_reports_list(phone_number=phone if phone else None, auth_token=login_token)
inject_reports_text = _resolve_reports_context(reports_raw)
if inject_reports_text.strip():
logger.info(
"[INJECT-REPORTS] ✅ REASONING: reports list READY wrappedLen=%d -> augmenting instructions + user_context before sending upstream.",
len(inject_reports_text),
)
else:
logger.info(
"[INJECT-REPORTS] ⏭️ REASONING: reports list skipped (context empty / upstream returned empty list / fetch failed)."
)
# ###################################################################
# 2) Latest summary fetch (optional — only if env is configured)
# ###################################################################
if needs_summary and phone:
if S2S_DEBUG_PROMPT:
logger.info(
"[INJECT-SUMMARY] starting REASONING fetch (reasoning summary... S2S_LATEST_SUMMARY configured; phoneLen=%d.",
len(phone),
)
latest_summary_raw = await _fetch_latest_summary_for_phone(phone, auth_token=login_token)
inject_summary_text = _resolve_latest_summary_context(latest_summary_raw)
if inject_summary_text.strip():
logger.info(
"[INJECT-SUMMARY] ✅ REASONING: latest summary READY phoneLen=%d wrappedLen=%d -> augmenting instructions + user_context before sending upstream.",
len(phone), len(inject_summary_text),
)
else:
logger.info(
"[INJECT-SUMMARY] ⏭️ REASONING: latest summary skipped (context empty / upstream returned empty summary / fetch failed.)"
)
user_context_parts: list[str] = []
if inject_profile_text:
user_context_parts.append(inject_profile_text)
if inject_reports_text:
user_context_parts.append(inject_reports_text)
if inject_summary_text:
user_context_parts.append(inject_summary_text)
original_user_context = (req.user_context or "").strip()
if original_user_context:
user_context_parts.append(original_user_context)
effective_user_context = "\n\n".join(user_context_parts)
if S2S_INJECT_PROFILE:
any_injected = bool(inject_profile_text or inject_reports_text or inject_summary_text)
if any_injected:
logger.info(
"[INJECT-PROFILE] REASONING user_context final parts=%d serverProfileLen=%d serverReportsLen=%d serverSummaryLen=%d browserUserCtxLen=%d joinedLen=%d",
len(user_context_parts),
len(inject_profile_text),
len(inject_reports_text),
len(inject_summary_text),
len(original_user_context),
len(effective_user_context),
)
if S2S_DEBUG_PROMPT and inject_reports_text.strip():
logger.info(
"[DEBUG-REPORTS][REASONING augment] FULL reports list block (len=%d):\n%s\n",
len(inject_reports_text), inject_reports_text,
)
if S2S_DEBUG_PROMPT and inject_summary_text.strip():
logger.info(
"[DEBUG-SUMMARY][REASONING augment] FULL latest summary block (len=%d):\n%s\n",
len(inject_summary_text), inject_summary_text,
)
effective_instructions = _augment_instructions_with_profile(
req.instructions or "", inject_profile_text, inject_reports_text, inject_summary_text
)
else:
effective_instructions = (req.instructions or "").strip()
# ────────────────────────────────────────────────────────────────────
# Modalità A: NODE_RED — inoltra TUTTO il payload a Node-RED
# Il flow Node-RED decide autonomamente.
# FASE 1 (LOG-ONLY): Node-RED risponde con {answer: "", log_only: true}
# => il client (main.js) NON interferisce, lascia parlare il backend S2S.
# FASI successive: Node-RED produce {answer: "...", log_only: false}
# Risposta attesa: {answer, log_only?, usage, debug, latency_ms, …}
# ────────────────────────────────────────────────────────────────────
if S2S_REASONING_TYPE == "NODE_RED":
if not NODE_RED_REASONING_URL:
raise HTTPException(
status_code=503,
detail="S2S_REASONING_NODE_RED_URL is not configured. Set the env var to the Node-RED endpoint (e.g. http://localhost:1881).",
)
logger.info(
"[INJECT-PROFILE] REASONING->NODE_RED body fields. instructionsLen=%d userCtxLen=%d phoneLen=%d",
len(effective_instructions), len(effective_user_context), len(phone),
)
upstream_body = {
"transcript": transcript,
"instructions": effective_instructions,
"language": (req.language or "").strip(),
"user_context": effective_user_context,
}
if S2S_DEBUG_PROMPT:
logger.info(
"[DEBUG-PROMPT][REASONING → NODE_RED] FULL upstream_body JSON:\n%s\n",
json.dumps(upstream_body, ensure_ascii=False, indent=2),
)
logger.debug(
"[EXT][NODE_RED] reasoning POST %s/api/reasoning lang=%s transcriptLen=%d",
NODE_RED_REASONING_URL,
(req.language or "")[:20] or "auto",
len(transcript),
)
try:
async with httpx.AsyncClient(timeout=120.0) as http:
resp = await http.post(
f"{NODE_RED_REASONING_URL}/api/reasoning",
json=upstream_body,
headers={
"Content-Type": "application/json",
"X-S2S-Caller": "demo-server",
},
)
except httpx.RequestError as exc:
logger.warning("[EXT][NODE_RED] unreachable: %r", exc)
raise HTTPException(status_code=502, detail="Node-RED reasoning engine is unreachable.")
logger.debug(
"[EXT][NODE_RED] reasoning RESP status=%s len=%dB",
resp.status_code, len(resp.content or b""),
)
if resp.status_code != 200:
logger.warning("[EXT][NODE_RED] failed %s: %s", resp.status_code, resp.text[:500])
raise HTTPException(
status_code=502,
detail=f"Node-RED reasoning failed (HTTP {resp.status_code}).",
)
try:
payload = resp.json()
except Exception:
logger.warning("[EXT][NODE_RED] invalid JSON: %s", resp.text[:500])
raise HTTPException(status_code=502, detail="Node-RED returned invalid JSON.")
answer = payload.get("answer") if isinstance(payload, dict) else None
log_only = bool(isinstance(payload, dict) and payload.get("log_only") is True)
if not isinstance(answer, str):
logger.warning("[EXT][NODE_RED] invalid answer type: %s", json.dumps(payload, ensure_ascii=True)[:800])
raise HTTPException(status_code=502, detail="Node-RED returned invalid answer field.")
answer_stripped = answer.strip()
# ── Log-only mode: answer vuota + flag log_only sono L'INTENZIONE ──
# Non solleviamo errore; il flag viene inoltrato al client.
if not answer_stripped and not log_only:
logger.warning("[EXT][NODE_RED] empty answer and !log_only: %s", json.dumps(payload, ensure_ascii=True)[:800])
raise HTTPException(status_code=502, detail="Node-RED returned no answer without log_only flag.")
usage = payload.get("usage") if isinstance(payload, dict) else None
if isinstance(usage, dict):
logger.debug(
"[EXT][NODE_RED] reasoning USAGE prompt=%d completion=%d total=%d",
usage.get("prompt_tokens", 0), usage.get("completion_tokens", 0), usage.get("total_tokens", 0),
)
extra = {}
for key in ("latency_ms", "intent", "caller_profile_keys", "debug"):
if isinstance(payload, dict) and key in payload:
extra[key] = payload[key]
if log_only:
logger.info(
"[Vocero REASONING] engine=NODE_RED mode=LOG_ONLY transcript=%s extra=%s",
transcript[:300],
json.dumps({k: v for k, v in extra.items() if k != "debug"}, ensure_ascii=True)[:400],
)
else:
logger.info(
"[Vocero REASONING] engine=NODE_RED transcript=%s answer=%s extra=%s",
transcript[:300], answer_stripped[:500],
json.dumps({k: v for k, v in extra.items() if k != "debug"}, ensure_ascii=True)[:400],
)
out = {"answer": answer_stripped}
if log_only:
out["log_only"] = True
if usage:
out["usage"] = usage
out.update({k: v for k, v in extra.items() if k != "debug" or logger.level <= logging.DEBUG})
return JSONResponse(out)
# ────────────────────────────────────────────────────────────────────
# Modalità B: OPENAI — chiamata diretta a OpenAI Chat Completions
# ────────────────────────────────────────────────────────────────────
if not OPENAI_API_KEY:
raise HTTPException(status_code=503, detail="OPENAI_API_KEY is not configured.")
system_parts = []
instructions = effective_instructions or ((req.instructions or "").strip())
if instructions:
system_parts.append(instructions)
if effective_user_context.strip():
system_parts.append(f"User context:\n{effective_user_context.strip()}")
if req.language.strip():
system_parts.append(
"Reply only with the final assistant answer text. "
f"Keep the answer natural and concise in this language: {req.language.strip()}."
)
else:
system_parts.append("Reply only with the final assistant answer text. Keep the answer natural and concise.")
logger.info(
"[INJECT-PROFILE] REASONING->OPENAI system message parts=%d instructionsLen=%d userCtxLen=%d phoneLen=%d",
len(system_parts),
len(effective_instructions),
len(effective_user_context),
len(phone),
)
system_msg = "\n\n".join(system_parts)
body = {
"model": OPENAI_REASONING_MODEL,
"messages": [
{"role": "system", "content": system_msg},
{"role": "user", "content": transcript},
],
}
if S2S_DEBUG_PROMPT:
logger.info(
"[DEBUG-PROMPT][REASONING → OPENAI] FULL SYSTEM message (len=%d):\n%s\n",
len(system_msg), system_msg,
)
logger.info(
"[DEBUG-PROMPT][REASONING → OPENAI] USER transcript (len=%d):\n%s\n",
len(transcript), transcript,
)
logger.debug(
"[EXT][OPENAI] reasoning POST %s/chat/completions model=%s systemLen=%d transcriptLen=%d language=%s",
OPENAI_BASE_URL, OPENAI_REASONING_MODEL,
len(system_parts[0]) if system_parts else 0, len(transcript),
(req.language or "")[:20] or "auto",
)
try:
async with httpx.AsyncClient(timeout=30.0) as http:
resp = await http.post(
f"{OPENAI_BASE_URL}/chat/completions",
headers=_openai_headers(),
json=body,
)
except httpx.RequestError as exc:
logger.warning("OpenAI reasoning unreachable: %r", exc)
raise HTTPException(status_code=502, detail="OpenAI reasoning unreachable.")
logger.debug(
"[EXT][OPENAI] reasoning RESP status=%s len=%dB",
resp.status_code, len(resp.content or b""),
)
if resp.status_code != 200:
logger.warning("OpenAI reasoning failed %s: %s", resp.status_code, resp.text[:400])
raise HTTPException(status_code=502, detail=f"OpenAI reasoning failed ({resp.status_code}).")
try:
payload = resp.json()
except Exception:
logger.warning("OpenAI reasoning returned invalid JSON: %s", resp.text[:400])
raise HTTPException(status_code=502, detail="OpenAI reasoning returned invalid JSON.")
answer = ""
choices = payload.get("choices")
if isinstance(choices, list) and choices:
message = choices[0].get("message") if isinstance(choices[0], dict) else None
if isinstance(message, dict):
content = message.get("content")
if isinstance(content, str):
answer = content.strip()
usage = payload.get("usage")
if isinstance(usage, dict):
logger.debug(
"[EXT][OPENAI] reasoning USAGE prompt=%d completion=%d total=%d",
usage.get("prompt_tokens", 0), usage.get("completion_tokens", 0), usage.get("total_tokens", 0),
)
if not answer:
logger.warning("OpenAI reasoning returned empty answer: %s", json.dumps(payload, ensure_ascii=True)[:800])
raise HTTPException(status_code=502, detail="OpenAI reasoning returned no answer.")
logger.info("[Vocero REASONING] engine=OPENAI transcript=%s answer=%s", transcript[:300], answer[:500])
return JSONResponse({"answer": answer})
async def _forward_agent_text_to_nodered(payload: dict) -> None:
"""Fire-and-forget: inoltra la risposta TTS del backend nativo a Node-RED
(endpoint /api/log/system-response) per il logging della FASE 1.
Se Node-RED è down o non è configurato: silent ignore, non propagare errori
al realtime loop (non vogliamo bloccare la connessione WS)."""
if S2S_REASONING_TYPE != "NODE_RED" or not NODE_RED_REASONING_URL:
return
url = f"{NODE_RED_REASONING_URL}/api/log/system-response"
logger.debug(
"[EXT][NODE_RED] forward system-response POST %s len=%d source=%s",
NODE_RED_REASONING_URL,
len(str(payload.get("tts", "")) or 0),
payload.get("source_event") or "-",
)
try:
async with httpx.AsyncClient(timeout=5.0) as http:
resp = await http.post(
url,
json=payload,
headers={
"Content-Type": "application/json",
"X-S2S-Caller": "demo-server",
},
)
if resp.status_code != 200:
logger.debug(
"[EXT][NODE_RED] forward system-response non-200 status=%s body=%s",
resp.status_code,
(resp.text or "")[:300],
)
except httpx.RequestError as exc:
logger.debug("[EXT][NODE_RED] forward system-response unreachable: %r", exc)
except Exception as exc: # noqa: BLE001 — vogliamo non bloccare mai
logger.debug("[EXT][NODE_RED] forward system-response unexpected: %r", exc)
@app.websocket("/v1/realtime")
async def realtime_proxy(ws: WebSocket):
"""Relay the browser's same-origin realtime WebSocket to the configured upstream."""
client_ip = ws.client.host if ws.client else "?"
if not S2S_PROXY_TARGET:
logger.debug("[EXT][S2S-PROXY] client=%s rejected: S2S_PROXY_TARGET not configured", client_ip)
await ws.accept()
await ws.close(code=1013, reason="Realtime proxy not configured.")
return
await ws.accept()
upstream_url = _proxy_upstream_url(ws)
subprotocols = _subprotocols(ws)
# ── S2S_INJECT_PROFILE: resolve caller profile + latest summary ──
# Query params (browser URL carries ?phoneNumber=...) are forwarded by the
# client WS library: we read ?phoneNumber= 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_profile_context = ""
injected_reports_context = ""
injected_summary_context = ""
inject_phone = ""
login_token: str | None = None
login_response_obj: dict | None = None
if not S2S_INJECT_PROFILE:
logger.info(
"[INJECT-PROFILE] ⏭️ WS connect: SKIPPED (S2S_INJECT_PROFILE=false). "
"No profile fetch; tunnel stays pure passthrough."
)
else:
needs_phone_authed = bool(VOCERO_USER_PHONE_AUTH_URL.strip())
needs_reports = bool(S2S_REPORT_LIST.strip())
needs_summary = bool(S2S_LATEST_SUMMARY.strip())
needs_login = bool(S2S_LOGIN_BY_PHONE.strip())
can_inject = needs_phone_authed or needs_reports or needs_summary or needs_login
if not can_inject:
logger.info(
"[INJECT-PROFILE] ⏭️ WS connect: SKIPPED (VOCERO_USER_PHONE_AUTH_URL, S2S_REPORT_LIST, S2S_LATEST_SUMMARY and S2S_LOGIN_BY_PHONE all empty). "
"Set at least one with S2S_INJECT_PROFILE=true to enable caller context / reports / login."
)
else:
client_query = dict(parse_qsl(ws.url.query, keep_blank_values=True))
inject_phone = (client_query.get("phoneNumber") or "").strip()
if (not inject_phone) and (not needs_reports):
# reports list can be fetched without phone (generic list)
logger.info(
"[INJECT-PROFILE] ⏭️ WS connect: SKIPPED (incoming WS query has NO ?phoneNumber=). "
"Make sure the browser opens the WS with the phone query param."
)
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 _login_by_phone(inject_phone)
# 1) Profile fetch (optional — only if env is set)
if needs_phone_authed and inject_phone:
profile_obj = await _fetch_user_profile_for_phone(inject_phone, auth_token=login_token)
injected_profile_context = _resolve_user_context_text(
inject_phone, profile_obj, loaded=True
)
if injected_profile_context.strip():
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:
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 S2S_DEBUG_PROMPT:
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 _fetch_reports_list(
phone_number=inject_phone if inject_phone else None,
auth_token=login_token,
)
injected_reports_context = _resolve_reports_context(reports_raw)
if injected_reports_context.strip():
logger.info(
"[INJECT-REPORTS] ✅ WS connect: reports list READY wrappedLen=%d -> will patch session.update frames in this tunnel.",
len(injected_reports_context),
)
else:
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:
if S2S_DEBUG_PROMPT:
logger.info(
"[INJECT-SUMMARY] WS connect: starting S2S_LATEST_SUMMARY fetch for phoneLen=%d...",
len(inject_phone),
)
latest_summary_raw = await _fetch_latest_summary_for_phone(inject_phone, auth_token=login_token)
injected_summary_context = _resolve_latest_summary_context(latest_summary_raw)
if injected_summary_context.strip():
logger.info(
"[INJECT-SUMMARY] ✅ WS connect: latest SUMMARY READY phoneLen=%d wrappedLen=%d -> will patch session.update frames in this tunnel.",
len(inject_phone), len(injected_summary_context),
)
else:
logger.info(
"[INJECT-SUMMARY] ⏭️ WS connect: latest summary skipped (context empty / upstream returned empty summary / fetch failed)."
)
logger.info(
"[INJECT-PROFILE] WS connect FINAL: injectProfile=%s injectReports=%s injectSummary=%s login=%s phoneLen=%d profileCtxLen=%d reportsCtxLen=%d summaryCtxLen=%d target=%s",
bool(injected_profile_context),
bool(injected_reports_context),
bool(injected_summary_context),
("token-present" if login_token else "skipped"),
len(inject_phone),
len(injected_profile_context),
len(injected_reports_context),
len(injected_summary_context),
upstream_url,
)
if S2S_DEBUG_PROMPT and injected_reports_context.strip():
logger.info(
"[DEBUG-REPORTS][WS connect] FULL reports list block loaded (len=%d):\n%s\n",
len(injected_reports_context), injected_reports_context,
)
if S2S_DEBUG_PROMPT and injected_summary_context.strip():
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=subprotocols or None,
open_timeout=15,
max_size=None,
) as upstream:
logger.info("[INJECT-PROFILE] WS tunnel ESTABLISHED client=%s", client_ip)
session_ctx = {
"last_user_transcript": "",
"last_item_id": None,
"injected_profile_context": injected_profile_context,
"injected_reports_context": injected_reports_context,
"injected_summary_context": injected_summary_context,
"inject_phone": inject_phone,
"login_token": login_token,
"login_response": login_response_obj,
}
client_to_upstream = asyncio.create_task(
_client_to_upstream(ws, upstream, client_ip, session_ctx)
)
upstream_to_client = asyncio.create_task(
_upstream_to_client(ws, upstream, client_ip, session_ctx)
)
done, pending = await asyncio.wait(
{client_to_upstream, upstream_to_client},
return_when=asyncio.FIRST_COMPLETED,
)
for task in pending:
task.cancel()
await asyncio.gather(*pending, return_exceptions=True)
for task in done:
exc = task.exception()
if exc:
raise exc
except WebSocketDisconnect:
logger.debug("[EXT][S2S-PROXY] client=%s disconnected (WSDisconnect)", client_ip)
return
except Exception as exc:
logger.warning("Realtime proxy failed: %r", exc)
logger.debug("[EXT][S2S-PROXY] client=%s ERROR %s", client_ip, type(exc).__name__)
try:
await ws.close(code=1011, reason="Upstream realtime unavailable.")
except Exception:
pass
finally:
logger.debug("[EXT][S2S-PROXY] client=%s TUNNEL CLOSED", client_ip)
@app.get("/api/user/phone")
async def user_by_phone(phoneNumber: str):
"""HTTP passthrough: fetches caller identity / profile from the external
info endpoint set via VOCERO_USER_PHONE_AUTH_URL and forwards the raw
JSON to the browser.
The proxy NEVER inspects this payload nor injects it into the realtime
WebSocket session. It is only used by:
- the browser UI to render a caller name header (#ident-user-name)
- the /api/reasoning HTTP body user_context field (browser's choice).
When VOCERO_USER_PHONE_AUTH_URL is unset we return 503 so the UI can
silently skip identity loading (backward compatible with generic deploys).
When S2S_LOGIN_BY_PHONE is set, the proxy first performs a login-by-phone
POST with the caller phone and (if a bearer token is extracted) forwards
it as Authorization: Bearer on the subsequent profile GET.
"""
from urllib.parse import urlencode
if not VOCERO_USER_PHONE_AUTH_URL:
raise HTTPException(
status_code=503,
detail="VOCERO_USER_PHONE_AUTH_URL is not configured on this deploy.",
)
phone = (phoneNumber or "").strip()
if not phone:
raise HTTPException(status_code=400, detail="phoneNumber is required.")
# 0) Login-by-phone (optional — only if S2S_LOGIN_BY_PHONE env is set).
login_token = None
if S2S_LOGIN_BY_PHONE:
login_token, _login_obj = await _login_by_phone(phone)
target_url = f"{VOCERO_USER_PHONE_AUTH_URL}?{urlencode({'phoneNumber': phone})}"
logger.debug(
"[EXT][USER-BY-PHONE] GET upstream=%s phoneLen=%d bearer=%s",
VOCERO_USER_PHONE_AUTH_URL,
len(phone),
"yes" if login_token else "no",
)
try:
headers = {}
if login_token:
headers["Authorization"] = f"Bearer {login_token}"
async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as http:
resp = await http.get(target_url, headers=headers or None)
except httpx.RequestError as exc:
logger.warning("[EXT][USER-BY-PHONE] unreachable: %r", exc)
raise HTTPException(status_code=502, detail="Upstream user-info endpoint is unreachable.")
# Forward the upstream status + JSON verbatim; never rewrite the payload.
if resp.status_code >= 500:
logger.warning(
"[EXT][USER-BY-PHONE] upstream HTTP %s: %.200s",
resp.status_code, resp.text,
)
try:
payload = resp.json()
except Exception:
raise HTTPException(
status_code=502,
detail="Upstream user-info endpoint returned invalid JSON.",
)
return JSONResponse(status_code=resp.status_code if resp.status_code != 204 else 200, content=payload)
@app.get("/api/me")
async def me(request: Request):
"""Login state, tier, and remaining daily budget. Only meaningful in LB mode;
sets the anonymous tracking cookie when first seen."""
client_ip = request.client.host if request.client else "?"
logger.debug(
"[INTERNAL] /api/me client=%s LIMITER=%s",
client_ip, LIMITER_ENABLED,
)
if not LIMITER_ENABLED:
return {"enabled": False}
view = auth.user_view(request)
tier, keys, set_cookie = auth.resolve_identity(request)
unlimited = limiter.budget_for(tier) is None
rem = None if unlimited else await asyncio.to_thread(limiter.remaining, keys, tier)
logger.debug(
"[INTERNAL] /api/me client=%s tier=%s unlimited=%s remaining=%s setCookie=%s",
client_ip, tier, unlimited, rem, bool(set_cookie),
)
out = {
"enabled": True,
"auth": AUTH_ENABLED,
**view,
"remainingSec": rem,
"limitSec": limiter.budget_for(tier),
"loginUrl": auth.OAUTH_LOGIN_PATH if AUTH_ENABLED else None,
"logoutUrl": auth.OAUTH_LOGOUT_PATH if AUTH_ENABLED else None,
}
resp = JSONResponse(out)
if set_cookie:
auth.set_anon_cookie(resp, set_cookie)
return resp
@app.post("/api/search")
async def search(req: SearchRequest):
"""Proxy a Google search via Serper.dev. The key stays on the server unless
the user brought their own (then theirs is used for this request only)."""
query = (req.query or "").strip()
if not query:
raise HTTPException(status_code=400, detail="Empty query.")
key = (req.key or "").strip() or SERPER_KEY
has_user_key = bool((req.key or "").strip())
if not key:
# No server key and the user didn't supply one — search is unavailable.
raise HTTPException(status_code=503, detail="Search is not configured.")
headers = {"X-API-KEY": key, "Content-Type": "application/json"}
payload = {"q": query, "num": MAX_RESULTS}
logger.debug(
"[EXT][SERPER] search POST %s query=%s userKey=%s num=%d",
SERPER_URL, json.dumps(query, ensure_ascii=True)[:200], has_user_key, MAX_RESULTS,
)
try:
async with httpx.AsyncClient(timeout=12.0) as http:
resp = await http.post(SERPER_URL, headers=headers, json=payload)
except httpx.RequestError as exc:
logger.warning("Serper unreachable: %r", exc)
raise HTTPException(status_code=502, detail="Search provider unreachable.")
logger.debug(
"[EXT][SERPER] search RESP status=%s len=%dB",
resp.status_code, len(resp.content or b""),
)
if resp.status_code != 200:
# Serper's error body carries the real reason (e.g. "Not enough
# credits") and contains no key, so it's safe to log and relay.
body = resp.text[:300]
logger.warning("Serper error %s: %s", resp.status_code, body)
msg = None
try:
msg = resp.json().get("message")
except Exception:
pass
detail = f"Search provider error ({resp.status_code})"
if msg:
detail += f": {msg}"
raise HTTPException(status_code=502, detail=detail)
data = resp.json()
results = []
for item in (data.get("organic") or [])[:MAX_RESULTS]:
results.append(
{
"title": item.get("title", ""),
"snippet": item.get("snippet", ""),
"url": item.get("link", ""),
}
)
# A direct answer when Google has one — saves the model a hop.
box = data.get("answerBox") or {}
answer = box.get("answer") or box.get("snippet") or None
if not answer:
kg = data.get("knowledgeGraph") or {}
answer = kg.get("description") or None
logger.debug(
"[EXT][SERPER] search PARSED hasAnswer=%s resultCount=%d titles=%s",
bool(answer), len(results), [r.get("title", "")[:40] for r in results],
)
return JSONResponse({"query": query, "answer": answer, "results": results})
@app.post("/api/session")
async def session(request: Request):
"""Proxy the session handshake to the load balancer, keeping its URL secret,
and meter conversation time by tier.
The browser POSTs here (same-origin); we resolve the caller's tier, refuse if
today's budget is already spent (402), otherwise POST <LOAD_BALANCER_URL>/session
and relay the JSON back. The LB body carries a per-session `connect_url`
(compute host + short-lived token) the browser must dial directly — that one
URL is unavoidably exposed, but the stable load-balancer address is not. On a
successful grant we reserve the first time chunk against the day's budget."""
client_ip = request.client.host if request.client else "?"
if not LOAD_BALANCER_URL:
# No LB configured — this deploy is direct-mode only; the browser should
# never call this. 404 so it's indistinguishable from a missing route.
raise HTTPException(status_code=404, detail="Not found.")
tier, keys, set_cookie = auth.resolve_identity(request)
# Metering runs only on the deployed Space; off-Space the LB still proxies but
# nothing is tracked. Within metering, unlimited tiers (pro, org) aren't either.
tracked = LIMITER_ENABLED and limiter.budget_for(tier) is not None
logger.debug(
"[EXT][LB] session: client=%s tier=%s tracked=%s LIMITER=%s",
client_ip, tier, tracked, LIMITER_ENABLED,
)
# Refuse before troubling the LB if the day's budget is already gone. Done
# here (at enqueue) so we never put a user who can't talk into the queue.
if tracked:
rem = await asyncio.to_thread(limiter.remaining, keys, tier)
logger.debug("[EXT][LB] session: client=%s budget remaining=%s", client_ip, rem)
if rem is not None and rem <= 0:
resp = JSONResponse(
{"tier": tier, "reason": "limit", "remainingSec": 0}, status_code=402
)
if set_cookie:
auth.set_anon_cookie(resp, set_cookie)
return resp
url = f"{LOAD_BALANCER_URL.rstrip('/')}/session"
logger.debug("[EXT][LB] session POST %s (handshake)", url)
try:
async with httpx.AsyncClient(timeout=15.0) as http:
lb = await http.post(url, headers={"Content-Type": "application/json"}, content="{}")
except httpx.RequestError as exc:
logger.warning("Load balancer unreachable: %r", exc)
raise HTTPException(status_code=502, detail="Speech service unreachable.")
logger.debug(
"[EXT][LB] session RESP status=%s len=%dB",
lb.status_code, len(lb.content or b""),
)
# The queue is full: the LB replies 503 {state:"at_capacity"}. Relay it as-is
# so the client shows a soft "try again shortly", not a hard error.
if lb.status_code == 503:
body = _safe_json(lb)
if body.get("state") == "at_capacity":
logger.debug("[EXT][LB] session: at_capacity — queue full")
resp = JSONResponse({"state": "at_capacity"}, status_code=503)
if set_cookie:
auth.set_anon_cookie(resp, set_cookie)
return resp
if lb.status_code != 200:
# The LB's error body may name the reason (e.g. capacity); it carries no
# secret, so relay a trimmed copy.
logger.warning("Session handshake failed %s: %s", lb.status_code, lb.text[:300])
raise HTTPException(status_code=502, detail=f"Session handshake failed ({lb.status_code}).")
data = lb.json()
state = data.get("state", "?")
connect_url = (data.get("connect_url") or data.get("connectUrl") or "")[:120]
session_id = data.get("session_id") or data.get("sessionId")
logger.debug(
"[EXT][LB] session PARSED state=%s sessionId=%s hasConnectUrl=%s connectUrl=%s",
state, session_id, bool(connect_url), connect_url,
)
# Busy pool: the LB queued us. Relay the ticket untouched — crucially with NO
# reservation, so waiting in line never costs the day's budget.
if data.get("state") == "queued":
data["tier"] = tier
logger.debug("[EXT][LB] session: queued queueId=%s position=%s",
data.get("queue_id") or data.get("queueId"),
data.get("position"))
resp = JSONResponse(data)
if set_cookie:
auth.set_anon_cookie(resp, set_cookie)
return resp
# A slot was free: reserve the first chunk now and return the grant.
return await _finalize_grant(data, keys, tier, tracked, set_cookie)
@app.get("/api/queue/{queue_id}")
async def queue_status(queue_id: str, request: Request):
"""Poll a waiting ticket: relay the position, or — when the head of the line
claims a freed slot — reserve the budget now and return the grant. Re-checks the
daily budget at claim, since a multi-minute wait could have spent it elsewhere."""
client_ip = request.client.host if request.client else "?"
if not LOAD_BALANCER_URL:
raise HTTPException(status_code=404, detail="Not found.")
tier, keys, set_cookie = auth.resolve_identity(request)
tracked = LIMITER_ENABLED and limiter.budget_for(tier) is not None
logger.debug(
"[EXT][LB] queue poll: client=%s queueId=%s tier=%s",
client_ip, queue_id[:16], tier,
)
url = f"{LOAD_BALANCER_URL.rstrip('/')}/queue/{queue_id}"
logger.debug("[EXT][LB] queue GET %s", url)
try:
async with httpx.AsyncClient(timeout=15.0) as http:
lb = await http.get(url)
except httpx.RequestError as exc:
logger.warning("Load balancer unreachable: %r", exc)
raise HTTPException(status_code=502, detail="Speech service unreachable.")
logger.debug(
"[EXT][LB] queue RESP status=%s len=%dB",
lb.status_code, len(lb.content or b""),
)
if lb.status_code == 404:
# Ticket unknown/expired (reaped after we stopped polling). Tell the client
# to start over rather than spin.
logger.debug("[EXT][LB] queue: queueId=%s EXPIRED (404)", queue_id[:16])
resp = JSONResponse({"state": "expired"}, status_code=404)
if set_cookie:
auth.set_anon_cookie(resp, set_cookie)
return resp
if lb.status_code != 200:
logger.warning("Queue poll failed %s: %s", lb.status_code, lb.text[:300])
raise HTTPException(status_code=502, detail=f"Queue poll failed ({lb.status_code}).")
data = lb.json()
state = data.get("state", "?")
position = data.get("position")
logger.debug(
"[EXT][LB] queue PARSED queueId=%s state=%s position=%s sessionId=%s",
queue_id[:16], state, position,
data.get("session_id") or data.get("sessionId"),
)
if data.get("state") == "queued":
data["tier"] = tier
resp = JSONResponse(data)
if set_cookie:
auth.set_anon_cookie(resp, set_cookie)
return resp
# Claimed a slot. Re-check the budget: it may have been spent in another tab
# during the wait. If so, refuse — the just-claimed slot is now a pending
# session on the LB and its pending-timeout reaper reclaims it shortly.
if tracked:
rem = await asyncio.to_thread(limiter.remaining, keys, tier)
logger.debug(
"[EXT][LB] queue CLAIMED queueId=%s budget remaining=%s",
queue_id[:16], rem,
)
if rem is not None and rem <= 0:
resp = JSONResponse(
{"tier": tier, "reason": "limit", "remainingSec": 0}, status_code=402
)
if set_cookie:
auth.set_anon_cookie(resp, set_cookie)
return resp
return await _finalize_grant(data, keys, tier, tracked, set_cookie)
@app.delete("/api/queue/{queue_id}")
async def queue_leave(queue_id: str):
"""Leave the queue from the explicit 'Leave queue' button (a real fetch)."""
if not LOAD_BALANCER_URL:
raise HTTPException(status_code=404, detail="Not found.")
logger.debug("[EXT][LB] queue LEAVE (button) queueId=%s", queue_id[:16])
await _lb_leave(queue_id)
return {"ok": True}
@app.post("/api/queue/end")
async def queue_end(request: Request):
"""Leave the queue on teardown/tab-close (navigator.sendBeacon, which can only
POST). Body: { queueId }. Best-effort; the LB reaps the ticket on TTL anyway."""
if not LOAD_BALANCER_URL:
raise HTTPException(status_code=404, detail="Not found.")
qid = await _queue_id(request)
if qid:
logger.debug("[EXT][LB] queue LEAVE (teardown) queueId=%s", qid[:16])
await _lb_leave(qid)
return {"ok": True}
async def _finalize_grant(data, keys, tier, tracked, set_cookie):
"""Shared grant tail (fast path or queue claim): reserve the first chunk, attach
the metering fields the client needs, and set the anon cookie."""
remaining = None
session_id = data.get("session_id")
if tracked and session_id:
logger.debug(
"[EXT][LB] grant FINALIZE sessionId=%s tier=%s tracked=%s",
session_id, tier, tracked,
)
await asyncio.to_thread(limiter.begin, session_id, keys, tier)
remaining = await asyncio.to_thread(limiter.remaining, keys, tier)
logger.debug(
"[EXT][LB] grant RESERVED sessionId=%s remaining=%s",
session_id, remaining,
)
data.update({
"tier": tier,
"limited": tracked,
"remainingSec": remaining,
"heartbeatSec": limiter.HEARTBEAT_SEC,
})
resp = JSONResponse(data)
if set_cookie:
auth.set_anon_cookie(resp, set_cookie)
return resp
async def _lb_leave(queue_id: str) -> None:
"""Best-effort: tell the LB to drop a waiting ticket."""
url = f"{LOAD_BALANCER_URL.rstrip('/')}/queue/{queue_id}"
logger.debug("[EXT][LB] queue DELETE %s", url)
try:
async with httpx.AsyncClient(timeout=5.0) as http:
resp = await http.delete(url)
logger.debug(
"[EXT][LB] queue DELETE RESP queueId=%s status=%s",
queue_id[:16], resp.status_code,
)
except httpx.RequestError as exc:
logger.warning("Queue leave failed: %r", exc)
def _safe_json(response) -> dict:
try:
body = response.json()
except Exception:
return {}
return body if isinstance(body, dict) else {}
def _proxy_public_url(request: Request) -> str:
if not S2S_PROXY_TARGET:
return ""
forwarded_proto = (request.headers.get("x-forwarded-proto") or request.url.scheme).split(",")[0].strip()
forwarded_host = (request.headers.get("x-forwarded-host") or request.headers.get("host") or request.url.netloc).split(",")[0].strip()
ws_scheme = "wss" if forwarded_proto in {"https", "wss"} else "ws"
root_path = (request.scope.get("root_path") or "").rstrip("/")
return f"{ws_scheme}://{forwarded_host}{root_path}/v1/realtime"
def _proxy_upstream_url(ws: WebSocket) -> str:
target = urlsplit(S2S_PROXY_TARGET)
merged_query = dict(parse_qsl(target.query, keep_blank_values=True))
merged_query.update(parse_qsl(ws.url.query, keep_blank_values=True))
return urlunsplit(
(
target.scheme,
target.netloc,
target.path or "/v1/realtime",
urlencode(merged_query, doseq=True),
target.fragment,
)
)
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, client_ip: str, session_ctx: dict) -> None:
while True:
message = await ws.receive()
if message["type"] == "websocket.disconnect":
# #region debug-point B:client-disconnect
await _report_server_debug(
"B",
"server.py:_client_to_upstream",
"client websocket disconnected",
{"code": message.get("code")},
)
# #endregion
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_event_summary(message["text"])
if summary:
await _report_server_debug(
"B",
"server.py:_client_to_upstream",
"forwarding client event to upstream",
summary,
)
event_type = summary.get("type")
# --- INPUT UTENTE lato client (text typed / 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:
logger.info(
"[INPUT UTENTE] client=%s text=%s",
client_ip, user_text,
)
if (event_type not in _NOISY_CLIENT_EVENT_TYPES
and event_type not in _NOISY_REALTIME_EVENT_TYPES):
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 / latest caller summary,
# patch instructions to include them before forwarding upstream.
# Pure no-op when S2S_INJECT_PROFILE=false (original pure
# passthrough behaviour).
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 ""
forward_text = _patch_client_session_update_with_profile(
message["text"], ctx_profile_context or "", ctx_reports_context or "", ctx_summary_context or ""
)
await upstream.send(forward_text)
elif message.get("bytes") is not None:
# #region debug-point B:client-bytes-forward
await _report_server_debug(
"B",
"server.py:_client_to_upstream",
"forwarding client binary frame to upstream",
{"bytes": len(message["bytes"])},
)
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, client_ip: str, session_ctx: dict) -> None:
async for message in upstream:
if isinstance(message, bytes):
# #region debug-point B:upstream-bytes-forward
await _report_server_debug(
"B",
"server.py:_upstream_to_client",
"forwarding upstream binary frame to client",
{"bytes": len(message)},
)
logger.debug(
"[EXT][S2S-PROXY] client=%s UPSTREAM→ bytes=%dB",
client_ip, len(message),
)
# #endregion
await ws.send_bytes(message)
else:
# #region debug-point B:upstream-text-forward
summary = _realtime_event_summary(message)
if summary:
await _report_server_debug(
"B",
"server.py:_upstream_to_client",
"forwarding upstream event to client",
summary,
)
event_type = summary.get("type")
# --- INPUT UTENTE (server-side STT emesso dall'upstream) ---
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:
logger.info(
"[INPUT UTENTE] 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:
logger.info(
"[INPUT UTENTE] 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
# --- OUTPUT AGENTE: tutte le varianti ---
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"}:
agent_text = (summary.get("text") or "").strip()
if agent_text:
logger.info(
"[OUTPUT AGENTE] client=%s tts=%s",
client_ip, agent_text,
)
# ── Forward fire-and-forget a Node-RED (FASE 1 LOG-ONLY) ──
asyncio.create_task(_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:
# Ispezione: questo evento custom non emette testo nel campo atteso
raw_preview = summary.get("_raw_preview") or json.dumps(summary, ensure_ascii=True)
logger.debug(
"[OUTPUT AGENTE][ISPEZIONE] client=%s %s ha 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:
logger.info(
"[OUTPUT AGENTE] client=%s content_part=%s",
client_ip, agent_text,
)
asyncio.create_task(_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:
logger.info(
"[OUTPUT AGENTE] client=%s output_item=%s",
client_ip, agent_text,
)
asyncio.create_task(_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:
logger.info(
"[OUTPUT AGENTE] client=%s created=%s",
client_ip, agent_text,
)
asyncio.create_task(_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 "",
}))
# Log dettagliato evento solo se non è un tipo rumoroso
if event_type not in _NOISY_REALTIME_EVENT_TYPES:
logger.debug(
"[EXT][S2S-PROXY] client=%s UPSTREAM→ event=%s summary=%s",
client_ip, event_type, json.dumps(summary, ensure_ascii=True)[:300],
)
# ── Upstream error filter: non propagare al browser errori interni
# causati dal nostro inject di tools[] (sono noise per l'utente finale).
# (strategy B non funziona attualmente (ai-server non supporta tools[]).
#
# Formati errore upstream gestiti (entrambi con controllo su message e su type annidato):
# (A) Wrapper standard (OSSERVATO NEI LOG REALI):
# {"type":"error","error":{"message":"Unknown or invalid event: session.update","type":"unknown_or_invalid_event","code":null,...}}
# (B) Flat fallback (per altri upstream compatibili):
# {"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":
# ── Estrazione robusta message + error_type: wrapper annidato PRIORITARIO
_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
# Fallback flat se non abbiamo trovato nel wrapper
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
# ── Decisioni soppressione
_suppress = False
_suppress_reason = ""
if "Unknown or invalid event: session.update" in _err_msg:
_suppress = True
_suppress_reason = (
"rifiuto del nostro inject session.update con tools[] — ai-server attualmente non supporta "
"tools[]. Modifica richiesta in arrivo. Loggo solo server-side e non inoltro al browser."
)
elif "session.update" in _err_msg and ("tool" in _err_msg.lower() or "invalid" in _err_msg.lower()):
_suppress = True
_suppress_reason = (
f"rifiuto di session.update da upstream (message): {_err_msg[:160]}. Non inoltro al browser."
)
elif _err_type == "unknown_or_invalid_event" and "session.update" in _err_msg:
_suppress = True
_suppress_reason = (
f"rifiuto di session.update da upstream (type=unknown_or_invalid_event): {_err_msg[:160]}. Non inoltro al browser."
)
if _suppress:
logger.info(
"[WS SUPPRESS] suppressed upstream error (non mandato al browser): %s",
_suppress_reason,
)
logger.debug(
"[WS SUPPRESS] full upstream suppressed error payload=%s",
json.dumps(_evt, ensure_ascii=True)[:600],
)
return
# ── Log ispezione: se arriviamo qua l'errore NON è stato soppresso.
# Facciamo sempre vedere message+type per capire se aggiungere nuove regole.
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)
async def _queue_id(request: Request) -> str:
"""Pull `queueId` from a JSON body, tolerating sendBeacon's blob posts."""
try:
data = await request.json()
except Exception:
return ""
return (data or {}).get("queueId", "") if isinstance(data, dict) else ""
async def _session_id(request: Request) -> str:
"""Pull `sessionId` from a JSON body, tolerating sendBeacon's blob posts."""
try:
data = await request.json()
except Exception:
return ""
return (data or {}).get("sessionId", "") if isinstance(data, dict) else ""
@app.post("/api/session/heartbeat")
async def session_heartbeat(request: Request):
"""Extend the live reservation one chunk at a time. `expired` once the day's
budget is spent — the client then tears down."""
client_ip = request.client.host if request.client else "?"
if not LIMITER_ENABLED:
raise HTTPException(status_code=404, detail="Not found.")
sid = await _session_id(request)
alive = bool(sid) and await asyncio.to_thread(limiter.heartbeat, sid)
logger.debug(
"[INTERNAL] /api/session/heartbeat client=%s sessionId=%s alive=%s",
client_ip, (sid or "")[:16], alive,
)
return {"expired": not alive}
@app.post("/api/session/end")
async def session_end(request: Request):
"""Clean teardown: reconcile to real elapsed time and refund the unused
chunk. Sent via navigator.sendBeacon, so it must succeed without a response."""
client_ip = request.client.host if request.client else "?"
if not LIMITER_ENABLED:
raise HTTPException(status_code=404, detail="Not found.")
sid = await _session_id(request)
logger.debug(
"[INTERNAL] /api/session/end client=%s sessionId=%s",
client_ip, (sid or "")[:16],
)
if sid:
await asyncio.to_thread(limiter.end, sid)
return {"ok": True}
@app.middleware("http")
async def no_cache_static_assets(request: Request, call_next):
response = await call_next(request)
path = request.url.path or ""
nocache_ext = (".html", ".js", ".mjs", ".css", ".wasm", ".worklet.js")
if any(path.endswith(ext) for ext in nocache_ext) or path == "/" or path.endswith("/"):
response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate, max-age=0"
response.headers["Pragma"] = "no-cache"
response.headers["Expires"] = "0"
return response
# Static front-end. Registered last so the /api routes win. `html=True` serves
# index.html at "/". The repo is public anyway, so serving the dir is fine.
app.mount("/", StaticFiles(directory=HERE, html=True), name="static")
Reference in New Issue View Git Blame Copy Permalink