""" db_5_process_session.py Short-term chat history management + long-term FAISS archiving (db5). Short-term window (chat_history_short): - Keep: last 10 days AND at least 40 messages - Trim when: oldest session > 21 days OR total messages > 200 - Trimmed closed sessions are chunked and archived to FAISS db5 FAISS db5 (long-term): - Embedding model: intfloat/e5-large-v2 (1024 dims, same as all other DBs) - Two documents per chunk: raw turns text + LLM summary (multi-vector) - Rich metadata: chunk_id, session_id, timestamps, character, turn_range - Long-term history recalled only on explicit memory-reference queries (see classify_memory_reference.py) """ import os from pathlib import Path from datetime import datetime, timedelta, timezone from tempfile import TemporaryDirectory from typing import Optional from langchain_huggingface import HuggingFaceEmbeddings from langchain_community.vectorstores import FAISS from langchain_core.documents import Document as LangchainDocument from supabase import create_client from config import SUPABASE_URL, SUPABASE_SERVICE_KEY, HF_EMBEDDING_MODEL from db_user import _load_history, _save_history from db_client import supabase # ── Short-term window ────────────────────────────────────────────────────────── _MIN_DAYS = 10 _MAX_DAYS = 21 _MIN_MESSAGES = 40 _MAX_MESSAGES = 200 _SESSION_IDLE_MINUTES = 360 # 6 hours # ── Chunking ─────────────────────────────────────────────────────────────────── _CHUNK_WINDOW = 8 # messages per chunk _CHUNK_OVERLAP = 2 # messages shared between adjacent chunks _CHUNK_STEP = _CHUNK_WINDOW - _CHUNK_OVERLAP # = 6 _MID_SESSION_SNAPSHOT_EVERY_N = 30 # ── Embedding model (lazy singleton) ────────────────────────────────────────── _embedding_model: Optional[HuggingFaceEmbeddings] = None def _get_embeddings() -> HuggingFaceEmbeddings: global _embedding_model if _embedding_model is None: _embedding_model = HuggingFaceEmbeddings( model_name=HF_EMBEDDING_MODEL, encode_kwargs={"normalize_embeddings": True}, ) return _embedding_model # ── Time helpers ─────────────────────────────────────────────────────────────── def _now_utc() -> datetime: return datetime.now(timezone.utc).replace(tzinfo=None) def _parse_ts(ts_str: Optional[str]) -> Optional[datetime]: if not ts_str: return None try: return datetime.fromisoformat(ts_str.replace("Z", "")) except Exception: return None def _age_days(dt: Optional[datetime]) -> float: if dt is None: return 0.0 return (_now_utc() - dt).total_seconds() / 86400.0 # ── Session helpers ──────────────────────────────────────────────────────────── def _new_session(session_id: int, now: str) -> dict: return { "session_id": session_id, "status": "open", "date_opened": now, "date_closed": None, "messages": [], } def _total_messages(sessions: list) -> int: return sum(len(s.get("messages", [])) for s in sessions) def close_current_session_if_expired( max_idle_minutes: int = _SESSION_IDLE_MINUTES, user_id: str = "None" ): """Close the current session if idle too long.""" short = _load_history("chat_history_short", user_id) if not short["sessions"]: return short current = short["sessions"][-1] msgs = current.get("messages", []) if not msgs: return short last_ts = msgs[-1].get("time") or msgs[-1].get("timestamp") last_dt = _parse_ts(last_ts) if last_dt and _now_utc() - last_dt > timedelta(minutes=max_idle_minutes): current["status"] = "closed" current["date_closed"] = _now_utc().isoformat() _save_history("chat_history_short", short, user_id) return short # ── Save message ─────────────────────────────────────────────────────────────── def save_chat_message( message_en: dict, message_user_language: dict, user_id: str = "None" ): """Append a message to both short-term and total history in Supabase.""" now = _now_utc().isoformat() if isinstance(message_user_language, str): message_user_language = {"lang": message_user_language} for msg in (message_en, message_user_language): if "time" not in msg: msg["time"] = now # total history (user language, for UI) total = _load_history("chat_history_total", user_id) if not total["sessions"]: total["sessions"].append(_new_session(1, now)) total["sessions"][-1]["messages"].append(message_user_language) _save_history("chat_history_total", total, user_id) # short history (English, for reasoning) short = _load_history("chat_history_short", user_id) if not short["sessions"]: short["sessions"].append(_new_session(1, now)) short["sessions"][-1]["messages"].append(message_en) _save_history("chat_history_short", short, user_id) # ── Short-term trimming ──────────────────────────────────────────────────────── def _trim_short_history_if_needed(user_id: str, username: str) -> None: """ Remove oldest closed sessions that exceed the time/count window and archive them to FAISS db5. Always keeps at least _MIN_DAYS of history and _MIN_MESSAGES messages. """ short = _load_history("chat_history_short", user_id) sessions = short["sessions"] changed = False while len(sessions) > 1: oldest = sessions[0] # only remove closed sessions if oldest.get("status") != "closed": break age = _age_days(_parse_ts(oldest.get("date_opened"))) total = _total_messages(sessions) remaining_after = total - len(oldest.get("messages", [])) must_trim = (age > _MAX_DAYS) or (total > _MAX_MESSAGES) # allow trim only if enough messages remain, or session is very old safe_to_trim = (remaining_after >= _MIN_MESSAGES) or (age > _MAX_DAYS * 2) if must_trim and safe_to_trim: _archive_sessions_to_faiss([oldest], username=username) sessions.pop(0) changed = True else: break if changed: short["sessions"] = sessions _save_history("chat_history_short", short, user_id) # ── Mid-session snapshot ─────────────────────────────────────────────────────── def _mid_session_faiss_snapshot_if_needed(user_id: str, username: str) -> None: """Snapshot the open session to FAISS every N messages so db5 stays fresh.""" try: short = _load_history("chat_history_short", user_id) if not short["sessions"]: return current = short["sessions"][-1] msg_count = len(current.get("messages", [])) if msg_count > 0 and msg_count % _MID_SESSION_SNAPSHOT_EVERY_N == 0: chunks = _make_chunks(current["messages"], session_id=current["session_id"]) if chunks: _add_chunks_to_faiss(chunks, username=username) print(f"[db5] Mid-session snapshot at {msg_count} messages for {user_id}") except Exception as e: print(f"[db5] Mid-session snapshot error: {e}") # ── Main tick ────────────────────────────────────────────────────────────────── def on_pre_message_tick(user_id: str, username: str): """Called before each new user message: expire, trim, snapshot.""" close_current_session_if_expired(user_id=user_id) _trim_short_history_if_needed(user_id=user_id, username=username) _mid_session_faiss_snapshot_if_needed(user_id=user_id, username=username) # ── Context retrieval (used by app_nn.py) ───────────────────────────────────── def _prepare_chat_history_for_retrieval(user_id: str = "None", max_turns: int = 20) -> list: """Return recent messages from the current session for prompt context.""" short = _load_history("chat_history_short", user_id) if not short["sessions"]: return [] return short["sessions"][-1]["messages"][-max_turns:] # ── Sliding-window chunker ───────────────────────────────────────────────────── def _format_turn(msg: dict) -> str: role = msg.get("role", "?") if role == "assistant": speaker = (msg.get("character_id") or "assistant").upper() else: speaker = "USER" content = msg.get("content", "").strip() return f"{speaker}: {content}" def _make_chunks(messages: list, session_id) -> list[dict]: """ Slice a message list into overlapping windows. Returns list of chunk dicts with text, metadata fields, and turn positions. """ if not messages: return [] n = len(messages) starts = list(range(0, max(1, n - _CHUNK_WINDOW + 1), _CHUNK_STEP)) if not starts: starts = [0] chunks = [] for pos in starts: window = messages[pos: pos + _CHUNK_WINDOW] text = "\n".join(_format_turn(m) for m in window) ts_start = window[0].get("time") or window[0].get("timestamp") or "" ts_end = window[-1].get("time") or window[-1].get("timestamp") or "" character = next( (m.get("character") for m in window if m.get("character")), "" ) or "" chunks.append({ "chunk_id": f"{session_id}_{pos}", "session_id": str(session_id), "turn_start": pos, "turn_end": pos + len(window) - 1, "timestamp_start": ts_start, "timestamp_end": ts_end, "character": character, "text": text, }) return chunks # ── Per-chunk LLM summary (Tier C / nano) ───────────────────────────────────── _SUMMARY_PROMPT = ( "Summarise this chat exchange in 2-3 sentences. " "Capture: main topic(s), any opinion or position the user expressed, " "and any tension or contradiction that emerged. Be concise.\n\n" ) def _summarise_chunk(chunk_text: str) -> str: try: from llm_client import client, get_model resp = client.chat.completions.create( model=get_model("C"), messages=[{"role": "user", "content": _SUMMARY_PROMPT + chunk_text}], temperature=0.0, max_tokens=150, ) return resp.choices[0].message.content.strip() except Exception as e: print(f"[db5] Summary LLM error: {e}") return chunk_text[:300] # graceful fallback # ── FAISS archive ────────────────────────────────────────────────────────────── def _archive_sessions_to_faiss(sessions: list, username: str) -> None: for session in sessions: chunks = _make_chunks( session.get("messages", []), session_id=session.get("session_id", 0), ) if chunks: _add_chunks_to_faiss(chunks, username=username) def _add_chunks_to_faiss(chunks: list[dict], username: str) -> None: """ Build LangChain Documents from chunks (raw + summary pair per chunk) and merge into the user's db5 FAISS index in Supabase. Two documents per chunk: raw — full formatted turns text; matched by exact phrasing summary — LLM 2-3 sentence summary; matched by topic/theme Both carry `raw_text` in metadata so callers always have the full exchange regardless of which document type was retrieved. """ if not chunks: return embeddings = _get_embeddings() docs = [] for chunk in chunks: summary = _summarise_chunk(chunk["text"]) base_meta = { "chunk_id": chunk["chunk_id"], "session_id": chunk["session_id"], "turn_start": chunk["turn_start"], "turn_end": chunk["turn_end"], "timestamp_start": chunk["timestamp_start"], "timestamp_end": chunk["timestamp_end"], "character": chunk["character"], "raw_text": chunk["text"], # always present for retrieval output } docs.append(LangchainDocument( page_content=chunk["text"], metadata={**base_meta, "chunk_type": "raw"}, )) docs.append(LangchainDocument( page_content=summary, metadata={**base_meta, "chunk_type": "summary"}, )) _merge_docs_into_db5(docs, embeddings, username) print(f"[db5] Archived {len(chunks)} chunks ({len(docs)} docs) for {username}") def _merge_docs_into_db5( docs: list[LangchainDocument], embeddings: HuggingFaceEmbeddings, username: str, ) -> None: """Create a new FAISS index from docs and merge it into the existing db5.""" user_folder = f"user_{username}" db_name = "db5" bucket = "Databases" path_prefix = f"users/{user_folder}/{db_name}" with TemporaryDirectory() as tmp_dir: tmp_path = Path(tmp_dir) new_db = FAISS.from_documents(docs, embeddings) new_db.save_local(str(tmp_path)) # try to download and merge with existing db5 existing_path = tmp_path / "existing" existing_path.mkdir(exist_ok=True) has_existing = True for fname in ["index.faiss", "index.pkl"]: try: data = supabase.storage.from_(bucket).download(f"{path_prefix}/{fname}") with open(existing_path / fname, "wb") as f: f.write(data) except Exception: has_existing = False break if has_existing: try: base = FAISS.load_local( str(existing_path), embeddings, allow_dangerous_deserialization=True, ) base.merge_from(new_db) base.save_local(str(tmp_path)) except Exception as e: print(f"[db5] Merge failed, uploading fresh index: {e}") # upload (remove first to avoid 409 conflict) for fname in ["index.faiss", "index.pkl"]: local = tmp_path / fname if not local.exists(): continue remote = f"{path_prefix}/{fname}" try: supabase.storage.from_(bucket).remove([remote]) except Exception: pass with open(local, "rb") as fh: supabase.storage.from_(bucket).upload( remote, fh, {"content-type": "application/octet-stream"}, )