feat(memory): verbatim transcript-chunk retrieval (flag-gated)

backend/database/vector_db.py

@@ -634,3 +634,119 @@ def delete_memory_vectors_batch(uid: str, memory_ids: List[str]) -> int:
             logger.warning(f'delete_memory_vectors_batch chunk failed uid={uid} chunk={i // 1000}')
     logger.info(f'delete_memory_vectors_batch uid={uid} total_deleted={total_deleted}')
     return total_deleted
+
+
+# ---------------------------------------------------------------------------
+# Transcript chunks ("ns_tchunks"): verbatim retrieval over raw conversation
+# transcripts. Conversation vectors (ns1) embed only the structured SUMMARY, so
+# specific details (exact dates, names, numbers, one-off mentions) are not
+# findable semantically. Chunk vectors make the raw transcript searchable.
+#
+# Privacy: chunk TEXT is embedded but never stored in Pinecone metadata —
+# transcripts are encrypted at rest in Firestore, and mirroring them as
+# plaintext metadata would bypass that. Readers re-hydrate the text from
+# Firestore via (conversation_id, chunk_index).
+TRANSCRIPT_CHUNKS_NAMESPACE = "ns_tchunks"
+
+
+def upsert_transcript_chunk_vectors(uid: str, conversation_id: str, chunks: List[dict]) -> int:
+    """chunks: [{'text': str, 'created_at': int unix ts, 'chunk_index': int}]"""
+    if index is None:
+        logger.warning('Pinecone index not initialized, skipping transcript chunk upsert')
+        return 0
+    chunks = [c for c in chunks if (c.get('text') or '').strip()]
+    if not chunks:
+        return 0
+
+    vectors = embeddings.embed_documents([c['text'] for c in chunks])
+    payload = []
+    for c, v in zip(chunks, vectors):
+        payload.append(
+            {
+                'id': f"{uid}-{conversation_id}-c{c['chunk_index']}",
+                'values': v,
+                'metadata': {
+                    'uid': uid,
+                    'conversation_id': conversation_id,
+                    'chunk_index': c['chunk_index'],
+                    'created_at': int(c['created_at']),
+                },
+            }
+        )
+
+    upserted = 0
+    for i in range(0, len(payload), 100):
+        index.upsert(vectors=payload[i : i + 100], namespace=TRANSCRIPT_CHUNKS_NAMESPACE)
+        upserted += len(payload[i : i + 100])
+    logger.info(f'upsert_transcript_chunk_vectors uid={uid} conversation={conversation_id} count={upserted}')
+    return upserted
+
+
+def search_transcript_chunks(
+    uid: str, query: str, limit: int = 20, starts_at: int = None, ends_at: int = None
+) -> List[dict]:
+    """Semantic search over transcript chunks. Returns chunk references
+    [{conversation_id, chunk_index, created_at, score}] — hydrate text from
+    Firestore (utils.conversations.transcript_chunks.hydrate_chunk_texts)."""
+    if index is None:
+        return []
+    vector = embeddings.embed_query(query)
+    filter_data = {'uid': uid}
+    if starts_at is not None and ends_at is not None:
+        filter_data['created_at'] = {'$gte': int(starts_at), '$lte': int(ends_at)}
+    xc = index.query(
+        vector=vector,
+        top_k=limit,
+        include_metadata=True,
+        filter=filter_data,
+        namespace=TRANSCRIPT_CHUNKS_NAMESPACE,
+    )
+    results = []
+    for m in xc.get('matches', []):
+        md = m.get('metadata') or {}
+        results.append(
+            {
+                'created_at': int(md['created_at']) if md.get('created_at') is not None else None,
+                'conversation_id': md.get('conversation_id'),
+                'chunk_index': int(md['chunk_index']) if md.get('chunk_index') is not None else None,
+                'score': m.get('score', 0),
+            }
+        )
+    return results
+
+
+def delete_transcript_chunk_vectors(uid: str, conversation_id: str):
+    """Delete all chunk vectors for one conversation (id-prefix listing on serverless)."""
+    if index is None:
+        return
+    prefix = f'{uid}-{conversation_id}-c'
+    try:
+        ids = []
+        for page in index.list(prefix=prefix, namespace=TRANSCRIPT_CHUNKS_NAMESPACE):
+            ids.extend(page if isinstance(page, list) else [page])
+        for i in range(0, len(ids), 1000):
+            index.delete(ids=ids[i : i + 1000], namespace=TRANSCRIPT_CHUNKS_NAMESPACE)
+        if ids:
+            logger.info(f'delete_transcript_chunk_vectors uid={uid} conversation={conversation_id} count={len(ids)}')
+    except Exception:
+        logger.warning(f'delete_transcript_chunk_vectors failed uid={uid} conversation={conversation_id}')
+
+
+def delete_transcript_chunk_vectors_batch(uid: str, conversation_ids: List[str]) -> int:
+    """Account-deletion purge: drop all transcript-chunk vectors for the user's conversations."""
+    if index is None or not conversation_ids:
+        return 0
+    deleted = 0
+    for conversation_id in conversation_ids:
+        prefix = f'{uid}-{conversation_id}-c'
+        try:
+            ids = []
+            for page in index.list(prefix=prefix, namespace=TRANSCRIPT_CHUNKS_NAMESPACE):
+                ids.extend(page if isinstance(page, list) else [page])
+            for i in range(0, len(ids), 1000):
+                index.delete(ids=ids[i : i + 1000], namespace=TRANSCRIPT_CHUNKS_NAMESPACE)
+            deleted += len(ids)
+        except Exception:
+            logger.warning(f'delete_transcript_chunk_vectors_batch failed uid={uid} conversation={conversation_id}')
+    logger.info(f'delete_transcript_chunk_vectors_batch uid={uid} total_deleted={deleted}')
+    return deleted

