"""Long-term memory. Memory is what lets a workflow resume, and what lets the AI know a document was chased three weeks ago. The tests below concentrate on the two properties that make it useful rather than decorative: recall is scoped and scored, and open gaps are answered definitely rather than by similarity. """ from __future__ import annotations import pytest from app.memory.embeddings import DIMENSIONS, HashingEmbedder, cosine_similarity from app.memory.records import MemoryKind, MemoryRecord from app.memory.repository import InMemoryRepository @pytest.fixture def repo() -> InMemoryRepository: return InMemoryRepository() class TestMemoryRecord: def test_a_blank_memory_is_refused(self): # An empty memory occupies a retrieval slot and tells a later run nothing. with pytest.raises(ValueError, match="needs content"): MemoryRecord(kind=MemoryKind.OBSERVATION, content=" ") def test_an_impossible_confidence_is_refused(self): with pytest.raises(ValueError, match="between 0 and 1"): MemoryRecord(kind=MemoryKind.DOCUMENT, content="x", confidence=1.4) def test_records_get_distinct_ids(self): a = MemoryRecord(kind=MemoryKind.OBSERVATION, content="one") b = MemoryRecord(kind=MemoryKind.OBSERVATION, content="two") assert a.id != b.id def test_a_gap_is_open_until_resolved(self): gap = MemoryRecord(kind=MemoryKind.MISSING, content="Bank statement not received") assert gap.is_open_gap gap.resolve() assert not gap.is_open_gap def test_resolving_keeps_the_record(self): # That a document was once outstanding is part of the account's history; # deleting it loses the fact that the AI chased it. gap = MemoryRecord(kind=MemoryKind.MISSING, content="CNIC not received") gap.resolve() assert gap.content == "CNIC not received" assert gap.resolved def test_only_missing_records_can_be_open_gaps(self): note = MemoryRecord(kind=MemoryKind.OBSERVATION, content="Client prefers WhatsApp") assert not note.is_open_gap class TestEmbedder: def test_it_produces_the_configured_dimensions(self): assert len(HashingEmbedder().embed("anything")) == DIMENSIONS def test_it_is_deterministic(self): embedder = HashingEmbedder() assert embedder.embed("bank statement") == embedder.embed("bank statement") def test_empty_text_gives_a_zero_vector(self): assert set(HashingEmbedder().embed("")) == {0.0} def test_repeated_text_is_more_similar_than_unrelated_text(self): embedder = HashingEmbedder() same = cosine_similarity( embedder.embed("bank statement for tax year 2024"), embedder.embed("bank statement for tax year 2024"), ) different = cosine_similarity( embedder.embed("bank statement for tax year 2024"), embedder.embed("client prefers to be contacted on WhatsApp"), ) assert same > different def test_word_order_is_partly_captured(self): # Bigrams are what stop "paid" and "not paid" collapsing together. embedder = HashingEmbedder() assert ( cosine_similarity(embedder.embed("invoice paid"), embedder.embed("invoice not paid")) < 1.0 ) def test_vectors_of_different_lengths_are_refused(self): with pytest.raises(ValueError, match="differ in length"): cosine_similarity([0.1, 0.2], [0.1, 0.2, 0.3]) class TestRecall: def test_it_returns_the_most_relevant_first(self, repo): repo.remember(MemoryRecord(kind=MemoryKind.DOCUMENT, content="Bank statement received")) repo.remember(MemoryRecord(kind=MemoryKind.PREFERENCE, content="Prefers WhatsApp contact")) results = repo.recall("bank statement") assert results[0][0].content == "Bank statement received" def test_it_returns_the_score_so_a_weak_match_can_be_judged(self, repo): # A bare list invites treating the top hit as relevant when everything # scored 0.02. repo.remember(MemoryRecord(kind=MemoryKind.DOCUMENT, content="Bank statement received")) _, score = repo.recall("bank statement")[0] assert 0.0 < score <= 1.0 def test_recall_can_be_scoped_to_one_client(self, repo): repo.remember(MemoryRecord(kind=MemoryKind.DOCUMENT, content="CNIC received", client_id=1)) repo.remember(MemoryRecord(kind=MemoryKind.DOCUMENT, content="CNIC received", client_id=2)) results = repo.recall("CNIC", client_id=1) assert len(results) == 1 assert results[0][0].client_id == 1 def test_filters_apply_before_ranking(self, repo): """Otherwise a scoped recall silently returns fewer results than asked for.""" for i in range(5): repo.remember( MemoryRecord(kind=MemoryKind.DOCUMENT, content=f"Bank statement {i}", client_id=99) ) repo.remember( MemoryRecord(kind=MemoryKind.DOCUMENT, content="Bank statement exact", client_id=1) ) results = repo.recall("bank statement", limit=3, client_id=99) assert len(results) == 3 assert all(r.client_id == 99 for r, _ in results) def test_recall_can_be_scoped_to_one_kind(self, repo): repo.remember(MemoryRecord(kind=MemoryKind.DOCUMENT, content="Statement filed")) repo.remember(MemoryRecord(kind=MemoryKind.DECISION, content="Statement filed under 2024")) results = repo.recall("statement", kind=MemoryKind.DECISION) assert all(r.kind is MemoryKind.DECISION for r, _ in results) def test_a_minimum_similarity_excludes_noise(self, repo): repo.remember(MemoryRecord(kind=MemoryKind.OBSERVATION, content="Entirely unrelated note")) assert repo.recall("bank statement", min_similarity=0.5) == [] def test_the_limit_is_honoured(self, repo): for i in range(10): repo.remember(MemoryRecord(kind=MemoryKind.DOCUMENT, content=f"Document {i}")) assert len(repo.recall("document", limit=3)) == 3 def test_recall_on_an_empty_store_returns_nothing(self, repo): assert repo.recall("anything") == [] class TestOpenGaps: def test_it_lists_what_is_still_missing(self, repo): repo.remember(MemoryRecord(kind=MemoryKind.MISSING, content="Bank statement", client_id=1)) repo.remember(MemoryRecord(kind=MemoryKind.DOCUMENT, content="CNIC received", client_id=1)) gaps = repo.open_gaps(client_id=1) assert [g.content for g in gaps] == ["Bank statement"] def test_a_resolved_gap_drops_out(self, repo): gap = repo.remember( MemoryRecord(kind=MemoryKind.MISSING, content="Salary slip", client_id=1) ) assert repo.open_gaps(client_id=1) gap.resolve() assert repo.open_gaps(client_id=1) == [] def test_gaps_are_oldest_first(self, repo): from datetime import UTC, datetime, timedelta older = MemoryRecord(kind=MemoryKind.MISSING, content="First", client_id=1) older.created_at = datetime.now(UTC) - timedelta(days=5) newer = MemoryRecord(kind=MemoryKind.MISSING, content="Second", client_id=1) repo.remember(newer) repo.remember(older) # The longest-outstanding request is the one most worth chasing. assert [g.content for g in repo.open_gaps(client_id=1)] == ["First", "Second"] def test_gaps_are_answered_definitely_not_by_similarity(self, repo): # Wording that matches no query term must still be returned — a gap # omitted because of phrasing is a document nobody chases. repo.remember( MemoryRecord(kind=MemoryKind.MISSING, content="zzzz unmatched phrasing", client_id=1) ) assert len(repo.open_gaps(client_id=1)) == 1 class TestLifecycle: def test_a_record_can_be_fetched_by_id(self, repo): record = repo.remember(MemoryRecord(kind=MemoryKind.OBSERVATION, content="Note")) assert repo.get(record.id) is record def test_forgetting_removes_the_record_and_its_vector(self, repo): record = repo.remember(MemoryRecord(kind=MemoryKind.OBSERVATION, content="Note")) assert repo.forget(record.id) assert repo.get(record.id) is None assert repo.recall("note") == [] def test_forgetting_something_absent_reports_it(self, repo): assert not repo.forget("no-such-id") class TestNoTenantConcept: """ADR-0001: one deployment per installation.""" def test_a_memory_record_has_no_tenant_field(self): # Its appearance would mean multi-tenancy had crept in — grounds to # reject the change, not a convenience. fields = MemoryRecord(kind=MemoryKind.OBSERVATION, content="x").__slots__ for forbidden in ("tenant_id", "customer_id", "installation_id", "firm_id"): assert forbidden not in fields def test_the_schema_declares_no_tenant_column(self): from pathlib import Path schema = Path(__file__).resolve().parents[1] / "app" / "memory" / "schema.sql" sql = schema.read_text(encoding="utf-8").lower() for forbidden in ("tenant_id", "customer_id", "firm_id"): assert forbidden not in sql, f"{forbidden} in schema — isolation must stay structural"