diff --git a/docs/memory.md b/docs/memory.md
index 89bce1b..7bbb1a8 100644
--- a/docs/memory.md
+++ b/docs/memory.md
@@ -221,7 +221,7 @@
 
 ### Automatic recall
 
-`ContextBuilder._memory_recall_msg` injects up to `memory_recall_k` (2) raw excerpts **once per turn, only when the context already holds a summary** — that is the moment when dropped history matters. The query is built from the last four user/assistant messages plus the incoming one, so the search follows the current topic. Hits below `memory_recall_min_similarity` (0.55 vector, or 2 matched terms on the FTS path) and hits marked `superseded_by` are dropped. Delivered chunk ids are remembered in `session.session_metadata["injected_chunk_ids"]`, so the same excerpt does not ride in every prompt; the model can still call `memory_search` for anything wider.
+`ContextBuilder._memory_recall_msg` injects up to `memory_recall_k` (2) raw excerpts **once per turn, only when the context already holds a summary** — that is the moment when dropped history matters. The query is built from the last four user/assistant messages plus the incoming one, so the search follows the current topic. Hits below `memory_recall_min_similarity` (0.55 vector, or 2 matched terms on the FTS path) and hits marked `superseded_by` are dropped. Delivered chunk ids are remembered in `session.session_metadata["injected_chunk_ids"]`, so the same excerpt does not ride in every prompt; the ledger keeps the last 100 ids (it is stored in the session row and must not grow forever), so a very old excerpt can be recalled again later. The model can still call `memory_search` for anything wider. The freshness labels are computed even though recall asks for raw only: an excerpt a newer fact has overridden is never injected.
 
 Both behaviours are switchable: `MEMORY_SEARCH_ENABLED` (indexing demoted messages) and `MEMORY_RECALL_ENABLED` (the automatic injection), see `docs/config.md`.
 
diff --git a/navi/core/compressor.py b/navi/core/compressor.py
index ac68257..e7cc534 100644
--- a/navi/core/compressor.py
+++ b/navi/core/compressor.py
@@ -1196,10 +1196,21 @@
         # context at every reload (the half-the-context-is-summaries symptom).
         new_context_ids = {id(m) for m in new_context}
         demoted: list[Message] = []
+        seen: set[int] = set()
         for msg in [*session.messages, *session.context]:
-            if msg.role != "system" and id(msg) not in new_context_ids:
-                msg.is_context = False
-                demoted.append(msg)
+            if msg.role == "system" or id(msg) in new_context_ids or id(msg) in seen:
+                continue
+            seen.add(id(msg))
+            if not msg.is_context:
+                # Dropped by an earlier compression. Iterating the display
+                # history surfaces it again on every later pass; indexing it
+                # again writes the same turns under a new chunk boundary (new
+                # hash) instead of once, forever. The current summary covers
+                # it; if it was dropped while indexing was off, that is the
+                # one turn range this trade-off leaves unindexed.
+                continue
+            msg.is_context = False
+            demoted.append(msg)
 
         # Rows that live only in session.context — a summary is is_display=False,
         # so after a reload it is absent from session.messages — are invisible to
diff --git a/navi/core/context_builder.py b/navi/core/context_builder.py
index fa73857..41403cc 100644
--- a/navi/core/context_builder.py
+++ b/navi/core/context_builder.py
@@ -121,6 +121,9 @@
     return out
 
 
