From 9af599466f981d1e69a6a7c3759bc91d74c5533f Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Fri, 12 Jun 2026 09:47:57 -0700 Subject: [PATCH] feat(chat): parallel agent inference (cross-thread + within-thread forks) (#3633) --- app/src/components/chat/ChatComposer.tsx | 15 +- app/src/lib/i18n/ar.ts | 2 + app/src/lib/i18n/bn.ts | 2 + app/src/lib/i18n/de.ts | 2 + app/src/lib/i18n/en.ts | 2 + app/src/lib/i18n/es.ts | 2 + app/src/lib/i18n/fr.ts | 2 + app/src/lib/i18n/hi.ts | 2 + app/src/lib/i18n/id.ts | 2 + app/src/lib/i18n/it.ts | 2 + app/src/lib/i18n/ko.ts | 2 + app/src/lib/i18n/pl.ts | 2 + app/src/lib/i18n/pt.ts | 2 + app/src/lib/i18n/ru.ts | 2 + app/src/lib/i18n/zh-CN.ts | 2 + app/src/pages/Conversations.tsx | 359 +++++++++++++----- .../Conversations.attachments.test.tsx | 6 +- .../__tests__/Conversations.render.test.tsx | 6 +- .../pages/__tests__/Conversations.test.tsx | 14 +- .../Conversations.welcomeLock.test.tsx | 20 +- app/src/providers/ChatRuntimeProvider.tsx | 123 +++++- .../__tests__/ChatRuntimeProvider.test.tsx | 70 +++- app/src/services/chatService.ts | 13 +- app/src/services/socketService.ts | 12 +- .../store/__tests__/chatRuntimeSlice.test.ts | 104 +++++ app/src/store/__tests__/threadSlice.test.ts | 50 ++- app/src/store/chatRuntimeSlice.ts | 74 +++- app/src/store/threadSlice.ts | 55 ++- docs/TEST-COVERAGE-MATRIX.md | 1 + src/openhuman/agent/harness/run_queue/mod.rs | 5 + .../agent/harness/run_queue/types.rs | 6 + src/openhuman/channels/providers/web/mod.rs | 4 + src/openhuman/channels/providers/web/ops.rs | 339 +++++++++++++++-- .../channels/providers/web/run_task.rs | 49 ++- src/openhuman/channels/providers/web/types.rs | 18 + src/openhuman/channels/providers/web_tests.rs | 228 ++++++++++- 36 files changed, 1403 insertions(+), 196 deletions(-) diff --git a/app/src/components/chat/ChatComposer.tsx b/app/src/components/chat/ChatComposer.tsx index 24315f8cf..e340397e7 100644 --- a/app/src/components/chat/ChatComposer.tsx +++ b/app/src/components/chat/ChatComposer.tsx @@ -16,6 +16,13 @@ export interface ChatComposerProps { fileInputRef: React.RefObject; composerInteractionBlocked: boolean; isSending: boolean; + /** + * When true, the selected thread has an in-flight turn but the user may still + * type and send a PARALLEL branch (Cmd/Ctrl+Enter). Keeps the textarea + * editable even though `composerInteractionBlocked` is set, and surfaces a + * hint. The normal Send button / plain Enter stay gated. + */ + allowParallelSend?: boolean; attachments: Attachment[]; onAttachFiles: (files: FileList | null) => Promise; onRemoveAttachment: (id: string) => void; @@ -46,6 +53,7 @@ export default function ChatComposer({ fileInputRef, composerInteractionBlocked, isSending, + allowParallelSend = false, attachments, onAttachFiles, onRemoveAttachment, @@ -61,6 +69,9 @@ export default function ChatComposer({ const { t } = useT(); const hasContent = inputValue.trim().length > 0 || attachments.length > 0 || isSending; + // The textarea stays editable when a parallel branch is allowed even though + // the primary composer interaction is blocked by an in-flight turn. + const textareaDisabled = (composerInteractionBlocked && !allowParallelSend) || isSending; // Auto-resize textarea: grow with content, cap at COMPOSER_MAX_HEIGHT, then scroll. useEffect(() => { @@ -150,9 +161,9 @@ export default function ChatComposer({ isComposingTextRef.current = false; }} onKeyDown={handleInputKeyDown} - placeholder={t('chat.typeMessage')} + placeholder={allowParallelSend ? t('chat.parallelBranchHint') : t('chat.typeMessage')} rows={1} - disabled={composerInteractionBlocked || isSending} + disabled={textareaDisabled} className="relative z-10 w-full resize-none border-0 bg-transparent py-0.5 px-0.5 text-sm leading-5 whitespace-pre-wrap break-words font-sans text-stone-900 dark:text-neutral-100 placeholder:text-stone-400 dark:placeholder:text-neutral-500 outline-none focus:outline-none focus-visible:outline-none focus:ring-0 focus-visible:ring-0 overflow-hidden disabled:opacity-50 disabled:cursor-not-allowed" /> diff --git a/app/src/lib/i18n/ar.ts b/app/src/lib/i18n/ar.ts index 383a78de5..0a65afa74 100644 --- a/app/src/lib/i18n/ar.ts +++ b/app/src/lib/i18n/ar.ts @@ -318,6 +318,8 @@ const messages: TranslationMap = { 'chat.newThread': 'محادثة جديدة', 'chat.typeMessage': 'كيف يمكنني مساعدتك اليوم؟', 'chat.send': 'إرسال الرسالة', + 'chat.parallelBranchHint': 'فرع متوازٍ — ⌘/Ctrl+Enter للإرسال', + 'chat.parallelBranchLabel': 'فرع متوازٍ', 'chat.thinking': 'جارٍ التفكير...', 'chat.noMessages': 'لا توجد رسائل بعد', 'chat.startConversation': 'ابدأ محادثة', diff --git a/app/src/lib/i18n/bn.ts b/app/src/lib/i18n/bn.ts index 9ef5b3fb2..f197e5b38 100644 --- a/app/src/lib/i18n/bn.ts +++ b/app/src/lib/i18n/bn.ts @@ -322,6 +322,8 @@ const messages: TranslationMap = { 'chat.newThread': 'নতুন থ্রেড', 'chat.typeMessage': 'আজ আমি আপনাকে কীভাবে সাহায্য করতে পারি?', 'chat.send': 'বার্তা পাঠান', + 'chat.parallelBranchHint': 'সমান্তরাল শাখা টাইপ করুন — পাঠাতে ⌘/Ctrl+Enter', + 'chat.parallelBranchLabel': 'সমান্তরাল শাখা', 'chat.thinking': 'ভাবছে...', 'chat.noMessages': 'এখনো কোনো বার্তা নেই', 'chat.startConversation': 'একটি কথোপকথন শুরু করুন', diff --git a/app/src/lib/i18n/de.ts b/app/src/lib/i18n/de.ts index 000f682d8..cf64d6499 100644 --- a/app/src/lib/i18n/de.ts +++ b/app/src/lib/i18n/de.ts @@ -333,6 +333,8 @@ const messages: TranslationMap = { 'chat.newThread': 'Neuer Thread', 'chat.typeMessage': 'Wie kann ich dir heute helfen?', 'chat.send': 'Nachricht senden', + 'chat.parallelBranchHint': 'Parallelen Zweig eingeben — ⌘/Strg+Enter zum Senden', + 'chat.parallelBranchLabel': 'Paralleler Zweig', 'chat.thinking': 'Denken...', 'chat.noMessages': 'Noch keine Nachrichten', 'chat.startConversation': 'Beginne ein Gespräch', diff --git a/app/src/lib/i18n/en.ts b/app/src/lib/i18n/en.ts index 531423a32..46cd4c523 100644 --- a/app/src/lib/i18n/en.ts +++ b/app/src/lib/i18n/en.ts @@ -359,6 +359,8 @@ const en: TranslationMap = { 'chat.newThread': 'New thread', 'chat.typeMessage': 'How can I help you today?', 'chat.send': 'Send message', + 'chat.parallelBranchHint': 'Type a parallel branch — ⌘/Ctrl+Enter to send', + 'chat.parallelBranchLabel': 'Parallel branch', 'chat.thinking': 'Thinking...', 'chat.noMessages': 'No messages yet', 'chat.startConversation': 'Start a conversation', diff --git a/app/src/lib/i18n/es.ts b/app/src/lib/i18n/es.ts index a14f27925..95b1b5222 100644 --- a/app/src/lib/i18n/es.ts +++ b/app/src/lib/i18n/es.ts @@ -335,6 +335,8 @@ const messages: TranslationMap = { 'chat.newThread': 'Nuevo hilo', 'chat.typeMessage': '¿En qué puedo ayudarte hoy?', 'chat.send': 'Enviar mensaje', + 'chat.parallelBranchHint': 'Escribe una rama paralela — ⌘/Ctrl+Enter para enviar', + 'chat.parallelBranchLabel': 'Rama paralela', 'chat.thinking': 'Pensando...', 'chat.noMessages': 'Sin mensajes aún', 'chat.startConversation': 'Inicia una conversación', diff --git a/app/src/lib/i18n/fr.ts b/app/src/lib/i18n/fr.ts index 7b3d4a9a3..e1fb283f1 100644 --- a/app/src/lib/i18n/fr.ts +++ b/app/src/lib/i18n/fr.ts @@ -334,6 +334,8 @@ const messages: TranslationMap = { 'chat.newThread': 'Nouveau fil', 'chat.typeMessage': "Comment puis-je t'aider aujourd'hui ?", 'chat.send': 'Envoyer le message', + 'chat.parallelBranchHint': 'Saisir une branche parallèle — ⌘/Ctrl+Entrée pour envoyer', + 'chat.parallelBranchLabel': 'Branche parallèle', 'chat.thinking': 'En train de réfléchir…', 'chat.noMessages': "Aucun message pour l'instant", 'chat.startConversation': 'Démarre une conversation', diff --git a/app/src/lib/i18n/hi.ts b/app/src/lib/i18n/hi.ts index cae9091a6..86f318732 100644 --- a/app/src/lib/i18n/hi.ts +++ b/app/src/lib/i18n/hi.ts @@ -321,6 +321,8 @@ const messages: TranslationMap = { 'chat.newThread': 'नई थ्रेड', 'chat.typeMessage': 'आज मैं आपकी कैसे मदद कर सकता हूँ?', 'chat.send': 'मैसेज भेजें', + 'chat.parallelBranchHint': 'समानांतर शाखा टाइप करें — भेजने के लिए ⌘/Ctrl+Enter', + 'chat.parallelBranchLabel': 'समानांतर शाखा', 'chat.thinking': 'सोच रहा है...', 'chat.noMessages': 'अभी कोई मैसेज नहीं', 'chat.startConversation': 'बातचीत शुरू करें', diff --git a/app/src/lib/i18n/id.ts b/app/src/lib/i18n/id.ts index 099127e7f..a5560ff23 100644 --- a/app/src/lib/i18n/id.ts +++ b/app/src/lib/i18n/id.ts @@ -325,6 +325,8 @@ const messages: TranslationMap = { 'chat.newThread': 'Thread baru', 'chat.typeMessage': 'Apa yang bisa saya bantu hari ini?', 'chat.send': 'Kirim pesan', + 'chat.parallelBranchHint': 'Ketik cabang paralel — ⌘/Ctrl+Enter untuk mengirim', + 'chat.parallelBranchLabel': 'Cabang paralel', 'chat.thinking': 'Berpikir...', 'chat.noMessages': 'Belum ada pesan', 'chat.startConversation': 'Mulai percakapan', diff --git a/app/src/lib/i18n/it.ts b/app/src/lib/i18n/it.ts index 04adae397..6a83e706c 100644 --- a/app/src/lib/i18n/it.ts +++ b/app/src/lib/i18n/it.ts @@ -330,6 +330,8 @@ const messages: TranslationMap = { 'chat.newThread': 'Nuovo thread', 'chat.typeMessage': 'Come posso aiutarti oggi?', 'chat.send': 'Invia messaggio', + 'chat.parallelBranchHint': 'Digita un ramo parallelo — ⌘/Ctrl+Invio per inviare', + 'chat.parallelBranchLabel': 'Ramo parallelo', 'chat.thinking': 'Sto pensando...', 'chat.noMessages': 'Nessun messaggio', 'chat.startConversation': 'Inizia una conversazione', diff --git a/app/src/lib/i18n/ko.ts b/app/src/lib/i18n/ko.ts index 4d6da4b30..14e112bbc 100644 --- a/app/src/lib/i18n/ko.ts +++ b/app/src/lib/i18n/ko.ts @@ -322,6 +322,8 @@ const messages: TranslationMap = { 'chat.newThread': '새 스레드', 'chat.typeMessage': '오늘 무엇을 도와드릴까요?', 'chat.send': '메시지 보내기', + 'chat.parallelBranchHint': '병렬 분기 입력 — 보내려면 ⌘/Ctrl+Enter', + 'chat.parallelBranchLabel': '병렬 분기', 'chat.thinking': '생각 중...', 'chat.noMessages': '아직 메시지가 없습니다', 'chat.startConversation': '대화 시작', diff --git a/app/src/lib/i18n/pl.ts b/app/src/lib/i18n/pl.ts index b5c48e4d6..0c1f0c57f 100644 --- a/app/src/lib/i18n/pl.ts +++ b/app/src/lib/i18n/pl.ts @@ -327,6 +327,8 @@ const messages: TranslationMap = { 'chat.newThread': 'Nowy wątek', 'chat.typeMessage': 'Jak mogę ci dziś pomóc?', 'chat.send': 'Wyślij wiadomość', + 'chat.parallelBranchHint': 'Wpisz równoległą gałąź — ⌘/Ctrl+Enter, aby wysłać', + 'chat.parallelBranchLabel': 'Równoległa gałąź', 'chat.thinking': 'Myślę...', 'chat.noMessages': 'Brak wiadomości', 'chat.startConversation': 'Rozpocznij rozmowę', diff --git a/app/src/lib/i18n/pt.ts b/app/src/lib/i18n/pt.ts index 8ba7251c7..2946414ec 100644 --- a/app/src/lib/i18n/pt.ts +++ b/app/src/lib/i18n/pt.ts @@ -334,6 +334,8 @@ const messages: TranslationMap = { 'chat.newThread': 'Nova conversa', 'chat.typeMessage': 'Como posso ajudá-lo hoje?', 'chat.send': 'Enviar mensagem', + 'chat.parallelBranchHint': 'Digite uma ramificação paralela — ⌘/Ctrl+Enter para enviar', + 'chat.parallelBranchLabel': 'Ramificação paralela', 'chat.thinking': 'Pensando...', 'chat.noMessages': 'Nenhuma mensagem ainda', 'chat.startConversation': 'Inicie uma conversa', diff --git a/app/src/lib/i18n/ru.ts b/app/src/lib/i18n/ru.ts index 98f2f15ba..8ded184ac 100644 --- a/app/src/lib/i18n/ru.ts +++ b/app/src/lib/i18n/ru.ts @@ -327,6 +327,8 @@ const messages: TranslationMap = { 'chat.newThread': 'Новый чат', 'chat.typeMessage': 'Чем я могу помочь тебе сегодня?', 'chat.send': 'Отправить сообщение', + 'chat.parallelBranchHint': 'Введите параллельную ветку — ⌘/Ctrl+Enter для отправки', + 'chat.parallelBranchLabel': 'Параллельная ветка', 'chat.thinking': 'Думаю...', 'chat.noMessages': 'Сообщений пока нет', 'chat.startConversation': 'Начни разговор', diff --git a/app/src/lib/i18n/zh-CN.ts b/app/src/lib/i18n/zh-CN.ts index c12e6f805..d4b2d6f55 100644 --- a/app/src/lib/i18n/zh-CN.ts +++ b/app/src/lib/i18n/zh-CN.ts @@ -307,6 +307,8 @@ const messages: TranslationMap = { 'chat.newThread': '新对话', 'chat.typeMessage': '今天我能帮您做什么？', 'chat.send': '发送', + 'chat.parallelBranchHint': '输入并行分支 — ⌘/Ctrl+Enter 发送', + 'chat.parallelBranchLabel': '并行分支', 'chat.thinking': '思考中...', 'chat.noMessages': '暂无消息', 'chat.startConversation': '开始对话', diff --git a/app/src/pages/Conversations.tsx b/app/src/pages/Conversations.tsx index a303d8b2b..fb36ee268 100644 --- a/app/src/pages/Conversations.tsx +++ b/app/src/pages/Conversations.tsx @@ -1,6 +1,6 @@ import { convertFileSrc } from '@tauri-apps/api/core'; import debugFactory from 'debug'; -import { useEffect, useMemo, useRef, useState } from 'react'; +import { useCallback, useEffect, useMemo, useRef, useState } from 'react'; import { useLocation, useNavigate } from 'react-router-dom'; import { type ChatSendError, chatSendError } from '../chat/chatSendError'; @@ -42,7 +42,7 @@ import { beginInferenceTurn, clearRuntimeForThread, fetchAndHydrateTurnState, - type InferenceStatus, + registerParallelRequest, setTaskBoardForThread, setToolTimelineForThread, } from '../store/chatRuntimeSlice'; @@ -50,12 +50,13 @@ import { useAppDispatch, useAppSelector } from '../store/hooks'; import { selectSocketStatus } from '../store/socketSelectors'; import { addMessageLocal, + clearThreadInferenceActive, createNewThread, deleteThread, loadThreadMessages, loadThreads, + markThreadInferenceActive, persistReaction, - setActiveThread, setSelectedThread, setThreadSidebarVisible, THREAD_NOT_FOUND_MESSAGE, @@ -138,11 +139,17 @@ interface ConversationsProps { composer?: 'text' | 'mic-cloud'; } +// Stable empty reference so the `activeThreadIds` selector returns the same +// object identity when the slice field is absent (narrow test stores), +// avoiding spurious re-renders. +const EMPTY_ACTIVE_THREADS: Record = {}; + export function isComposerInteractionBlocked(args: { - activeThreadId: string | null; + /** Whether the *currently selected* thread has an in-flight inference turn. */ + selectedThreadActive: boolean; rustChat: boolean; }): boolean { - return !args.rustChat || Boolean(args.activeThreadId); + return !args.rustChat || args.selectedThreadActive; } interface ImeKeyboardEventLike { @@ -198,8 +205,19 @@ const Conversations = ({ messages, isLoadingMessages, messagesError, - activeThreadId, } = useAppSelector(state => state.thread); + // Optional-chain + default: narrow test stores may omit `activeThreadIds`. + const activeThreadIds = useAppSelector( + state => state.thread.activeThreadIds ?? EMPTY_ACTIVE_THREADS + ); + // Per-thread inference tracking (parallel inference): the selected thread's + // own in-flight state gates the composer; a turn running on a *different* + // thread no longer locks this one. `firstActiveThreadId` is a best-effort + // fallback for thread-scoped chips/panels when no thread is selected. + const selectedThreadActive = selectedThreadId + ? Boolean(activeThreadIds[selectedThreadId]) + : false; + const firstActiveThreadId = Object.keys(activeThreadIds)[0] ?? null; const [inputValue, setInputValue] = useState(''); const [attachments, setAttachments] = useState([]); @@ -223,7 +241,28 @@ const Conversations = ({ const [attachError, setAttachError] = useState(null); const [sendAdvisory, setSendAdvisory] = useState(null); const [openRouterStatus, setOpenRouterStatus] = useState<'idle' | 'saving' | 'error'>('idle'); - const [pendingSendingThreadId, setPendingSendingThreadId] = useState(null); + // Threads whose send is mid-flight (dispatched locally, backend not yet + // accepted). A Set so concurrent sends to different threads each track their + // own pending state instead of clobbering a single slot. + const [pendingSendingThreadIds, setPendingSendingThreadIds] = useState>( + () => new Set() + ); + const addPendingSendingThread = useCallback((threadId: string) => { + setPendingSendingThreadIds(prev => { + if (prev.has(threadId)) return prev; + const next = new Set(prev); + next.add(threadId); + return next; + }); + }, []); + const removePendingSendingThread = useCallback((threadId: string) => { + setPendingSendingThreadIds(prev => { + if (!prev.has(threadId)) return prev; + const next = new Set(prev); + next.delete(threadId); + return next; + }); + }, []); const socketStatus = useAppSelector(selectSocketStatus); const agentProfiles = useAppSelector(selectAgentProfiles); const selectedAgentProfileId = useAppSelector(selectActiveAgentProfileId); @@ -244,6 +283,9 @@ const Conversations = ({ const streamingAssistantByThread = useAppSelector( state => state.chatRuntime.streamingAssistantByThread ); + const parallelStreamsByThread = useAppSelector( + state => state.chatRuntime.parallelStreamsByThread + ); const agentMessageViewMode = useAppSelector( state => state.theme?.agentMessageViewMode ?? 'bubbles' ); @@ -312,7 +354,10 @@ const Conversations = ({ const textInputRef = useRef(null); const isComposingTextRef = useRef(false); - const pendingSendRef = useRef(null); + // Threads with an in-flight send, guarding against double-submit to the SAME + // thread. Per-thread (a Set) so a send to thread B isn't blocked by an + // in-flight send to thread A. + const pendingSendsRef = useRef>(new Set()); const mediaRecorderRef = useRef(null); const mediaStreamRef = useRef(null); const audioChunksRef = useRef([]); @@ -320,17 +365,21 @@ const Conversations = ({ const lastSpokenMessageIdRef = useRef(null); const autocompleteDebounceRef = useRef(null); const autocompleteRequestSeqRef = useRef(0); - const sendingTimeoutRef = useRef | null>(null); - // Thread id whose send started the current silence timer. Tracked separately - // from `selectedThreadId` so switching threads mid-turn doesn't move the - // timer's reference point. - const sendingThreadIdRef = useRef(null); + // Per-thread silence timers. Each in-flight turn gets its own 120s safety + // timer keyed by thread id, so concurrent turns on different threads don't + // share (and clobber) a single timeout. + const sendingTimeoutsRef = useRef>>(new Map()); // Ref so the mount-time dictation event handler can call the latest send fn. const handleSendMessageRef = useRef<((text?: string) => Promise) | null>(null); - // Previous inference status for the sending thread; lets the rearm effect - // distinguish "status was just cleared (chat_done / chat_error)" from - // "status was never set yet (in-flight turn pre-status)". - const prevInferenceStatusRef = useRef(undefined); + // Per-thread "turn signature": the last-seen tuple of progress-slice + // references [inferenceStatus, streamingAssistant, toolTimeline, taskBoard] + // for each thread that owns a live silence timer. Redux Toolkit (immer) + // only produces new references for the thread whose slice actually changed, + // so comparing references lets the rearm effect (a) detect a turn completing + // (status defined → undefined) and (b) rearm a thread's timer ONLY when that + // thread's own state changed — unrelated threads' activity must not keep a + // foreground turn's timer alive. + const turnSignatureByThreadRef = useRef>(new Map()); const getAudioExtension = (mimeType: string): string => { const lower = mimeType.toLowerCase(); @@ -524,24 +573,30 @@ const Conversations = ({ } }, [inputValue, sendAdvisory, sendError]); + const clearSilenceTimer = useCallback((threadId: string) => { + const existing = sendingTimeoutsRef.current.get(threadId); + if (existing) { + clearTimeout(existing); + sendingTimeoutsRef.current.delete(threadId); + } + }, []); + const armSilenceTimer = (threadId: string) => { - if (sendingTimeoutRef.current) clearTimeout(sendingTimeoutRef.current); - sendingThreadIdRef.current = threadId; - sendingTimeoutRef.current = setTimeout(() => { - debug('armSilenceTimer: no inference signal for 120s — clearing runtime'); + clearSilenceTimer(threadId); + const timeout = setTimeout(() => { + debug(`armSilenceTimer: no inference signal for 120s — clearing runtime (${threadId})`); setSendError(chatSendError('safety_timeout', t('chat.safetyTimeout'))); dispatch(clearRuntimeForThread({ threadId })); - dispatch(setActiveThread(null)); - sendingTimeoutRef.current = null; - sendingThreadIdRef.current = null; - // Reset so the NEXT send starts from a clean "never had a status" - // baseline — otherwise the rearm effect could read this turn's last - // status as a stale "previous" and falsely treat the next send's - // first signal as a chat-done transition. - prevInferenceStatusRef.current = undefined; - pendingSendRef.current = null; - setPendingSendingThreadId(null); + dispatch(clearThreadInferenceActive(threadId)); + sendingTimeoutsRef.current.delete(threadId); + // Reset so the NEXT send to this thread starts from a clean baseline — + // otherwise the rearm effect could read this turn's last signature as a + // stale "previous" and mis-handle the next send's first signal. + turnSignatureByThreadRef.current.delete(threadId); + pendingSendsRef.current.delete(threadId); + removePendingSendingThread(threadId); }, 120_000); + sendingTimeoutsRef.current.set(threadId, timeout); }; // Rearm the silence timer on every inference signal for the sending @@ -556,42 +611,47 @@ const Conversations = ({ // (chat_done / chat_error), drop the timer — the completion handlers // own UI cleanup. // - // `prevInferenceStatusRef` distinguishes "status was just cleared - // (chat_done / chat_error transition: defined → undefined)" from "status - // was never set yet (the Send handler also dispatches - // `setToolTimelineForThread({ entries: [] })` to reset the timeline, - // which fires this effect immediately after `armSilenceTimer` — at - // that instant the inference status hasn't been published yet)". Only - // the real transition should clear our timer. + // Rearm each live silence timer when its OWN thread shows progress, and drop + // it when that thread's turn completes. With parallel inference several + // timers may be live at once, so we iterate every thread that currently owns + // a timer. Per-thread reference comparison (see `turnSignatureByThreadRef`) + // ensures an unrelated thread's activity does NOT rearm this thread's timer, + // while still catching pure-text streams and sub-agent tool/board activity + // that bump the other slices without re-emitting a top-level status. + // + // The done-transition (status defined → undefined) is detected per thread to + // distinguish "turn just finished (chat_done / chat_error)" from "status + // never set yet" — the Send handler dispatches `setToolTimelineForThread([])` + // immediately after arming, firing this effect before any status publishes. useEffect(() => { - const threadId = sendingThreadIdRef.current; - if (!threadId || !sendingTimeoutRef.current) return; - const status = inferenceStatusByThread[threadId]; - if (status === undefined && prevInferenceStatusRef.current !== undefined) { - clearTimeout(sendingTimeoutRef.current); - sendingTimeoutRef.current = null; - sendingThreadIdRef.current = null; - prevInferenceStatusRef.current = undefined; - return; + for (const threadId of Array.from(sendingTimeoutsRef.current.keys())) { + const current = [ + inferenceStatusByThread[threadId], + streamingAssistantByThread[threadId], + toolTimelineByThread[threadId], + taskBoardByThread[threadId], + ] as const; + const previous = turnSignatureByThreadRef.current.get(threadId); + const status = current[0]; + const previousStatus = previous?.[0]; + if (status === undefined && previousStatus !== undefined) { + clearSilenceTimer(threadId); + turnSignatureByThreadRef.current.delete(threadId); + continue; + } + const changed = !previous || previous.some((value, index) => value !== current[index]); + if (!changed) continue; + turnSignatureByThreadRef.current.set(threadId, current); + armSilenceTimer(threadId); } - prevInferenceStatusRef.current = status; - armSilenceTimer(threadId); - // Scope the dependencies to the SENDING thread's slices only, keyed by the - // reactive `activeThreadId` (set on send, cleared on done/error/timeout — - // so it tracks the in-flight turn for the timer's whole lifetime, unlike - // `pendingSendingThreadId` which is released the moment the backend accepts - // the send). Depending on the whole maps would rearm this thread's timer - // whenever ANY other thread's state changed — unrelated background activity - // shouldn't keep a foreground turn's timer alive. armSilenceTimer is stable - // (refs + dispatch), so listing the per-thread values is enough to rearm on - // every progress event for this thread. + // armSilenceTimer / clearSilenceTimer are stable (refs + dispatch); + // depending on the progress maps rearms live timers on every signal. // eslint-disable-next-line react-hooks/exhaustive-deps }, [ - activeThreadId, - activeThreadId ? inferenceStatusByThread[activeThreadId] : undefined, - activeThreadId ? streamingAssistantByThread[activeThreadId] : undefined, - activeThreadId ? toolTimelineByThread[activeThreadId] : undefined, - activeThreadId ? taskBoardByThread[activeThreadId] : undefined, + inferenceStatusByThread, + streamingAssistantByThread, + toolTimelineByThread, + taskBoardByThread, ]); useEffect(() => { @@ -599,7 +659,7 @@ const Conversations = ({ !isTauri() || !rustChat || inputMode !== 'text' || - Boolean(activeThreadId) || + selectedThreadActive || inputValue.trim().length < AUTOCOMPLETE_MIN_CONTEXT_CHARS ) { setInlineSuggestionValue(''); @@ -631,7 +691,7 @@ const Conversations = ({ autocompleteDebounceRef.current = null; } }; - }, [activeThreadId, inputValue, inputMode, rustChat]); + }, [selectedThreadActive, inputValue, inputMode, rustChat]); useEffect(() => { return () => { @@ -739,7 +799,9 @@ const Conversations = ({ }; const handleSendMessage = async (text?: string) => { - if (pendingSendRef.current) return; + // Guard double-submit to the SAME thread only; a send to another thread + // may proceed concurrently. + if (selectedThreadId && pendingSendsRef.current.has(selectedThreadId)) return; const normalized = text ?? inputValue; const trimmedInput = normalized.trim(); @@ -783,8 +845,8 @@ const Conversations = ({ const sendingThreadId = selectedThreadId; if (!sendingThreadId) return; - pendingSendRef.current = sendingThreadId; - setPendingSendingThreadId(sendingThreadId); + pendingSendsRef.current.add(sendingThreadId); + addPendingSendingThread(sendingThreadId); const pendingAttachments = attachments.slice(); const modelOverride = agentProfiles.find(p => p.id === selectedAgentProfileId)?.modelOverride ?? CHAT_MODEL_HINT; @@ -818,14 +880,14 @@ const Conversations = ({ // unrelated errors whose `.toString()` happens to equal the sentinel. if (error === THREAD_NOT_FOUND_MESSAGE) { setSendError(null); - pendingSendRef.current = null; - setPendingSendingThreadId(null); + pendingSendsRef.current.delete(sendingThreadId); + removePendingSendingThread(sendingThreadId); return; } const msg = error instanceof Error ? error.message : String(error); setSendError(chatSendError('cloud_send_failed', msg)); - pendingSendRef.current = null; - setPendingSendingThreadId(null); + pendingSendsRef.current.delete(sendingThreadId); + removePendingSendingThread(sendingThreadId); return; } setInputValue(''); @@ -841,11 +903,11 @@ const Conversations = ({ // Fresh send: clear the previous-status baseline before arming so the // first inference signal of this turn isn't misread as a chat-done // transition (defined → undefined) left over from the prior turn. - prevInferenceStatusRef.current = undefined; + turnSignatureByThreadRef.current.delete(sendingThreadId); armSilenceTimer(sendingThreadId); dispatch(setToolTimelineForThread({ threadId: sendingThreadId, entries: [] })); dispatch(beginInferenceTurn({ threadId: sendingThreadId })); - dispatch(setActiveThread(sendingThreadId)); + dispatch(markThreadInferenceActive(sendingThreadId)); // ── Cloud socket path ───────────────────────────────────────────────────── // Always route primary chat through the cloud backend via socket. @@ -863,18 +925,14 @@ const Conversations = ({ // Backend accepted the send; lifecycle ('started' → 'streaming') now // owns the `isSending` UI lock. Release the pending guard so the next // user turn isn't blocked by a stale ref/state. - pendingSendRef.current = null; - setPendingSendingThreadId(null); + pendingSendsRef.current.delete(sendingThreadId); + removePendingSendingThread(sendingThreadId); // Active-thread reset happens in the global ChatRuntimeProvider events. } catch (err) { // Chat loop errors are emitted via socket events; this catch handles emit-level failures. - if (sendingTimeoutRef.current) { - clearTimeout(sendingTimeoutRef.current); - sendingTimeoutRef.current = null; - } - sendingThreadIdRef.current = null; - prevInferenceStatusRef.current = undefined; + clearSilenceTimer(sendingThreadId); + turnSignatureByThreadRef.current.delete(sendingThreadId); const msg = err instanceof Error ? err.message : String(err); if ( msg.toLowerCase().includes('blocked by a security policy') || @@ -888,14 +946,83 @@ const Conversations = ({ setSendError(chatSendError('cloud_send_failed', msg)); } dispatch(clearRuntimeForThread({ threadId: sendingThreadId })); - dispatch(setActiveThread(null)); - pendingSendRef.current = null; - setPendingSendingThreadId(null); + dispatch(clearThreadInferenceActive(sendingThreadId)); + pendingSendsRef.current.delete(sendingThreadId); + removePendingSendingThread(sendingThreadId); } }; handleSendMessageRef.current = handleSendMessage; + // Send a PARALLEL (forked) turn on the selected thread — runs concurrently + // with the in-flight turn instead of interrupting it (queue_mode 'parallel'). + // Kept separate from `handleSendMessage` so it never touches the primary + // turn's lifecycle (silence timer, active marker, pending guard); the forked + // turn streams into its own lane (registered via `registerParallelRequest`) + // and renders as an interleaved branch bubble. + const handleSendParallel = async (text?: string) => { + if (!rustChat || !selectedThreadId) return; + const threadId = selectedThreadId; + const normalized = (text ?? inputValue).trim(); + if (!normalized && attachments.length === 0) return; + + const pendingAttachments = attachments.slice(); + const modelOverride = + agentProfiles.find(p => p.id === selectedAgentProfileId)?.modelOverride ?? CHAT_MODEL_HINT; + const messageText = buildMessageWithAttachments(normalized, pendingAttachments); + const userMessage: ThreadMessage = { + id: `msg_${globalThis.crypto.randomUUID()}`, + content: normalized, + type: 'text', + extraMetadata: + pendingAttachments.length > 0 + ? { + attachmentCount: pendingAttachments.length, + attachmentNames: pendingAttachments.map(a => a.file.name), + attachmentKinds: pendingAttachments.map(a => a.kind), + attachmentDataUris: pendingAttachments + .filter(a => a.kind === 'image') + .map(a => a.previewUri ?? a.dataUri), + attachmentCompressed: pendingAttachments.map(a => a.compressed), + parallelBranch: true, + } + : { parallelBranch: true }, + sender: 'user', + createdAt: new Date().toISOString(), + }; + + try { + await dispatch(addMessageLocal({ threadId, message: userMessage })).unwrap(); + } catch (error) { + if (error === THREAD_NOT_FOUND_MESSAGE) return; + const msg = error instanceof Error ? error.message : String(error); + setSendError(chatSendError('cloud_send_failed', msg)); + return; + } + + setInputValue(''); + setAttachments([]); + setSendError(null); + + try { + const requestId = await chatSend({ + threadId, + message: messageText, + model: modelOverride, + profileId: selectedAgentProfileId, + locale: uiLocale, + queueMode: 'parallel', + }); + if (requestId) { + dispatch(registerParallelRequest({ threadId, requestId })); + } + trackEvent('chat_parallel_message_sent'); + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); + setSendError(chatSendError('cloud_send_failed', msg)); + } + }; + const transcribeAndSendAudio = async (mimeType: string) => { setIsRecording(false); mediaRecorderRef.current = null; @@ -958,7 +1085,7 @@ const Conversations = ({ }; const handleVoiceRecordToggle = async () => { - if (!rustChat || Boolean(activeThreadId) || isTranscribing) return; + if (!rustChat || selectedThreadActive || isTranscribing) return; if (!canUseMicrophoneApi) { setSendError( chatSendError( @@ -1106,6 +1233,18 @@ const Conversations = ({ return; } + // Cmd/Ctrl+Enter sends a PARALLEL branch when the selected thread already + // has a turn in flight (otherwise it behaves like a normal send). + if (e.key === 'Enter' && (e.metaKey || e.ctrlKey)) { + e.preventDefault(); + if (selectedThreadActive) { + void handleSendParallel(); + } else { + void handleSendMessage(); + } + return; + } + if (e.key === 'Enter' && !e.shiftKey) { e.preventDefault(); void handleSendMessage(); @@ -1151,10 +1290,18 @@ const Conversations = ({ const selectedStreamingAssistant = selectedThreadId ? (streamingAssistantByThread[selectedThreadId] ?? null) : null; + // Live streams for concurrent parallel (forked) turns on the selected thread, + // rendered as separate interleaved branch bubbles. + const selectedParallelStreams = selectedThreadId + ? Object.values(parallelStreamsByThread[selectedThreadId] ?? {}) + : []; const inlineCompletionSuffix = getInlineCompletionSuffix(inputValue, inlineSuggestionValue); // Blocks all composer interaction while a turn is in-flight or Rust chat is unavailable. // isSending: the *selected* thread is in-flight (drives selected-thread UI only). - const composerInteractionBlocked = isComposerInteractionBlocked({ activeThreadId, rustChat }); + const composerInteractionBlocked = isComposerInteractionBlocked({ + selectedThreadActive, + rustChat, + }); // Auto-focus the composer when a thread becomes selected and the composer // isn't blocked. Without this, navigating into a thread from elsewhere in // the app (e.g. acting on a subconscious reflection in the Intelligence @@ -1193,7 +1340,7 @@ const Conversations = ({ const isSending = Boolean( selectedThreadId && - (pendingSendingThreadId === selectedThreadId || + (pendingSendingThreadIds.has(selectedThreadId) || inferenceTurnLifecycleByThread[selectedThreadId] === 'started' || inferenceTurnLifecycleByThread[selectedThreadId] === 'streaming') ); @@ -1560,8 +1707,8 @@ const Conversations = ({ {t('chat.agentProfile.reasoning')} - {(selectedThreadId ?? activeThreadId) && ( - + {(selectedThreadId ?? firstActiveThreadId) && ( + )}