[NOTICKET] fix query now use orchestrator msg, rework db pipeline replace ingestion logic

src/api/v1/chat.py

@@ -242,7 +242,7 @@ async def chat_stream(request: ChatRequest, db: AsyncSession = Depends(get_db)):
                     results=retrieval_objects,
                     user_id=request.user_id,
                     db=db,
-                    question=request.message,
+                    question=intent_result.get("search_query") or request.message,
                 )
                 query_context = _format_query_results(query_results)
                 if query_context:

src/pipeline/db_pipeline/db_pipeline_service.py

@@ -195,7 +195,20 @@ class DbPipelineService:
             all_docs.extend(docs)
             logger.info("profiled table", table=table_name, count=len(docs))
 
-        # Delete only after all docs are ready
+        # Insert new chunks first; only delete stale chunks after the insert succeeds.
+        # Prevents data loss if aadd_documents fails — old embeddings stay queryable
+        # until they're proven replaceable. Stale rows are identified by an older
+        # updated_at than this run.
+        if not all_docs:
+            logger.warning(
+                "no docs produced from schema; skipping delete to preserve existing embeddings",
+                user_id=user_id,
+                client_id=client_id,
+            )
+            return 0
+
+        await vector_store.aadd_documents(all_docs)
+
         async with _pgvector_engine.begin() as conn:
             result = await conn.execute(
                 text(
@@ -203,16 +216,14 @@ class DbPipelineService:
                     "WHERE cmetadata->>'user_id' = :user_id "
                     "  AND cmetadata->>'source_type' = 'database' "
                     "  AND cmetadata->>'database_client_id' = :client_id "
+                    "  AND cmetadata->>'updated_at' < :updated_at "
                     "  AND collection_id = ("
                     "    SELECT uuid FROM langchain_pg_collection WHERE name = 'document_embeddings'"
                     "  )"
                 ),
-                {"user_id": user_id, "client_id": client_id},
+                {"user_id": user_id, "client_id": client_id, "updated_at": updated_at},
             )
-            logger.info("cleared old db embeddings", user_id=user_id, deleted=result.rowcount)
-
-        if all_docs:
-            await vector_store.aadd_documents(all_docs)
+            logger.info("cleared stale db embeddings", user_id=user_id, deleted=result.rowcount)
 
         logger.info("db pipeline complete", user_id=user_id, total=len(all_docs))
         return len(all_docs)