Sync wip branches

DSL/CronManager/script/service_enrichment.sh

@@ -37,7 +37,7 @@ echo "[PACKAGES] Installing required packages..."
 "$UV_BIN" pip install --python "$VENV_PATH/bin/python3" "httpx>=0.27.0" || exit 1
 "$UV_BIN" pip install --python "$VENV_PATH/bin/python3" "pydantic>=2.11.7" || exit 1
 "$UV_BIN" pip install --python "$VENV_PATH/bin/python3" "qdrant-client>=1.15.1" || exit 1
-"$UV_BIN" pip install --python "$VENV_PATH/bin/python3" "loguru>=0.7.3" || exit 1
+"$UV_BIN" pip install --python "$VENV_PATH/bin/python3" "requests>=2.32.5" || exit 1
 
 echo "[PACKAGES] All packages installed successfully"
 

DSL/Ruuter.public/rag-search/POST/api-tools/index.yml

@@ -80,13 +80,7 @@ execute_indexing:
       params: ${params}
   result: indexing_result
   on_error: handle_cron_error
-  next: check_indexing_status
-
-check_indexing_status:
-  switch:
-    - condition: ${200 <= indexing_result.response.statusCodeValue && indexing_result.response.statusCodeValue < 300}
-      next: assign_success
-  next: assign_cron_failure
+  next: assign_success
 
 handle_cron_error:
   log: "ERROR: Failed to queue api_tool indexing job - ${indexing_result.error || 'CronManager unreachable'}"
@@ -98,7 +92,7 @@ assign_cron_failure:
       success: false
       error: "INDEXING_QUEUE_FAILED"
       message: "Failed to queue indexing job. CronManager may be unavailable."
-      details: ${indexing_result.error}
+      details: ${indexing_result.error || 'CronManager unreachable'}
   next: return_server_error
 
 assign_success:

DSL/Ruuter.public/rag-search/POST/services/enrich.yml

@@ -75,13 +75,7 @@ execute_enrichment:
       is_common: ${service_is_common}
   result: enrichment_result
   on_error: handle_cron_error
-  next: check_enrichment_status
-
-check_enrichment_status:
-  switch:
-    - condition: ${200 <= enrichment_result.response.statusCodeValue && enrichment_result.response.statusCodeValue < 300}
-      next: assign_success
-  next: assign_cron_failure
+  next: assign_success
 
 handle_cron_error:
   log: "ERROR: Failed to queue enrichment job - ${enrichment_result.error || 'CronManager unreachable'}"
@@ -93,7 +87,7 @@ assign_cron_failure:
       success: false
       error: "ENRICHMENT_QUEUE_FAILED"
       message: "Failed to queue enrichment job. CronManager may be unavailable."
-      details: ${enrichment_result.error}
+      details: ${enrichment_result.error || 'CronManager unreachable'}
   next: return_server_error
 
 assign_success:

src/api_tool_indexer/constants.py

@@ -23,7 +23,9 @@ class ApiToolIndexerConstants:
     # LLM / Embedding API
     DEFAULT_API_BASE_URL = "http://llm-orchestration-service:8100"
     DEFAULT_ENVIRONMENT = "production"
-    DEFAULT_CONNECTION_ID = ""
+    # None → orchestration service resolves the embedding model via the
+    # DB-fetched vault UUID (path: embeddings/connections/{provider}/{vault_uuid}).
+    DEFAULT_CONNECTION_ID = None
 
     # Retry Configuration
     MAX_RETRIES = 3

src/intent_data_enrichment/api_client.py

