admin/openhuman
1
0
Fork 0
mirror of https://github.com/tinyhumansai/openhuman.git synced 2026-07-30 23:14:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #172 from graycyrus/develop

Browse Source 
fix: resolve race condition and message persistence bugs in conversat…
This commit is contained in:
Cyrus Gray
2026-03-13 18:28:34 +05:30
committed by GitHub
parent ddc326bb26 0233e721c0
commit 0a9d5c538a
3 changed files with 102 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pages/Conversations.tsx
+70 -13
View File
@@ -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<string, ThreadMessage[]> } };
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 = () => {
</div>
</div>
))}
{/* Typing indicator (#14) */}
{isSending && (
{/* Typing indicator (#14) - Only show for the active thread */}
{activeThreadId === selectedThreadId && isSending && (
<div className="flex justify-start">
<div className="bg-white/5 rounded-2xl rounded-bl-md px-4 py-3">
<div className="flex items-center gap-1">
@@ -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}
</button>
@@ -941,6 +989,14 @@ const Conversations = () => {
{/* Message Input */}
<div className="flex-shrink-0 border-t border-white/10 px-4 py-3">
{/* Show warning if another thread is active */}
{activeThreadId && activeThreadId !== selectedThreadId && (
<div className="mb-3 p-2 rounded-lg bg-amber-500/10 border border-amber-500/20">
<p className="text-xs text-amber-400">
Another conversation is active. Please wait for it to complete before sending messages here.
</p>
</div>
)}
{/* Model selector */}
<div className="flex items-center gap-2 mb-2">
{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"
/>
<button
onClick={() => handleSendMessage()}
disabled={!inputValue.trim() || isSending}
disabled={!inputValue.trim() || isSending || !!(activeThreadId && activeThreadId !== selectedThreadId)}
className="p-2.5 rounded-xl bg-primary-600 hover:bg-primary-500 disabled:opacity-40 disabled:cursor-not-allowed transition-colors flex-shrink-0">
{isSending ? (
<svg className="w-4 h-4 animate-spin" fill="none" viewBox="0 0 24 24">
src/store/index.ts
+1
View File
@@ -47,6 +47,7 @@ const aiPersistConfig = { key: 'ai', storage, whitelist: ['config'] };
const skillsPersistConfig = { key: 'skills', storage, whitelist: ['skills'] };
// Persist config for thread data and UI prefs (includes threads and messages)
// Note: activeThreadId is intentionally excluded as it's transient state
const threadPersistConfig = {
key: 'thread',
storage,
src/store/threadSlice.ts
+31 -37
View File
@@ -11,6 +11,7 @@ interface ThreadState {
selectedThreadId: string | null;
panelWidth: number;
lastViewedAt: Record<string, number>;
activeThreadId: string | null; // Track which thread is currently sending/receiving
// NEW: Add efficient message storage
messagesByThreadId: Record<string, ThreadMessage[]>;
@@ -35,6 +36,7 @@ const initialState: ThreadState = {
selectedThreadId: null,
panelWidth: 320,
lastViewedAt: {},
activeThreadId: null,
messagesByThreadId: {},
messages: [],
isLoadingMessages: false,
@@ -200,7 +202,7 @@ const threadSlice = createSlice({
createdAt: new Date().toISOString(),
});
},
addInferenceResponse: (state, action: { payload: { content: string } }) => {
addInferenceResponse: (state, action: { payload: { content: string; threadId?: string } }) => {
const aiMessage: ThreadMessage = {
id: `inference-${Date.now()}`,
content: action.payload.content,
@@ -210,52 +212,33 @@ const threadSlice = createSlice({
createdAt: new Date().toISOString(),
};
// Add to current messages view
state.messages.push(aiMessage);
// Use provided threadId or fall back to activeThreadId to ensure response goes to correct thread
const targetThreadId = action.payload.threadId || state.activeThreadId;
// Also add to persistent storage
if (state.selectedThreadId) {
if (!state.messagesByThreadId[state.selectedThreadId]) {
state.messagesByThreadId[state.selectedThreadId] = [];
if (targetThreadId) {
// Ensure messagesByThreadId exists for this thread
if (!state.messagesByThreadId[targetThreadId]) {
state.messagesByThreadId[targetThreadId] = [];
}
// CRITICAL FIX: Ensure the preceding user message is also persisted
// Find the last user message that might not be in persistent storage yet
const lastUserMessage = state.messages.filter(m => m.sender === 'user').pop();
// Add the AI response to persistent storage
state.messagesByThreadId[targetThreadId].push(aiMessage);
if (lastUserMessage) {
const persistedMessages = state.messagesByThreadId[state.selectedThreadId];
const userMessageExists = persistedMessages.some(m => m.id === lastUserMessage.id);
// If user message isn't persisted yet, add it first.
// Replace optimistic- prefix with a stable id so historySnapshot
// (which filters out optimistic- ids) includes it on future sends.
if (!userMessageExists) {
const stableMessage = lastUserMessage.id.startsWith('optimistic-')
? { ...lastUserMessage, id: `msg_${Date.now()}_user` }
: lastUserMessage;
persistedMessages.push(stableMessage);
// Keep state.messages in sync with the stable id
const messages = state.messages;
for (let i = messages.length - 1; i >= 0; i--) {
if (messages[i].id === lastUserMessage.id) {
messages[i] = stableMessage;
break;
}
}
}
// Add to current messages view only if it's the currently selected thread
if (targetThreadId === state.selectedThreadId) {
state.messages.push(aiMessage);
}
// Now add the AI response
state.messagesByThreadId[state.selectedThreadId].push(aiMessage);
// Update thread metadata
const thread = state.threads.find(t => t.id === state.selectedThreadId);
const thread = state.threads.find(t => t.id === targetThreadId);
if (thread) {
thread.messageCount = state.messagesByThreadId[state.selectedThreadId].length;
thread.messageCount = state.messagesByThreadId[targetThreadId].length;
thread.lastMessageAt = aiMessage.createdAt;
}
}
// Clear active thread when response is received
state.activeThreadId = null;
},
removeOptimisticMessages: state => {
state.messages = state.messages.filter(m => !m.id.startsWith('optimistic-'));
@@ -331,6 +314,10 @@ const threadSlice = createSlice({
state.selectedThreadId = null;
state.messages = [];
state.lastViewedAt = {};
state.activeThreadId = null;
},
setActiveThread: (state, action: { payload: string | null }) => {
state.activeThreadId = action.payload;
},
},
extraReducers: builder => {
@@ -356,22 +343,28 @@ const threadSlice = createSlice({
state.selectedThreadId = null;
state.messages = [];
state.lastViewedAt = {};
state.activeThreadId = null;
})
// sendMessage
.addCase(sendMessage.pending, state => {
.addCase(sendMessage.pending, (state, action) => {
state.sendStatus = 'loading';
state.sendError = null;
// Set the active thread when message sending starts
state.activeThreadId = action.meta.arg.threadId;
})
.addCase(sendMessage.fulfilled, state => {
state.sendStatus = 'success';
state.suggestedQuestions = [];
state.suggestError = null;
// Don't clear activeThreadId here - let addInferenceResponse handle it
})
.addCase(sendMessage.rejected, (state, action) => {
state.sendStatus = 'error';
state.sendError = action.payload as string;
// Remove optimistic messages so the user doesn't see phantom messages
state.messages = state.messages.filter(m => !m.id.startsWith('optimistic-'));
// Clear active thread on error
state.activeThreadId = null;
})
// fetchSuggestedQuestions
.addCase(fetchSuggestedQuestions.pending, state => {
@@ -407,5 +400,6 @@ export const {
deleteThreadLocal,
updateMessagesForThread,
clearAllThreads,
setActiveThread,
} = threadSlice.actions;
export default threadSlice.reducer;