Refactor caching logic to improve cacheability determination based on conversation context; ensure session turns are persisted even on cache hits.

routes/agents/agents_orchestration.py

@@ -462,7 +462,13 @@ async def _run_pipeline(
         if route == RouteType.SQL_QUERY and token_user_id:
             access_scope = await run_in_threadpool(get_user_source_access, token_user_id) or ""
         cache_key = _cache_key(route, token_user_id, payload, access_scope)
-        cacheable = route not in _UNCACHEABLE_ROUTES and not session_id and not history
+        # Cacheability depends on conversation *context*, not on session identity.
+        # `history` is already resolved above (client-provided, or the Redis session
+        # fallback for a session_id with empty client history), so any turn with real
+        # prior context is non-empty here and stays uncacheable; a contextless first
+        # turn — even one carrying a session_id — is cacheable. The cache key does not
+        # include history, which is exactly why we only cache when it's empty.
+        cacheable = route not in _UNCACHEABLE_ROUTES and not history
         if cacheable:
             try:
                 cached = await get_redis_client().get(cache_key)
@@ -652,12 +658,15 @@ async def multi_agent_chat(
     result = await _run_pipeline(payload, token_user_id, trace_id, start)
     await _write_cache(result)
 
+    # Persist the session turn even on a cache hit — a cacheable first turn may now
+    # carry a session_id, and skipping this would leave it out of the Redis session
+    # store. Logging stays gated on a fresh (non-cached) answer.
+    if result.session_id and result.response.answer:
+        background_tasks.add_task(
+            _persist_session_turn, token_user_id, result.session_id,
+            payload.query, result.response.answer,
+        )
     if not result.from_cache:
-        if result.session_id and result.response.answer:
-            background_tasks.add_task(
-                _persist_session_turn, token_user_id, result.session_id,
-                payload.query, result.response.answer,
-            )
         background_tasks.add_task(
             log_chat_event, **_chat_log_kwargs(result, payload, token_user_id, trace_id)
         )
@@ -702,11 +711,13 @@ async def multi_agent_chat_stream(
             # The client gets the result first; bookkeeping below is best-effort.
             await queue.put(("result", json.loads(result.response.model_dump_json())))
             await _write_cache(result)
+            # Persist the session turn even on a cache hit (see JSON endpoint);
+            # logging stays gated on a fresh answer.
+            if result.session_id and result.response.answer:
+                await _persist_session_turn(
+                    token_user_id, result.session_id, payload.query, result.response.answer
+                )
             if not result.from_cache:
-                if result.session_id and result.response.answer:
-                    await _persist_session_turn(
-                        token_user_id, result.session_id, payload.query, result.response.answer
-                    )
                 await log_chat_event(**_chat_log_kwargs(result, payload, token_user_id, trace_id))
         except HTTPException as exc:  # already logged inside the pipeline
             await queue.put(("error", {"status_code": exc.status_code, "detail": exc.detail, "trace_id": trace_id}))