+_INJECTED_IDS_LIMIT = 100
+
+
 def _recall_query(context: list, user_message: str, max_chars: int = 1200) -> str:
     """A search query made of the recent turn window, not just the last message.
 
@@ -489,6 +492,13 @@
         if injected_ids is not None:
             for hit in selected:
                 injected_ids.append(hit.chunk_id)
+            # The ledger lives in `session_metadata`, which is JSON-encoded into
+            # the sessions row on every save — an unbounded list grows the row
+            # for the life of the session. Older ids are dropped; a chunk that
+            # falls out after ~100 injections may be recalled again, which is
+            # a much smaller cost than a row that never stops growing.
+            if len(injected_ids) > _INJECTED_IDS_LIMIT:
+                del injected_ids[: len(injected_ids) - _INJECTED_IDS_LIMIT]
 
         lines = [
             "## Recovered from earlier in this conversation",
diff --git a/navi/core/pg_session_store.py b/navi/core/pg_session_store.py
index e1e221d..816fcc2 100644
--- a/navi/core/pg_session_store.py
+++ b/navi/core/pg_session_store.py
@@ -1007,13 +1007,22 @@
             result = await conn.execute("DELETE FROM sessions WHERE id = $1", session_id)
             # Raw chunks have no FK on purpose (memory DDL may run before the
             # sessions table exists), so deleting a chat must clean them here —
-            # otherwise a deleted conversation stays searchable.
+            # otherwise a deleted conversation stays searchable. Best-effort:
+            # the session row is already gone, and a failure here must not turn
+            # a successful delete into a 500 (or leave the caller thinking the
+            # delete did not happen).
             try:
                 await conn.execute(
                     "DELETE FROM memory_chunks WHERE session_id = $1", session_id
                 )
             except asyncpg.UndefinedTableError:
                 pass
+            except Exception:
+                structlog.get_logger().warning(
+                    "session.chunk_cleanup_failed",
+                    session_id=session_id,
+                    exc_info=True,
+                )
         return pending is not None or result == "DELETE 1"
 
     async def get_or_create_share(self, session_id: str, owner_id: str | None) -> dict | None:
diff --git a/navi/memory/_chunks.py b/navi/memory/_chunks.py
index 47088da..f5c79ca 100644
--- a/navi/memory/_chunks.py
+++ b/navi/memory/_chunks.py
@@ -26,6 +26,7 @@
 from __future__ import annotations
 
 import hashlib
+import re
 import uuid
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING
@@ -33,7 +34,7 @@
 import structlog
 
 from ._embeddings import _vector_to_str
-from ._facts import _normalize_query
+from ._facts import STOPWORDS, _normalize_query
 
 if TYPE_CHECKING:
     from navi.llm.base import Message
@@ -56,6 +57,13 @@
 # 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",
@@ -238,14 +246,30 @@
                         vec_str, user_id, session_id, limit,
                     )
                 except Exception:
-                    log.warning("memory.chunk_vector_failed", query=query[:40], exc_info=True)
-        if rows:
-            log.debug("memory.chunk_vector_search", hits=len(rows), query=query[:40])
+                    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
-        rows = await self._lexical_chunks(query, user_id, session_id, limit)
-        if rows:
-            log.debug("memory.chunk_lexical_search", hits=len(rows), query=query[:40])
-        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,
@@ -290,7 +314,7 @@
         session_id: str | None,
         limit: int,
     ) -> list[dict]:
-        terms = _normalize_query(query)
+        terms = _lexical_terms(query)
         where, params = _chunk_filter(user_id, session_id)
         if not terms:
             return await self._recent_chunks(where, params, limit)
@@ -321,11 +345,11 @@
                 *params,
             )
         hits = []
-        normalized_terms = {t.lower() for t in terms if len(t) > 2}
+        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'])}".lower()
-            hit["terms_hit"] = sum(1 for term in normalized_terms if term in haystack)
+            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
@@ -345,9 +369,12 @@
                 *params, limit,
             )
         hits = []
-        for index, row in enumerate(rows):
+        for row in rows:
             hit = _row_to_chunk(row)
-            hit["score"] = 1.0 - index * 0.05
+            # 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
@@ -358,7 +385,7 @@
         pool = await self._get_pool()
         async with pool.acquire() as conn:
             try:
-                return await conn.fetchval(
+                hot = await conn.fetchval(
                     "SELECT max(created_at) FROM session_messages "
                     "WHERE session_id = $1 AND is_summary = TRUE",
                     session_id,
@@ -366,6 +393,20 @@
             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)
@@ -376,6 +417,33 @@
             ) 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 = []
diff --git a/navi/memory/_embeddings.py b/navi/memory/_embeddings.py
index 2bf02fa..d560dde 100644
--- a/navi/memory/_embeddings.py
+++ b/navi/memory/_embeddings.py
@@ -75,7 +75,7 @@
             if vectors and vectors[0]:
                 return vectors[0]
         except Exception:
-            log.warning("memory.embed_failed", text=text[:60], exc_info=True)
+            log.warning("memory.embed_failed", text_len=len(text), exc_info=True)
         return None
 
     async def _generate_embeddings(self, texts: list[str]) -> list[list[float] | None]:
diff --git a/navi/memory/_facts.py b/navi/memory/_facts.py
index f9be273..6639e7a 100644
--- a/navi/memory/_facts.py
+++ b/navi/memory/_facts.py
@@ -19,6 +19,26 @@
 _AUTO_DUMP_THRESHOLD = 60
 _VECTOR_CUTOFF_DISTANCE = 0.3
 
+# Words that carry no retrieval signal. They are not removed by
+# `_normalize_query` — fact keys and sketch text legitimately contain short
+# words ("was", "the") and the ILIKE paths are cheap enough at that scale — but
+# anything that ranks or gates matches (chunk overlap, freshness conflicts)
+# must ignore them, or "was" in one line and "other" in another count as
+# agreement.
+STOPWORDS = frozenset({
+    # Short function words survive `_normalize_query` (len > 1) and would
+    # otherwise become ILIKE patterns and "matched terms" on their own.
+    "a", "i", "of", "in", "to", "is", "it", "on", "at", "by", "or", "as", "an",
+    "be", "we", "if", "so", "no", "do", "up", "my", "me", "us", "am", "go",
+    "all", "any", "one", "out", "off", "per", "via", "vs",
+    "the", "and", "for", "was", "were", "that", "this", "with", "from", "have",
+    "has", "had", "not", "but", "you", "your", "our", "are", "what", "when",
+    "where", "which", "how", "why", "did", "does", "can", "could", "would",
+    "should", "about", "there", "then", "than", "into", "they", "them", "its",
+    "это", "что", "как", "где", "когда", "был", "была", "были", "для", "или",
+    "там", "тут", "уже", "ещё", "еще", "если", "чтобы", "только", "можно",
+})
+
 # Leading tokens that carry no meaning of their own: `preferred_language`,
 # `ui_language` and `language` are one key, and the extractor writes all three.
 _MERGE_PREFIX_TOKENS = frozenset({"preferred", "preference", "pref", "user", "my", "ui", "response"})
@@ -195,6 +215,7 @@
                             rows = await conn.fetch(
                                 """SELECT id, category, key, value, updated_at,
                                           source, confidence, expires_at, source_context, merged_into,
