Spaces:

alesamodio
/

Socrates_docker

Sleeping

App Files Files Community

Socrates_docker / db_5_process_session.py

AlessandroAmodioNGI

Claude Opus 4.8 (1M context)

fix: centralize Supabase client on HTTP/1.1 to stop RemoteProtocolError 500s

6d936a1 4 months ago

Raw History Blame Contribute Delete

15.9 kB

"""
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"},
                )

	"""
	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"},
	)