diff --git a/src/everos/memory/cascade/handlers/knowledge_topic.py b/src/everos/memory/cascade/handlers/knowledge_topic.py index 80e2cdfd..0f4604b6 100644 --- a/src/everos/memory/cascade/handlers/knowledge_topic.py +++ b/src/everos/memory/cascade/handlers/knowledge_topic.py @@ -29,6 +29,7 @@ from __future__ import annotations +import dataclasses import json from typing import Any, ClassVar @@ -91,8 +92,25 @@ async def handle_added_or_modified(self, md_path: str) -> HandlerOutcome: } ) + topic_payload = self._build_sqlite_payload(fields, md_path) + prior = await knowledge_topic_repo.get_by_id(fields["node_id"]) if prior is not None and prior.content_sha256 == digest: + rows = await knowledge_topic_sqlite_repo.get_topics_by_ids( + [fields["node_id"]] + ) + if not rows or not self._sqlite_row_matches_payload( + rows[0], + topic_payload, + ): + await knowledge_topic_sqlite_repo.upsert_from_handler(topic_payload) + return HandlerOutcome( + md_path=md_path, + kind=self.kind, + upserted=1, + deleted=0, + skipped=0, + ) return HandlerOutcome( md_path=md_path, kind=self.kind, @@ -104,7 +122,6 @@ async def handle_added_or_modified(self, md_path: str) -> HandlerOutcome: row = await self._build_lance_row(fields, digest, md_path) await knowledge_topic_repo.upsert([row]) - topic_payload = self._build_sqlite_payload(fields, md_path) await knowledge_topic_sqlite_repo.upsert_from_handler(topic_payload) return HandlerOutcome( @@ -115,6 +132,18 @@ async def handle_added_or_modified(self, md_path: str) -> HandlerOutcome: skipped=0, ) + def _sqlite_row_matches_payload( + self, + row: Any, + payload: TopicUpsertPayload, + ) -> bool: + """Return whether SQLite already mirrors the parsed topic payload.""" + for key, value in dataclasses.asdict(payload).items(): + current = row.get(key) if isinstance(row, dict) else getattr(row, key, None) + if current != value: + return False + return True + # ------------------------------------------------------------------ # Private helpers # ------------------------------------------------------------------ diff --git a/tests/unit/test_memory/test_cascade/test_handler_knowledge_topic.py b/tests/unit/test_memory/test_cascade/test_handler_knowledge_topic.py index cef0f659..8a2f5bb8 100644 --- a/tests/unit/test_memory/test_cascade/test_handler_knowledge_topic.py +++ b/tests/unit/test_memory/test_cascade/test_handler_knowledge_topic.py @@ -86,6 +86,9 @@ async def upsert_from_handler(self, payload: TopicUpsertPayload) -> None: self.upserts.append(data) self.rows[payload.node_id] = data + async def get_topics_by_ids(self, node_ids: list[str]) -> list[dict]: + return [self.rows[nid] for nid in node_ids if nid in self.rows] + async def delete_by_md_path(self, md_path: str) -> int: self.deletes.append(md_path) before = len(self.rows) @@ -230,7 +233,7 @@ async def test_handle_added_or_modified_upserts_to_both_stores( assert sq["md_path"] == md_path -async def test_same_digest_skips( +async def test_same_digest_skips_when_sqlite_matches( memory_root: MemoryRoot, fake_lance: _FakeLanceRepo, fake_sqlite: _FakeSqliteRepo, @@ -250,6 +253,54 @@ async def test_same_digest_skips( assert len(fake_sqlite.upserts) == 1 +async def test_same_digest_repairs_missing_sqlite_row( + memory_root: MemoryRoot, + fake_lance: _FakeLanceRepo, + fake_sqlite: _FakeSqliteRepo, +) -> None: + """If LanceDB is current but SQLite lost the row, rehydrate SQLite.""" + md_path = _write_topic_md(memory_root) + handler = _handler(memory_root) + + first = await handler.handle_added_or_modified(md_path) + assert first.upserted == 1 + + fake_sqlite.rows.clear() + fake_sqlite.upserts.clear() + + second = await handler.handle_added_or_modified(md_path) + + assert second.upserted == 1 + assert second.skipped == 0 + assert len(fake_lance.upserts) == 1 + assert len(fake_sqlite.upserts) == 1 + assert fake_sqlite.upserts[0]["node_id"] == "node_001" + + +async def test_same_digest_repairs_stale_sqlite_row( + memory_root: MemoryRoot, + fake_lance: _FakeLanceRepo, + fake_sqlite: _FakeSqliteRepo, +) -> None: + """If LanceDB is current but SQLite differs, refresh SQLite.""" + md_path = _write_topic_md(memory_root) + handler = _handler(memory_root) + + first = await handler.handle_added_or_modified(md_path) + assert first.upserted == 1 + + fake_sqlite.rows["node_001"]["summary"] = "stale summary" + fake_sqlite.upserts.clear() + + second = await handler.handle_added_or_modified(md_path) + + assert second.upserted == 1 + assert second.skipped == 0 + assert len(fake_lance.upserts) == 1 + assert len(fake_sqlite.upserts) == 1 + assert fake_sqlite.upserts[0]["summary"] == "Overview of budget planning practices." + + async def test_wrong_type_skips( memory_root: MemoryRoot, fake_lance: _FakeLanceRepo,