"""One search across everything Navi remembers.
Memory is stored in three different shapes, and they are different on purpose:
* **facts** — curated, rewritten by key, user lifetime. "What is true about the
user" (``memory_facts``).
* **sketches** — a session's working notes, updated in place while a task runs
(``session_store`` scope ``scratchpad``).
* **raw** — verbatim messages that context compression dropped
(``memory_chunks``), append-only and immutable.
Writing them is deliberately separate. Reading them is not: an agent asked "what
did we decide about the port" should not have to know which of the three holds
the answer, or that it is even a different store. This module merges the three
into one ranked answer with provenance attached to every item.
The merge is tiered, not a blind mix: curated facts first, then raw excerpts by
relevance, then sketches from the current session. A fact that was rewritten
after a raw excerpt wins on freshness — the raw line is still returned, labelled
as outdated, because "we changed our mind, the old value was X" is often exactly
the question.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime
import structlog
from ._facts import STOPWORDS as _STOPWORDS
from ._facts import _normalize_query
log = structlog.get_logger()
SCOPES = ("facts", "raw", "sketches", "all")
_MAX_K = 20
_HIT_TEXT_CHARS = 1200
@dataclass
class MemoryHit:
source_type: str # "fact" | "raw" | "sketch"
text: str
session_id: str | None = None
timestamp: datetime | None = None
key: str | None = None # fact key / sketch section
category: str | None = None # fact category
start_seq: int = -1 # raw: first dropped turn
end_seq: int = -1 # raw: last dropped turn
chunk_id: str | None = None
score: float | None = None # vector similarity, 0..1 when available
terms_hit: int = 0 # lexical overlap, for the FTS fallback
superseded_by: str | None = None # newer curated fact that overrides this
before_summary: bool = False # raw older than the session's last summary
breadcrumbs: tuple[str, ...] = field(default_factory=tuple)
class UnifiedMemorySearch:
"""Read-side facade over facts, raw chunks and sketches."""
def __init__(self, memory_store=None, kv_store=None) -> None:
self._memory = memory_store
self._kv = kv_store
async def latest_summary_at(self, session_id: str) -> datetime | None:
"""When `session_id` was last compressed, for the freshness label."""
if self._memory is None or not session_id:
return None
try:
return await self._memory.latest_summary_at(session_id)
except Exception:
return None
async def search(
self,
query: str,
*,
scope: str = "all",
k: int = 5,
user_id: str | None = None,
session_id: str | None = None,
current_session_id: str | None = None,
summary_at: datetime | None = None,
) -> list[MemoryHit]:
"""Search the requested stores and return a tiered, provenance-labelled list.
`session_id` narrows raw chunks to one conversation; `current_session_id`
only orders sketches (the current session's pad first). `summary_at` is the
timestamp of the session's latest summary — raw chunks older than it are
the verbatim record of history the summary already curated.
"""
query = (query or "").strip()
if not query:
return []
scope = scope if scope in SCOPES else "all"
k = max(1, min(int(k or 5), _MAX_K))
wanted = {"facts", "raw", "sketches"} if scope == "all" else {scope}
facts: list[MemoryHit] = []
raw: list[MemoryHit] = []
sketches: list[MemoryHit] = []
freshness_facts: list[MemoryHit] = []
if "facts" in wanted:
facts = await self._facts(query, user_id, k)
freshness_facts = facts
elif "raw" in wanted:
# Raw excerpts are labelled against curated facts even when the
# caller asked for raw only: an outdated excerpt shown as
# authoritative is worse than a slower answer. The facts are used
# for the label and are not returned.
freshness_facts = await self._facts(query, user_id, min(k, 5))
if "raw" in wanted:
raw = await self._raw(query, user_id, session_id, k)
if "sketches" in wanted:
sketches = await self._sketches(query, user_id, k, current_session_id or session_id)
_mark_freshness(freshness_facts, raw, summary_at, current_session_id or session_id)
hits = facts + raw + sketches
# The query is conversation text — often the user's own words, and
# recall runs it on every turn once a session has a summary. Its
# length is enough for diagnostics; the text would journal it.
log.info(
"memory_search.done",
query_len=len(query),
scope=scope,
facts=len(facts),
raw=len(raw),
sketches=len(sketches),
)
return hits
async def _facts(self, query: str, user_id: str | None, k: int) -> list[MemoryHit]:
if self._memory is None:
return []
try:
rows = await self._memory.search_facts(query, user_id=user_id, limit=k)
except Exception:
log.warning("memory_search.facts_failed", query_len=len(query), exc_info=True)
return []
hits = []
for row in rows:
hits.append(MemoryHit(
source_type="fact",
text=f"{row.get('key')}: {row.get('value')}",
session_id=row.get("source_session_id") or None,
timestamp=_as_datetime(row.get("updated_at")),
key=row.get("key"),
category=row.get("category"),
score=None,
))
return hits
async def _raw(
self, query: str, user_id: str | None, session_id: str | None, k: int,
) -> list[MemoryHit]:
if self._memory is None:
return []
try:
rows = await self._memory.search_chunks(
query, user_id=user_id, session_id=session_id, limit=k,
)
except Exception:
log.warning("memory_search.raw_failed", query_len=len(query), exc_info=True)
return []
hits = []
for row in rows:
hits.append(MemoryHit(
source_type="raw",
text=row.get("content") or "",
session_id=row.get("session_id"),
timestamp=row.get("ended_at") or row.get("created_at"),
start_seq=row.get("start_seq", -1),
end_seq=row.get("end_seq", -1),
chunk_id=row.get("id"),
score=row.get("score"),
terms_hit=row.get("terms_hit", 0),
breadcrumbs=tuple(row.get("breadcrumbs") or ()),
))
return hits
async def _sketches(
self, query: str, user_id: str | None, k: int, prefer_session_id: str | None,
) -> list[MemoryHit]:
if self._kv is None:
return []
terms = _normalize_query(query)
try:
rows = await self._kv.search_values(
user_id,
"scratchpad",
terms,
limit=k,
prefer_session_id=prefer_session_id,
)
except Exception:
log.warning("memory_search.sketches_failed", query_len=len(query), exc_info=True)
return []
hits = []
for row in rows:
value = (row.get("value") or "").strip()
if not value:
continue
hits.append(MemoryHit(
source_type="sketch",
text=value,
session_id=row.get("session_id"),
timestamp=_as_datetime(row.get("updated_at")),
key=row.get("key"),
))
return hits
def _as_datetime(value) -> datetime | None:
"""Timestamps come back in two shapes: asyncpg returns `datetime`, the facts
store hands out ISO strings (`_facts._row_to_fact` formats for the API). The
freshness comparison runs across both, so normalize at the boundary rather
than letting `str <= datetime` blow up the whole search."""
if value is None or isinstance(value, datetime):
return value
try:
return datetime.fromisoformat(str(value))
except (TypeError, ValueError):
return None
def _mark_freshness(
facts: list[MemoryHit],
raw: list[MemoryHit],
summary_at: datetime | None,
summary_session_id: str | None,
) -> None:
"""Label raw excerpts that a newer curated fact has already overridden.
Only facts *newer* than the excerpt can supersede it — an old fact next to a
newer verbatim line is the lossy copy, not the truth. Overlap is judged on
content words, not on the user's query: the conflict has to be between the
two records themselves.
`summary_at` belongs to one session's summary; a chunk from another session
says nothing about it, so the comparison is scoped to that session.
"""
newest: MemoryHit | None = None
for fact in facts:
if fact.timestamp is None:
continue
if newest is None or fact.timestamp > newest.timestamp:
newest = fact
for hit in raw:
if (
summary_at is not None
and summary_session_id is not None
and hit.session_id == summary_session_id
and hit.timestamp is not None
):
hit.before_summary = hit.timestamp < summary_at
if newest is None or newest.timestamp is None or hit.timestamp is None:
continue
if newest.timestamp <= hit.timestamp:
continue
if _shared_terms(newest.text, hit.text) >= 2:
label = newest.key or "a newer fact"
hit.superseded_by = f"{label} (updated {_fmt_time(newest.timestamp)})"
def _shared_terms(left: str, right: str) -> int:
left_terms = {
t for t in _normalize_query(left)
if len(t) > 2 and t not in _STOPWORDS
}
right_terms = set(_normalize_query(right))
return len(left_terms & right_terms)
def _fmt_time(value: datetime | None) -> str:
if value is None:
return ""
try:
return value.strftime("%Y-%m-%d %H:%M")
except (AttributeError, ValueError):
return str(value)[:16]
def format_hits(hits: list[MemoryHit], query: str, scope: str) -> str:
"""Render search results for a tool result — provenance on every line."""
header = f'memory_search "{query}" (scope={scope}): {len(hits)} hit(s)'
if not hits:
return header + "\nNo matches in any store. Try fewer, different words."
lines = [header]
for hit in hits:
if hit.source_type == "fact":
where = f"fact [{hit.category or 'other'}]"
if hit.session_id:
where += f" · session {hit.session_id}"
if hit.timestamp is not None:
where += f" · updated {_fmt_time(hit.timestamp)}"
lines.append(f"\n• {where}\n {hit.text}")
continue
if hit.source_type == "raw":
turn = (
f"turn {hit.start_seq}" if hit.start_seq == hit.end_seq
else f"turns {hit.start_seq}–{hit.end_seq}"
)
if hit.end_seq < 0:
turn = "turn unknown"
where = f"raw · session {hit.session_id} · {turn}"
if hit.timestamp is not None:
where += f" · {_fmt_time(hit.timestamp)}"
lines.append(f"\n• {where}\n {_excerpt(hit.text)}")
if hit.superseded_by:
lines.append(
f" ⚠ outdated: a newer curated fact overrides this — {hit.superseded_by}. "
"Trust the fact; quote this only as history."
)
elif hit.before_summary:
lines.append(
" (verbatim record from before the current summary — exact details, "
"but the summary is the curated version.)"
)
if hit.breadcrumbs:
lines.append(f" keys: {', '.join(hit.breadcrumbs[:12])}")
continue
where = f"sketch · session {hit.session_id} · {hit.key or 'main'}"
if hit.timestamp is not None:
where += f" · updated {_fmt_time(hit.timestamp)}"
lines.append(f"\n• {where}\n {_excerpt(hit.text)}")
return "\n".join(lines)
def _excerpt(text: str) -> str:
text = (text or "").strip()
if len(text) <= _HIT_TEXT_CHARS:
return text
return text[:_HIT_TEXT_CHARS].rstrip() + " …"