+                                          source_session_id,
                                           embedding <=> $1::vector AS distance
                                    FROM memory_facts
                                    WHERE (expires_at IS NULL OR expires_at > now())
@@ -208,6 +229,7 @@
                             rows = await conn.fetch(
                                 """SELECT id, category, key, value, updated_at,
                                           source, confidence, expires_at, source_context, merged_into,
+                                          source_session_id,
                                           embedding <=> $1::vector AS distance
                                    FROM memory_facts
                                    WHERE (expires_at IS NULL OR expires_at > now())
@@ -223,15 +245,15 @@
                             if r["distance"] < _VECTOR_CUTOFF_DISTANCE
                         ]
                         if results:
-                            log.debug("memory.vector_search", hits=len(results), query=query[:40])
+                            log.debug("memory.vector_search", hits=len(results), query_len=len(query))
                             return results
                 except Exception:
-                    log.warning("memory.vector_search_failed", query=query[:40], exc_info=True)
+                    log.warning("memory.vector_search_failed", query_len=len(query), exc_info=True)
             else:
-                log.warning("memory.vector_fallback_to_ilike", reason="embed_failed", query=query[:40])
+                log.warning("memory.vector_fallback_to_ilike", reason="embed_failed", query_len=len(query))
         else:
             reason = "no_embedding_backend" if not self._embedding_backend else "no_pgvector"
-            log.warning("memory.vector_fallback_to_ilike", reason=reason, query=query[:40])
+            log.warning("memory.vector_fallback_to_ilike", reason=reason, query_len=len(query))
 
         # 2. Text fallback (original ILIKE logic)
         terms = _normalize_query(query)
@@ -266,7 +288,7 @@
         async with pool.acquire() as conn:
             rows = await conn.fetch(
                 f"SELECT id, category, key, value, updated_at, source, confidence, "
-                f"expires_at, source_context, merged_into FROM memory_facts "
+                f"expires_at, source_context, merged_into, source_session_id FROM memory_facts "
                 f"WHERE {user_filter} AND {and_where} ORDER BY updated_at DESC LIMIT ${limit_idx}",
                 *base_params, limit,
             )
@@ -278,7 +300,8 @@
             )
             rows = await conn.fetch(
                 f"SELECT id, category, key, value, updated_at, source, confidence, "
-                f"expires_at, source_context, merged_into, ({score_expr}) AS score "
+                f"expires_at, source_context, merged_into, source_session_id, "
+                f"({score_expr}) AS score "
                 f"FROM memory_facts WHERE {user_filter} AND ({or_where}) "
                 f"ORDER BY score DESC, updated_at DESC LIMIT ${limit_idx}",
                 *base_params, limit,
@@ -349,7 +372,7 @@
     ) -> list[dict]:
         q = (
             "SELECT id, category, key, value, updated_at, source, confidence, "
-            "expires_at, source_context, merged_into FROM memory_facts "
+            "expires_at, source_context, merged_into, source_session_id FROM memory_facts "
         )
         params: list = []
         param_idx = 0
@@ -459,4 +482,7 @@
         "expires_at": expires_str,
         "source_context": row.get("source_context", ""),
         "merged_into": row.get("merged_into"),
+        # Provenance: which conversation taught this fact. `memory_search`
+        # surfaces it; older rows may predate the column's use and carry NULL.
+        "source_session_id": row.get("source_session_id"),
     }
diff --git a/navi/memory/search.py b/navi/memory/search.py
index db650b3..e326caa 100644
--- a/navi/memory/search.py
+++ b/navi/memory/search.py
@@ -28,6 +28,7 @@
 
 import structlog
 
+from ._facts import STOPWORDS as _STOPWORDS
 from ._facts import _normalize_query
 
 log = structlog.get_logger()
@@ -35,14 +36,6 @@
 SCOPES = ("facts", "raw", "sketches", "all")
 _MAX_K = 20
 _HIT_TEXT_CHARS = 1200
-_STOPWORDS = frozenset({
-    "the", "and", "for", "was", "were", "that", "this", "with", "from", "have",
-    "has", "had", "not", "but", "you", "your", "our", "are", "what", "when",
-    "where", "which", "how", "why", "did", "does", "can", "could", "would",
-    "should", "about", "there", "then", "than", "into", "they", "them", "its",
-    "это", "что", "как", "где", "когда", "был", "была", "были", "для", "или",
-    "там", "тут", "уже", "ещё", "еще", "если", "чтобы", "только", "можно",
-})
 
 
 @dataclass
@@ -108,18 +101,29 @@
         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(facts, raw, summary_at, 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=query[:60],
+            query_len=len(query),
             scope=scope,
             facts=len(facts),
             raw=len(raw),
@@ -133,7 +137,7 @@
         try:
             rows = await self._memory.search_facts(query, user_id=user_id, limit=k)
         except Exception:
-            log.warning("memory_search.facts_failed", query=query[:40], exc_info=True)
+            log.warning("memory_search.facts_failed", query_len=len(query), exc_info=True)
             return []
         hits = []
         for row in rows:
@@ -158,7 +162,7 @@
                 query, user_id=user_id, session_id=session_id, limit=k,
             )
         except Exception:
-            log.warning("memory_search.raw_failed", query=query[:40], exc_info=True)
+            log.warning("memory_search.raw_failed", query_len=len(query), exc_info=True)
             return []
         hits = []
         for row in rows:
@@ -191,7 +195,7 @@
                 prefer_session_id=prefer_session_id,
             )
         except Exception:
-            log.warning("memory_search.sketches_failed", query=query[:40], exc_info=True)
+            log.warning("memory_search.sketches_failed", query_len=len(query), exc_info=True)
             return []
         hits = []
         for row in rows:
diff --git a/tests/unit/core/test_agent.py b/tests/unit/core/test_agent.py
index 432a1b4..cf2ff47 100644
--- a/tests/unit/core/test_agent.py
+++ b/tests/unit/core/test_agent.py
@@ -1514,3 +1514,51 @@
         # json-encodes on every save — a set there would break persistence.
         assert session.session_metadata["injected_chunk_ids"] == ["c1"]
         json.dumps(session.session_metadata)
+
+    @pytest.mark.asyncio
+    async def test_the_injected_ledger_stays_bounded(self, monkeypatch):
+        from datetime import datetime, timezone
+
+        from navi.core.context_builder import _INJECTED_IDS_LIMIT
+        from navi.memory.search import MemoryHit
+
+        class FakeSearch:
+            async def search(self, query, **kwargs):
+                return [MemoryHit(
+                    source_type="raw", text="the deploy port is 8099",
+                    session_id="s1", start_seq=3, end_seq=4, chunk_id="c1", score=0.9,
+                    timestamp=datetime(2026, 11, 9, tzinfo=timezone.utc),
+                )]
+
+            async def latest_summary_at(self, session_id):
+                return None
+
+        sessions = InMemorySessionStore()
+        profiles = ProfileRegistry()
+        profiles.register(make_profile("test"))
+        backends = BackendRegistry()
+        backends.register("ollama", FakeLLMBackend(responses=["hello"]))
+        agent = Agent(
+            session_store=sessions,
+            profile_registry=profiles,
+            tool_registry=make_registry_with_tools(),
+            backend_registry=backends,
+            memory_search=FakeSearch(),
+        )
+        session = await sessions.create(profile_id="test")
+        session.id = "s1"
+        session.context.append(Message(role="user", content="summary", is_summary=True))
+        # More history than the ledger keeps: the oldest entries must fall out
+        # rather than growing the persisted `session_metadata` forever.
+        session.session_metadata = {
+            "injected_chunk_ids": [f"old{i}" for i in range(_INJECTED_IDS_LIMIT)]
+        }
+        await sessions.save(session)
+
+        await agent.run(session.id, "what was the port?")
+
+        ids = session.session_metadata["injected_chunk_ids"]
+        assert len(ids) == _INJECTED_IDS_LIMIT
+        assert ids[-1] == "c1"
+        assert "old0" not in ids
+        json.dumps(session.session_metadata)
diff --git a/tests/unit/core/test_compression_breadcrumbs.py b/tests/unit/core/test_compression_breadcrumbs.py
index 43c9606..cc081aa 100644
--- a/tests/unit/core/test_compression_breadcrumbs.py
+++ b/tests/unit/core/test_compression_breadcrumbs.py
@@ -6,6 +6,7 @@
 import navi.core.compressor as _cm
 from navi.config import Settings
 from navi.core.compressor import ContextCompressor, compress_context
+from navi.core.session import InMemorySessionStore
 from navi.llm.base import Message
 from tests.conftest_factory import FakeLLMBackend
 
@@ -122,3 +123,57 @@
         store = FakeChunkStore()
         await ContextCompressor()._index_demoted(store, SimpleNamespace(user_id="u1"), "s1", [], None)
         assert store.calls == []
+
+
+class TestDemotedTurnsAreIndexedOnce:
+    """`session.messages` and `session.context` share Message objects (that is how
+    `PgSessionStore.get` builds them), and a turn dropped by one compression stays
+    in the display history forever. Demoting the union indexed every dropped turn
+    twice in the same pass and again on every later pass, under a new chunk
+    boundary each time — duplicate rows and repeated embedding work."""
+
+    @staticmethod
+    def _turns(start: int, count: int) -> list[Message]:
+        out: list[Message] = []
+        for i in range(count):
+            out.append(Message(role="user", content=f"turn-{start + i} question",
+                               sequence_number=(start + i) * 2))
+            out.append(Message(role="assistant", content=f"turn-{start + i} answer",
+                               sequence_number=(start + i) * 2 + 1))
+        return out
+
+    async def test_a_dropped_turn_is_written_once_and_never_reindexed(self):
+        store = FakeChunkStore()
+        sessions = InMemorySessionStore()
+        session = await sessions.create(profile_id="test")
+        session.context = self._turns(0, 8)
+        session.messages = list(session.context)
+        compressor = ContextCompressor()
+        backend = FakeLLMBackend(responses=["summary one", "summary two"])
+
+        async def compress():
+            return await compressor.compress_and_save_session(
+                session=session,
+                session_store=sessions,
+                llm=backend,
+                model="m",
+                temperature=0.3,
+                session_id=session.id,
+                reason="pytest",
+                keep_recent=1,
+                window=65_536,
+                memory_store=store,
+            )
+
+        await compress()
+        first_text = "\n".join(c["content"] for c in store.calls[0][0])
+        assert first_text.count("turn-0 question") == 1
+
+        added = self._turns(8, 4)
+        session.context.extend(added)
+        session.messages.extend(added)
+        await compress()
+        second_text = "\n".join(c["content"] for c in store.calls[1][0])
+        assert "turn-8 question" in second_text
+        assert "turn-0 question" not in second_text
+        assert second_text.count("turn-8 question") == 1
diff --git a/tests/unit/core/test_pg_session_store.py b/tests/unit/core/test_pg_session_store.py
index 27c7958..293f937 100644
--- a/tests/unit/core/test_pg_session_store.py
+++ b/tests/unit/core/test_pg_session_store.py
@@ -661,3 +661,28 @@
 
     # Every table was still created; only the index was given up on.
     assert len(conn.calls) == 4
+
+
+@pytest.mark.asyncio
+async def test_delete_cleans_the_chat_chunks():
+    conn = FakeConnection()
+    conn.enqueue("DELETE 1")
+    conn.enqueue("DELETE 3")
+    store = PgSessionStore(FakePool(conn))
+    store._initialized = True
+
+    assert await store.delete("s1") is True
+    assert any("memory_chunks" in call[1] for call in conn.calls)
+
+
+@pytest.mark.asyncio
+async def test_delete_succeeds_even_if_chunk_cleanup_fails():
+    # The session row is already gone; a cleanup failure must not turn a
+    # successful delete into a 500 (nor imply the delete did not happen).
+    conn = FakeConnection()
+    conn.enqueue("DELETE 1")
+    conn.enqueue(RuntimeError("chunk cleanup blew up"))
+    store = PgSessionStore(FakePool(conn))
+    store._initialized = True
+
+    assert await store.delete("s1") is True
diff --git a/tests/unit/memory/test_chunks.py b/tests/unit/memory/test_chunks.py
index 8570182..2ad0ea6 100644
--- a/tests/unit/memory/test_chunks.py
+++ b/tests/unit/memory/test_chunks.py
@@ -3,7 +3,7 @@
 from datetime import datetime, timezone
 
 from navi.llm.base import Message
-from navi.memory._chunks import build_chunks
+from navi.memory._chunks import _lexical_terms, _matched_terms, build_chunks
 from tests.conftest_factory import FakeConnection, make_store_with_pool
 
 
@@ -129,6 +129,25 @@
         assert conn.calls == []
 
 
+class TestLexicalTerms:
+    """The chunk query is scored, so its terms must be content words only."""
+
+    def test_function_words_are_not_terms(self):
+        assert _lexical_terms("the port was on the other host") == ["port", "other", "host"]
+
+    def test_matching_is_word_boundary_not_substring(self):
+        # "the" inside "other" used to count as a matched query term.
+        assert _matched_terms("other things", {"the"}) == 0
+        assert _matched_terms("the port", {"the", "port"}) == 2
+
+    def test_terms_are_capped_and_deduplicated(self):
+        query = "port " * 5 + " ".join(f"word{i}" for i in range(30))
+        terms = _lexical_terms(query)
+        assert len(terms) == 12
+        assert terms[0] == "port"
+        assert terms.count("port") == 1
+
+
 class TestSearchChunks:
     async def test_lexical_fallback_ranks_by_matched_terms(self):
         conn = FakeConnection()
@@ -175,6 +194,55 @@
         ])
         hits = await store.search_chunks("!!!", user_id="u1", limit=2)
         assert hits[0]["id"] == "x"
+        # "Most recent" is not "most similar": a fabricated 1.0 used to pass
+        # recall's similarity threshold and inject an unrelated turn.
+        assert hits[0]["score"] is None
+
+    async def test_a_short_vector_result_is_topped_up_lexically(self):
+        # A chunk with a NULL embedding is invisible to `ORDER BY embedding <=>`
+        # — it must still be reachable through the lexical path.
+        class Backend:
+            async def embed(self, texts, model=None, route=None):
+                return [[1.0, 0.0]]
+
+        conn = FakeConnection()
+        store = make_store_with_pool(conn)
+        store._pgvector_checked = True
+        store._pgvector_available = True
+        store._embedding_backend = Backend()
+        conn.enqueue([{
+            "distance": 0.1, "id": "vec", "session_id": "s1", "start_seq": 1,
+            "end_seq": 2, "started_at": None, "ended_at": None,
+            "content": "vector match", "breadcrumbs": [], "created_at": None,
+        }])
+        conn.enqueue([{
+            "rank": 0.2, "id": "lex", "session_id": "s1", "start_seq": 3,
+            "end_seq": 4, "started_at": None, "ended_at": None,
+            "content": "lexical match port", "breadcrumbs": [], "created_at": None,
+        }])
+
+        hits = await store.search_chunks("port", user_id="u1", limit=2)
+        assert [h["id"] for h in hits] == ["vec", "lex"]
+
+    async def test_a_full_vector_result_skips_the_lexical_query(self):
+        class Backend:
+            async def embed(self, texts, model=None, route=None):
+                return [[1.0, 0.0]]
+
+        conn = FakeConnection()
+        store = make_store_with_pool(conn)
+        store._pgvector_checked = True
+        store._pgvector_available = True
+        store._embedding_backend = Backend()
+        conn.enqueue([{
+            "distance": 0.1, "id": "vec", "session_id": "s1", "start_seq": 1,
+            "end_seq": 2, "started_at": None, "ended_at": None,
+            "content": "vector match", "breadcrumbs": [], "created_at": None,
+        }])
+
+        hits = await store.search_chunks("port", user_id="u1", limit=1)
+        assert [h["id"] for h in hits] == ["vec"]
+        assert len(conn.calls) == 1
 
 
 class TestLatestSummaryAt:
@@ -191,3 +259,23 @@
         store = make_store_with_pool(conn)
         conn.enqueue(RuntimeError("boom"))
         assert await store.latest_summary_at("s1") is None
+
+    async def test_the_archive_is_consulted_too(self):
+        # Long sessions archive old rows, summaries included; the hot table then
+        # has no summary at all and `before_summary` would be lost.
+        conn = FakeConnection()
+        store = make_store_with_pool(conn)
+        hot = datetime(2026, 2, 1, tzinfo=timezone.utc)
+        archived = datetime(2026, 2, 3, tzinfo=timezone.utc)
+        conn.enqueue(hot)
+        conn.enqueue(archived)
+        assert await store.latest_summary_at("s1") == archived
+        assert "session_messages_archive" in conn.calls[1][1]
+
+    async def test_a_missing_archive_table_keeps_the_hot_value(self):
+        conn = FakeConnection()
+        store = make_store_with_pool(conn)
+        hot = datetime(2026, 2, 1, tzinfo=timezone.utc)
+        conn.enqueue(hot)
+        conn.enqueue(RuntimeError("no such table"))
+        assert await store.latest_summary_at("s1") == hot
diff --git a/tests/unit/memory/test_memory_search.py b/tests/unit/memory/test_memory_search.py
index 465a812..186095e 100644
--- a/tests/unit/memory/test_memory_search.py
+++ b/tests/unit/memory/test_memory_search.py
@@ -136,6 +136,18 @@
         assert hit.superseded_by is not None
         assert "server_port" in hit.superseded_by
 
+    async def test_raw_scope_still_labels_a_newer_fact(self):
+        # Recall asks for raw only, but an outdated excerpt shown as
+        # authoritative is worse than the extra facts query. The facts are used
+        # for the label and are not returned.
+        memory = FakeMemory(
+            facts=[_fact("server_port", "8099 is the server port", updated=NOW)],
+            chunks=[_chunk("the server port was 8098", ended=NOW - timedelta(days=3))],
+        )
+        hits = await UnifiedMemorySearch(memory, FakeKv()).search("server port", scope="raw")
+        assert [h.source_type for h in hits] == ["raw"]
+        assert hits[0].superseded_by is not None
+
     async def test_an_older_fact_does_not_supersede_a_newer_excerpt(self):
         memory = FakeMemory(
             facts=[_fact("server_port", "8099 is the server port", updated=NOW - timedelta(days=3))],
diff --git a/tests/unit/memory/test_store.py b/tests/unit/memory/test_store.py
index 3b17049..be51068 100644
--- a/tests/unit/memory/test_store.py
+++ b/tests/unit/memory/test_store.py
@@ -368,6 +368,28 @@
         # Should have done get_all_facts instead of ILIKE
         assert "ORDER BY category" in conn.calls[1][1]
 
+    async def test_a_fact_keeps_its_source_session(self):
+        # `memory_search` shows "fact · session …"; without this column the
+        # provenance was silently None for every fact on the vector and ILIKE
+        # paths, while the tests' fakes supplied it by hand.
+        conn = FakeConnection()
+        conn.enqueue([FakeRecord(id="1", category="technical", key="server_port",
+                                  value="8099", updated_at=None, source="conversation",
+                                  confidence=90, expires_at=None, source_context="",
+                                  source_session_id="sess-42", distance=0.1)])
+        store = make_store_with_pool(conn)
+        store._pgvector_checked = True
+        store._pgvector_available = True
+        store._embedding_backend = object()
+
+        async def _fake_embed(text: str):
+            return [0.1] * 8
+
+        store._generate_embedding = _fake_embed
+        results = await store.search_facts("port", limit=5)
+        assert results[0]["source_session_id"] == "sess-42"
+        assert "source_session_id" in conn.calls[0][1]
+
     async def test_fallback_no_terms(self):
         conn = FakeConnection()
         conn.enqueue([])  # get_all_facts returns empty