fix(core): handle legacy pgvector storage status

src/basic_memory/services/project_service.py

@@ -1094,7 +1094,8 @@ async def get_embedding_status(self, project_id: int) -> EmbeddingStatus:
         if is_postgres:
             table_check_sql = text(
                 "SELECT table_name FROM information_schema.tables "
-                "WHERE table_name IN ('search_vector_chunks', 'search_vector_embeddings')"
+                "WHERE table_schema = ANY (current_schemas(false)) "
+                "AND table_name IN ('search_vector_chunks', 'search_vector_embeddings')"
             )
         else:
             table_check_sql = text(
@@ -1108,9 +1109,6 @@ async def get_embedding_status(self, project_id: int) -> EmbeddingStatus:
             existing_vector_tables = {str(name) for name in table_result.scalars().all()}
             manifest_exists = "search_vector_chunks" in existing_vector_tables
             storage_exists = "search_vector_embeddings" in existing_vector_tables
-            vector_tables_exist = manifest_exists and (
-                storage_exists or not uses_builtin_vector_storage
-            )
 
             manifest_schema_current = manifest_exists
             if manifest_exists and not is_postgres:
@@ -1126,6 +1124,23 @@ async def get_embedding_status(self, project_id: int) -> EmbeddingStatus:
                     "embedding_status",
                 }.issubset(manifest_columns)
 
+            storage_schema_current = storage_exists
+            if storage_exists and is_postgres and vector_index == "pgvector":
+                columns_result = await self.repository.execute_query(
+                    session,
+                    text(
+                        "SELECT 1 FROM pg_attribute "
+                        "WHERE attrelid = 'search_vector_embeddings'::regclass "
+                        "AND attname = 'source_hash'"
+                    ),
+                    {},
+                )
+                storage_schema_current = columns_result.scalar_one_or_none() is not None
+
+            vector_tables_exist = manifest_exists and (
+                not uses_builtin_vector_storage or (storage_exists and storage_schema_current)
+            )
+
             if not manifest_schema_current or not vector_tables_exist:
                 # Count distinct entities in search index for the recommendation message
                 si_result = await self.repository.execute_query(
@@ -1142,6 +1157,12 @@ async def get_embedding_status(self, project_id: int) -> EmbeddingStatus:
                     reindex_reason = (
                         "Vector manifest schema is outdated — run: bm reindex --embeddings"
                     )
+                elif storage_exists and not storage_schema_current:
+                    # Legacy pgvector tables are repaired by index initialization. Status is a
+                    # read path, so it only reports the required rebuild instead of mutating data.
+                    reindex_reason = (
+                        "Vector storage schema is outdated — run: bm reindex --embeddings"
+                    )
                 elif manifest_schema_current:
                     reindex_reason = "Vector storage not initialized — run: bm reindex --embeddings"
                 else:

tests/services/test_project_service_embedding_status.py

@@ -195,6 +195,62 @@ async def test_embedding_status_treats_legacy_sqlite_manifest_as_unavailable(
     assert "schema is outdated" in (status.reindex_reason or "")
 
 
+@pytest.mark.asyncio
+async def test_embedding_status_treats_legacy_pgvector_storage_as_unavailable(
+    project_service: ProjectService,
+    test_graph,
+    test_project,
+):
+    """Legacy pgvector storage should recommend rebuild before querying source_hash."""
+    if not _is_postgres():
+        pytest.skip("The pgvector physical storage schema only applies to Postgres.")
+
+    await _create_embeddings_stub(project_service)
+    await _execute(project_service, text("CREATE SCHEMA embedding_status_shadow"), {})
+    await _execute(
+        project_service,
+        text(
+            "CREATE TABLE embedding_status_shadow.search_vector_embeddings ("
+            "chunk_id INTEGER PRIMARY KEY)"
+        ),
+        {},
+    )
+
+    original_execute_query = project_service.repository.execute_query
+
+    async def _execute_with_legacy_storage_first(session, query, params=None):
+        await session.execute(text("SET LOCAL search_path TO embedding_status_shadow, public"))
+        return await original_execute_query(session, query, params or {})
+
+    try:
+        with (
+            patch.object(
+                type(project_service),
+                "config_manager",
+                new_callable=lambda: property(
+                    lambda self: _config_manager_with(semantic_search_enabled=True)
+                ),
+            ),
+            patch.object(
+                project_service.repository,
+                "execute_query",
+                side_effect=_execute_with_legacy_storage_first,
+            ),
+        ):
+            status = await project_service.get_embedding_status(test_project.id)
+    finally:
+        await _drop_embeddings_stub(project_service)
+        await _execute(
+            project_service,
+            text("DROP SCHEMA embedding_status_shadow CASCADE"),
+            {},
+        )
+
+    assert status.vector_tables_exist is False
+    assert status.reindex_recommended is True
+    assert "Vector storage schema is outdated" in (status.reindex_reason or "")
+
+
 @pytest.mark.asyncio
 async def test_embedding_status_entities_without_chunks(
     project_service: ProjectService, test_graph, test_project