fix(memory/vector): skip a corrupt vec_metadata row instead of aborting the search

The sqlite-vec search loop rebuilt VectorMetadata inline with raw
datetime.fromisoformat(row["created_at"]) and json.loads(row["metadata_json"]).
A single row with an unparseable datetime or malformed JSON (schema drift across
an upgrade, a partially written row) raised mid-loop and aborted the whole
search, losing every other result. #2947 healed the entity_refs shape for this
exact reason but left created_at/metadata_json unguarded. Route row construction
through a _row_to_metadata helper that skips an unreadable row and normalizes
entity_refs, so one bad row no longer takes down the search.

Author: abhay-codes07

headroom/memory/adapters/sqlite_vector.py

@@ -648,6 +648,42 @@ async def remove_batch(self, memory_ids: list[str]) -> int:
                 conn.commit()
                 return len(rowids)
 
+    def _row_to_metadata(self, row: Any) -> VectorMetadata | None:
+        """Build ``VectorMetadata`` from a ``vec_metadata`` row, skipping a bad row.
+
+        A single row with an unparseable ``created_at`` / ``valid_until`` or
+        malformed ``entity_refs`` / ``metadata_json`` JSON (schema drift across an
+        upgrade, a partially written row) otherwise raises mid-loop and aborts
+        the *whole* search, losing every other result. #2947 healed the
+        ``entity_refs`` shape for exactly this reason ("took down whole memory
+        searches rather than the one bad row"), but ``created_at`` and
+        ``metadata_json`` in this loop were left unguarded. Return ``None`` so
+        the caller skips just the offending row.
+        """
+        try:
+            created_at = datetime.fromisoformat(row["created_at"])
+            valid_until = datetime.fromisoformat(row["valid_until"]) if row["valid_until"] else None
+            entity_refs = normalize_entity_refs(json.loads(row["entity_refs"]))
+            metadata = json.loads(row["metadata_json"])
+        except (ValueError, TypeError):
+            logger.warning(
+                "Skipping unreadable vec_metadata row (memory_id=%s)",
+                row["memory_id"],
+            )
+            return None
+        return VectorMetadata(
+            memory_id=row["memory_id"],
+            user_id=row["user_id"],
+            session_id=row["session_id"],
+            agent_id=row["agent_id"],
+            importance=row["importance"],
+            created_at=created_at,
+            valid_until=valid_until,
+            entity_refs=entity_refs,
+            content=row["content"],
+            metadata=metadata,
+        )
+
     async def search(self, filter: VectorFilter) -> list[VectorSearchResult]:
         """Search for similar vectors.
 
@@ -722,23 +758,11 @@ async def search(self, filter: VectorFilter) -> list[VectorSearchResult]:
                     if similarity < filter.min_similarity:
                         continue
 
-                    # Build metadata for filtering
-                    meta = VectorMetadata(
-                        memory_id=row["memory_id"],
-                        user_id=row["user_id"],
-                        session_id=row["session_id"],
-                        agent_id=row["agent_id"],
-                        importance=row["importance"],
-                        created_at=datetime.fromisoformat(row["created_at"]),
-                        valid_until=(
-                            datetime.fromisoformat(row["valid_until"])
-                            if row["valid_until"]
-                            else None
-                        ),
-                        entity_refs=json.loads(row["entity_refs"]),
-                        content=row["content"],
-                        metadata=json.loads(row["metadata_json"]),
-                    )
+                    # Build metadata for filtering. A single corrupt/legacy row
+                    # is skipped rather than aborting the whole search (#2947).
+                    meta = self._row_to_metadata(row)
+                    if meta is None:
+                        continue
 
                     # Apply filters
                     if not self._passes_filter(meta, filter):

tests/test_sqlite_vector_index.py

@@ -77,6 +77,36 @@ async def test_index_and_search(self, index):
         assert results[0].memory.id == memories[0].id
         assert results[0].similarity > 0.99  # Should be ~1.0 for exact match
 
+    @pytest.mark.asyncio
+    async def test_search_skips_corrupt_row_instead_of_aborting(self, index):
+        """A single vec_metadata row with an unparseable created_at (or malformed
+        metadata_json) must not abort the whole search. #2947 healed the
+        entity_refs shape for this reason, but created_at/metadata_json in the
+        same loop were unguarded, so one bad row lost every other result."""
+        np.random.seed(1)
+        good = Memory(
+            content="good", user_id="alice", embedding=np.random.randn(384).astype(np.float32)
+        )
+        bad = Memory(
+            content="bad", user_id="alice", embedding=np.random.randn(384).astype(np.float32)
+        )
+        await index.index(good)
+        await index.index(bad)
+
+        # Simulate a schema-drift / partially written row on disk.
+        conn = index._get_conn()
+        conn.execute(
+            "UPDATE vec_metadata SET created_at=? WHERE memory_id=?", ("not-a-date", bad.id)
+        )
+        conn.commit()
+
+        filter = VectorFilter(query_vector=good.embedding, top_k=5, user_id="alice")
+        results = await index.search(filter)  # must not raise
+
+        ids = {r.memory.id for r in results}
+        assert good.id in ids  # the good result survives
+        assert bad.id not in ids  # the corrupt row is skipped, not fatal
+
     @pytest.mark.asyncio
     async def test_true_delete(self, index):
         """Test that delete actually removes entries."""