@@ -20,7 +20,7 @@ def __init__(
         self,
         api_base_url: str = EnrichmentConstants.DEFAULT_API_BASE_URL,
         environment: str = EnrichmentConstants.DEFAULT_ENVIRONMENT,
-        connection_id: str = EnrichmentConstants.DEFAULT_CONNECTION_ID,
+        connection_id: Optional[str] = EnrichmentConstants.DEFAULT_CONNECTION_ID,
         max_retries: int = EnrichmentConstants.MAX_RETRIES,
         retry_delay_base: int = EnrichmentConstants.RETRY_DELAY_BASE,
         timeout: int = EnrichmentConstants.REQUEST_TIMEOUT,

src/intent_data_enrichment/constants.py

@@ -10,7 +10,9 @@ class EnrichmentConstants:
     # API Configuration
     DEFAULT_API_BASE_URL = "http://llm-orchestration-service:8100"
     DEFAULT_ENVIRONMENT = "production"
-    DEFAULT_CONNECTION_ID = "gpt-4o-mini"
+    # None → orchestration service resolves the embedding model via the
+    # DB-fetched vault UUID (path: embeddings/connections/{provider}/{vault_uuid}).
+    DEFAULT_CONNECTION_ID = None
 
     # Retry Configuration
     MAX_RETRIES = 3

src/llm_orchestration_service.py

@@ -60,6 +60,9 @@
 from src.utils.language_detector import detect_language, get_language_name
 from src.utils.prompt_config_loader import PromptConfigurationLoader
 from src.utils.query_validator import validate_query_basic
+from src.utils.sse_utils import extract_content_from_sse
+from src.utils.conversation_history_store import should_save_history, save_history_round
+from src.utils.conversation_history_helpers import get_conversation_history
 from src.guardrails import NeMoRailsAdapter, GuardrailCheckResult
 from src.contextual_retrieval import ContextualRetriever
 from src.contextual_retrieval.bm25_search import SmartBM25Search
@@ -68,13 +71,27 @@
     ContextualRetrievalFailureError,
 )
 from src.llm_orchestrator_config.feature_flags import FeatureFlags
-from src.tool_classifier import ToolClassifier
+from src.tool_classifier import ToolClassifier, WorkflowType
 from src.tool_classifier.constants import SERVICE_STEP_PREFIXES
 from src.tool_classifier.workflows.service_workflow import ServiceWorkflowExecutor
 
 # Initialize Loki logger for orchestration service
 logger = LokiLogger(service_name="llm-orchestration-service")
 
+# Set of content strings that must NOT be persisted in conversation history.
+# Covers all multilingual error / OOS / guardrail-violation messages so that
+# failed or blocked exchanges are never written to Redis.
+_HISTORY_EXCLUDED_MESSAGES: frozenset[str] = frozenset(
+    {
+        *OUT_OF_SCOPE_MESSAGES.values(),
+        *TECHNICAL_ISSUE_MESSAGES.values(),
+        *INPUT_GUARDRAIL_VIOLATION_MESSAGES.values(),
+        *OUTPUT_GUARDRAIL_VIOLATION_MESSAGES.values(),
+        *QUERY_VALIDATION_FAILED_MESSAGES.values(),
+        STREAM_TOKEN_LIMIT_MESSAGE,
+    }
+)
+
 
 class LangfuseConfig:
     """Configuration for Langfuse integration."""
@@ -152,6 +169,11 @@ def __init__(self) -> None:
         # Workflow executors access it via self.orchestration_service.session_store.
         self.session_store: Any = None
 
+        # Redis-backed conversation history store.
+        # Set to None here; the FastAPI lifespan injects the live store after
+        # Redis initialises (app.state.orchestration_service.conversation_history_store = ...).
+        self.conversation_history_store: Any = None
+
         # Shared BM25 search index pre-warmed at startup.
         # Populated by _prewarm_shared_bm25() which is called from the FastAPI
         # lifespan so it runs inside the async event loop.  Until then it is None
@@ -503,6 +525,18 @@ async def process_orchestration_request(
                     },
                 )
                 langfuse.flush()
+
+            # Persist successful exchange to conversation history (non-streaming)
+            if should_save_history(
+                self.conversation_history_store, response, _HISTORY_EXCLUDED_MESSAGES
+            ):
+                await save_history_round(
+                    self.conversation_history_store,
+                    request.chatId,
+                    request.message,
+                    response.content,
+                )
+
             return response
 
         except Exception as e:
@@ -731,14 +765,51 @@ async def stream_orchestration_response(
                         )
                         time_metric["classifier.route"] = time.time() - start_time
 