backend/routers/conversations.py

@@ -9,7 +9,7 @@
 import database.memories as memories_db
 import database.redis_db as redis_db
 import database.users as users_db
-from database.vector_db import delete_vector, delete_memory_vector
+from database.vector_db import delete_vector, delete_memory_vector, delete_transcript_chunk_vectors
 from utils.other.storage import delete_conversation_audio_files
 from models.calendar_context import CalendarMeetingContext
 from models.conversation import (
@@ -389,6 +389,7 @@ def delete_conversation(
     logger.info(f'delete_conversation {conversation_id} {uid} cascade={cascade}')
     conversations_db.delete_conversation(uid, conversation_id)
     delete_vector(uid, conversation_id)
+    delete_transcript_chunk_vectors(uid, conversation_id)
 
     if cascade:
         # Delete audio files

backend/routers/tools.py

@@ -21,7 +21,9 @@
 from fastapi import APIRouter, Depends, Query
 from pydantic import BaseModel, Field
 
+import database.vector_db as vector_db
 from utils.other.endpoints import get_current_user_uid, with_rate_limit
+from utils.conversations.transcript_chunks import hydrate_chunk_texts
 from utils.retrieval.tool_services.conversations import get_conversations_text, search_conversations_text
 from utils.retrieval.tool_services.memories import get_memories_text, search_memories_text
 from utils.retrieval.tool_services.action_items import (
@@ -115,6 +117,32 @@ def search_conversations(
     return _ok("search_conversations", result)
 
 
+class SearchChunksRequest(BaseModel):
+    query: str = Field(description="Semantic search query")
+    limit: int = Field(default=20, ge=1, le=30)
+
+
+@router.post("/v1/tools/conversations/search-chunks", response_model=ToolResponse)
+def search_conversation_chunks(
+    body: SearchChunksRequest,
+    uid: str = Depends(with_rate_limit(get_current_user_uid, "tools:search")),
+):
+    """Semantic search over RAW transcript chunks (verbatim evidence with dates).
+
+    Complements /conversations/search, which matches against conversation summaries:
+    summaries drop specifics (exact dates, names, numbers), so detail questions need
+    this verbatim layer. Returns chunks newest-relevant with their conversation date.
+    """
+    rows = vector_db.search_transcript_chunks(uid, body.query, limit=body.limit)
+    rows = hydrate_chunk_texts(uid, rows)
+    if not rows:
+        return _ok("search_conversation_chunks", f"No transcript excerpts found matching '{body.query}'.")
+    parts = []
+    for i, r in enumerate(rows, 1):
+        parts.append(f"Excerpt {i} (relevance: {r['score']:.2f}):\n{r['text']}")
+    return _ok("search_conversation_chunks", "\n\n".join(parts))
+
+
 # --------------- memory endpoints ---------------
 
 

backend/routers/users.py

@@ -109,6 +109,7 @@
 from database.screen_activity import get_screen_activity_ids
 from database.vector_db import (
     delete_conversation_vectors_batch,
+    delete_transcript_chunk_vectors_batch,
     delete_memory_vectors_batch,
     delete_action_item_vectors_batch,
     delete_screen_activity_vectors,
@@ -159,7 +160,8 @@ def _purge_derived_user_data(uid: str):
     blocks the others or the subsequent Firestore deletion. IDs are read via lightweight IDs-only
     queries (no decryption).
 
-    Scope: conversation (ns1), memory (ns2), action-item (ns4) and screen-activity (ns3) vectors,
+    Scope: conversation (ns1), memory (ns2), action-item (ns4), screen-activity (ns3) and
+    transcript-chunk (ns_tchunks) vectors,
     plus conversation recordings. Known follow-ups NOT covered here: X-post vectors (no delete helper
     yet), speech-profile / person-sample / private-cloud-sync / chat-upload GCS blobs, and the
     externally-indexed Typesense collection.
@@ -171,6 +173,13 @@ def _purge_derived_user_data(uid: str):
     except Exception as e:
         logger.error(f'delete_account purge conversation vectors failed for {uid}: {sanitize(str(e))}')
 
+    try:
+        conversation_ids = get_conversation_ids(uid)
+        if conversation_ids:
+            delete_transcript_chunk_vectors_batch(uid, conversation_ids)
+    except Exception as e:
+        logger.error(f'delete_account purge transcript chunk vectors failed for {uid}: {sanitize(str(e))}')
+
     try:
         memory_ids = get_memory_ids(uid)
         if memory_ids:

backend/utils/conversations/process_conversation.py

@@ -31,7 +31,8 @@
 )
 from utils.llm.memories import resolve_memory_conflict
 from database.apps import record_app_usage, get_omi_personas_by_uid_db, get_app_by_id_db
-from database.vector_db import upsert_vector2, update_vector_metadata
+from database.vector_db import upsert_vector2, update_vector_metadata, upsert_transcript_chunk_vectors
+from utils.conversations.transcript_chunks import build_transcript_chunks
 from models.app import App, UsageHistoryType
 from models.memories import MemoryDB, Memory
 from models.calendar_context import CalendarMeetingContext
@@ -655,6 +656,18 @@ def _run_auto_sync():
         )
 
 
+# Verbatim transcript-chunk indexing (ns_tchunks). Off by default: enables semantic
+# retrieval over raw transcript text, which the summary-only conversation vectors miss.
+TRANSCRIPT_CHUNK_INDEXING_ENABLED = os.getenv('TRANSCRIPT_CHUNK_INDEXING_ENABLED', 'false').lower() == 'true'
+
+
+def save_transcript_chunk_vectors(uid: str, conversation: Conversation):
+    segments = [s.dict() if hasattr(s, 'dict') else s for s in (conversation.transcript_segments or [])]
+    chunks = build_transcript_chunks(segments, conversation.started_at or conversation.created_at)
+    if chunks:
+        upsert_transcript_chunk_vectors(uid, conversation.id, chunks)
+
+
 def save_structured_vector(uid: str, conversation: Conversation, update_only: bool = False):
     vector = generate_embedding(str(conversation.structured)) if not update_only else None
     tz = notification_db.get_user_time_zone(uid)
@@ -846,6 +859,8 @@ def process_conversation(
         )
         if not is_reprocess:
             submit_with_context(postprocess_executor, save_structured_vector, uid, conversation)
+            if TRANSCRIPT_CHUNK_INDEXING_ENABLED:
+                submit_with_context(postprocess_executor, save_transcript_chunk_vectors, uid, conversation)
         submit_with_context(postprocess_executor, _extract_memories, uid, conversation)
         submit_with_context(postprocess_executor, _extract_trends, uid, conversation)
         submit_with_context(postprocess_executor, _save_action_items, uid, conversation)

backend/utils/conversations/transcript_chunks.py

@@ -0,0 +1,98 @@
+"""Build verbatim transcript chunks for vector indexing.
+
+Conversation vectors (ns1) embed only the structured summary, so specific details
+(exact dates, names, numbers, one-off mentions) are unfindable semantically. These
+chunks slice the raw transcript into overlapping windows, each prefixed with the
+conversation date, so semantic search can land on the verbatim evidence.
+"""
+
+from datetime import datetime
+from typing import List, Optional
+
+import database.conversations as conversations_db
+
+# ~8 segments per chunk with 2-segment overlap keeps chunks small enough to embed
+# precisely while not splitting answers across a hard boundary.
+CHUNK_WINDOW = 8
+CHUNK_STRIDE = 6
+
+
+def _speaker_label(seg: dict, people_by_id: Optional[dict] = None) -> str:
+    if seg.get('is_user'):
+        return 'User'
+    person_id = seg.get('person_id')
+    if person_id and people_by_id and person_id in people_by_id:
+        return people_by_id[person_id]
+    speaker_id = seg.get('speaker_id')
+    return f"Speaker {speaker_id}" if speaker_id is not None else 'Speaker'
+
+
+def build_transcript_chunks(
+    segments: List[dict],
+    started_at: Optional[datetime],
+    window: int = CHUNK_WINDOW,
+    stride: int = CHUNK_STRIDE,
+    people_by_id: Optional[dict] = None,
+) -> List[dict]:
+    """segments: transcript_segment dicts ({'text','is_user','speaker_id','person_id',...}).
+
+    Returns [{'text', 'created_at' (unix ts), 'chunk_index'}] ready for
+    vector_db.upsert_transcript_chunk_vectors.
+    """
+    lines = []
+    for seg in segments or []:
+        text = (seg.get('text') or '').strip()
+        if not text:
+            continue
+        lines.append(f"{_speaker_label(seg, people_by_id)}: {text}")
+    if not lines:
+        return []
+
+    date_header = ''
+    created_ts = 0
+    if started_at is not None:
+        date_header = f"[Conversation on {started_at.strftime('%d %b %Y, %H:%M')}]\n"
+        created_ts = int(started_at.timestamp())
+
+    chunks = []
+    idx = 0
+    pos = 0
+    while pos < len(lines):
+        piece = lines[pos : pos + window]
+        chunks.append(
+            {
+                'text': date_header + "\n".join(piece),
+                'created_at': created_ts,
+                'chunk_index': idx,
+            }
+        )
+        if pos + window >= len(lines):
+            break
+        pos += stride
+        idx += 1
+    return chunks
+
+
+def hydrate_chunk_texts(uid: str, rows: List[dict]) -> List[dict]:
+    """Attach verbatim text to chunk references returned by vector search.
+
+    Re-reads the conversations from Firestore (decrypted by the db layer) and rebuilds
+    the deterministic chunking, so transcript text never has to live in Pinecone.
+    Rows whose conversation/chunk no longer exists are dropped.
+    """
+    conv_ids = list({r['conversation_id'] for r in rows if r.get('conversation_id')})
+    if not conv_ids:
+        return []
+    conversations = conversations_db.get_conversations_by_id(uid, conv_ids)
+    chunks_by_conv = {}
+    for c in conversations:
+        segs = c.get('transcript_segments') or []
+        started = c.get('started_at') or c.get('created_at')
+        chunks_by_conv[c['id']] = {ch['chunk_index']: ch['text'] for ch in build_transcript_chunks(segs, started)}
+
+    hydrated = []
+    for r in rows:
+        text = chunks_by_conv.get(r.get('conversation_id'), {}).get(r.get('chunk_index'))
+        if text:
+            hydrated.append({**r, 'text': text})
+    return hydrated