from __future__ import annotations import asyncio import base64 import hashlib import json import re import threading import time import urllib.parse import uuid from contextlib import closing, suppress from typing import Any import websockets from flask import current_app, session from flask_sock import Sock from .audio_protocol import AudioProtocolError, decode_audio_frame, normalise_vad from .db import connect, json_text, transaction from .models import build_profiles, normalise_preferences from .provider import ProviderError, build_payload, fake_stream, stream_together def uid(prefix: str) -> str: return f"{prefix}_{uuid.uuid4().hex}" def _speech_text(text: str, model_key: str) -> str: value = str(text or "") if model_key == "k2.7-code": value = re.sub(r"```[\s\S]*?```", " The code is available on screen. ", value) value = re.sub(r"(?m)^[-+]{3}\s.*$", "", value) value = re.sub(r"(?m)^@@.*$", "", value) value = re.sub(r"https?://\S+", "a link shown on screen", value) value = re.sub(r"(?m)^Traceback \(most recent call last\):[\s\S]*?(?=\n\n|\Z)", "A stack trace is shown on screen.", value) return re.sub(r"\s+", " ", value).strip() def _ensure_conversation(app, user_id: str, conversation_id: str | None, model_key: str) -> str: with closing(connect(app.config["DATABASE"])) as db: if conversation_id: row = db.execute("SELECT id FROM conversations WHERE id=? AND user_id=? AND deleted_at IS NULL", (conversation_id, user_id)).fetchone() if row: return str(row["id"]) conversation_id = uid("cnv") db.execute("INSERT INTO conversations(id,user_id,title,model_key,version) VALUES(?,?,?, ?,1)", (conversation_id, user_id, "Voice conversation", model_key)) db.execute("INSERT INTO audit_log(user_id,action,target_type,target_id,detail_json) VALUES(?,?,?,?,?)", (user_id, "audio.conversation_created", "conversation", conversation_id, "{}")) return conversation_id def _context_messages(app, conversation_id: str, model_key: str) -> list[dict[str, Any]]: with closing(connect(app.config["DATABASE"])) as db: rows = db.execute( """ SELECT role,content,reasoning FROM ( SELECT role,content,reasoning,created_at,rowid AS message_order FROM messages WHERE conversation_id=? AND status IN ('complete','streaming') ORDER BY created_at DESC,rowid DESC LIMIT ? ) ORDER BY created_at,message_order """, (conversation_id, int(app.config["MAX_CONTEXT_MESSAGES"])), ).fetchall() messages = [] for row in rows: item = {"role": row["role"], "content": row["content"]} if model_key == "k2.7-code" and row["role"] == "assistant" and row["reasoning"]: item["reasoning_content"] = row["reasoning"] messages.append(item) return messages def _begin_turn(app, user_id: str, conversation_id: str, transcript: str, model_key: str, turn_id: str) -> tuple[str, str]: user_message_id, assistant_message_id = uid("msg"), uid("msg") with closing(connect(app.config["DATABASE"])) as db, transaction(db): conversation = db.execute("SELECT * FROM conversations WHERE id=? AND user_id=?", (conversation_id, user_id)).fetchone() title = conversation["title"] if title == "Voice conversation": title = transcript[:72] + ("…" if len(transcript) > 72 else "") db.execute("INSERT INTO messages(id,conversation_id,role,content,status) VALUES(?,?, 'user',?,'complete')", (user_message_id, conversation_id, transcript)) db.execute("INSERT INTO messages(id,conversation_id,role,content,status) VALUES(?,?, 'assistant','','streaming')", (assistant_message_id, conversation_id)) db.execute("UPDATE conversations SET title=?,model_key=?,version=version+1,updated_at=CURRENT_TIMESTAMP WHERE id=?", (title, model_key, conversation_id)) db.execute("INSERT INTO audit_log(user_id,action,target_type,target_id,detail_json) VALUES(?,?,?,?,?)", (user_id, "audio.turn_started", "conversation", conversation_id, json_text({"turn_id": turn_id, "model_key": model_key}))) return user_message_id, assistant_message_id def _finish_turn(app, user_id: str, assistant_message_id: str, content: str, reasoning: str, status: str, turn_id: str) -> None: with closing(connect(app.config["DATABASE"])) as db: db.execute("UPDATE messages SET content=?,reasoning=?,status=? WHERE id=?", (content.strip(), reasoning.strip(), status, assistant_message_id)) db.execute("INSERT INTO audit_log(user_id,action,target_type,target_id,detail_json) VALUES(?,?,?,?,?)", (user_id, f"audio.turn_{status}", "message", assistant_message_id, json_text({"turn_id": turn_id}))) class ConversationSession: def __init__(self, app, ws, user_id: str): self.app = app self.ws = ws self.user_id = user_id self.loop: asyncio.AbstractEventLoop | None = None self.send_lock = asyncio.Lock() self.closed = asyncio.Event() self.mode = "conversation" self.model_key = "k2.6" self.voice = str(app.config["TTS_DEFAULT_VOICE"]) self.conversation_id: str | None = None self.audio_session_id = uid("aud") self.turn_id: str | None = None self.model_cancel: threading.Event | None = None self.model_task: asyncio.Task | None = None self.model_responses: dict[str, Any] = {} self.stt_ws = None self.tts_ws = None self.state = "idle" self.session_started = False self.vad = normalise_vad(None, app.config["AUDIO_VAD_DEFAULTS"]) self.speed = 1.0 self.seen_transcripts: dict[str, float] = {} self.stats: dict[str, int] = {"input_audio_bytes": 0, "output_audio_bytes": 0, "transcripts": 0, "turns": 0, "barge_ins": 0, "provider_errors": 0} def bump(self, key: str, amount: int = 1) -> None: self.stats[key] = int(self.stats.get(key, 0)) + int(amount) def persist_stats(self) -> None: with closing(connect(self.app.config["DATABASE"])) as db: db.execute("UPDATE audio_sessions SET stats_json=?,updated_at=CURRENT_TIMESTAMP WHERE id=?", (json_text(self.stats), self.audio_session_id)) async def update_session(self, data: dict[str, Any]) -> None: if self.turn_id and any(key in data for key in {"model_key", "mode"}): await self.send("error", code="KIMU_AUDIO_TURN_ACTIVE", message="Model or mode cannot change during an active turn.") return mode = str(data.get("mode") or self.mode) model_key = str(data.get("model_key") or self.model_key) if mode not in {"conversation", "dictation"} or model_key not in {"k2.6", "k2.7-code"}: await self.send("error", code="KIMU_AUDIO_CONFIGURATION", message="Unsupported audio mode or model profile.") return if model_key == "k2.7-code" and not build_profiles(self.app.config)[model_key].configured: await self.send("error", code="KIMU_MODEL_NOT_CONFIGURED", message="Kimi K2.7 Coder is not configured.") return voice = str(data.get("voice") or self.voice)[:100] if voice not in {str(value) for value in self.app.config["TTS_VOICES"]}: await self.send("error", code="KIMU_AUDIO_VOICE", message="The selected Sonic 3 voice is not allowed.") return try: speed = min(2.0, max(0.5, float(data.get("speed") if data.get("speed") is not None else self.speed))) except (TypeError, ValueError): speed = self.speed supplied_vad = data.get("vad") if isinstance(supplied_vad, dict): self.vad = normalise_vad(supplied_vad, self.vad) if self.stt_ws: await self.stt_ws.send(json.dumps({"type": "transcription_session.updated", "session": {"turn_detection": self.vad}})) if self.tts_ws and (voice != self.voice or speed != self.speed): await self.tts_ws.send(json.dumps({"type": "tts_session.updated", "session": {"voice": voice, "speed": speed}})) self.mode, self.model_key, self.voice, self.speed = mode, model_key, voice, speed with closing(connect(self.app.config["DATABASE"])) as db: db.execute("UPDATE audio_sessions SET model_key=?,mode=?,voice=?,stats_json=?,updated_at=CURRENT_TIMESTAMP WHERE id=?", (self.model_key, self.mode, self.voice, json_text(self.stats), self.audio_session_id)) await self.send("session.updated", mode=self.mode, model_key=self.model_key, voice=self.voice, speed=self.speed, vad=self.vad) async def send(self, event_type: str, **payload: Any) -> None: if self.closed.is_set(): return message = json.dumps({"type": event_type, "audio_session_id": self.audio_session_id, **payload}, separators=(",", ":"), ensure_ascii=False) async with self.send_lock: try: await asyncio.to_thread(self.ws.send, message) except Exception: self.closed.set() async def set_state(self, state: str, turn_id: str | None = None) -> None: self.state = state with closing(connect(self.app.config["DATABASE"])) as db: db.execute("UPDATE audio_sessions SET state=?,current_turn_id=?,updated_at=CURRENT_TIMESTAMP WHERE id=?", (state, turn_id, self.audio_session_id)) await self.send("state", state=state, turn_id=turn_id) async def connect_upstreams(self) -> None: if self.app.config["FAKE_PROVIDER"]: await self.send("provider.ready", fake=True, stt_model=self.app.config["WHISPER_MODEL_ID"], tts_model=self.app.config["SONIC_MODEL_ID"]) return api_key = self.app.config["TOGETHER_API_KEY"] if not api_key: raise RuntimeError("Together API key is not configured.") headers = {"Authorization": f"Bearer {api_key}", "OpenAI-Beta": "realtime=v1"} turn_detection = self.vad if isinstance(self.vad, dict) else {} stt_params = { "model": self.app.config["WHISPER_MODEL_ID"], "input_audio_format": "pcm_s16le_16000", "turn_detection": "server_vad", "threshold": turn_detection.get("threshold", 0.3), "min_silence_duration_ms": turn_detection.get("min_silence_duration_ms", 550), "min_speech_duration_ms": turn_detection.get("min_speech_duration_ms", 200), "max_speech_duration_s": turn_detection.get("max_speech_duration_s", 45), "speech_pad_ms": turn_detection.get("speech_pad_ms", 250), } stt_url = self.app.config["TOGETHER_STT_WS_URL"] + "?" + urllib.parse.urlencode(stt_params) tts_url = self.app.config["TOGETHER_TTS_WS_URL"] + "?" + urllib.parse.urlencode({"model": self.app.config["SONIC_MODEL_ID"], "voice": self.voice, "response_format": "pcm", "sample_rate": int(self.app.config["TTS_SAMPLE_RATE"]), "speed": self.speed, "language": "en-gb", "alignment": "word", "segment": "sentence"}) self.stt_ws = await websockets.connect(stt_url, additional_headers=headers, max_size=8 * 1024 * 1024, ping_interval=20, ping_timeout=20) self.tts_ws = await websockets.connect(tts_url, additional_headers={"Authorization": f"Bearer {api_key}"}, max_size=8 * 1024 * 1024, ping_interval=20, ping_timeout=20) await self.send("provider.ready", fake=False, stt_model=self.app.config["WHISPER_MODEL_ID"], tts_model=self.app.config["SONIC_MODEL_ID"]) async def cancel_turn(self, source: str = "user") -> None: previous_turn = self.turn_id if self.model_cancel: self.model_cancel.set() if self.model_task and not self.model_task.done(): self.model_task.cancel() response = self.model_responses.pop(previous_turn, None) if previous_turn else None if response is not None: with suppress(Exception): await asyncio.to_thread(response.close) if self.tts_ws and previous_turn: with suppress(Exception): await self.tts_ws.send(json.dumps({"type": "context.cancel", "context_id": previous_turn})) await self.tts_ws.send(json.dumps({"type": "input_text_buffer.clear", "context_id": previous_turn})) if source == "barge_in": self.bump("barge_ins") await self.send("audio.clear", turn_id=previous_turn, source=source) await self.send("turn.cancelled", turn_id=previous_turn, source=source) self.turn_id = None self.model_cancel = None await self.set_state("listening") async def handle_transcript(self, transcript: str, source_id: str | None = None) -> None: transcript = str(transcript or "").strip() if not transcript: return now = time.monotonic() signature = source_id or hashlib.sha256(transcript.encode("utf-8")).hexdigest() self.seen_transcripts = {key: seen for key, seen in self.seen_transcripts.items() if now - seen < 120} if signature in self.seen_transcripts or any(key.startswith("text:") and key == f"text:{transcript}" and now - seen < 2 for key, seen in self.seen_transcripts.items()): await self.send("transcript.duplicate_ignored", transcript=transcript, source_id=source_id) return self.seen_transcripts[signature] = now self.seen_transcripts[f"text:{transcript}"] = now self.bump("transcripts") await self.send("transcript.final", transcript=transcript, mode=self.mode) if self.mode == "dictation": await self.set_state("listening") return if self.turn_id: await self.cancel_turn("new_turn") self.bump("turns") self.turn_id = uid("turn") current_turn = self.turn_id self.conversation_id = _ensure_conversation(self.app, self.user_id, self.conversation_id, self.model_key) _, assistant_message_id = _begin_turn(self.app, self.user_id, self.conversation_id, transcript, self.model_key, current_turn) await self.send("turn.started", turn_id=current_turn, conversation_id=self.conversation_id, transcript=transcript, model_key=self.model_key) await self.set_state("thinking", current_turn) cancel = threading.Event() self.model_cancel = cancel self.model_task = asyncio.create_task(self.generate_assistant(current_turn, assistant_message_id, cancel)) async def generate_assistant(self, turn_id: str, assistant_message_id: str, cancel: threading.Event) -> None: content, reasoning = "", "" profiles = build_profiles(self.app.config) profile = profiles[self.model_key] preferences = normalise_preferences(profile, None) messages = _context_messages(self.app, self.conversation_id, self.model_key) payload = build_payload(profile.model_id, messages, preferences, self.model_key) queue: asyncio.Queue[dict[str, str] | None] = asyncio.Queue() loop = asyncio.get_running_loop() def provider_response(response) -> None: if response is None: self.model_responses.pop(turn_id, None) else: self.model_responses[turn_id] = response def producer() -> None: try: iterator = fake_stream(self.model_key, messages[-1]["content"] if messages else "") if self.app.config["FAKE_PROVIDER"] else stream_together( self.app.config["TOGETHER_BASE_URL"], self.app.config["TOGETHER_API_KEY"], payload, cancel_event=cancel, response_callback=provider_response, ) for item in iterator: if cancel.is_set(): break asyncio.run_coroutine_threadsafe(queue.put(item), loop).result() except Exception as exc: asyncio.run_coroutine_threadsafe(queue.put({"type": "error", "text": str(exc), "code": getattr(exc, "code", "KIMU_AUDIO_MODEL_ERROR")}), loop).result() finally: asyncio.run_coroutine_threadsafe(queue.put(None), loop).result() producer_task = asyncio.create_task(asyncio.to_thread(producer)) try: while True: item = await queue.get() if item is None: break if cancel.is_set() or turn_id != self.turn_id: break if item["type"] == "reasoning": reasoning += item["text"] await self.send("assistant.reasoning.delta", turn_id=turn_id, delta=item["text"]) elif item["type"] == "content": content += item["text"] await self.send("assistant.text.delta", turn_id=turn_id, delta=item["text"]) speech = _speech_text(item["text"], self.model_key) if speech and self.tts_ws: await self.tts_ws.send(json.dumps({"type": "input_text_buffer.append", "text": speech, "context_id": turn_id})) elif speech and self.app.config["FAKE_PROVIDER"]: await self.send("tts.text", turn_id=turn_id, text=speech) elif item["type"] == "error": raise ProviderError(item.get("code", "KIMU_AUDIO_MODEL_ERROR"), item["text"]) if cancel.is_set() or turn_id != self.turn_id: _finish_turn(self.app, self.user_id, assistant_message_id, content, reasoning, "cancelled", turn_id) return if self.tts_ws: await self.tts_ws.send(json.dumps({"type": "input_text_buffer.commit", "context_id": turn_id})) await self.set_state("speaking", turn_id) _finish_turn(self.app, self.user_id, assistant_message_id, content, reasoning, "complete", turn_id) await self.send("assistant.done", turn_id=turn_id, content=content.strip(), conversation_id=self.conversation_id) if self.app.config["FAKE_PROVIDER"]: await asyncio.sleep(0.05) await self.send("audio.done", turn_id=turn_id) self.turn_id = None await self.set_state("listening") except asyncio.CancelledError: cancel.set() _finish_turn(self.app, self.user_id, assistant_message_id, content, reasoning, "cancelled", turn_id) except Exception as exc: _finish_turn(self.app, self.user_id, assistant_message_id, content, reasoning, "failed", turn_id) await self.send("error", code=getattr(exc, "code", "KIMU_AUDIO_MODEL_ERROR"), message=str(exc), turn_id=turn_id) self.turn_id = None await self.set_state("listening") finally: with suppress(Exception): await producer_task async def client_reader(self) -> None: while not self.closed.is_set(): message = await asyncio.to_thread(self.ws.receive) if message is None: self.closed.set() break if isinstance(message, bytes): if not self.session_started: await self.send("error", code="KIMU_AUDIO_SESSION_REQUIRED", message="Start the audio session before sending microphone data.") continue maximum = int(self.app.config["MAX_AUDIO_FRAME_BYTES"]) if len(message) > maximum: await self.send("error", code="KIMU_AUDIO_FRAME_TOO_LARGE", message="The audio frame exceeds the configured limit.") continue if len(message) % 2: await self.send("error", code="KIMU_AUDIO_FRAME_ALIGNMENT", message="PCM16 audio frames must contain an even number of bytes.") continue self.bump("input_audio_bytes", len(message)) if self.stt_ws: await self.stt_ws.send(json.dumps({"type": "input_audio_buffer.append", "audio": base64.b64encode(message).decode("ascii")})) continue try: data = json.loads(message) except json.JSONDecodeError: await self.send("error", code="KIMU_AUDIO_PROTOCOL", message="Invalid WebSocket JSON message.") continue event_type = data.get("type") if not self.session_started and event_type not in {"session.start", "ping", "session.stop"}: await self.send("error", code="KIMU_AUDIO_SESSION_REQUIRED", message="Start the audio session before sending this event.") continue if event_type == "session.start": if self.session_started: await self.send("error", code="KIMU_AUDIO_SESSION_STARTED", message="The audio session is already active.") continue supplied_csrf = str(data.get("csrf_token") or "") expected_csrf = str(session.get("csrf_token") or "") if not expected_csrf or supplied_csrf != expected_csrf: await self.send("error", code="KIMU_CSRF_FAILED", message="The Conversation Mode security token is invalid.") self.closed.set() break mode = str(data.get("mode") or "conversation") model_key = str(data.get("model_key") or "k2.6") if mode not in {"conversation", "dictation"} or model_key not in {"k2.6", "k2.7-code"}: await self.send("error", code="KIMU_AUDIO_CONFIGURATION", message="Unsupported audio mode or model profile.") continue if model_key == "k2.7-code" and not build_profiles(self.app.config)[model_key].configured: await self.send("error", code="KIMU_MODEL_NOT_CONFIGURED", message="Kimi K2.7 Coder is not configured.") continue self.mode, self.model_key = mode, model_key requested_voice = str(data.get("voice") or self.voice)[:100] allowed_voices = {str(value) for value in self.app.config["TTS_VOICES"]} if requested_voice not in allowed_voices: await self.send("error", code="KIMU_AUDIO_VOICE", message="The selected Sonic 3 voice is not allowed.") continue self.voice = requested_voice try: self.speed = min(2.0, max(0.5, float(data.get("speed") or 1.0))) except (TypeError, ValueError): self.speed = 1.0 self.conversation_id = str(data.get("conversation_id") or "") or None supplied_vad = data.get("vad") if isinstance(supplied_vad, dict): self.vad = normalise_vad(supplied_vad, self.vad) with closing(connect(self.app.config["DATABASE"])) as db: db.execute("INSERT INTO audio_sessions(id,user_id,conversation_id,model_key,mode,state,voice,stats_json) VALUES(?,?,?,?,?,'connecting',?,'{}')", (self.audio_session_id, self.user_id, self.conversation_id, self.model_key, self.mode, self.voice)) try: await self.connect_upstreams() except Exception as exc: await self.send("error", code=getattr(exc, "code", "KIMU_AUDIO_PROVIDER_UNAVAILABLE"), message=str(exc)) self.closed.set() break self.session_started = True await self.set_state("listening") await self.send("session.ready", mode=self.mode, model_key=self.model_key, voice=self.voice, sample_rate=16000, tts_sample_rate=int(self.app.config["TTS_SAMPLE_RATE"]), conversation_id=self.conversation_id) elif event_type == "session.update": await self.update_session(data) elif event_type == "audio.append": audio = data.get("audio") if isinstance(audio, str): try: decoded = decode_audio_frame(audio, int(self.app.config["MAX_AUDIO_FRAME_BYTES"])) except AudioProtocolError as exc: await self.send("error", code=exc.code, message=str(exc)) continue self.bump("input_audio_bytes", len(decoded)) if self.stt_ws: await self.stt_ws.send(json.dumps({"type": "input_audio_buffer.append", "audio": audio})) elif event_type == "audio.commit": if self.stt_ws: await self.stt_ws.send(json.dumps({"type": "input_audio_buffer.commit"})) elif event_type == "barge_in": await self.cancel_turn("barge_in") elif event_type == "transcript.commit": await self.handle_transcript(str(data.get("transcript") or ""), str(data.get("operation_id") or uid("manual"))) elif event_type == "tts.preview": if self.turn_id: await self.send("error", code="KIMU_AUDIO_TURN_ACTIVE", message="Voice preview is unavailable while a response is active.") continue preview = _speech_text(str(data.get("text") or "This is the selected KIMU voice."), self.model_key)[:500] preview_id = uid("preview") if self.tts_ws and preview: await self.tts_ws.send(json.dumps({"type": "input_text_buffer.append", "text": preview, "context_id": preview_id})) await self.tts_ws.send(json.dumps({"type": "input_text_buffer.commit", "context_id": preview_id})) elif self.app.config["FAKE_PROVIDER"]: await self.send("tts.text", turn_id=preview_id, text=preview, preview=True) await self.send("audio.done", turn_id=preview_id, preview=True) elif event_type == "session.stop": self.closed.set() break elif event_type == "ping": await self.send("pong", at=time.time()) async def stt_reader(self) -> None: while not self.closed.is_set() and not self.stt_ws: await asyncio.sleep(0.05) if self.closed.is_set() or not self.stt_ws: return async for raw in self.stt_ws: if self.closed.is_set(): break data = json.loads(raw) event_type = data.get("type") if event_type == "conversation.item.input_audio_transcription.delta": await self.send("transcript.delta", delta=data.get("delta", "")) elif event_type == "conversation.item.input_audio_transcription.completed": source_id = str(data.get("item_id") or data.get("event_id") or "") or None await self.handle_transcript(data.get("transcript", ""), source_id) elif event_type == "conversation.item.input_audio_transcription.failed": error = data.get("error") or {} await self.send("error", code=error.get("code", "KIMU_STT_FAILED"), message=error.get("message", "Whisper transcription failed.")) elif event_type in {"session.created", "transcription_session.updated"}: await self.send("stt.status", provider_event=event_type) async def tts_reader(self) -> None: while not self.closed.is_set() and not self.tts_ws: await asyncio.sleep(0.05) if self.closed.is_set() or not self.tts_ws: return async for raw in self.tts_ws: if self.closed.is_set(): break data = json.loads(raw) event_type = data.get("type") context_id = data.get("context_id") or self.turn_id if context_id and self.turn_id and context_id != self.turn_id: continue if event_type == "conversation.item.audio_output.delta": delta = data.get("delta", "") with suppress(Exception): self.bump("output_audio_bytes", len(base64.b64decode(delta))) await self.send("audio.delta", turn_id=context_id, audio=delta, sample_rate=int(self.app.config["TTS_SAMPLE_RATE"]), format="pcm_s16le") elif event_type == "conversation.item.audio_output.done": await self.send("audio.done", turn_id=context_id) if context_id == self.turn_id: self.turn_id = None await self.set_state("listening") elif event_type == "context.cancelled": await self.send("tts.cancelled", turn_id=context_id) elif event_type in {"conversation.item.tts.failed", "error"}: error = data.get("error") or {} await self.send("error", code=error.get("code", "KIMU_TTS_FAILED"), message=error.get("message", data.get("message", "Sonic speech generation failed.")), turn_id=context_id) async def run(self) -> None: self.loop = asyncio.get_running_loop() tasks = [asyncio.create_task(self.client_reader()), asyncio.create_task(self.stt_reader()), asyncio.create_task(self.tts_reader())] try: await self.closed.wait() finally: if self.model_cancel: self.model_cancel.set() for task in tasks: task.cancel() if self.model_task: self.model_task.cancel() for task in tasks: with suppress(asyncio.CancelledError, Exception): await task if self.stt_ws: with suppress(Exception): await self.stt_ws.close() if self.tts_ws: with suppress(Exception): await self.tts_ws.close() for response in list(self.model_responses.values()): with suppress(Exception): await asyncio.to_thread(response.close) self.model_responses.clear() with closing(connect(self.app.config["DATABASE"])) as db: db.execute("UPDATE audio_sessions SET state='closed',stats_json=?,updated_at=CURRENT_TIMESTAMP WHERE id=?", (json_text(self.stats), self.audio_session_id)) def register_conversation_gateway(app) -> None: sock = Sock(app) @sock.route("/api/v1/audio/conversation/ws") def conversation_socket(ws): user_id = session.get("user_id") if not user_id: ws.send(json.dumps({"type": "error", "code": "KIMU_AUTH_REQUIRED", "message": "Sign in is required."})) ws.close() return with closing(connect(app.config["DATABASE"])) as db: user = db.execute("SELECT id FROM users WHERE id=?", (user_id,)).fetchone() if not user: ws.close() return asyncio.run(ConversationSession(app, ws, str(user_id)).run())