+                        # Accumulate content for history only on non-RAG workflows;
+                        # RAG routes through _stream_rag_pipeline which has its own hook.
+                        _save_classifier_history = (
+                            self.conversation_history_store is not None
+                            and classification.workflow != WorkflowType.RAG
+                        )
+                        _classifier_accumulated: list[str] = []
+                        # Tracks whether an excluded marker (OOS / guardrail violation /
+                        # error) was observed at any point during the stream.  When True
+                        # the entire accumulated buffer is discarded so no partial content
+                        # from before the blocked marker is ever written to Redis.
+                        _history_blocked = False
+
                         async for sse_chunk in stream_result:
                             yield sse_chunk
+                            if _save_classifier_history and not _history_blocked:
+                                extracted = extract_content_from_sse(sse_chunk)
+                                if extracted is not None and extracted != "END":
+                                    if extracted in _HISTORY_EXCLUDED_MESSAGES:
+                                        # Excluded marker observed — discard any partial
+                                        # content accumulated before this point and stop
+                                        # accumulating for the rest of the stream.
+                                        _classifier_accumulated.clear()
+                                        _history_blocked = True
+                                    else:
+                                        _classifier_accumulated.append(extracted)
 
                         # Successfully completed streaming through classifier
                         logger.info(
                             f"[{request.chatId}] [{stream_ctx.stream_id}] Tool classifier streaming completed"
                         )
 
+                        # Persist conversation history (classifier streaming, non-RAG workflows)
+                        if (
+                            _save_classifier_history
+                            and not _history_blocked
+                            and _classifier_accumulated
+                        ):
+                            await save_history_round(
+                                self.conversation_history_store,
+                                request.chatId,
+                                request.message,
+                                "".join(_classifier_accumulated),
+                            )
+
                         # Log costs and timings
                         self.log_costs(costs_metric)
                         log_step_timings(time_metric, request.chatId)
@@ -868,10 +939,16 @@ async def _stream_rag_pipeline(
         )
 
         start_time = time.time()
+        conversation_history, conversation_summary = await get_conversation_history(
+            chat_id=request.chatId,
+            store=self.conversation_history_store,
+            fallback=request.conversationHistory,
+        )
         refined_output, refiner_usage = self._refine_user_prompt(
             llm_manager=components["llm_manager"],
             original_message=request.message,
-            conversation_history=request.conversationHistory,
+            conversation_history=conversation_history,
+            conversation_summary=conversation_summary,
         )
         time_metric["prompt_refiner"] = time.time() - start_time
         costs_metric["prompt_refiner"] = refiner_usage
@@ -1173,6 +1250,17 @@ async def bot_response_generator() -> AsyncIterator[str]:
                     f"Storage failed for chat_id: {request.chatId}, environment: {request.environment} - {str(storage_error)}"
                 )
 
+            # Persist conversation history (RAG streaming)
+            if self.conversation_history_store is not None:
+                _rag_bot_message = "".join(accumulated_response)
+                if _rag_bot_message not in _HISTORY_EXCLUDED_MESSAGES:
+                    await save_history_round(
+                        self.conversation_history_store,
+                        request.chatId,
+                        request.message,
+                        _rag_bot_message,
+                    )
+
             # Mark stream as completed successfully
             stream_ctx.mark_completed()
 
