"""Raw chunks: building them from dropped messages, storing them once, finding them."""
from datetime import datetime, timezone
from navi.llm.base import Message
from navi.memory._chunks import _lexical_terms, _matched_terms, build_chunks
from tests.conftest_factory import FakeConnection, make_store_with_pool
def _msg(role: str, content: str, seq: int = 0, **kwargs) -> Message:
return Message(role=role, content=content, sequence_number=seq, **kwargs)
class TestBuildChunks:
def test_marks_summary_and_compression_rows_are_not_indexed(self):
messages = [
_msg("user", "hello", 1),
_msg("user", "old summary", 2, is_summary=True),
_msg("system", "compression marker", 3, is_compression=True),
_msg("system", "plain system", 4),
_msg("assistant", "hi", 5),
]
chunks = build_chunks(messages)
assert len(chunks) == 1
assert "old summary" not in chunks[0]["content"]
assert "compression marker" not in chunks[0]["content"]
assert "hello" in chunks[0]["content"]
assert "hi" in chunks[0]["content"]
def test_turns_are_packed_until_the_size_cap(self):
messages = [_msg("user", "x" * 60, index) for index in range(10)]
chunks = build_chunks(messages, max_chars=150)
assert len(chunks) == 5 # two messages per chunk
assert all(len(c["content"]) <= 150 for c in chunks)
assert [c["start_seq"] for c in chunks] == [0, 2, 4, 6, 8]
def test_the_turn_range_tracks_the_packed_messages(self):
messages = [_msg("user", "a", 7), _msg("assistant", "b", 8), _msg("tool", "c", 9, name="grep")]
chunk = build_chunks(messages)[0]
assert chunk["start_seq"] == 7
assert chunk["end_seq"] == 9
assert "Tool result (grep): c" in chunk["content"]
def test_a_huge_message_is_elided_in_the_middle(self):
body = "HEAD " + "x" * 20_000 + " TAIL"
chunk = build_chunks([_msg("tool", body, 1)])[0]
assert len(chunk["content"]) < 13_000
assert chunk["content"].startswith("Tool result: HEAD")
assert chunk["content"].endswith("TAIL")
assert "middle elided" in chunk["content"]
def test_timestamps_come_from_the_messages(self):
early = datetime(2026, 1, 1, tzinfo=timezone.utc)
late = datetime(2026, 1, 2, tzinfo=timezone.utc)
chunk = build_chunks([
_msg("user", "a", 1, created_at=early),
_msg("assistant", "b", 2, created_at=late),
])[0]
assert chunk["started_at"] == early
assert chunk["ended_at"] == late
def test_the_hash_is_stable_and_content_addressed(self):
first = build_chunks([_msg("user", "same", 1)])[0]
second = build_chunks([_msg("user", "same", 99)])[0]
assert first["content_hash"] == second["content_hash"]
def test_an_assistant_tool_request_without_text_is_recorded(self):
from navi.llm.base import ToolCallRequest
msg = Message(
role="assistant",
content="",
sequence_number=1,
tool_calls=[ToolCallRequest(id="1", name="grep", arguments={})],
)
chunks = build_chunks([msg])
assert chunks and "grep" in chunks[0]["content"]
class TestAddChunks:
async def test_chunks_are_inserted_with_scope_and_breadcrumbs(self):
conn = FakeConnection()
store = make_store_with_pool(conn)
store._pgvector_checked = True
store._pgvector_available = True
conn.enqueue("INSERT 0 1")
conn.enqueue("INSERT 0 1")
chunks = build_chunks(
[_msg("user", "first", 1), _msg("user", "second", 2)], max_chars=10
)
for chunk in chunks:
chunk["breadcrumbs"] = ["port 8099"]
written = await store.add_chunks(chunks, user_id="u1", session_id="s1")
assert written == 2
assert all("memory_chunks" in call[1] for call in conn.calls)
assert "ON CONFLICT (session_id, content_hash) DO NOTHING" in conn.calls[0][1]
assert conn.calls[0][2][1] == "u1"
assert conn.calls[0][2][2] == "s1"
assert conn.calls[0][2][8] == ["port 8099"]
async def test_without_pgvector_the_write_is_text_only(self):
# `::vector` would fail on a plain Postgres; the row must not name the column.
conn = FakeConnection()
store = make_store_with_pool(conn)
store._pgvector_checked = True
store._pgvector_available = False
conn.enqueue("INSERT 0 1")
written = await store.add_chunks(
build_chunks([_msg("user", "no vectors here", 1)]), session_id="s1"
)
assert written == 1
assert "embedding" not in conn.calls[0][1]
async def test_a_conflict_is_not_counted_as_written(self):
conn = FakeConnection()
store = make_store_with_pool(conn)
store._pgvector_checked = True
store._pgvector_available = True
conn.enqueue("INSERT 0 0")
written = await store.add_chunks(
build_chunks([_msg("user", "again", 1)]), session_id="s1"
)
assert written == 0
async def test_no_session_means_nothing_is_written(self):
conn = FakeConnection()
store = make_store_with_pool(conn)
assert await store.add_chunks(build_chunks([_msg("user", "x", 1)])) == 0
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()
store = make_store_with_pool(conn)
conn.enqueue([
{
"rank": 0.5, "id": "a", "session_id": "s1", "start_seq": 1, "end_seq": 2,
"started_at": None, "ended_at": datetime(2026, 1, 1),
"content": "the port is 8099 and the host is localhost",
"breadcrumbs": ["port"], "created_at": datetime(2026, 1, 1),
},
{
"rank": 0.1, "id": "b", "session_id": "s1", "start_seq": 3, "end_seq": 4,
"started_at": None, "ended_at": datetime(2026, 1, 1),
"content": "some port talk without the value",
"breadcrumbs": [], "created_at": datetime(2026, 1, 1),
},
])
hits = await store.search_chunks("port host localhost", user_id="u1", limit=5)
assert [h["id"] for h in hits] == ["a", "b"]
assert hits[0]["terms_hit"] == 3
assert hits[1]["terms_hit"] == 1
assert hits[0]["score"] is None
async def test_the_scope_filter_is_applied(self):
conn = FakeConnection()
store = make_store_with_pool(conn)
conn.enqueue([])
await store.search_chunks("q", user_id=None, session_id="s1", limit=3)
query, args = conn.calls[0][1], conn.calls[0][2]
assert "user_id IS NULL" in query
assert "session_id = $1" in query
assert args[0] == "s1"
async def test_an_empty_term_list_falls_back_to_recent(self):
conn = FakeConnection()
store = make_store_with_pool(conn)
conn.enqueue([
{
"id": "x", "session_id": "s1", "start_seq": 1, "end_seq": 1,
"started_at": None, "ended_at": None, "content": "body",
"breadcrumbs": [], "created_at": None,
},
])
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:
async def test_returns_the_timestamp(self):
conn = FakeConnection()
store = make_store_with_pool(conn)
when = datetime(2026, 2, 3, tzinfo=timezone.utc)
conn.enqueue(when)
assert await store.latest_summary_at("s1") == when
assert "is_summary = TRUE" in conn.calls[0][1]
async def test_a_failure_is_swallowed(self):
conn = FakeConnection()
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