feat(backend): implement async pipeline and SSE streaming, remove rate-limit lock

CHANGES.md

@@ -1,5 +1,25 @@
 # Changelog
 
+## [2026-09-11]
+
+### Changed
+- **Asynchronous LLM Client & Non-Blocking Scheduling (`src/rag/llm.py`)**:
+  - Removed in-process `threading.Lock()` and blocking `time.sleep()` in `LLMClient` to eliminate request serialization bottleneck across worker threads.
+  - Added `AsyncOpenAI` client for non-blocking asynchronous model invocations.
+  - Added `StreamJsonExtractor` state machine to parse and extract JSON response text deltas progressively in real time.
+  - Added `complete_async()` and `complete_stream()` methods with non-blocking rate limiting and async fallback cascades.
+- **Asynchronous Orchestrator Turn Pipeline (`src/agent/orchestrator.py`)**:
+  - Added `process_turn_async` running CPU/IO retrieval non-blockingly via `asyncio.to_thread`.
+  - Added `process_turn_stream` async generator emitting structured SSE events (`start`, `delta`, `replace`, `citations`, `done`) with safety guardrail evaluation and post-stream citation validation.
+- **Server-Sent Events (SSE) Streaming API (`backend/main.py`)**:
+  - Added `POST /api/chat/stream` endpoint returning `StreamingResponse(..., media_type="text/event-stream")`.
+  - Preserved synchronous `POST /api/chat` calling `orchestrator.process_turn` for backward compatibility with synchronous callers and test suites.
+- **Frontend Real-Time Token Streaming & Animated Feedback (`frontend/src/api.js`, `frontend/src/App.jsx`, `frontend/src/App.css`)**:
+  - Added `streamMessage` using `ReadableStreamDefaultReader` supporting CRLF/LF packet parsing and trailing buffer flushes.
+  - Fixed React 18 state batching race in `App.jsx` using atomic message existence checks.
+  - Added real-time pipeline status telemetry (`Searching OrbitMesh documentation...` -> `Generating diagnostic response...`) with an animated 3-dot pulse indicator and streaming cursor.
+
+
 ## [2026-09-04]
 
 ### Fixed

backend/main.py

@@ -4,7 +4,9 @@
 from pathlib import Path
 from typing import List, Optional
 
+import json
 from fastapi import Depends, FastAPI, HTTPException, Security, status
+from fastapi.responses import StreamingResponse
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.security.api_key import APIKeyHeader
 from pydantic import BaseModel, Field
@@ -144,6 +146,34 @@ def process_chat(request: ChatRequest, _: Optional[str] = Depends(verify_api_key
     )
 
 
+@app.post("/api/chat/stream")
+async def process_chat_stream(request: ChatRequest, _: Optional[str] = Depends(verify_api_key)):
+    session_id = request.session_id.strip() if request.session_id else None
+    if not session_id:
+        session_id = f"web-{uuid.uuid4().hex[:8]}"
+
+    async def event_generator():
+        try:
+            async for event_packet in orchestrator.process_turn_stream(session_id, request.message):
+                event_name = event_packet.get("event", "message")
+                data_payload = json.dumps(event_packet.get("data", {}))
+                yield f"event: {event_name}\ndata: {data_payload}\n\n"
+        except Exception as e:
+            logger.error(f"Stream generation failed for session '{session_id}': {e}", exc_info=True)
+            err_payload = json.dumps({"error": "Stream generation failed"})
+            yield f"event: error\ndata: {err_payload}\n\n"
+
+    return StreamingResponse(
+        event_generator(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+            "X-Accel-Buffering": "no",
+        }
+    )
+
+
 if __name__ == "__main__":
     import uvicorn
 

frontend/src/App.css

@@ -565,3 +565,65 @@
     padding-right: 16px;
   }
 }
+
+/* Streaming token cursor */
+.streaming-cursor {
+  display: inline-block;
+  width: 2px;
+  height: 1.1em;
+  background-color: #3182ce;
+  margin-left: 2px;
+  vertical-align: text-bottom;
+  animation: cursor-blink 0.8s infinite;
+}
+
+@keyframes cursor-blink {
+  0%, 100% { opacity: 1; }
+  50% { opacity: 0; }
+}
+
+/* Loading status and animated typing indicator */
+.loading-status-wrap {
+  display: flex;
+  align-items: center;
+  gap: 10px;
+}
+
+.typing-indicator {
+  display: inline-flex;
+  align-items: center;
+  gap: 4px;
+}
+
+.typing-indicator .dot {
+  width: 6px;
+  height: 6px;
+  background-color: #3182ce;
+  border-radius: 50%;
+  animation: dot-pulse 1.4s infinite ease-in-out both;
+}
+
+.typing-indicator .dot:nth-child(1) {
+  animation-delay: -0.32s;
+}
+
+.typing-indicator .dot:nth-child(2) {
+  animation-delay: -0.16s;
+}
+
+@keyframes dot-pulse {
+  0%, 80%, 100% {
+    transform: scale(0.6);
+    opacity: 0.4;
+  }
+  40% {
+    transform: scale(1);
+    opacity: 1;
+  }
+}
+
+.loading-text {
+  font-size: 0.88rem;
+  color: #4a5568;
+  font-style: italic;
+}

frontend/src/App.jsx

@@ -1,5 +1,5 @@
 import { useState, useRef, useEffect } from 'react';
