alvaro
4b0108ecf0
Opening a report or podcast link now establishes an authenticated starting reference for the conversation. Users can select other accessible content or compare reports while retaining the original link, and TALKFIRST greetings can name the verified report or podcast. Release notes: - Added authenticated report and episode link resolution with bounded lookups and unavailable-content handling - Added explicit conversation selection without restricting searches or changing the reference during comparisons - Added contextual TALKFIRST greetings through both voice-proxy and HTTP reasoning paths - Preserved backend permission checks, conversation isolation, cancellation, and the existing speech protocol Reviewed-on: AquaPortal/s2s#2
128 lines
4.7 KiB
Python
128 lines
4.7 KiB
Python
"""Bounded, process-local conversation state for the single-worker Vocero API."""
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import secrets
|
|
from dataclasses import dataclass, field
|
|
from time import monotonic
|
|
from typing import Any
|
|
from urllib.parse import urlsplit
|
|
|
|
from starlette.requests import HTTPConnection
|
|
|
|
from vocero_api.core import config as settings
|
|
from vocero_api.models.content import ConversationContent
|
|
from vocero_api.models.mcp import BackendLogin, CallerIdentity, IntegrationError, McpToolDefinition
|
|
|
|
COOKIE_NAME = "vocero_browser"
|
|
CONVERSATION_HEADER = "x-vocero-conversation"
|
|
|
|
|
|
@dataclass
|
|
class Conversation:
|
|
id: str
|
|
owner: str = field(repr=False)
|
|
backend_token: str = field(repr=False)
|
|
profile: CallerIdentity
|
|
expires_at: float
|
|
linked_content: ConversationContent = field(default_factory=ConversationContent)
|
|
current_content: ConversationContent = field(default_factory=ConversationContent)
|
|
greeting_pending: bool = False
|
|
catalog: list[McpToolDefinition] = field(default_factory=list)
|
|
history: list[list[dict[str, str]]] = field(default_factory=list, repr=False)
|
|
generation: int = 0
|
|
client_turn: int = 0
|
|
pending: asyncio.Task[Any] | None = field(default=None, repr=False)
|
|
socket_attached: bool = False
|
|
closed: asyncio.Event = field(default_factory=asyncio.Event, repr=False)
|
|
|
|
def require_active(self) -> None:
|
|
if self.closed.is_set() or monotonic() >= self.expires_at:
|
|
raise IntegrationError(401, "La conversación expiró. Inicia una nueva sesión.")
|
|
|
|
def interrupt(self) -> int:
|
|
self.generation += 1
|
|
if self.pending is not None and not self.pending.done() and self.pending is not asyncio.current_task():
|
|
self.pending.cancel()
|
|
self.pending = None
|
|
return self.generation
|
|
|
|
def close(self) -> None:
|
|
self.interrupt()
|
|
self.closed.set()
|
|
self.backend_token = ""
|
|
self.history.clear()
|
|
self.catalog.clear()
|
|
self.linked_content = ConversationContent()
|
|
self.current_content = ConversationContent()
|
|
self.greeting_pending = False
|
|
|
|
|
|
class ConversationStore:
|
|
def __init__(self) -> None:
|
|
self._items: dict[str, Conversation] = {}
|
|
|
|
def prune(self) -> None:
|
|
for item in list(self._items.values()):
|
|
if item.closed.is_set() or monotonic() >= item.expires_at:
|
|
self.remove(item.id)
|
|
|
|
def create(
|
|
self, browser: str, login: BackendLogin, linked_content: ConversationContent | None = None
|
|
) -> Conversation:
|
|
self.prune()
|
|
if len(self._items) >= settings.S2S_MAX_CONVERSATIONS:
|
|
raise IntegrationError(503, "No hay capacidad para otra conversación. Intenta más tarde.")
|
|
item = Conversation(
|
|
id=secrets.token_urlsafe(24),
|
|
owner=self._owner(browser),
|
|
backend_token=login.sessionToken.get_secret_value(),
|
|
profile=login.user,
|
|
expires_at=monotonic() + settings.S2S_CONVERSATION_TTL_SEC,
|
|
linked_content=linked_content or ConversationContent(),
|
|
current_content=linked_content or ConversationContent(),
|
|
greeting_pending=settings.S2S_START_SESSION_MODE == "TALKFIRST",
|
|
)
|
|
self._items[item.id] = item
|
|
return item
|
|
|
|
@staticmethod
|
|
def _owner(browser: str) -> str:
|
|
return hashlib.sha256(browser.encode()).hexdigest()
|
|
|
|
def get(self, browser: str, conversation_id: str) -> Conversation:
|
|
self.prune()
|
|
item = self._items.get(conversation_id)
|
|
if not browser or item is None or not secrets.compare_digest(item.owner, self._owner(browser)):
|
|
raise IntegrationError(401, "La conversación no es válida. Inicia una nueva sesión.")
|
|
item.require_active()
|
|
return item
|
|
|
|
def remove(self, conversation_id: str) -> None:
|
|
item = self._items.pop(conversation_id, None)
|
|
if item is not None:
|
|
item.close()
|
|
|
|
def clear(self) -> None:
|
|
for conversation_id in list(self._items):
|
|
self.remove(conversation_id)
|
|
|
|
|
|
store = ConversationStore()
|
|
|
|
|
|
def check_origin(connection: HTTPConnection) -> None:
|
|
origin = connection.headers.get("origin")
|
|
if origin:
|
|
parsed = urlsplit(origin)
|
|
if parsed.scheme not in {"http", "https"} or parsed.netloc != connection.headers.get("host"):
|
|
raise IntegrationError(403, "El origen de la solicitud no está permitido.")
|
|
|
|
|
|
def resolve(connection: HTTPConnection) -> Conversation:
|
|
check_origin(connection)
|
|
conversation_id = connection.headers.get(CONVERSATION_HEADER, "")
|
|
if connection.scope["type"] == "websocket":
|
|
conversation_id = connection.query_params.get("conversation_id", "")
|
|
return store.get(connection.cookies.get(COOKIE_NAME, ""), conversation_id)
|