From 0233e721c0105a195f88fa8af8a308124610276f Mon Sep 17 00:00:00 2001 From: cyrus Date: Fri, 13 Mar 2026 18:21:41 +0530 Subject: [PATCH] fix: resolve race condition and message persistence bugs in conversations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Fix race condition where agent responses went to wrong threads when users switched conversations - Add activeThreadId state to track which thread is currently sending/receiving messages - Implement single active conversation pattern to prevent concurrent message sending - Fix optimistic message persistence bug where user messages disappeared after conversation switch - Replace optimistic messaging with immediate persistent storage for user messages - Add clear UI feedback when other conversations are active with disabled inputs - Ensure responses always go to original sending thread regardless of current selection - Simplify addInferenceResponse logic by removing complex retroactive user message handling - Add proper error handling for failed message sends with cleanup of persisted messages 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude --- src/pages/Conversations.tsx | 83 +++++++++++++++++++++++++++++++------ src/store/index.ts | 1 + src/store/threadSlice.ts | 68 ++++++++++++++---------------- 3 files changed, 102 insertions(+), 50 deletions(-) diff --git a/src/pages/Conversations.tsx b/src/pages/Conversations.tsx index 453f32750..a209fbf77 100644 --- a/src/pages/Conversations.tsx +++ b/src/pages/Conversations.tsx @@ -22,7 +22,7 @@ import { useAppDispatch, useAppSelector } from '../store/hooks'; import type { NotionPageSummary, NotionSummary, NotionUserProfile } from '../store/notionSlice'; import { addInferenceResponse, - addOptimisticMessage, + addMessageLocal, clearDeleteStatus, clearPurgeStatus, clearSelectedThread, @@ -30,11 +30,13 @@ import { deleteThreadLocal, fetchSuggestedQuestions, purgeThreads, - removeOptimisticMessages, + setActiveThread, setLastViewed, setPanelWidth, setSelectedThread, + updateMessagesForThread, } from '../store/threadSlice'; +import type { ThreadMessage } from '../types/thread'; const MIN_PANEL_WIDTH = 200; const MAX_PANEL_WIDTH = 480; @@ -113,6 +115,7 @@ const Conversations = () => { lastViewedAt, suggestedQuestions, isLoadingSuggestions, + activeThreadId, } = useAppSelector(state => state.thread); const skillsState = useAppSelector(state => state.skills); @@ -306,14 +309,43 @@ const Conversations = () => { const trimmed = text ?? inputValue.trim(); if (!trimmed || !selectedThreadId || isSending) return; - // Snapshot history before the optimistic update (exclude stale optimistic msgs) - const historySnapshot = messages.filter(m => !m.id.startsWith('optimistic-')); + // Check if another thread is already sending + if (activeThreadId && activeThreadId !== selectedThreadId) { + return; // Block sending from non-active threads + } + + // Store the original thread ID to ensure response goes to correct thread + const sendingThreadId = selectedThreadId; + + // Create stable user message and persist immediately + const userMessage: ThreadMessage = { + id: `msg_${Date.now()}_${Math.random()}`, + content: trimmed, + type: 'text', + extraMetadata: {}, + sender: 'user', + createdAt: new Date().toISOString(), + }; + + // Immediately persist user message to both current view and persistent storage + dispatch(addMessageLocal({ threadId: sendingThreadId, message: userMessage })); + + // Update current view if this is the selected thread + if (sendingThreadId === selectedThreadId) { + // Message is already added to persistent storage, reload current view + dispatch(setSelectedThread(sendingThreadId)); + } + + // Snapshot history for AI request (excluding the just-added user message since we'll add it manually) + const historySnapshot = messages.filter(m => !m.id.startsWith('optimistic-') && m.id !== userMessage.id); - dispatch(addOptimisticMessage({ content: trimmed })); setInputValue(''); setSendError(null); setIsSending(true); + // Set this thread as active + dispatch(setActiveThread(sendingThreadId)); + try { // Process user message with SOUL + TOOLS injection let processedUserContent = trimmed; @@ -491,14 +523,30 @@ const Conversations = () => { break; } - dispatch(addInferenceResponse({ content: finalContent })); + // Pass the original sending thread ID to ensure response goes to correct thread + dispatch(addInferenceResponse({ content: finalContent, threadId: sendingThreadId })); } catch (err) { - dispatch(removeOptimisticMessages()); + // Remove the user message from persistent storage on error + // We'll use a thunk-like approach to access current state + dispatch((dispatch, getState) => { + const state = getState() as { thread: { messagesByThreadId: Record } }; + const persistedMessages = state.thread.messagesByThreadId[sendingThreadId] || []; + const currentMessages = persistedMessages.filter(m => m.id !== userMessage.id); + dispatch(updateMessagesForThread({ threadId: sendingThreadId, messages: currentMessages })); + + // Also remove from current view if this is the selected thread + if (sendingThreadId === selectedThreadId) { + dispatch(setSelectedThread(sendingThreadId)); + } + }); + const msg = err && typeof err === 'object' && 'error' in err ? String((err as { error: unknown }).error) : 'Failed to get response'; setSendError(msg); + // Clear active thread on error + dispatch(setActiveThread(null)); } finally { setIsSending(false); } @@ -900,8 +948,8 @@ const Conversations = () => { ))} - {/* Typing indicator (#14) */} - {isSending && ( + {/* Typing indicator (#14) - Only show for the active thread */} + {activeThreadId === selectedThreadId && isSending && (
@@ -930,7 +978,7 @@ const Conversations = () => { key={i} type="button" onClick={() => handleSendMessage(s.text)} - disabled={isSending} + disabled={isSending || !!(activeThreadId && activeThreadId !== selectedThreadId)} className="flex-shrink-0 px-3 py-1.5 rounded-lg text-[12px] whitespace-nowrap bg-white/5 text-stone-400 hover:bg-white/10 transition-colors disabled:opacity-50 disabled:cursor-not-allowed"> {s.text} @@ -941,6 +989,14 @@ const Conversations = () => { {/* Message Input */}
+ {/* Show warning if another thread is active */} + {activeThreadId && activeThreadId !== selectedThreadId && ( +
+

+ Another conversation is active. Please wait for it to complete before sending messages here. +

+
+ )} {/* Model selector */}
{isLoadingModels ? ( @@ -983,13 +1039,14 @@ const Conversations = () => { value={inputValue} onChange={e => setInputValue(e.target.value)} onKeyDown={handleInputKeyDown} - placeholder="Type a message..." + placeholder={activeThreadId && activeThreadId !== selectedThreadId ? "Another conversation is active..." : "Type a message..."} rows={1} - className="flex-1 resize-none bg-white/5 border border-white/10 rounded-xl px-4 py-2.5 text-sm placeholder:text-stone-500 focus:outline-none focus:ring-1 focus:ring-primary-500/50 focus:border-primary-500/50 transition-all max-h-32" + disabled={!!(activeThreadId && activeThreadId !== selectedThreadId)} + className="flex-1 resize-none bg-white/5 border border-white/10 rounded-xl px-4 py-2.5 text-sm placeholder:text-stone-500 focus:outline-none focus:ring-1 focus:ring-primary-500/50 focus:border-primary-500/50 transition-all max-h-32 disabled:opacity-50 disabled:cursor-not-allowed" />