fix(cascade): repair knowledge topic SQLite parity

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
     # ------------------------------------------------------------------

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,