@@ -1422,10 +1510,16 @@ async def _execute_orchestration_pipeline(
 
         # Step 1: Refine user prompt
         start_time = time.time()
+        conversation_history, conversation_summary = await get_conversation_history(
+            chat_id=request.chatId,
+            store=self.conversation_history_store,
+            fallback=request.conversationHistory,
+        )
         refined_output, refiner_usage = self._refine_user_prompt(
             llm_manager=components["llm_manager"],
             original_message=request.message,
-            conversation_history=request.conversationHistory,
+            conversation_history=conversation_history,
+            conversation_summary=conversation_summary,
         )
         timing_key = f"{prefix}.prompt_refiner" if prefix else "prompt_refiner"
         time_metric[timing_key] = time.time() - start_time
@@ -2304,6 +2398,7 @@ def _refine_user_prompt(
         llm_manager: LLMManager,
         original_message: str,
         conversation_history: List[ConversationItem],
+        conversation_summary: Optional[str] = None,
     ) -> tuple[PromptRefinerOutput, Dict[str, Any]]:
         """
         Refine user prompt using loaded LLM configuration and return usage info.
@@ -2312,6 +2407,10 @@ def _refine_user_prompt(
             llm_manager: The LLM manager instance to use
             original_message: The original user message to refine
             conversation_history: Previous conversation context
+            conversation_summary: Optional summary of earlier conversation rounds
+                that were evicted from Redis.  When provided it is prepended to the
+                DSPy history as a ``system`` turn so the refiner can use it for
+                context without re-summarising via an LLM call.
 
         Returns:
             Tuple of (PromptRefinerOutput, usage_dict): The refined prompt output and usage info
@@ -2324,8 +2423,16 @@ def _refine_user_prompt(
         logger.info("Starting prompt refinement process")
 
         try:
-            # Convert conversation history to DSPy format
+            # Convert conversation history to DSPy format, optionally prepending
+            # a pre-computed summary of earlier (evicted) conversation rounds.
             history: List[Dict[str, str]] = []
+            if conversation_summary:
+                history.append(
+                    {
+                        "role": "system",
+                        "content": f"Summary of earlier conversation: {conversation_summary}",
+                    }
+                )
             for item in conversation_history:
                 role = "assistant" if item.authorRole == "bot" else item.authorRole
                 history.append({"role": role, "content": item.message})

src/llm_orchestration_service_api.py

@@ -1,5 +1,6 @@
 """LLM Orchestration Service API - FastAPI application."""
 
+import asyncio
 import os
 import logging
 from contextlib import asynccontextmanager
@@ -12,12 +13,15 @@
 import uvicorn
 
 from llm_orchestration_service import LLMOrchestrationService
+from llm_orchestrator_config.llm_manager import LLMManager
 from src.utils.redis_client import (
     init_redis_client,
     close_redis_client,
     check_redis_health,
 )
 from src.utils.api_tool_session_store import APIToolSessionStore
+from src.utils.conversation_history_store import ConversationHistoryStore
+from src.utils.conversation_summary_generator import create_incremental_summarizer
 from src.llm_orchestrator_config.llm_ochestrator_constants import (
     STREAMING_ALLOWED_ENVS,
     STREAM_TIMEOUT_MESSAGE,
@@ -98,23 +102,50 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     try:
         await init_redis_client()
         app.state.session_store = APIToolSessionStore()
+
+        # Wire an incremental summarizer if the LLM manager singleton is available.
+        summarizer = None
+        try:
+            summarizer = create_incremental_summarizer(LLMManager())
+            logger.info("Incremental conversation summarizer initialized")
+        except Exception as e:
+            logger.warning(
+                f"Could not create incremental summarizer, continuing without it: {e}"
+            )
+
+        app.state.conversation_history_store = ConversationHistoryStore(
+            summarizer=summarizer
+        )
         logger.info("Redis session store initialized successfully")
     except Exception as e:
         logger.warning(f"Redis session store unavailable, continuing without it: {e}")
         app.state.session_store = None
+        app.state.conversation_history_store = None
 
-    # Expose session_store on the orchestration service so workflow executors
-    # (e.g. APIToolWorkflowExecutor) can reach it via self.orchestration_service.
+    # Expose session_store and conversation_history_store on the orchestration
+    # service so downstream components can reach them via self.orchestration_service.
     if (
         hasattr(app.state, "orchestration_service")
         and app.state.orchestration_service is not None
     ):
         app.state.orchestration_service.session_store = app.state.session_store
+        app.state.orchestration_service.conversation_history_store = (
+            app.state.conversation_history_store
+        )
 
     yield
 
     # Shutdown
     logger.info("Shutting down LLM Orchestration Service API")
+
+    # Await any in-flight incremental summary tasks to avoid lost work.
+    store = getattr(app.state, "conversation_history_store", None)
+    if store is not None and store._pending_tasks:
+        logger.info(
+            f"Waiting for {len(store._pending_tasks)} pending summary task(s) to complete..."
+        )
+        await asyncio.gather(*store._pending_tasks, return_exceptions=True)
+
     if (
         hasattr(app.state, "orchestration_service")
         and app.state.orchestration_service is not None