feat: Production-grade CoT Hardening - Priority 1 & 2 Defenses

backend/pytest.ini

@@ -85,7 +85,9 @@ env =
     MILVUS_HOST=milvus-standalone
 
 # Test Selection Patterns
-norecursedirs = volumes data .git .tox playwright
+norecursedirs = volumes data .git .tox
+# Explicitly ignore playwright tests (requires separate dependencies)
+collect_ignore = ../tests/playwright
 
 # Filter warnings
 filterwarnings =

backend/rag_solution/generation/providers/factory.py

@@ -218,3 +218,15 @@ def list_providers(cls) -> dict[str, type[LLMBase]]:
         with cls._lock:
             logger.debug(f"Listing providers: {cls._providers}")
             return cls._providers.copy()  # Return a copy to prevent modification
+
+    @classmethod
+    def clear_providers(cls) -> None:
+        """
+        Clear all registered providers (primarily for testing).
+
+        This method is useful for test isolation to prevent provider
+        registration errors across test modules.
+        """
+        with cls._lock:
+            cls._providers.clear()
+            logger.debug("Cleared all registered providers")

backend/rag_solution/models/collection.py

@@ -4,7 +4,7 @@
 
 import uuid
 from datetime import datetime
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, ClassVar
 
 from sqlalchemy import Boolean, DateTime, Enum, String
 from sqlalchemy.dialects.postgresql import UUID
@@ -29,6 +29,7 @@ class Collection(Base):  # pylint: disable=too-few-public-methods
     """
 
     __tablename__ = "collections"
+    __table_args__: ClassVar[dict] = {"extend_existing": True}
 
     # 🆔 Identification
     id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=IdentityService.generate_id)

backend/rag_solution/models/question.py

@@ -4,7 +4,7 @@
 
 import uuid
 from datetime import datetime
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, ClassVar
 
 from sqlalchemy import JSON, DateTime, ForeignKey, String
 from sqlalchemy.dialects.postgresql import UUID
@@ -31,6 +31,7 @@ class SuggestedQuestion(Base):
     """
 
     __tablename__ = "suggested_questions"
+    __table_args__: ClassVar[dict] = {"extend_existing": True}
 
     id: Mapped[uuid.UUID] = mapped_column(
         UUID(as_uuid=True),

backend/rag_solution/models/token_warning.py

@@ -2,6 +2,7 @@
 
 import uuid
 from datetime import datetime
+from typing import ClassVar
 
 from sqlalchemy import DateTime, Float, Integer, String
 from sqlalchemy.dialects.postgresql import UUID
@@ -19,6 +20,7 @@ class TokenWarning(Base):
     """
 
     __tablename__ = "token_warnings"
+    __table_args__: ClassVar[dict] = {"extend_existing": True}
 
     id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=IdentityService.generate_id)
     user_id: Mapped[uuid.UUID | None] = mapped_column(UUID(as_uuid=True), nullable=True, index=True)

backend/rag_solution/services/answer_synthesizer.py

@@ -18,40 +18,58 @@ def __init__(self, llm_service: LLMBase | None = None, settings: Settings | None
         self.llm_service = llm_service
         self.settings = settings or get_settings()
 
-    def synthesize(self, original_question: str, reasoning_steps: list[ReasoningStep]) -> str:
+    def synthesize(self, original_question: str, reasoning_steps: list[ReasoningStep]) -> str:  # noqa: ARG002
         """Synthesize a final answer from reasoning steps.
 
+        NOTE: Since we now use structured output parsing in chain_of_thought_service.py,
+        the intermediate_answer already contains only the clean final answer (from <answer> tags).
+        We no longer need to add prefixes like "Based on the analysis of..." as this was
+        causing CoT reasoning leakage.
+
         Args:
-            original_question: The original question.
+            original_question: The original question (not used, kept for API compatibility).
             reasoning_steps: The reasoning steps taken.
 
         Returns:
             The synthesized final answer.
         """
+        import logging
+
+        logger = logging.getLogger(__name__)
+
         if not reasoning_steps:
             return "Unable to generate an answer due to insufficient information."
 
-        # Combine intermediate answers
+        # Extract intermediate answers (already cleaned by structured output parsing)
         intermediate_answers = [step.intermediate_answer for step in reasoning_steps if step.intermediate_answer]
 
         if not intermediate_answers:
             return "Unable to synthesize an answer from the reasoning steps."
 
-        # Simple synthesis (in production, this would use an LLM)
+        # DEBUG: Log what we receive from CoT
+        logger.debug("=" * 80)
+        logger.debug("📝 ANSWER SYNTHESIZER DEBUG")
+        logger.debug("Number of intermediate answers: %d", len(intermediate_answers))
+        for i, answer in enumerate(intermediate_answers):
+            logger.debug("Intermediate answer %d (first 300 chars): %s", i + 1, answer[:300])
+        logger.debug("=" * 80)
+
+        # For single answer, return it directly (already clean from XML parsing)
         if len(intermediate_answers) == 1:
-            return intermediate_answers[0]
+            final = intermediate_answers[0]
+            logger.debug("🎯 FINAL ANSWER (single step, first 300 chars): %s", final[:300])
+            return final
 
-        # Combine multiple answers
-        synthesis = f"Based on the analysis of {original_question}: "
+        # For multiple answers, combine cleanly without contaminating prefixes
+        # The LLM already provided clean answers via <answer> tags
+        synthesis = intermediate_answers[0]
 
-        for i, answer in enumerate(intermediate_answers):
-            if i == 0:
-                synthesis += answer
-            elif i == len(intermediate_answers) - 1:
-                synthesis += f" Additionally, {answer.lower()}"
-            else:
-                synthesis += f" Furthermore, {answer.lower()}"
+        for answer in intermediate_answers[1:]:
+            # Only add if it provides new information (avoid duplicates)
+            if answer.lower() not in synthesis.lower():
+                synthesis += f" {answer}"
 
+        logger.debug("🎯 FINAL SYNTHESIZED ANSWER (first 300 chars): %s", synthesis[:300])
         return synthesis
 
     async def synthesize_answer(self, original_question: str, reasoning_steps: list[ReasoningStep]) -> SynthesisResult: