Persist streamed assistant answers even if the client disconnects
The first message in a new conversation navigated to /c/:id, which remounted ChatScreen (two separate routes) and aborted the in-flight /stream-query fetch. Server-side that surfaced as asyncio.CancelledError, which isn't an Exception, so the except branch missed it and skipped the add_message_to_conversation save — the fully-streamed answer was never persisted. Backend: handle CancelledError and persist via a detached, shielded task so a disconnect at any point (navigation, tab close, network blip) still commits the answer. Frontend: serve "/" and "/c/:id" from a single catch-all ChatScreen route so first-message navigation no longer remounts and aborts the stream; derive the conversation id via useMatch instead of useParams. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -1,3 +1,4 @@
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import uuid
|
||||
@@ -31,6 +32,18 @@ conversation_blueprint = Blueprint(
|
||||
"conversation_api", __name__, url_prefix="/api/conversation"
|
||||
)
|
||||
|
||||
# Detached persistence tasks. A strong reference keeps them from being garbage
|
||||
# collected mid-flight when the SSE stream is cancelled by a client disconnect.
|
||||
_persist_tasks: set[asyncio.Task] = set()
|
||||
|
||||
|
||||
def _persist_in_background(coro) -> asyncio.Task:
|
||||
task = asyncio.create_task(coro)
|
||||
_persist_tasks.add(task)
|
||||
task.add_done_callback(_persist_tasks.discard)
|
||||
return task
|
||||
|
||||
|
||||
_SYSTEM_PROMPT = SIMBA_SYSTEM_PROMPT
|
||||
|
||||
|
||||
@@ -200,6 +213,15 @@ async def stream_query():
|
||||
async def event_generator():
|
||||
final_message = None
|
||||
streamed = ""
|
||||
|
||||
async def persist(text: str) -> None:
|
||||
await add_message_to_conversation(
|
||||
conversation=conversation,
|
||||
message=text,
|
||||
speaker="simba",
|
||||
user=user,
|
||||
)
|
||||
|
||||
try:
|
||||
async for event in main_agent.astream_events(
|
||||
payload, version="v2", config=agent_config
|
||||
@@ -229,17 +251,24 @@ async def stream_query():
|
||||
if isinstance(content, str) and content:
|
||||
final_message = content
|
||||
|
||||
except asyncio.CancelledError:
|
||||
# The client disconnected mid-stream (e.g. a route change aborts the
|
||||
# fetch). CancelledError is not an Exception, so without this branch
|
||||
# persistence below would be skipped and the answer lost. Save what
|
||||
# was generated in a detached task, then honor the cancellation.
|
||||
text = final_message or streamed
|
||||
if text:
|
||||
_persist_in_background(persist(text))
|
||||
raise
|
||||
except Exception as e:
|
||||
yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
|
||||
|
||||
final_message = final_message or streamed
|
||||
if final_message:
|
||||
await add_message_to_conversation(
|
||||
conversation=conversation,
|
||||
message=final_message,
|
||||
speaker="simba",
|
||||
user=user,
|
||||
)
|
||||
# Persist via a detached, shielded task so a disconnect at this
|
||||
# point (after generation completes) still commits the answer.
|
||||
task = _persist_in_background(persist(final_message))
|
||||
await asyncio.shield(task)
|
||||
yield f"data: {json.dumps({'type': 'response', 'message': final_message})}\n\n"
|
||||
else:
|
||||
yield f"data: {json.dumps({'type': 'error', 'message': 'No response generated'})}\n\n"
|
||||
|
||||
+10
-10
@@ -1,4 +1,4 @@
|
||||
import { Navigate, Route, Routes } from "react-router-dom";
|
||||
import { Route, Routes } from "react-router-dom";
|
||||
import "./App.css";
|
||||
import { AuthProvider } from "./contexts/AuthContext";
|
||||
import { ChatScreen } from "./components/ChatScreen";
|
||||
@@ -31,16 +31,16 @@ const AppContainer = () => {
|
||||
|
||||
return (
|
||||
<Routes>
|
||||
<Route
|
||||
path="/"
|
||||
element={<ChatScreen setAuthenticated={setAuthenticated} isAdmin={isAdmin} />}
|
||||
/>
|
||||
<Route
|
||||
path="/c/:conversationId"
|
||||
element={<ChatScreen setAuthenticated={setAuthenticated} isAdmin={isAdmin} />}
|
||||
/>
|
||||
<Route path="/conversations" element={<ConversationsPage />} />
|
||||
<Route path="*" element={<Navigate to="/" replace />} />
|
||||
{/* One ChatScreen instance serves both the home ("/") and per-conversation
|
||||
("/c/:id") routes. Using a single catch-all route keeps that instance
|
||||
mounted across the navigate() that fires when the first message creates
|
||||
a conversation — a remount here would abort the in-flight SSE stream
|
||||
(and drop the answer before it's persisted). */}
|
||||
<Route
|
||||
path="*"
|
||||
element={<ChatScreen setAuthenticated={setAuthenticated} isAdmin={isAdmin} />}
|
||||
/>
|
||||
</Routes>
|
||||
);
|
||||
};
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { useState, useCallback, useEffect, useRef } from "react";
|
||||
import { useParams } from "react-router-dom";
|
||||
import { useMatch } from "react-router-dom";
|
||||
import { conversationService } from "../api/conversationService";
|
||||
import type { Conversation } from "./useConversations";
|
||||
|
||||
@@ -42,7 +42,10 @@ export function useChat({
|
||||
onSessionExpired,
|
||||
scrollToBottom,
|
||||
}: UseChatOptions) {
|
||||
const { conversationId } = useParams();
|
||||
// ChatScreen renders under a single catch-all route so it isn't remounted
|
||||
// when the first message navigates to /c/:id (which would abort the stream),
|
||||
// so the id comes from matching the URL rather than useParams.
|
||||
const conversationId = useMatch("/c/:conversationId")?.params.conversationId;
|
||||
const [messages, setMessages] = useState<Message[]>([]);
|
||||
const [isLoading, setIsLoading] = useState(false);
|
||||
const [messagesLoading, setMessagesLoading] = useState(false);
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { useState, useCallback, useEffect, useMemo } from "react";
|
||||
import { useNavigate, useParams } from "react-router-dom";
|
||||
import { useMatch, useNavigate } from "react-router-dom";
|
||||
import { conversationService } from "../api/conversationService";
|
||||
|
||||
export type Conversation = {
|
||||
@@ -13,7 +13,9 @@ const SIDEBAR_LIMIT = 10;
|
||||
|
||||
export function useConversations() {
|
||||
const navigate = useNavigate();
|
||||
const { conversationId } = useParams();
|
||||
// ChatScreen now renders under a single catch-all route, so useParams no
|
||||
// longer captures the id. Derive it from the URL directly instead.
|
||||
const conversationId = useMatch("/c/:conversationId")?.params.conversationId;
|
||||
const [conversations, setConversations] = useState<Conversation[]>([]);
|
||||
|
||||
const refreshConversations = useCallback(async () => {
|
||||
|
||||
Reference in New Issue
Block a user