-import { sendMessage } from './api';
+import { sendMessage, streamMessage } from './api';
 import './App.css';
 
 function generateSessionId() {
@@ -91,6 +91,7 @@ export default function App() {
   const [isSidebarOpen, setIsSidebarOpen] = useState(true);
   const [input, setInput] = useState('');
   const [loading, setLoading] = useState(false);
+  const [agentStatus, setAgentStatus] = useState('Assistant is analyzing...');
   const [error, setError] = useState(null);
 
   const messagesEndRef = useRef(null);
@@ -184,54 +185,124 @@ export default function App() {
 
     setInput('');
     setLoading(true);
+    setAgentStatus('Assistant is analyzing...');
     setError(null);
 
+    const assistantMsgId = 'assistant-' + Date.now();
+
     try {
-      const data = await sendMessage(activeSessionId, textToSend);
-
-      const assistantMessage = {
-        id: 'assistant-' + Date.now(),
-        sender: 'assistant',
-        text: data.response || 'No response returned.',
-        citations: data.citations || [],
-        action: data.action || 'instruct',
-        timestamp: new Date().toLocaleTimeString([], { hour: '2-digit', minute: '2-digit' }),
-      };
-
-      setSessions((prev) =>
-        prev.map((s) => {
-          if (s.id === activeSessionId) {
-            return {
-              ...s,
-              messages: [...s.messages, assistantMessage],
-            };
-          }
-          return s;
-        })
-      );
-    } catch (err) {
-      setError(err.message || 'Failed to send message');
-      setSessions((prev) =>
-        prev.map((s) => {
-          if (s.id === activeSessionId) {
-            return {
-              ...s,
-              messages: [
-                ...s.messages,
-                {
-                  id: 'error-' + Date.now(),
+      await streamMessage(activeSessionId, textToSend, {
+        onStatus: (statusText) => {
+          setAgentStatus(statusText);
+        },
+        onChunk: (delta, isReplace) => {
+          setLoading(false);
+          setSessions((prev) =>
+            prev.map((s) => {
+              if (s.id !== activeSessionId) return s;
+              const exists = s.messages.some((m) => m.id === assistantMsgId);
+              if (!exists) {
+                const newMsg = {
+                  id: assistantMsgId,
                   sender: 'assistant',
-                  text: 'Error: Unable to connect to backend server. Ensure backend is running.',
+                  text: delta,
                   citations: [],
-                  action: 'error',
+                  action: null,
+                  timestamp: new Date().toLocaleTimeString([], { hour: '2-digit', minute: '2-digit' }),
+                  isStreaming: true,
+                };
+                return { ...s, messages: [...s.messages, newMsg] };
+              }
+              return {
+                ...s,
+                messages: s.messages.map((m) => {
+                  if (m.id !== assistantMsgId) return m;
+                  return {
+                    ...m,
+                    text: isReplace ? delta : m.text + delta,
+                  };
+                }),
+              };
+            })
+          );
+        },
+        onCitations: (citations) => {
+          setSessions((prev) =>
+            prev.map((s) => {
+              if (s.id !== activeSessionId) return s;
+              return {
+                ...s,
+                messages: s.messages.map((m) => {
+                  if (m.id !== assistantMsgId) return m;
+                  return { ...m, citations: citations || [] };
+                }),
+              };
+            })
+          );
+        },
+        onDone: (doneData) => {
+          setLoading(false);
+          setSessions((prev) =>
+            prev.map((s) => {
+              if (s.id !== activeSessionId) return s;
+              const exists = s.messages.some((m) => m.id === assistantMsgId);
+              if (!exists) {
+                const newMsg = {
+                  id: assistantMsgId,
+                  sender: 'assistant',
+                  text: doneData.response || '',
+                  citations: doneData.citations || [],
+                  action: doneData.action || 'instruct',
                   timestamp: new Date().toLocaleTimeString([], { hour: '2-digit', minute: '2-digit' }),
-                },
-              ],
-            };
-          }
-          return s;
-        })
-      );
+                  isStreaming: false,
+                };
+                return { ...s, messages: [...s.messages, newMsg] };
+              }
+              return {
+                ...s,
+                messages: s.messages.map((m) => {
+                  if (m.id !== assistantMsgId) return m;
+                  return {
+                    ...m,
+                    text: doneData.response || m.text,
+                    action: doneData.action || 'instruct',
+                    citations: (m.citations && m.citations.length > 0) ? m.citations : (doneData.citations || []),
+                    isStreaming: false,
+                  };
+                }),
+              };
+            })
+          );
+        },
+        onError: (err) => {
+          setError(err.message || 'Stream connection error');
+          setLoading(false);
+          setSessions((prev) =>
+            prev.map((s) => {
+              if (s.id !== activeSessionId) return s;
+              const exists = s.messages.some((m) => m.id === assistantMsgId);
+              if (exists) return s;
+              return {
+                ...s,
+                messages: [
+                  ...s.messages,
+                  {
+                    id: 'error-' + Date.now(),
+                    sender: 'assistant',
+                    text: 'Error: Unable to connect to backend server. Ensure backend is running.',
+                    citations: [],
+                    action: 'error',
+                    timestamp: new Date().toLocaleTimeString([], { hour: '2-digit', minute: '2-digit' }),
+                  },
+                ],
+              };
+            })
+          );
+        },
+      });
+    } catch (err) {
+      setError(err.message || 'Failed to send message');
+      setLoading(false);
     } finally {
       setLoading(false);
     }
@@ -364,7 +435,10 @@ export default function App() {
                     <span className="message-time">{msg.timestamp}</span>
                   </div>
 
-                  <div className="message-content">{msg.text}</div>
+                  <div className="message-content">
+                    {msg.text}
+                    {msg.isStreaming && <span className="streaming-cursor" />}
+                  </div>
 
                   {msg.action && msg.sender === 'assistant' && (
                     <div className="message-action">
@@ -393,7 +467,14 @@ export default function App() {
             {loading && (
               <div className="message-row assistant">
                 <div className="message-bubble loading-bubble">
-                  <span>Assistant is analyzing...</span>
+                  <div className="loading-status-wrap">
+                    <span className="typing-indicator">
+                      <span className="dot" />
+                      <span className="dot" />
+                      <span className="dot" />
+                    </span>
+                    <span className="loading-text">{agentStatus}</span>
+                  </div>
                 </div>
               </div>
             )}