from collections import defaultdict from typing import Dict, List from app.config import config # In-memory store: { uid: { "stt": [...], "text": [...] } } _store: Dict[str, Dict[str, List[dict]]] = defaultdict( lambda: {"stt": [], "text": []} ) def get_history(uid: str, channel: str) -> List[dict]: """Return conversation history for a user on a given channel (stt | text).""" return _store[uid][channel] def append_turn(uid: str, channel: str, role: str, content: str) -> None: """Append a single turn and enforce the history cap.""" history = _store[uid][channel] history.append({"role": role, "content": content}) # Keep only the last N turns (each turn = 1 message, cap covers both roles) max_messages = config.MAX_HISTORY_TURNS * 2 if len(history) > max_messages: _store[uid][channel] = history[-max_messages:] def clear_session(uid: str) -> None: """Clear all history for a user across both channels.""" _store.pop(uid, None)