"""Raw conversation chunks — the append-only record of what compression dropped.
Long-term memory used to have three shapes: curated facts, per-session sketches,
and the live context. When the context was compressed, everything that did not
fit the summary was simply gone — the message rows stay in the database, but
nothing searched them by meaning, and after a reload the session carried only the
summary. "We discussed this earlier" was unanswerable.
This table is the missing store: every message that compression demotes is
written here once, verbatim (a very large single message is elided in the
middle), with the turn range and timestamps it came from and the breadcrumb keys
extracted for it. Three properties are load-bearing:
* **Append-only and immutable.** A row is never updated. A correction is a new
chunk written later; `UNIQUE(session_id, content_hash)` only prevents the same
dropped run from being indexed twice, which is what makes a compression replay
idempotent.
* **Read together with facts and sketches**, through `UnifiedMemorySearch`, so an
agent has one place to ask. Writing stays separate: chunks are written only by
the compressor, never by a tool.
* **Provenance is part of the record** — session, turn range, timestamps — because
an answer built on a raw excerpt must say when it was said, and a curated fact
from last week must win over a raw line from last month.
"""
from __future__ import annotations
import hashlib
import re
import uuid
from datetime import datetime, timezone
from typing import TYPE_CHECKING
import structlog
from ._embeddings import _vector_to_str
from ._facts import STOPWORDS, _normalize_query
if TYPE_CHECKING:
from navi.llm.base import Message
log = structlog.get_logger()
# One chunk is one searchable unit. 4000 chars is roughly a long tool result or
# a dozen chat messages — small enough that a hit is about one topic, large
# enough that the answer's context is not split across three rows.
CHUNK_MAX_CHARS = 4000
# A single message can be megabytes (a full file read). Keep its head and tail:
# the middle of a dump is the part nobody asks for later, while the command line
# at the top and the error at the bottom are usually the breadcrumbs.
CHUNK_MESSAGE_MAX_CHARS = 12_000
# Cosine distance cutoff for a chunk match. Facts use 0.3 because a fact value is
# one short sentence; a 4 KB passage against a short query naturally sits farther
# away, and 0.45 is where the pilot's paraphrased queries still landed. Anything
# past it goes to the lexical fallback instead — a wrong vector hit is worse than
# a keyword hit for a long passage.
VECTOR_CUTOFF_DISTANCE = 0.45
# Recall builds its query from a window of recent turns (up to 1200 chars), and
# every term used to become its own ILIKE pattern and its own `terms_hit` point.
# A dozen content words is what a query is actually about; the rest is prose.
_MAX_LEXICAL_TERMS = 12
_WORD = re.compile(r"\w+", re.UNICODE)
_ROLE_PREFIX = {
"user": "User",
"assistant": "Assistant",
"tool": "Tool result",
}
def build_chunks(
messages: list[Message],
breadcrumbs: list[str] | None = None,
max_chars: int = CHUNK_MAX_CHARS,
) -> list[dict]:
"""Group dropped messages into immutable chunks.
Consecutive messages are packed together until `max_chars`, so a turn's user
message, tool calls and the assistant reply land in one chunk. Summary and
compression marker messages are skipped — a summary is not raw history, and
indexing old summaries again would make each compression index its own past.
"""
chunks: list[dict] = []
current: list[str] = []
current_len = 0
start_seq: int | None = None
end_seq: int | None = None
started_at: datetime | None = None
ended_at: datetime | None = None
keys = list(breadcrumbs or [])
def flush() -> None:
nonlocal current, current_len, start_seq, end_seq, started_at, ended_at
if not current:
return
text = "\n\n".join(current).strip()
if text:
chunks.append({
"content": text,
"content_hash": hashlib.sha256(text.encode("utf-8", "replace")).hexdigest(),
"start_seq": start_seq if start_seq is not None else -1,
"end_seq": end_seq if end_seq is not None else -1,
"started_at": started_at,
"ended_at": ended_at,
"breadcrumbs": keys,
})
current, current_len = [], 0
start_seq = end_seq = None
started_at = ended_at = None
for msg in messages:
if msg.role == "system" or msg.is_summary or msg.is_compression:
continue
text = _format_message(msg)
if not text:
continue
if current and current_len + len(text) > max_chars:
flush()
current.append(text)
current_len += len(text) + 2
if start_seq is None:
start_seq = msg.sequence_number
started_at = msg.created_at
if msg.sequence_number >= 0:
end_seq = msg.sequence_number
if msg.created_at is not None:
ended_at = msg.created_at
flush()
return chunks
def _format_message(msg: Message) -> str:
content = (msg.content or "").strip()
if not content:
# An assistant message that only requested tool calls still records what
# was asked for; the result follows in its own message.
if msg.tool_calls:
names = ", ".join(tc.name for tc in msg.tool_calls if getattr(tc, "name", None))
return f"Assistant requested tools: {names}" if names else ""
return ""
if len(content) > CHUNK_MESSAGE_MAX_CHARS:
head = CHUNK_MESSAGE_MAX_CHARS * 3 // 4
tail = CHUNK_MESSAGE_MAX_CHARS - head - 40
content = f"{content[:head]}\n…[middle elided]…\n{content[-tail:]}"
prefix = _ROLE_PREFIX.get(msg.role, msg.role.title())
if msg.role == "tool" and msg.name:
prefix = f"Tool result ({msg.name})"
return f"{prefix}: {content}"
class ChunkMixin:
"""Storage and search for dropped-raw-history chunks.
Expected on the composite class:
_get_pool() -> asyncpg.Pool
_generate_embedding(text) -> list[float] | None
_generate_embeddings(texts) -> list[list[float] | None]
_has_pgvector() -> bool
"""
async def add_chunks(
self,
chunks: list[dict],
user_id: str | None = None,
session_id: str | None = None,
) -> int:
"""Write chunks, skipping ones already indexed for this session."""
if not chunks or not session_id:
return 0
# The embedding column exists only where pgvector does; on a plain Postgres
# the `::vector` cast in the INSERT would fail the whole write, so the row
# is written text-only and found through FTS.
use_vectors = await self._has_pgvector()
embeddings = (
await self._generate_embeddings([c["content"] for c in chunks])
if use_vectors else [None] * len(chunks)
)
now = datetime.now(timezone.utc)
pool = await self._get_pool()
written = 0
async with pool.acquire() as conn:
for chunk, embedding in zip(chunks, embeddings):
vec_str = _vector_to_str(embedding) if embedding else None
if use_vectors:
result = await conn.execute(
"""INSERT INTO memory_chunks
(id, user_id, session_id, start_seq, end_seq, started_at, ended_at,
content, breadcrumbs, content_hash, created_at, embedding)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12::vector)
ON CONFLICT (session_id, content_hash) DO NOTHING""",
str(uuid.uuid4()), user_id, session_id,
chunk["start_seq"], chunk["end_seq"],
chunk.get("started_at"), chunk.get("ended_at"),
chunk["content"], list(chunk.get("breadcrumbs") or []),
chunk["content_hash"], now, vec_str,
)
else:
result = await conn.execute(
"""INSERT INTO memory_chunks
(id, user_id, session_id, start_seq, end_seq, started_at, ended_at,
content, breadcrumbs, content_hash, created_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
ON CONFLICT (session_id, content_hash) DO NOTHING""",
str(uuid.uuid4()), user_id, session_id,
chunk["start_seq"], chunk["end_seq"],
chunk.get("started_at"), chunk.get("ended_at"),
chunk["content"], list(chunk.get("breadcrumbs") or []),
chunk["content_hash"], now,
)
if result == "INSERT 0 1":
written += 1
if written:
log.info(
"memory.chunks_written",
session_id=session_id,
written=written,
skipped=len(chunks) - written,
)
return written
async def search_chunks(
self,
query: str,
user_id: str | None = None,
session_id: str | None = None,
limit: int = 8,
) -> list[dict]:
"""Search raw chunks: vector first, lexical fallback when that misses.
`session_id` narrows to one conversation; omitting it searches every
session of the user. Anonymous rows (`user_id IS NULL`) are their own
scope, exactly like facts.
"""
if limit <= 0:
return []
rows: list[dict] = []
if self._embedding_backend and await self._has_pgvector():
query_embedding = await self._generate_embedding(query)
vec_str = _vector_to_str(query_embedding) if query_embedding else None
if vec_str:
try:
rows = await self._vector_chunks(
vec_str, user_id, session_id, limit,
)
except Exception:
log.warning("memory.chunk_vector_failed", query_len=len(query), exc_info=True)
if len(rows) >= limit:
log.debug("memory.chunk_vector_search", hits=len(rows))
return rows
if rows and not _lexical_terms(query):
# Nothing lexical to top up with; the empty-term fallback returns
# *recent* chunks, which are not matches and must not ride along
# behind a vector hit.
log.debug("memory.chunk_vector_search", hits=len(rows))
return rows
# Vector-first, but not vector-only: a chunk whose embedding is NULL (a
# failed embedding batch, or a row written while pgvector was absent) is
# invisible to `ORDER BY embedding <=>` and would never come back once
# any vector hit exists. Top up the short result with the lexical
# matches, which see every row.
lexical = await self._lexical_chunks(query, user_id, session_id, limit)
if not rows:
if lexical:
log.debug("memory.chunk_lexical_search", hits=len(lexical))
return lexical
seen = {row["id"] for row in rows}
rows.extend(row for row in lexical if row["id"] not in seen)
log.debug("memory.chunk_vector_search", hits=len(rows))
return rows[:limit]
async def _vector_chunks(
self,
vec_str: str,
user_id: str | None,
session_id: str | None,
limit: int,
) -> list[dict]:
where, params = _chunk_filter(user_id, session_id)
params.append(vec_str)
vec_idx = len(params)
params.append(limit * 3)
lim_idx = len(params)
pool = await self._get_pool()
async with pool.acquire() as conn:
rows = await conn.fetch(
f"""SELECT id, session_id, start_seq, end_seq, started_at, ended_at,
content, breadcrumbs, created_at,
embedding <=> ${vec_idx}::vector AS distance
FROM memory_chunks
WHERE {where}
ORDER BY embedding <=> ${vec_idx}::vector
LIMIT ${lim_idx}""",
*params,
)
hits: list[dict] = []
for row in rows:
distance = row["distance"]
if distance is None or distance >= VECTOR_CUTOFF_DISTANCE:
continue
hit = _row_to_chunk(row)
hit["score"] = max(0.0, 1.0 - float(distance))
hits.append(hit)
if len(hits) >= limit:
break
return hits
async def _lexical_chunks(
self,
query: str,
user_id: str | None,
session_id: str | None,
limit: int,
) -> list[dict]:
terms = _lexical_terms(query)
where, params = _chunk_filter(user_id, session_id)
if not terms:
return await self._recent_chunks(where, params, limit)
ts_query = " ".join(terms)
like_patterns = [f"%{term}%" for term in terms]
params.append(ts_query)
ts_idx = len(params)
params.append(like_patterns)
like_idx = len(params)
params.append(limit * 4)
lim_idx = len(params)
match_clause = (
f"(to_tsvector('simple', content) @@ plainto_tsquery('simple', ${ts_idx}) "
f"OR content ILIKE ANY(${like_idx}::text[]) "
f"OR array_to_string(breadcrumbs, ' ') ILIKE ANY(${like_idx}::text[]))"
)
pool = await self._get_pool()
async with pool.acquire() as conn:
rows = await conn.fetch(
f"""SELECT id, session_id, start_seq, end_seq, started_at, ended_at,
content, breadcrumbs, created_at,
ts_rank(to_tsvector('simple', content),
plainto_tsquery('simple', ${ts_idx})) AS rank
FROM memory_chunks
WHERE {where} AND {match_clause}
ORDER BY rank DESC, ended_at DESC NULLS LAST
LIMIT ${lim_idx}""",
*params,
)
hits = []
normalized_terms = {t for t in terms if len(t) > 2}
for row in rows:
hit = _row_to_chunk(row)
haystack = f"{hit['content']} {' '.join(hit['breadcrumbs'])}"
hit["terms_hit"] = _matched_terms(haystack, normalized_terms)
hit["score"] = None
hits.append(hit)
# A row matched one long term by accident; a row matching several query
# terms is the one the caller meant. Rank on that before the DB order.
hits.sort(key=lambda h: (-h["terms_hit"], -_ts(h.get("rank"))))
return hits[:limit]
async def _recent_chunks(self, where: str, params: list, limit: int) -> list[dict]:
pool = await self._get_pool()
async with pool.acquire() as conn:
rows = await conn.fetch(
f"""SELECT id, session_id, start_seq, end_seq, started_at, ended_at,
content, breadcrumbs, created_at
FROM memory_chunks
WHERE {where}
ORDER BY ended_at DESC NULLS LAST
LIMIT ${len(params) + 1}""",
*params, limit,
)
hits = []
for row in rows:
hit = _row_to_chunk(row)
# No relevance was computed here — "most recent" is not "most
# similar". A fabricated score (1.0, 0.95, …) used to sail past
# recall's similarity threshold and inject unrelated turns.
hit["score"] = None
hit["terms_hit"] = 0
hits.append(hit)
return hits
async def latest_summary_at(self, session_id: str) -> datetime | None:
"""When this session was last compressed — raw chunks before it are verbatim
history that the current summary has already curated."""
pool = await self._get_pool()
async with pool.acquire() as conn:
try:
hot = await conn.fetchval(
"SELECT max(created_at) FROM session_messages "
"WHERE session_id = $1 AND is_summary = TRUE",
session_id,
)
except Exception:
log.debug("memory.latest_summary_failed", session_id=session_id, exc_info=True)
return None
# Long sessions archive old rows, summaries included (the archive
# move is by sequence number, not by kind). Looking only at the hot
# table made the newest summary invisible there and the
# `before_summary` label wrong.
try:
archived = await conn.fetchval(
"SELECT max(created_at) FROM session_messages_archive "
"WHERE session_id = $1 AND is_summary = TRUE",
session_id,
)
except Exception:
archived = None
values = [value for value in (hot, archived) if value is not None]
return max(values) if values else None
async def chunk_count(self, session_id: str | None = None, user_id: str | None = None) -> int:
where, params = _chunk_filter(user_id, session_id)
pool = await self._get_pool()
async with pool.acquire() as conn:
return int(await conn.fetchval(
f"SELECT count(*) FROM memory_chunks WHERE {where}", *params
) or 0)
def _lexical_terms(query: str) -> list[str]:
"""Content words of the query, first occurrence wins, capped.
`_normalize_query` deliberately keeps function words — the fact and sketch
lookups use them too and their ILIKE paths are cheap — but a chunk query is
*scored* on term overlap, and "the" inside "other" must not count. The cap
matters because recall passes a window of recent conversation (up to 1200
chars); without it one query becomes hundreds of ILIKE patterns and matches
everything.
"""
seen: set[str] = set()
terms: list[str] = []
for term in _normalize_query(query):
if term in STOPWORDS or term in seen:
continue
seen.add(term)
terms.append(term)
if len(terms) >= _MAX_LEXICAL_TERMS:
break
return terms
def _matched_terms(text: str, terms: set[str]) -> int:
"""How many of `terms` occur as whole words in `text`."""
return len(terms & set(_WORD.findall(text.lower())))
def _chunk_filter(user_id: str | None, session_id: str | None) -> tuple[str, list]:
"""WHERE clause for the (user, session) scope. Anonymous rows are NULL-owned."""
clauses = []
params: list = []
if user_id is None:
clauses.append("user_id IS NULL")
else:
params.append(user_id)
clauses.append(f"user_id = ${len(params)}")
if session_id is not None:
params.append(session_id)
clauses.append(f"session_id = ${len(params)}")
return " AND ".join(clauses) or "TRUE", params
def _row_to_chunk(row) -> dict:
return {
"rank": row["rank"] if "rank" in row.keys() else None,
"id": row["id"],
"session_id": row["session_id"],
"start_seq": row["start_seq"],
"end_seq": row["end_seq"],
"started_at": row["started_at"],
"ended_at": row["ended_at"],
"content": row["content"],
"breadcrumbs": list(row["breadcrumbs"] or []),
"created_at": row["created_at"],
"terms_hit": 0,
}
def _ts(rank) -> float:
try:
return float(rank or 0.0)
except (TypeError, ValueError):
return 0.0
