feat(chat): parallel agent inference (cross-thread + within-thread forks) (#3633)

app/src/components/chat/ChatComposer.tsx

@@ -16,6 +16,13 @@ export interface ChatComposerProps {
   fileInputRef: React.RefObject<HTMLInputElement | null>;
   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<void>;
   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"
           />
         </div>

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': 'ابدأ محادثة',

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': 'একটি কথোপকথন শুরু করুন',

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',

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',

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',

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',

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': 'बातचीत शुरू करें',

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',

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',

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': '대화 시작',

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ę',

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',

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': 'Начни разговор',

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': '开始对话',

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<string, true> = {};
+
 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<Attachment[]>([]);
@@ -223,7 +241,28 @@ const Conversations = ({
   const [attachError, setAttachError] = useState<ChatSendError | null>(null);
   const [sendAdvisory, setSendAdvisory] = useState<string | null>(null);
   const [openRouterStatus, setOpenRouterStatus] = useState<'idle' | 'saving' | 'error'>('idle');
-  const [pendingSendingThreadId, setPendingSendingThreadId] = useState<string | null>(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<ReadonlySet<string>>(
+    () => 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<HTMLTextAreaElement>(null);
   const isComposingTextRef = useRef(false);
-  const pendingSendRef = useRef<string | null>(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<Set<string>>(new Set());
   const mediaRecorderRef = useRef<MediaRecorder | null>(null);
   const mediaStreamRef = useRef<MediaStream | null>(null);
   const audioChunksRef = useRef<Blob[]>([]);
@@ -320,17 +365,21 @@ const Conversations = ({
   const lastSpokenMessageIdRef = useRef<string | null>(null);
   const autocompleteDebounceRef = useRef<number | null>(null);
   const autocompleteRequestSeqRef = useRef(0);
-  const sendingTimeoutRef = useRef<ReturnType<typeof setTimeout> | 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<string | null>(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<Map<string, ReturnType<typeof setTimeout>>>(new Map());
   // Ref so the mount-time dictation event handler can call the latest send fn.
   const handleSendMessageRef = useRef<((text?: string) => Promise<void>) | 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<InferenceStatus | undefined>(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<Map<string, readonly unknown[]>>(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')}
                 </button>
               </div>
-              {(selectedThreadId ?? activeThreadId) && (
-                <ChatFilesChip threadId={(selectedThreadId ?? activeThreadId) as string} />
+              {(selectedThreadId ?? firstActiveThreadId) && (
+                <ChatFilesChip threadId={(selectedThreadId ?? firstActiveThreadId) as string} />
               )}
               <button
                 type="button"
@@ -1980,6 +2127,33 @@ const Conversations = ({
                     </div>
                   </div>
                 )}
+              {/* Parallel (forked) branch streams — concurrent turns on this
+                  thread, each its own labeled bubble so they don't collide with
+                  the primary stream above. */}
+              {selectedParallelStreams.map(
+                branch =>
+                  (branch.content.length > 0 || branch.thinking.length > 0) && (
+                    <div key={branch.requestId} className="flex justify-start">
+                      <div className="relative w-fit max-w-[75%]">
+                        <div className="mb-1 flex items-center gap-1.5 text-[10px] font-medium uppercase tracking-wide text-primary-500 dark:text-primary-400">
+                          <span className="inline-block w-1.5 h-1.5 rounded-full bg-primary-400 animate-pulse" />
+                          <span>{t('chat.parallelBranchLabel')}</span>
+                        </div>
+                        {branch.content.length > 0 && (
+                          <div className="rounded-2xl rounded-bl-md px-3 py-1.5 bg-stone-200/80 dark:bg-neutral-800 text-stone-900 dark:text-neutral-100 border-l-2 border-primary-400/60">
+                            <p className="text-xs text-stone-700 dark:text-neutral-200 font-mono whitespace-pre-wrap break-words leading-snug">
+                              {branch.content.length > STREAMING_PREVIEW_CHARS && (
+                                <span className="text-stone-400 dark:text-neutral-500">…</span>
+                              )}
+                              {branch.content.slice(-STREAMING_PREVIEW_CHARS)}
+                              <span className="inline-block w-1 h-3 ml-0.5 align-middle bg-primary-400 animate-pulse" />
+                            </p>
+                          </div>
+                        )}
+                      </div>
+                    </div>
+                  )
+              )}
               {/* Inference status indicator.
                   For the tool_use / subagent phases this line just restates the
                   active row already shown in the agentic-task-insights timeline,
@@ -2200,7 +2374,7 @@ const Conversations = ({
           {(() => {
             // Surface a parked ApprovalGate request for the shown thread just
             // above the composer, so it stays visible regardless of scroll.
-            const approvalThreadId = selectedThreadId ?? activeThreadId;
+            const approvalThreadId = selectedThreadId ?? firstActiveThreadId;
             const pendingApproval = approvalThreadId
               ? pendingApprovalByThread[approvalThreadId]
               : undefined;
@@ -2225,7 +2399,7 @@ const Conversations = ({
             // tool call) is tracked in follow-up issue #3162. The
             // failed-card UI still surfaces the truncated error reason;
             // the button just stays hidden until #3162 lands.
-            const artifactThreadId = selectedThreadId ?? activeThreadId;
+            const artifactThreadId = selectedThreadId ?? firstActiveThreadId;
             const all = artifactThreadId ? (artifactsByThread[artifactThreadId] ?? []) : [];
             const live = all.filter(a => a.status !== 'ready');
             if (live.length === 0) return null;
@@ -2260,6 +2434,7 @@ const Conversations = ({
               fileInputRef={fileInputRef}
               composerInteractionBlocked={composerInteractionBlocked}
               isSending={isSending}
+              allowParallelSend={selectedThreadActive}
               attachments={attachments}
               onAttachFiles={handleAttachFiles}
               onRemoveAttachment={id => setAttachments(prev => prev.filter(a => a.id !== id))}

app/src/pages/__tests__/Conversations.attachments.test.tsx

@@ -231,7 +231,7 @@ async function renderWithSelectedThread() {
     thread: {
       threads: [thread],
       selectedThreadId: thread.id,
-      activeThreadId: null,
+      activeThreadIds: {},
       welcomeThreadId: null,
       messagesByThreadId: { [thread.id]: [] },
       messages: [],
@@ -548,7 +548,7 @@ describe('Conversations — attachment feature', () => {
       thread: {
         threads: [thread],
         selectedThreadId: thread.id,
-        activeThreadId: null,
+        activeThreadIds: {},
         welcomeThreadId: null,
         messagesByThreadId: { [thread.id]: [message] },
         messages: [message],
@@ -597,7 +597,7 @@ describe('Conversations — attachment feature', () => {
       thread: {
         threads: [thread],
         selectedThreadId: thread.id,
-        activeThreadId: null,
+        activeThreadIds: {},
         welcomeThreadId: null,
         messagesByThreadId: { [thread.id]: [message] },
         messages: [message],

app/src/pages/__tests__/Conversations.render.test.tsx

@@ -221,7 +221,7 @@ const emptyThreadState = {
   threads: [],
   selectedThreadId: null,
   threadSidebarVisible: false,
-  activeThreadId: null,
+  activeThreadIds: {},
   welcomeThreadId: null,
   messagesByThreadId: {},
   messages: [],
@@ -870,8 +870,8 @@ describe('Conversations — smoke render (#1123 welcome-lock removal)', () => {
     let resolveSend: (() => void) | undefined;
     vi.mocked(chatSend).mockImplementationOnce(
       () =>
-        new Promise<void>(resolve => {
-          resolveSend = resolve;
+        new Promise<string | undefined>(resolve => {
+          resolveSend = () => resolve(undefined);
         })
     );
     const { textarea, thread } = await renderSelectedConversation();

app/src/pages/__tests__/Conversations.test.tsx

@@ -7,16 +7,20 @@ import {
 } from '../Conversations';
 
 describe('isComposerInteractionBlocked', () => {
-  it('blocks composer interaction while a thread is actively running', () => {
-    expect(isComposerInteractionBlocked({ activeThreadId: 'thread-1', rustChat: true })).toBe(true);
+  it('blocks composer interaction while the selected thread is actively running', () => {
+    expect(isComposerInteractionBlocked({ selectedThreadActive: true, rustChat: true })).toBe(true);
   });
 
-  it('allows composer interaction when chat is idle and ready', () => {
-    expect(isComposerInteractionBlocked({ activeThreadId: null, rustChat: true })).toBe(false);
+  it('allows composer interaction when the selected thread is idle and ready', () => {
+    expect(isComposerInteractionBlocked({ selectedThreadActive: false, rustChat: true })).toBe(
+      false
+    );
   });
 
   it('blocks composer interaction when rust chat is unavailable', () => {
-    expect(isComposerInteractionBlocked({ activeThreadId: null, rustChat: false })).toBe(true);
+    expect(isComposerInteractionBlocked({ selectedThreadActive: false, rustChat: false })).toBe(
+      true
+    );
   });
 });
 

app/src/pages/__tests__/Conversations.welcomeLock.test.tsx

@@ -20,17 +20,19 @@ describe('[#1123] Conversations — unlocked flow (welcome-lock removed)', () =>
     // The welcome-lock previously would have been active here
     // (chatOnboardingCompleted=false → welcomeLocked=true → composer blocked).
     // After #1123 there is no welcomeLocked state, so the composer is unblocked.
-    expect(isComposerInteractionBlocked({ activeThreadId: null, rustChat: true })).toBe(false);
-  });
-
-  it('still blocks when an agent thread is actively running (not a welcome-lock concern)', () => {
-    expect(isComposerInteractionBlocked({ activeThreadId: 'thread-xyz', rustChat: true })).toBe(
-      true
+    expect(isComposerInteractionBlocked({ selectedThreadActive: false, rustChat: true })).toBe(
+      false
     );
   });
 
+  it('still blocks when an agent thread is actively running (not a welcome-lock concern)', () => {
+    expect(isComposerInteractionBlocked({ selectedThreadActive: true, rustChat: true })).toBe(true);
+  });
+
   it('blocks when rust chat transport is unavailable', () => {
-    expect(isComposerInteractionBlocked({ activeThreadId: null, rustChat: false })).toBe(true);
+    expect(isComposerInteractionBlocked({ selectedThreadActive: false, rustChat: false })).toBe(
+      true
+    );
   });
 
   it('resolves thread display title to thread title (no "Onboarding" override)', () => {
@@ -43,6 +45,8 @@ describe('[#1123] Conversations — unlocked flow (welcome-lock removed)', () =>
     // The title override was in the component body (not exported separately)
     // so this test simply confirms the exported composer gate does not
     // special-case any thread as a "welcome thread".
-    expect(isComposerInteractionBlocked({ activeThreadId: null, rustChat: true })).toBe(false);
+    expect(isComposerInteractionBlocked({ selectedThreadActive: false, rustChat: true })).toBe(
+      false
+    );
   });
 });

app/src/providers/ChatRuntimeProvider.tsx

@@ -23,6 +23,7 @@ import { store } from '../store';
 import {
   appendSubagentStreamDelta,
   clearInferenceStatusForThread,
+  clearParallelRequest,
   clearPendingApprovalForThread,
   clearStreamingAssistantForThread,
   endInferenceTurn,
@@ -31,6 +32,7 @@ import {
   recordSubagentTranscriptTool,
   resolveSubagentTranscriptTool,
   setInferenceStatusForThread,
+  setParallelStream,
   setPendingApprovalForThread,
   setStreamingAssistantForThread,
   setTaskBoardForThread,
@@ -45,6 +47,7 @@ import { useAppDispatch, useAppSelector } from '../store/hooks';
 import { selectSocketStatus } from '../store/socketSelectors';
 import {
   addInferenceResponse,
+  clearThreadInferenceActive,
   createNewThread,
   generateThreadTitleIfNeeded,
   setActiveThread,
@@ -259,10 +262,12 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => {
       }
 
       const state = store.getState().thread;
-      // Resolution priority: selected > active (in-flight inference) > first thread.
-      // `activeThreadId` tracks the currently running inference thread.
+      // Resolution priority: selected > any in-flight inference thread > first
+      // thread. With parallel inference there may be several active threads;
+      // any one is an acceptable proactive target when nothing is selected.
+      const firstActiveThreadId = Object.keys(state.activeThreadIds)[0] ?? null;
       const targetFromState =
-        state.selectedThreadId ?? state.activeThreadId ?? state.threads[0]?.id ?? null;
+        state.selectedThreadId ?? firstActiveThreadId ?? state.threads[0]?.id ?? null;
       if (targetFromState) {
         return targetFromState;
       }
@@ -319,7 +324,7 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => {
       });
       refetchSnapshot();
       dispatch(endInferenceTurn({ threadId: event.thread_id }));
-      dispatch(setActiveThread(null));
+      dispatch(clearThreadInferenceActive(event.thread_id));
     };
 
     rtLog('subscribe_chat_events', { socket: socketStatus });
@@ -695,6 +700,22 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => {
       },
       onTextDelta: event => {
         const cr = store.getState().chatRuntime;
+        // A parallel (forked) turn streams into its own lane so it doesn't
+        // clobber the primary turn's stream on the same thread.
+        if (cr.parallelRequestThreads[event.request_id] !== undefined) {
+          const prev = cr.parallelStreamsByThread[event.thread_id]?.[event.request_id];
+          dispatch(
+            setParallelStream({
+              threadId: event.thread_id,
+              streaming: {
+                requestId: event.request_id,
+                content: `${prev?.content ?? ''}${event.delta}`,
+                thinking: prev?.thinking ?? '',
+              },
+            })
+          );
+          return;
+        }
         const existing = cr.streamingAssistantByThread[event.thread_id];
         let streaming: StreamingAssistantState;
         if (existing && existing.requestId !== event.request_id) {
@@ -710,6 +731,20 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => {
       },
       onThinkingDelta: event => {
         const cr = store.getState().chatRuntime;
+        if (cr.parallelRequestThreads[event.request_id] !== undefined) {
+          const prev = cr.parallelStreamsByThread[event.thread_id]?.[event.request_id];
+          dispatch(
+            setParallelStream({
+              threadId: event.thread_id,
+              streaming: {
+                requestId: event.request_id,
+                content: prev?.content ?? '',
+                thinking: `${prev?.thinking ?? ''}${event.delta}`,
+              },
+            })
+          );
+          return;
+        }
         const existing = cr.streamingAssistantByThread[event.thread_id];
         let streaming: StreamingAssistantState;
         if (existing && existing.requestId !== event.request_id) {
@@ -878,6 +913,52 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => {
           output_tokens: event.total_output_tokens,
         });
 
+        // Parallel (forked) turn: resolve only its own lane. The primary turn's
+        // stream / status / lifecycle / active marker may still be running, so
+        // we must NOT clear them here. Segmented parallel turns already
+        // persisted via `onSegment` (keyed by thread+request); a single-bubble
+        // parallel turn persists its full response now.
+        if (
+          event.request_id !== undefined &&
+          store.getState().chatRuntime.parallelRequestThreads[event.request_id] !== undefined
+        ) {
+          const parallelRequestId = event.request_id;
+          dispatch(
+            recordChatTurnUsage({
+              inputTokens: event.total_input_tokens,
+              outputTokens: event.total_output_tokens,
+            })
+          );
+          if (!event.segment_total && event.full_response.length > 0) {
+            void (async () => {
+              try {
+                await dispatch(
+                  addInferenceResponse({
+                    content: event.full_response,
+                    threadId: event.thread_id,
+                    extraMetadata: chatDoneExtraMetadata(event),
+                  })
+                ).unwrap();
+                void dispatch(
+                  generateThreadTitleIfNeeded({
+                    threadId: event.thread_id,
+                    assistantMessage: event.full_response,
+                  })
+                );
+              } catch (error) {
+                rtLog('parallel_chat_done_append_failed', {
+                  thread: event.thread_id,
+                  request: event.request_id,
+                  error: error instanceof Error ? error.message : String(error),
+                });
+              }
+            })();
+          }
+          dispatch(clearParallelRequest({ requestId: parallelRequestId }));
+          requestUsageRefresh();
+          return;
+        }
+
         const deliveryKey = segmentDeliveryKey(event.thread_id, event.request_id);
         const segmentDelivery = takeSegmentDelivery(segmentDeliveriesRef.current, deliveryKey);
         const completeSegmentDelivery = hasCompleteSegmentDelivery(event, segmentDelivery);
@@ -983,6 +1064,28 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => {
           err: event.error_type,
         });
 
+        // Parallel (forked) turn error: resolve only its lane, leaving the
+        // primary turn untouched. Surface a non-cancellation error as a message
+        // so the failed branch is visible.
+        if (
+          event.request_id !== undefined &&
+          store.getState().chatRuntime.parallelRequestThreads[event.request_id] !== undefined
+        ) {
+          deleteSegmentDelivery(
+            segmentDeliveriesRef.current,
+            segmentDeliveryKey(event.thread_id, event.request_id)
+          );
+          if (event.error_type !== 'cancelled') {
+            const errorContent = event.message || USER_FACING_AGENT_ERROR_MESSAGE;
+            void dispatch(
+              addInferenceResponse({ content: errorContent, threadId: event.thread_id })
+            );
+            requestUsageRefresh();
+          }
+          dispatch(clearParallelRequest({ requestId: event.request_id }));
+          return;
+        }
+
         deleteSegmentDelivery(
           segmentDeliveriesRef.current,
           segmentDeliveryKey(event.thread_id, event.request_id)
@@ -1026,7 +1129,7 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => {
         }
 
         dispatch(endInferenceTurn({ threadId: event.thread_id }));
-        dispatch(setActiveThread(null));
+        dispatch(clearThreadInferenceActive(event.thread_id));
       },
     });
 
@@ -1055,12 +1158,12 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => {
     const state = store.getState();
     const lifecycles = state.chatRuntime.inferenceTurnLifecycleByThread;
     const threadIds = Object.keys(lifecycles);
-    const activeThreadId = state.thread.activeThreadId;
-    if (threadIds.length === 0 && !activeThreadId) return;
+    const activeThreadIds = Object.keys(state.thread.activeThreadIds);
+    if (threadIds.length === 0 && activeThreadIds.length === 0) return;
     rtLog('socket_disconnect_reconcile', {
       socket: socketStatus,
       inFlight: threadIds.length,
-      active: activeThreadId,
+      active: activeThreadIds.length,
     });
     for (const threadId of threadIds) {
       dispatch(clearInferenceStatusForThread({ threadId }));
@@ -1069,7 +1172,9 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => {
       dispatch(clearPendingApprovalForThread({ threadId }));
       dispatch(endInferenceTurn({ threadId }));
     }
-    if (activeThreadId) {
+    // A disconnect kills every in-flight turn on the dead session, so clear all
+    // active markers (setActiveThread(null) clears the whole set).
+    if (activeThreadIds.length > 0) {
       dispatch(setActiveThread(null));
     }
   }, [socketStatus, dispatch]);

app/src/providers/__tests__/ChatRuntimeProvider.test.tsx

@@ -6,7 +6,11 @@ import { beforeEach, describe, expect, it, vi } from 'vitest';
 import * as chatService from '../../services/chatService';
 import { threadApi } from '../../services/api/threadApi';
 import { store } from '../../store';
-import { clearAllChatRuntime } from '../../store/chatRuntimeSlice';
+import {
+  clearAllChatRuntime,
+  registerParallelRequest,
+  resetSessionTokenUsage,
+} from '../../store/chatRuntimeSlice';
 import { setStatusForUser } from '../../store/socketSlice';
 import { clearAllThreads, loadThreads, setSelectedThread } from '../../store/threadSlice';
 import ChatRuntimeProvider, { findPendingDelegationContext } from '../ChatRuntimeProvider';
@@ -64,6 +68,10 @@ function resetRuntimeState() {
   // selection that clears ambient state.
   store.dispatch(clearAllThreads());
   store.dispatch(clearAllChatRuntime());
+  // `clearAllChatRuntime` intentionally preserves cumulative session token
+  // usage; reset it here so usage-recording tests stay isolated regardless of
+  // run order.
+  store.dispatch(resetSessionTokenUsage());
   store.dispatch(setStatusForUser({ userId: '__pending__', status: 'disconnected' }));
 }
 
@@ -300,6 +308,60 @@ describe('ChatRuntimeProvider — dedupe, proactive resolution, mid-turn invaria
       expect(row?.subagent?.transcript).toEqual([]);
     });
 
+    it('routes a parallel (forked) turn into its own lane, leaving the primary stream untouched', () => {
+      const listeners = renderProvider();
+
+      // Primary turn streams on the thread.
+      act(() => {
+        listeners.onTextDelta?.({
+          thread_id: 't-par',
+          request_id: 'primary',
+          round: 0,
+          delta: 'P',
+        });
+      });
+      // A parallel turn is registered and streams concurrently on the SAME thread.
+      act(() => {
+        store.dispatch(registerParallelRequest({ threadId: 't-par', requestId: 'branch' }));
+        listeners.onTextDelta?.({
+          thread_id: 't-par',
+          request_id: 'branch',
+          round: 0,
+          delta: 'B1',
+        });
+        listeners.onTextDelta?.({
+          thread_id: 't-par',
+          request_id: 'branch',
+          round: 0,
+          delta: 'B2',
+        });
+      });
+
+      const mid = store.getState().chatRuntime;
+      // Primary stream is not clobbered by the parallel branch.
+      expect(mid.streamingAssistantByThread['t-par']?.content).toBe('P');
+      expect(mid.parallelStreamsByThread['t-par']?.['branch']?.content).toBe('B1B2');
+
+      // The parallel turn's chat_done resolves ONLY its lane; the primary
+      // stream and its (still-running) state survive.
+      act(() => {
+        listeners.onDone?.({
+          thread_id: 't-par',
+          request_id: 'branch',
+          full_response: 'branch done',
+          rounds_used: 1,
+          total_input_tokens: 0,
+          total_output_tokens: 0,
+          segment_total: 0,
+        });
+      });
+
+      const after = store.getState().chatRuntime;
+      expect(after.parallelStreamsByThread['t-par']).toBeUndefined();
+      expect(after.parallelRequestThreads['branch']).toBeUndefined();
+      expect(after.streamingAssistantByThread['t-par']?.content).toBe('P');
+    });
+
     it('drops duplicate chat_done events with the same thread/request', async () => {
       const listeners = renderProvider();
 
@@ -992,7 +1054,7 @@ describe('ChatRuntimeProvider — dedupe, proactive resolution, mid-turn invaria
         );
       });
 
-      expect(store.getState().thread.activeThreadId).toBe(threadId);
+      expect(store.getState().thread.activeThreadIds[threadId]).toBe(true);
       expect(store.getState().chatRuntime.inferenceTurnLifecycleByThread[threadId]).toBe('started');
 
       await act(async () => {
@@ -1000,7 +1062,7 @@ describe('ChatRuntimeProvider — dedupe, proactive resolution, mid-turn invaria
       });
 
       await waitFor(() => {
-        expect(store.getState().thread.activeThreadId).toBeNull();
+        expect(store.getState().thread.activeThreadIds[threadId]).toBeUndefined();
         expect(
           store.getState().chatRuntime.inferenceTurnLifecycleByThread[threadId]
         ).toBeUndefined();
@@ -1031,7 +1093,7 @@ describe('ChatRuntimeProvider — dedupe, proactive resolution, mid-turn invaria
       });
 
       await waitFor(() => {
-        expect(store.getState().thread.activeThreadId).toBeNull();
+        expect(store.getState().thread.activeThreadIds[threadId]).toBeUndefined();
       });
       expect(store.getState().chatRuntime.streamingAssistantByThread[threadId]).toMatchObject({
         content: 'Hello there, partial',

app/src/services/chatService.ts

@@ -943,7 +943,7 @@ export function subscribeChatEvents(listeners: ChatEventListeners): () => void {
   };
 }
 
-export type QueueMode = 'interrupt' | 'steer' | 'followup' | 'collect';
+export type QueueMode = 'interrupt' | 'steer' | 'followup' | 'collect' | 'parallel';
 
 export interface ChatSendParams {
   threadId: string;
@@ -987,15 +987,19 @@ export interface ChatSendParams {
  * The Rust core spawns the agent loop asynchronously and streams events
  * (tool_call, tool_result, chat_done, chat_error) back over the socket
  * connection using the `client_id` (socket ID) for routing.
+ *
+ * Returns the turn's `request_id` (from the RPC ack) when the core provides
+ * one — used by `parallel` sends to register the forked turn's stream lane.
+ * `undefined` if the ack carried no id.
  */
-export async function chatSend(params: ChatSendParams): Promise<void> {
+export async function chatSend(params: ChatSendParams): Promise<string | undefined> {
   const socket = socketService.getSocket();
   const clientId = socket?.id;
   if (!clientId) {
     throw new Error('Socket not connected — no client ID for event routing');
   }
 
-  await callCoreRpc({
+  const result = await callCoreRpc({
     method: 'openhuman.channel_web_chat',
     params: {
       client_id: clientId,
@@ -1010,6 +1014,9 @@ export async function chatSend(params: ChatSendParams): Promise<void> {
       queue_mode: params.queueMode ?? undefined,
     },
   });
+
+  const requestId = (result as { request_id?: unknown } | null)?.request_id;
+  return typeof requestId === 'string' ? requestId : undefined;
 }
 
 /**

app/src/services/socketService.ts

@@ -266,10 +266,16 @@ class SocketService {
       // room and a per-thread room (see socketio.rs `emit_web_channel_event`);
       // because a reconnect produces a NEW client_id, the new socket must
       // re-subscribe to the thread room to keep receiving the stream.
+      // With parallel inference several threads may be streaming at once, so
+      // re-subscribe to every active thread room (plus the selected thread) —
+      // not just a single "active" thread — to keep all in-flight streams alive.
       const threadState = store.getState().thread;
-      const activeThreadId = threadState?.selectedThreadId ?? threadState?.activeThreadId;
-      if (activeThreadId) {
-        this.socket?.emit('thread:subscribe', { thread_id: activeThreadId });
+      const roomThreadIds = new Set<string>(Object.keys(threadState?.activeThreadIds ?? {}));
+      if (threadState?.selectedThreadId) {
+        roomThreadIds.add(threadState.selectedThreadId);
+      }
+      for (const threadId of roomThreadIds) {
+        this.socket?.emit('thread:subscribe', { thread_id: threadId });
       }
     });
 

app/src/store/__tests__/chatRuntimeSlice.test.ts

@@ -6,6 +6,7 @@ import reducer, {
   clearAllChatRuntime,
   clearArtifactsForThread,
   clearInferenceStatusForThread,
+  clearParallelRequest,
   clearPendingApprovalForThread,
   clearRuntimeForThread,
   clearStreamingAssistantForThread,
@@ -15,8 +16,10 @@ import reducer, {
   hydrateRuntimeFromRunLedger,
   hydrateRuntimeFromSnapshot,
   markInferenceTurnStreaming,
+  registerParallelRequest,
   removeArtifactForThread,
   setInferenceStatusForThread,
+  setParallelStream,
   setPendingApprovalForThread,
   setStreamingAssistantForThread,
   setTaskBoardForThread,
@@ -773,4 +776,105 @@ describe('chatRuntimeSlice', () => {
       expect(cleared.artifactsByThread).toEqual({});
     });
   });
+
+  describe('parallel (forked) turn lane', () => {
+    it('registers a parallel request and streams into its own lane keyed by requestId', () => {
+      let state = reducer(
+        undefined,
+        registerParallelRequest({ threadId: 't-1', requestId: 'req-a' })
+      );
+      expect(state.parallelRequestThreads['req-a']).toBe('t-1');
+
+      state = reducer(
+        state,
+        setParallelStream({
+          threadId: 't-1',
+          streaming: { requestId: 'req-a', content: 'hi', thinking: '' },
+        })
+      );
+      expect(state.parallelStreamsByThread['t-1']['req-a'].content).toBe('hi');
+    });
+
+    it('keeps two concurrent same-thread branches separate', () => {
+      let state = reducer(undefined, registerParallelRequest({ threadId: 't-1', requestId: 'r1' }));
+      state = reducer(state, registerParallelRequest({ threadId: 't-1', requestId: 'r2' }));
+      state = reducer(
+        state,
+        setParallelStream({
+          threadId: 't-1',
+          streaming: { requestId: 'r1', content: 'one', thinking: '' },
+        })
+      );
+      state = reducer(
+        state,
+        setParallelStream({
+          threadId: 't-1',
+          streaming: { requestId: 'r2', content: 'two', thinking: '' },
+        })
+      );
+      expect(Object.keys(state.parallelStreamsByThread['t-1'])).toEqual(['r1', 'r2']);
+      expect(state.parallelStreamsByThread['t-1']['r1'].content).toBe('one');
+      expect(state.parallelStreamsByThread['t-1']['r2'].content).toBe('two');
+    });
+
+    it('clearParallelRequest removes one branch and drops the thread bucket when empty', () => {
+      let state = reducer(undefined, registerParallelRequest({ threadId: 't-1', requestId: 'r1' }));
+      state = reducer(state, registerParallelRequest({ threadId: 't-1', requestId: 'r2' }));
+      state = reducer(
+        state,
+        setParallelStream({
+          threadId: 't-1',
+          streaming: { requestId: 'r1', content: 'one', thinking: '' },
+        })
+      );
+      state = reducer(
+        state,
+        setParallelStream({
+          threadId: 't-1',
+          streaming: { requestId: 'r2', content: 'two', thinking: '' },
+        })
+      );
+
+      state = reducer(state, clearParallelRequest({ requestId: 'r1' }));
+      expect(state.parallelRequestThreads['r1']).toBeUndefined();
+      expect(state.parallelStreamsByThread['t-1']['r1']).toBeUndefined();
+      expect(state.parallelStreamsByThread['t-1']['r2'].content).toBe('two');
+
+      state = reducer(state, clearParallelRequest({ requestId: 'r2' }));
+      expect(state.parallelStreamsByThread['t-1']).toBeUndefined();
+      expect(state.parallelRequestThreads).toEqual({});
+    });
+
+    it('clearRuntimeForThread drops the thread parallel streams and their request mappings', () => {
+      let state = reducer(undefined, registerParallelRequest({ threadId: 't-1', requestId: 'r1' }));
+      state = reducer(
+        state,
+        setParallelStream({
+          threadId: 't-1',
+          streaming: { requestId: 'r1', content: 'one', thinking: '' },
+        })
+      );
+      // An unrelated thread's parallel branch must survive.
+      state = reducer(state, registerParallelRequest({ threadId: 't-2', requestId: 'r9' }));
+      state = reducer(
+        state,
+        setParallelStream({
+          threadId: 't-2',
+          streaming: { requestId: 'r9', content: 'keep', thinking: '' },
+        })
+      );
+
+      state = reducer(state, clearRuntimeForThread({ threadId: 't-1' }));
+      expect(state.parallelStreamsByThread['t-1']).toBeUndefined();
+      expect(state.parallelRequestThreads['r1']).toBeUndefined();
+      expect(state.parallelStreamsByThread['t-2']['r9'].content).toBe('keep');
+      expect(state.parallelRequestThreads['r9']).toBe('t-2');
+    });
+
+    it('clearParallelRequest is a no-op for an unknown requestId', () => {
+      const state = reducer(undefined, clearParallelRequest({ requestId: 'nope' }));
+      expect(state.parallelStreamsByThread).toEqual({});
+      expect(state.parallelRequestThreads).toEqual({});
+    });
+  });
 });

app/src/store/__tests__/threadSlice.test.ts

@@ -10,9 +10,11 @@ import threadReducer, {
   clearAllThreads,
   clearSelectedThread,
   clearStaleThread,
+  clearThreadInferenceActive,
   generateThreadTitleIfNeeded,
   loadThreadMessages,
   loadThreads,
+  markThreadInferenceActive,
   setActiveThread,
   setSelectedThread,
   setWelcomeThreadId,
@@ -77,7 +79,7 @@ describe('threadSlice synchronous reducers', () => {
     const state = store.getState().thread;
     expect(state.threads).toEqual([]);
     expect(state.selectedThreadId).toBeNull();
-    expect(state.activeThreadId).toBeNull();
+    expect(state.activeThreadIds).toEqual({});
     expect(state.messagesByThreadId).toEqual({});
     expect(state.messages).toEqual([]);
     expect(state.isLoadingThreads).toBe(false);
@@ -133,12 +135,22 @@ describe('threadSlice synchronous reducers', () => {
     expect(state.messagesByThreadId['t-1']).toHaveLength(1);
   });
 
-  it('setActiveThread only touches the active id', () => {
+  it('setActiveThread marks a thread active; null clears the whole set', () => {
     const store = createStore();
     store.dispatch(setActiveThread('t-active'));
-    expect(store.getState().thread.activeThreadId).toBe('t-active');
+    expect(store.getState().thread.activeThreadIds).toEqual({ 't-active': true });
     store.dispatch(setActiveThread(null));
-    expect(store.getState().thread.activeThreadId).toBeNull();
+    expect(store.getState().thread.activeThreadIds).toEqual({});
+  });
+
+  it('tracks multiple concurrently-active threads independently', () => {
+    const store = createStore();
+    store.dispatch(markThreadInferenceActive('t-1'));
+    store.dispatch(markThreadInferenceActive('t-2'));
+    expect(store.getState().thread.activeThreadIds).toEqual({ 't-1': true, 't-2': true });
+    // Clearing one leaves the other running — the core of parallel inference.
+    store.dispatch(clearThreadInferenceActive('t-1'));
+    expect(store.getState().thread.activeThreadIds).toEqual({ 't-2': true });
   });
 
   it('clearAllThreads wipes threads, messages, and selection', async () => {
@@ -161,7 +173,7 @@ describe('threadSlice synchronous reducers', () => {
     expect(state.threads).toEqual([]);
     expect(state.messagesByThreadId).toEqual({});
     expect(state.selectedThreadId).toBeNull();
-    expect(state.activeThreadId).toBeNull();
+    expect(state.activeThreadIds).toEqual({});
     expect(state.messages).toEqual([]);
   });
 
@@ -186,7 +198,7 @@ describe('threadSlice synchronous reducers', () => {
     expect(state.threads.map(thread => thread.id)).toEqual(['t-2']);
     expect(state.messagesByThreadId['t-1']).toBeUndefined();
     expect(state.selectedThreadId).toBeNull();
-    expect(state.activeThreadId).toBeNull();
+    expect(state.activeThreadIds).toEqual({});
     expect(state.messages).toEqual([]);
   });
 });
@@ -352,7 +364,7 @@ describe('threadSlice addMessageLocal thunk', () => {
     expect(mockedThreadApi.generateTitleIfNeeded).not.toHaveBeenCalled();
     expect(mockedThreadApi.getThreads).toHaveBeenCalledTimes(2);
     expect(store.getState().thread.selectedThreadId).toBeNull();
-    expect(store.getState().thread.activeThreadId).toBeNull();
+    expect(store.getState().thread.activeThreadIds).toEqual({});
   });
 });
 
@@ -361,7 +373,7 @@ describe('threadSlice addInferenceResponse thunk', () => {
     vi.clearAllMocks();
   });
 
-  it('appends to the supplied thread even when activeThreadId is null', async () => {
+  it('appends to the supplied thread regardless of active/selected state', async () => {
     const store = createStore();
     const persisted = makeMessage({ id: 'srv-1', sender: 'agent', content: 'ack' });
     mockedThreadApi.appendMessage.mockResolvedValueOnce(persisted);
@@ -371,25 +383,26 @@ describe('threadSlice addInferenceResponse thunk', () => {
     expect(result.type).toBe('thread/addInferenceResponse/fulfilled');
     const state = store.getState().thread;
     expect(state.messagesByThreadId['t-1']).toEqual([persisted]);
-    // activeThreadId must not be mutated by this thunk — only ChatRuntimeProvider clears it.
-    expect(state.activeThreadId).toBeNull();
+    // Active markers must not be mutated by this thunk — only ChatRuntimeProvider clears them.
+    expect(state.activeThreadIds).toEqual({});
   });
 
-  it('falls back to activeThreadId when no threadId is supplied', async () => {
+  it('falls back to the selected thread when no threadId is supplied', async () => {
+    // Under parallel inference there is no single "active" thread to fall back
+    // to, so the legacy fallback target is now the selected thread.
     const store = createStore();
-    store.dispatch(setActiveThread('t-active'));
+    store.dispatch(setSelectedThread('t-selected'));
     mockedThreadApi.appendMessage.mockResolvedValueOnce(makeMessage({ sender: 'agent' }));
 
     await store.dispatch(addInferenceResponse({ content: 'ack' }));
     expect(mockedThreadApi.appendMessage).toHaveBeenCalledWith(
-      't-active',
+      't-selected',
       expect.objectContaining({ sender: 'agent', content: 'ack' })
     );
-    // activeThreadId must not be cleared by this thunk — ChatRuntimeProvider owns that.
-    expect(store.getState().thread.activeThreadId).toBe('t-active');
+    expect(store.getState().thread.selectedThreadId).toBe('t-selected');
   });
 
-  it('rejects cleanly when neither threadId nor activeThreadId is set', async () => {
+  it('rejects cleanly when neither threadId nor a selected thread is set', async () => {
     const store = createStore();
     const result = await store.dispatch(addInferenceResponse({ content: 'ack' }));
     expect(result.type).toBe('thread/addInferenceResponse/rejected');
@@ -398,7 +411,8 @@ describe('threadSlice addInferenceResponse thunk', () => {
 
   it('clears stale active thread when assistant append returns ThreadNotFound', async () => {
     const store = createStore();
-    store.dispatch(setActiveThread('t-active'));
+    store.dispatch(setSelectedThread('t-active'));
+    store.dispatch(markThreadInferenceActive('t-active'));
     mockedThreadApi.appendMessage.mockRejectedValueOnce(
       new CoreRpcError('thread t-active not found', 'thread_not_found', undefined, {
         kind: 'ThreadNotFound',
@@ -412,7 +426,7 @@ describe('threadSlice addInferenceResponse thunk', () => {
     expect(result.type).toBe('thread/addInferenceResponse/rejected');
     expect(result.payload).toBe(THREAD_NOT_FOUND_MESSAGE);
     expect(mockedThreadApi.appendMessage).toHaveBeenCalledTimes(1);
-    expect(store.getState().thread.activeThreadId).toBeNull();
+    expect(store.getState().thread.activeThreadIds).toEqual({});
   });
 });
 

app/src/store/chatRuntimeSlice.ts

@@ -244,8 +244,13 @@ export interface ArtifactSnapshot {
   updatedAt: number;
 }
 
-/** Queue behavior when a turn is already in flight for a thread. */
-export type QueueMode = 'interrupt' | 'steer' | 'followup' | 'collect';
+/**
+ * Queue behavior when a turn is already in flight for a thread.
+ * `parallel` runs an independent concurrent (forked) turn on the same thread
+ * instead of interrupting/queueing — its stream is tracked separately (see
+ * `parallelStreamsByThread`) so it renders as its own interleaved branch.
+ */
+export type QueueMode = 'interrupt' | 'steer' | 'followup' | 'collect' | 'parallel';
 
 /**
  * Per-thread UI state for an in-flight agent turn (socket events while the user
@@ -256,6 +261,21 @@ export type QueueMode = 'interrupt' | 'steer' | 'followup' | 'collect';
 interface ChatRuntimeState {
   inferenceStatusByThread: Record<string, InferenceStatus>;
   streamingAssistantByThread: Record<string, StreamingAssistantState>;
+  /**
+   * Live streams for concurrent PARALLEL (forked) turns on a thread, nested
+   * `threadId -> requestId -> stream`. A separate lane from
+   * `streamingAssistantByThread` (the single primary stream) so two same-thread
+   * turns don't clobber each other — each renders as its own interleaved
+   * branch bubble. Populated only for turns sent with `queueMode: 'parallel'`.
+   */
+  parallelStreamsByThread: Record<string, Record<string, StreamingAssistantState>>;
+  /**
+   * Maps a parallel turn's `requestId -> threadId`. Lets socket event handlers
+   * recognise a forked turn's events (and find its thread) so they route to the
+   * parallel lane instead of the primary stream. Entries are added on send and
+   * removed on that turn's `chat_done` / `chat_error`.
+   */
+  parallelRequestThreads: Record<string, string>;
   toolTimelineByThread: Record<string, ToolTimelineEntry[]>;
   taskBoardByThread: Record<string, TaskBoard>;
   inferenceTurnLifecycleByThread: Record<string, InferenceTurnLifecycle>;
@@ -283,6 +303,8 @@ export interface QueueStatus {
 const initialState: ChatRuntimeState = {
   inferenceStatusByThread: {},
   streamingAssistantByThread: {},
+  parallelStreamsByThread: {},
+  parallelRequestThreads: {},
   toolTimelineByThread: {},
   taskBoardByThread: {},
   inferenceTurnLifecycleByThread: {},
@@ -449,6 +471,40 @@ const chatRuntimeSlice = createSlice({
     clearStreamingAssistantForThread: (state, action: PayloadAction<{ threadId: string }>) => {
       delete state.streamingAssistantByThread[action.payload.threadId];
     },
+    /**
+     * Register a parallel (forked) turn so its socket events route to the
+     * parallel lane. Called when a `queueMode: 'parallel'` send is accepted.
+     */
+    registerParallelRequest: (
+      state,
+      action: PayloadAction<{ threadId: string; requestId: string }>
+    ) => {
+      state.parallelRequestThreads[action.payload.requestId] = action.payload.threadId;
+    },
+    /** Upsert the live stream for a parallel (forked) turn, keyed by requestId. */
+    setParallelStream: (
+      state,
+      action: PayloadAction<{ threadId: string; streaming: StreamingAssistantState }>
+    ) => {
+      const { threadId, streaming } = action.payload;
+      (state.parallelStreamsByThread[threadId] ??= {})[streaming.requestId] = streaming;
+    },
+    /**
+     * Tear down a parallel turn's lane state on its terminal event
+     * (chat_done / chat_error). Removes the stream and the request→thread entry.
+     */
+    clearParallelRequest: (state, action: PayloadAction<{ requestId: string }>) => {
+      const { requestId } = action.payload;
+      const threadId = state.parallelRequestThreads[requestId];
+      delete state.parallelRequestThreads[requestId];
+      if (threadId === undefined) return;
+      const streams = state.parallelStreamsByThread[threadId];
+      if (!streams) return;
+      delete streams[requestId];
+      if (Object.keys(streams).length === 0) {
+        delete state.parallelStreamsByThread[threadId];
+      }
+    },
     setToolTimelineForThread: (
       state,
       action: PayloadAction<{ threadId: string; entries: ToolTimelineEntry[] }>
@@ -702,6 +758,15 @@ const chatRuntimeSlice = createSlice({
     clearRuntimeForThread: (state, action: PayloadAction<{ threadId: string }>) => {
       delete state.inferenceStatusByThread[action.payload.threadId];
       delete state.streamingAssistantByThread[action.payload.threadId];
+      // Drop any parallel (forked) streams for this thread and their
+      // request→thread mappings — a hard per-thread reset covers every branch.
+      const parallelStreams = state.parallelStreamsByThread[action.payload.threadId];
+      if (parallelStreams) {
+        for (const requestId of Object.keys(parallelStreams)) {
+          delete state.parallelRequestThreads[requestId];
+        }
+        delete state.parallelStreamsByThread[action.payload.threadId];
+      }
       delete state.toolTimelineByThread[action.payload.threadId];
       delete state.taskBoardByThread[action.payload.threadId];
       delete state.inferenceTurnLifecycleByThread[action.payload.threadId];
@@ -716,6 +781,8 @@ const chatRuntimeSlice = createSlice({
     clearAllChatRuntime: state => {
       state.inferenceStatusByThread = {};
       state.streamingAssistantByThread = {};
+      state.parallelStreamsByThread = {};
+      state.parallelRequestThreads = {};
       state.toolTimelineByThread = {};
       state.taskBoardByThread = {};
       state.inferenceTurnLifecycleByThread = {};
@@ -837,6 +904,9 @@ export const {
   clearInferenceStatusForThread,
   setStreamingAssistantForThread,
   clearStreamingAssistantForThread,
+  registerParallelRequest,
+  setParallelStream,
+  clearParallelRequest,
   setToolTimelineForThread,
   clearToolTimelineForThread,
   appendSubagentStreamDelta,

app/src/store/threadSlice.ts

@@ -12,7 +12,15 @@ interface ThreadState {
   threads: Thread[];
   selectedThreadId: string | null;
   threadSidebarVisible: boolean;
-  activeThreadId: string | null;
+  /**
+   * Set of threads that currently have an in-flight inference turn, keyed by
+   * thread id. Replaces the legacy single `activeThreadId` so that turns on
+   * different threads can run concurrently — each thread tracks its own
+   * in-flight state and the composer is gated per-thread (see
+   * `isComposerInteractionBlocked`) rather than globally. Stored as a plain
+   * object (not a `Set`) to stay redux-persist serializable.
+   */
+  activeThreadIds: Record<string, true>;
   /**
    * @deprecated — welcome-agent replaced by Joyride walkthrough. Always null
    * for new users; retained for redux-persist deserialization compatibility.
@@ -29,7 +37,7 @@ const initialState: ThreadState = {
   threads: [],
   selectedThreadId: null,
   threadSidebarVisible: false,
-  activeThreadId: null,
+  activeThreadIds: {},
   welcomeThreadId: null,
   messagesByThreadId: {},
   messages: [],
@@ -189,7 +197,11 @@ export const addInferenceResponse = createAsyncThunk(
     { dispatch, getState, rejectWithValue }
   ) => {
     const state = getState() as { thread: ThreadState };
-    const targetThreadId = payload.threadId ?? state.thread.activeThreadId;
+    // All real callers pass an explicit `threadId` (the event's `thread_id`).
+    // The fallback to the selected thread is a last-resort for legacy callers
+    // that omit it; under parallel inference there is no single "active" thread
+    // to fall back to, so we cannot use `activeThreadIds` here.
+    const targetThreadId = payload.threadId ?? state.thread.selectedThreadId;
     if (!targetThreadId) return rejectWithValue('No target thread');
 
     const message: ThreadMessage = {
@@ -336,8 +348,27 @@ const threadSlice = createSlice({
       state.messages = [];
       state.messagesError = null;
     },
+    /**
+     * Back-compat shim retained for existing call sites:
+     *  - `setActiveThread(threadId)` marks that thread's inference turn active.
+     *  - `setActiveThread(null)` clears ALL in-flight markers (legacy global
+     *    clear). Prefer `clearThreadInferenceActive(threadId)` when the
+     *    finishing thread is known so unrelated concurrent turns are untouched.
+     */
     setActiveThread: (state, action: { payload: string | null }) => {
-      state.activeThreadId = action.payload;
+      if (action.payload === null) {
+        state.activeThreadIds = {};
+      } else {
+        state.activeThreadIds[action.payload] = true;
+      }
+    },
+    /** Mark a single thread as having an in-flight inference turn. */
+    markThreadInferenceActive: (state, action: PayloadAction<string>) => {
+      state.activeThreadIds[action.payload] = true;
+    },
+    /** Clear the in-flight marker for a single thread (on done/error/timeout). */
+    clearThreadInferenceActive: (state, action: PayloadAction<string>) => {
+      delete state.activeThreadIds[action.payload];
     },
     setThreadSidebarVisible: (state, action: PayloadAction<boolean>) => {
       state.threadSidebarVisible = action.payload;
@@ -351,9 +382,7 @@ const threadSlice = createSlice({
         state.messages = [];
         state.messagesError = null;
       }
-      if (state.activeThreadId === threadId) {
-        state.activeThreadId = null;
-      }
+      delete state.activeThreadIds[threadId];
       if (state.welcomeThreadId === threadId) {
         state.welcomeThreadId = null;
       }
@@ -363,7 +392,7 @@ const threadSlice = createSlice({
       state.messagesByThreadId = {};
       state.selectedThreadId = null;
       state.messages = [];
-      state.activeThreadId = null;
+      state.activeThreadIds = {};
       state.welcomeThreadId = null;
     },
     // Like `clearAllThreads` but keeps `selectedThreadId`. Used on the
@@ -376,7 +405,7 @@ const threadSlice = createSlice({
       state.threads = [];
       state.messagesByThreadId = {};
       state.messages = [];
-      state.activeThreadId = null;
+      state.activeThreadIds = {};
       state.welcomeThreadId = null;
     },
     // [#1123] True no-op — welcome-agent onboarding replaced by Joyride walkthrough.
@@ -400,8 +429,10 @@ const threadSlice = createSlice({
           state.messages = [];
           state.messagesError = null;
         }
-        if (state.activeThreadId && !liveThreadIds.has(state.activeThreadId)) {
-          state.activeThreadId = null;
+        for (const activeId of Object.keys(state.activeThreadIds)) {
+          if (!liveThreadIds.has(activeId)) {
+            delete state.activeThreadIds[activeId];
+          }
         }
       })
       .addCase(loadThreads.rejected, state => {
@@ -464,6 +495,8 @@ export const {
   setSelectedThread,
   clearSelectedThread,
   setActiveThread,
+  markThreadInferenceActive,
+  clearThreadInferenceActive,
   setThreadSidebarVisible,
   clearStaleThread,
   clearAllThreads,

docs/TEST-COVERAGE-MATRIX.md

@@ -181,6 +181,7 @@ Canonical mapping of every product feature to its test source(s). Drives gap-fil
 | 4.2.1 | User Message Handling  | WD+RI | `conversations-web-channel-flow.spec.ts`, `tests/json_rpc_e2e.rs` | ✅     |                             |
 | 4.2.2 | AI Response Generation | WD    | `agent-review.spec.ts`                                            | ✅     | Mock LLM                    |
 | 4.2.3 | Streaming Responses    | RI    | `tests/json_rpc_e2e.rs`                                           | 🟡     | UI streaming assertion thin |
+| 4.2.4 | Parallel inference (cross-thread + within-thread forked turns) | RU+VU | `src/openhuman/channels/providers/web_tests.rs`, `app/src/store/__tests__/chatRuntimeSlice.test.ts`, `app/src/providers/__tests__/ChatRuntimeProvider.test.tsx` | 🟡 | Concurrent same-/cross-thread dispatch, cooperative `CancellationToken` teardown, and parallel-lane stream routing covered; dedicated WD E2E is a follow-up |
 
 ### 4.3 Tool Invocation
 

src/openhuman/agent/harness/run_queue/mod.rs

@@ -52,6 +52,11 @@ impl RunQueue {
                     "[run_queue] interrupt-mode message pushed to queue — should have been handled by caller"
                 );
             }
+            QueueMode::Parallel => {
+                log::warn!(
+                    "[run_queue] parallel-mode message pushed to queue — should have spawned a forked turn at the caller"
+                );
+            }
         }
     }
 

src/openhuman/agent/harness/run_queue/types.rs

@@ -17,6 +17,11 @@ pub enum QueueMode {
     /// Silently collect the message as additional context; the agent sees it
     /// at the next iteration boundary but does not treat it as a new instruction.
     Collect,
+    /// Run as an independent concurrent turn on the same thread. The new turn
+    /// forks the thread's history-at-start (snapshot) and runs alongside any
+    /// in-flight turn instead of interrupting or queueing — its result is
+    /// appended to the conversation on completion (snapshot + append).
+    Parallel,
 }
 
 impl Default for QueueMode {
@@ -32,6 +37,7 @@ impl fmt::Display for QueueMode {
             Self::Steer => write!(f, "steer"),
             Self::Followup => write!(f, "followup"),
             Self::Collect => write!(f, "collect"),
+            Self::Parallel => write!(f, "parallel"),
         }
     }
 }

src/openhuman/channels/providers/web/mod.rs

@@ -27,6 +27,8 @@ pub use event_bus::{
 };
 
 // Public API — operations
+#[cfg(any(test, debug_assertions))]
+pub use ops::parallel_in_flight_entries_for_test;
 pub use ops::{
     cancel_chat, channel_web_cancel, channel_web_chat, channel_web_queue_clear,
     channel_web_queue_status, in_flight_entries_for_test, invalidate_thread_sessions, start_chat,
@@ -51,6 +53,8 @@ pub(crate) use schemas::{
 // Test helpers (debug/test builds only)
 #[cfg(any(test, debug_assertions))]
 pub use ops::set_test_forced_run_chat_task_error;
+#[cfg(any(test, debug_assertions))]
+pub use ops::{set_test_run_chat_task_block, TestRunChatTaskBlock};
 
 #[cfg(any(test, debug_assertions))]
 pub(crate) use ops::THREAD_SESSIONS;

src/openhuman/channels/providers/web/ops.rs

@@ -1,8 +1,10 @@
 use std::collections::HashMap;
+use std::time::Duration;
 
 use once_cell::sync::Lazy;
 use serde_json::{json, Value};
 use tokio::sync::Mutex;
+use tokio_util::sync::CancellationToken;
 use uuid::Uuid;
 
 use crate::core::event_bus::DomainEvent;
@@ -14,7 +16,7 @@ use crate::rpc::RpcOutcome;
 
 use super::event_bus::publish_web_channel_event;
 use super::run_task::run_chat_task;
-use super::types::{ChatRequestMetadata, InFlightEntry, SessionEntry};
+use super::types::{ChatRequestMetadata, InFlightEntry, ParallelEntry, SessionEntry};
 use super::web_errors::classify_inference_error;
 
 pub(crate) static THREAD_SESSIONS: Lazy<Mutex<HashMap<String, SessionEntry>>> =
@@ -23,10 +25,59 @@ pub(crate) static THREAD_SESSIONS: Lazy<Mutex<HashMap<String, SessionEntry>>> =
 pub(super) static IN_FLIGHT: Lazy<Mutex<HashMap<String, InFlightEntry>>> =
     Lazy::new(|| Mutex::new(HashMap::new()));
 
+/// Parallel (forked) turns, keyed by `request_id`. A separate lane from
+/// `IN_FLIGHT` (which holds one primary, interrupt-able turn per thread) so any
+/// number of concurrent `QueueMode::Parallel` turns can run on the same thread
+/// without touching interrupt/steer/queue semantics. See `QueueMode::Parallel`.
+pub(super) static PARALLEL_IN_FLIGHT: Lazy<Mutex<HashMap<String, ParallelEntry>>> =
+    Lazy::new(|| Mutex::new(HashMap::new()));
+
 #[cfg(any(test, debug_assertions))]
 pub(super) static TEST_FORCED_RUN_CHAT_TASK_ERROR: Lazy<Mutex<Option<String>>> =
     Lazy::new(|| Mutex::new(None));
 
+/// Test hook handles: when set, `run_chat_task` parks on a long sleep instead
+/// of doing real work, keeping the turn in-flight so concurrency / cancellation
+/// can be observed. `started` is flipped once the turn has actually parked (so
+/// a test can cancel only after the turn future is live), and a `Drop` guard
+/// inside the parked future flips `dropped`, proving cooperative cancellation
+/// tears the turn future down (vs. a hard `abort()` that never runs the Drop).
+#[cfg(any(test, debug_assertions))]
+#[derive(Clone)]
+pub struct TestRunChatTaskBlock {
+    pub started: std::sync::Arc<std::sync::atomic::AtomicBool>,
+    pub dropped: std::sync::Arc<std::sync::atomic::AtomicBool>,
+}
+
+#[cfg(any(test, debug_assertions))]
+pub(super) static TEST_RUN_CHAT_TASK_BLOCK: Lazy<Mutex<Option<TestRunChatTaskBlock>>> =
+    Lazy::new(|| Mutex::new(None));
+
+/// Cooperatively cancel an in-flight turn, with a hard `abort()` backstop.
+///
+/// Cancelling the token makes the turn's `tokio::select!` arm fire, dropping
+/// the turn future at its next await point (cancelling the in-flight LLM
+/// request and releasing locks cleanly). The detached backstop hard-aborts the
+/// task only if it has not finished unwinding within a short grace period, so a
+/// wedged turn can never leak. Returns the cancelled turn's request id.
+fn cancel_in_flight_gracefully(entry: InFlightEntry) -> String {
+    let request_id = entry.request_id.clone();
+    entry.cancel_token.cancel();
+    let mut handle = entry.handle;
+    tokio::spawn(async move {
+        tokio::select! {
+            _ = &mut handle => {}
+            _ = tokio::time::sleep(Duration::from_secs(5)) => {
+                log::warn!(
+                    "[web-channel] cooperative cancel did not finish within grace period — hard-aborting backstop"
+                );
+                handle.abort();
+            }
+        }
+    });
+    request_id
+}
+
 pub(crate) fn key_for(thread_id: &str) -> String {
     thread_id.to_string()
 }
@@ -57,6 +108,15 @@ pub async fn set_test_forced_run_chat_task_error(message: Option<&str>) {
     *slot = message.map(str::to_string);
 }
 
+/// Test hook: when `block` is `Some`, the next `run_chat_task` invocations park
+/// on a long sleep (staying in-flight), flip `started` once parked, and flip
+/// `dropped` when their future is torn down. Pass `None` to clear.
+#[cfg(any(test, debug_assertions))]
+pub async fn set_test_run_chat_task_block(block: Option<TestRunChatTaskBlock>) {
+    let mut slot = TEST_RUN_CHAT_TASK_BLOCK.lock().await;
+    *slot = block;
+}
+
 pub async fn start_chat(
     client_id: &str,
     thread_id: &str,
@@ -216,9 +276,37 @@ pub async fn start_chat(
         Some("steer") => crate::openhuman::agent::harness::run_queue::QueueMode::Steer,
         Some("followup") => crate::openhuman::agent::harness::run_queue::QueueMode::Followup,
         Some("collect") => crate::openhuman::agent::harness::run_queue::QueueMode::Collect,
+        Some("parallel") => crate::openhuman::agent::harness::run_queue::QueueMode::Parallel,
         _ => crate::openhuman::agent::harness::run_queue::QueueMode::Interrupt,
     };
 
+    // Parallel mode: spawn an independent forked turn that runs alongside any
+    // in-flight turn for this thread. It does not touch IN_FLIGHT (no
+    // interrupt/steer/queue) — it lives in its own request-keyed lane.
+    if matches!(
+        parsed_mode,
+        crate::openhuman::agent::harness::run_queue::QueueMode::Parallel
+    ) {
+        log::info!(
+            "[web-channel] starting PARALLEL forked turn thread_id={} request_id={}",
+            thread_id,
+            request_id
+        );
+        spawn_parallel_turn(
+            &client_id,
+            &thread_id,
+            request_id.clone(),
+            &message,
+            model_override,
+            temperature,
+            profile_id,
+            locale,
+            metadata,
+        )
+        .await;
+        return Ok(request_id);
+    }
+
     // Non-interrupt modes: push into the running turn's queue and return.
     if !matches!(
         parsed_mode,
@@ -275,16 +363,15 @@ pub async fn start_chat(
         let mut in_flight = IN_FLIGHT.lock().await;
 
         if let Some(existing) = in_flight.remove(&map_key) {
-            let cancelled_id = existing.request_id.clone();
-            existing.handle.abort();
+            let cancelled_id = cancel_in_flight_gracefully(existing);
             log::info!(
                 "[web-channel] interrupted in-flight turn thread_id={} cancelled_request_id={}",
                 thread_id,
-                existing.request_id
+                cancelled_id
             );
             crate::core::event_bus::publish_global(DomainEvent::RunQueueInterrupted {
                 thread_id: thread_id.clone(),
-                cancelled_request_id: existing.request_id.clone(),
+                cancelled_request_id: cancelled_id.clone(),
             });
             publish_web_channel_event(WebChannelEvent {
                 event: "chat_error".to_string(),
@@ -306,6 +393,12 @@ pub async fn start_chat(
     let request_id_task = request_id.clone();
     let map_key_task = map_key.clone();
 
+    // Cooperative cancellation for this turn. The token lives in the
+    // `InFlightEntry`; interrupt / cancel paths cancel it to tear the turn
+    // future down gracefully at the next await point.
+    let cancel_token = CancellationToken::new();
+    let task_cancel_token = cancel_token.clone();
+
     let user_message = message.clone();
     let handle = tokio::spawn(async move {
         let approval_ctx = crate::openhuman::approval::ApprovalChatContext {
@@ -316,25 +409,54 @@ pub async fn start_chat(
             thread_id: thread_id_task.clone(),
             client_id: client_id_task.clone(),
         };
-        let result = crate::openhuman::agent::turn_origin::with_origin(
-            origin,
-            crate::openhuman::approval::APPROVAL_CHAT_CONTEXT.scope(
-                approval_ctx,
-                run_chat_task(
-                    &client_id_task,
-                    &thread_id_task,
-                    &request_id_task,
-                    &user_message,
-                    model_override,
-                    temperature,
-                    profile_id,
-                    locale,
-                    turn_run_queue_task,
-                    metadata,
+        // `None` => the turn was cancelled cooperatively before producing a
+        // result; the interrupting/cancelling side already emitted the
+        // user-facing `chat_error`, so we just unwind quietly here.
+        let result = tokio::select! {
+            biased;
+            _ = task_cancel_token.cancelled() => None,
+            res = crate::openhuman::agent::turn_origin::with_origin(
+                origin,
+                crate::openhuman::approval::APPROVAL_CHAT_CONTEXT.scope(
+                    approval_ctx,
+                    run_chat_task(
+                        &client_id_task,
+                        &thread_id_task,
+                        &request_id_task,
+                        &user_message,
+                        model_override,
+                        temperature,
+                        profile_id,
+                        locale,
+                        turn_run_queue_task,
+                        metadata,
+                        /* fork */ false,
+                    ),
                 ),
-            ),
-        )
-        .await;
+            ) => Some(res),
+        };
+
+        let result = match result {
+            Some(res) => res,
+            None => {
+                log::info!(
+                    "[web-channel] turn cancelled cooperatively client_id={} thread_id={} request_id={}",
+                    client_id_task,
+                    thread_id_task,
+                    request_id_task
+                );
+                // Release any in-flight slot we still own and stop. The
+                // `request_id` guard below prevents clobbering a newer turn that
+                // replaced us on the interrupt path.
+                let mut in_flight = IN_FLIGHT.lock().await;
+                if let Some(current) = in_flight.get(&map_key_task) {
+                    if current.request_id == request_id_task {
+                        in_flight.remove(&map_key_task);
+                    }
+                }
+                return;
+            }
+        };
 
         match result {
             Ok(chat_result) => {
@@ -444,6 +566,7 @@ pub async fn start_chat(
                 request_id: request_id.clone(),
                 handle,
                 run_queue: turn_run_queue,
+                cancel_token,
             },
         );
     }
@@ -477,6 +600,155 @@ fn dispatch_followups(followups: Vec<crate::openhuman::agent::harness::run_queue
     }
 }
 
+/// Spawn an independent, forked (`QueueMode::Parallel`) turn. It snapshots the
+/// thread's history-at-start (inside `run_chat_task` with `fork = true`), runs
+/// concurrently with any other turn on the thread, and on completion delivers
+/// its response (append-only) and removes itself from `PARALLEL_IN_FLIGHT`.
+/// Emits the same per-`request_id` stream events as a primary turn, so the UI
+/// can render it as an interleaved branch.
+#[allow(clippy::too_many_arguments)]
+async fn spawn_parallel_turn(
+    client_id: &str,
+    thread_id: &str,
+    request_id: String,
+    message: &str,
+    model_override: Option<String>,
+    temperature: Option<f64>,
+    profile_id: Option<String>,
+    locale: Option<String>,
+    metadata: ChatRequestMetadata,
+) {
+    let cancel_token = CancellationToken::new();
+    let task_cancel_token = cancel_token.clone();
+
+    let client_id_task = client_id.to_string();
+    let thread_id_task = thread_id.to_string();
+    let request_id_task = request_id.clone();
+    let user_message = message.to_string();
+    // Forked turns don't participate in the steer/followup/collect queue, but
+    // `run_chat_task` requires a queue handle — give each its own.
+    let run_queue = crate::openhuman::agent::harness::run_queue::RunQueue::new();
+
+    let handle = tokio::spawn(async move {
+        let approval_ctx = crate::openhuman::approval::ApprovalChatContext {
+            thread_id: thread_id_task.clone(),
+            client_id: client_id_task.clone(),
+        };
+        let origin = crate::openhuman::agent::turn_origin::AgentTurnOrigin::WebChat {
+            thread_id: thread_id_task.clone(),
+            client_id: client_id_task.clone(),
+        };
+        let result = tokio::select! {
+            biased;
+            _ = task_cancel_token.cancelled() => None,
+            res = crate::openhuman::agent::turn_origin::with_origin(
+                origin,
+                crate::openhuman::approval::APPROVAL_CHAT_CONTEXT.scope(
+                    approval_ctx,
+                    run_chat_task(
+                        &client_id_task,
+                        &thread_id_task,
+                        &request_id_task,
+                        &user_message,
+                        model_override,
+                        temperature,
+                        profile_id,
+                        locale,
+                        run_queue,
+                        metadata,
+                        /* fork */ true,
+                    ),
+                ),
+            ) => Some(res),
+        };
+
+        match result {
+            Some(Ok(chat_result)) => {
+                crate::openhuman::channels::providers::presentation::deliver_response(
+                    &client_id_task,
+                    &thread_id_task,
+                    &request_id_task,
+                    &chat_result.full_response,
+                    &user_message,
+                    &chat_result.citations,
+                )
+                .await;
+            }
+            Some(Err(err)) => {
+                log::warn!(
+                    "[web-channel] parallel run_chat_task failed client_id={} thread_id={} request_id={} error={}",
+                    client_id_task,
+                    thread_id_task,
+                    request_id_task,
+                    err
+                );
+                let classified = classify_inference_error(&err);
+                publish_web_channel_event(WebChannelEvent {
+                    event: "chat_error".to_string(),
+                    client_id: client_id_task.clone(),
+                    thread_id: thread_id_task.clone(),
+                    request_id: request_id_task.clone(),
+                    message: Some(classified.message),
+                    error_type: Some(classified.error_type.to_string()),
+                    error_source: Some(classified.source.to_string()),
+                    error_retryable: Some(classified.retryable),
+                    error_retry_after_ms: classified.retry_after_ms,
+                    error_provider: classified.provider,
+                    error_fallback_available: classified.fallback_available,
+                    ..Default::default()
+                });
+            }
+            None => {
+                log::info!(
+                    "[web-channel] parallel turn cancelled cooperatively thread_id={} request_id={}",
+                    thread_id_task,
+                    request_id_task
+                );
+            }
+        }
+
+        PARALLEL_IN_FLIGHT.lock().await.remove(&request_id_task);
+    });
+
+    PARALLEL_IN_FLIGHT.lock().await.insert(
+        request_id,
+        ParallelEntry {
+            thread_id: thread_id.to_string(),
+            handle,
+            cancel_token,
+        },
+    );
+}
+
+/// Cooperatively cancel every parallel turn on a thread. Returns the cancelled
+/// request ids. Used by the thread-level cancel paths so a cancel/stop also
+/// tears down any concurrent forked turns, not just the primary turn.
+async fn cancel_parallel_turns_for_thread(thread_id: &str) -> Vec<String> {
+    let mut cancelled = Vec::new();
+    let mut parallel = PARALLEL_IN_FLIGHT.lock().await;
+    let request_ids: Vec<String> = parallel
+        .iter()
+        .filter(|(_, entry)| entry.thread_id == thread_id)
+        .map(|(request_id, _)| request_id.clone())
+        .collect();
+    for request_id in request_ids {
+        if let Some(entry) = parallel.remove(&request_id) {
+            entry.cancel_token.cancel();
+            let mut handle = entry.handle;
+            tokio::spawn(async move {
+                tokio::select! {
+                    _ = &mut handle => {}
+                    _ = tokio::time::sleep(Duration::from_secs(5)) => {
+                        handle.abort();
+                    }
+                }
+            });
+            cancelled.push(request_id);
+        }
+    }
+    cancelled
+}
+
 pub async fn invalidate_thread_sessions(thread_id: &str) {
     let mut sessions = THREAD_SESSIONS.lock().await;
     let keys_to_remove: Vec<String> = sessions
@@ -504,6 +776,16 @@ pub async fn in_flight_entries_for_test() -> Vec<(String, String)> {
         .collect()
 }
 
+/// Test accessor: `(request_id, thread_id)` for every in-flight parallel turn.
+#[cfg(any(test, debug_assertions))]
+pub async fn parallel_in_flight_entries_for_test() -> Vec<(String, String)> {
+    let guard = PARALLEL_IN_FLIGHT.lock().await;
+    guard
+        .iter()
+        .map(|(request_id, entry)| (request_id.clone(), entry.thread_id.clone()))
+        .collect()
+}
+
 pub async fn cancel_chat(client_id: &str, thread_id: &str) -> Result<Option<String>, String> {
     let client_id = client_id.trim();
     let thread_id = thread_id.trim();
@@ -521,12 +803,17 @@ pub async fn cancel_chat(client_id: &str, thread_id: &str) -> Result<Option<Stri
     {
         let mut in_flight = IN_FLIGHT.lock().await;
         if let Some(existing) = in_flight.remove(&map_key) {
-            removed_request_id = Some(existing.request_id.clone());
-            existing.handle.abort();
+            removed_request_id = Some(cancel_in_flight_gracefully(existing));
         }
     }
 
-    if let Some(request_id) = removed_request_id.clone() {
+    // Also tear down any concurrent parallel (forked) turns on the thread so a
+    // cancel/stop covers every in-flight turn, not just the primary one.
+    let cancelled_parallel = cancel_parallel_turns_for_thread(thread_id).await;
+
+    // Emit a cancelled chat_error for each cancelled turn (primary + parallels)
+    // so every interleaved branch's UI is resolved.
+    for request_id in removed_request_id.iter().cloned().chain(cancelled_parallel) {
         publish_web_channel_event(WebChannelEvent {
             event: "chat_error".to_string(),
             client_id: client_id.to_string(),

src/openhuman/channels/providers/web/run_task.rs

@@ -30,6 +30,11 @@ pub(crate) async fn run_chat_task(
     locale: Option<String>,
     run_queue: Arc<crate::openhuman::agent::harness::run_queue::RunQueue>,
     metadata: ChatRequestMetadata,
+    // When true, run as an isolated fork: build a fresh agent seeded from the
+    // thread's history-at-start and never touch the shared `THREAD_SESSIONS`
+    // cache, so a concurrent same-thread (parallel) turn cannot clobber — or be
+    // clobbered by — the primary turn's cached agent. See `QueueMode::Parallel`.
+    fork: bool,
 ) -> Result<WebChatTaskResult, String> {
     #[cfg(any(test, debug_assertions))]
     {
@@ -45,6 +50,40 @@ pub(crate) async fn run_chat_task(
         }
     }
 
+    // Test hook: park the turn in-flight so concurrency / cooperative
+    // cancellation can be observed. A `Drop` guard flips the supplied flag if
+    // this future is dropped (i.e. cancelled) before the sleep elapses, proving
+    // the turn was torn down cooperatively rather than left running.
+    #[cfg(any(test, debug_assertions))]
+    {
+        let block = {
+            let slot = super::ops::TEST_RUN_CHAT_TASK_BLOCK.lock().await;
+            slot.clone()
+        };
+        if let Some(block) = block {
+            struct DropGuard(std::sync::Arc<std::sync::atomic::AtomicBool>);
+            impl Drop for DropGuard {
+                fn drop(&mut self) {
+                    self.0.store(true, std::sync::atomic::Ordering::SeqCst);
+                }
+            }
+            let _guard = DropGuard(block.dropped.clone());
+            log::debug!(
+                "[web-channel][test] parking run_chat_task thread_id={} request_id={}",
+                thread_id,
+                request_id
+            );
+            // Signal that the turn future is live and parked, so a test can
+            // cancel only after the guard exists (otherwise a `biased` cancel
+            // could short-circuit before this future is ever polled).
+            block
+                .started
+                .store(true, std::sync::atomic::Ordering::SeqCst);
+            tokio::time::sleep(std::time::Duration::from_secs(30)).await;
+            return Err("test block elapsed".to_string());
+        }
+    }
+
     let config = config_rpc::load_config_with_timeout().await?;
     let (_profiles_state, profile) =
         AgentProfileStore::new(config.workspace_dir.clone()).resolve(profile_id.as_deref())?;
@@ -62,7 +101,11 @@ pub(crate) async fn run_chat_task(
         provider_role,
     );
 
-    let prior = {
+    // A forked (parallel) turn never reuses or evicts the shared cached agent —
+    // it always builds fresh from the history snapshot below.
+    let prior = if fork {
+        None
+    } else {
         let mut sessions = THREAD_SESSIONS.lock().await;
         sessions.remove(&map_key)
     };
@@ -242,7 +285,9 @@ pub(crate) async fn run_chat_task(
 
     agent.set_on_progress(None);
 
-    {
+    // Only the primary (non-fork) turn writes its agent back to the shared
+    // cache; a fork is fully isolated and lets its agent drop here.
+    if !fork {
         let mut sessions = THREAD_SESSIONS.lock().await;
         sessions.insert(
             map_key,

src/openhuman/channels/providers/web/types.rs

@@ -37,6 +37,24 @@ pub(super) struct InFlightEntry {
     pub(super) request_id: String,
     pub(super) handle: tokio::task::JoinHandle<()>,
     pub(super) run_queue: std::sync::Arc<crate::openhuman::agent::harness::run_queue::RunQueue>,
+    /// Cooperative cancellation for this turn. Cancelling it makes the turn's
+    /// `tokio::select!` arm fire and drops the in-flight turn future (which
+    /// cancels the in-flight LLM request and releases locks at a safe await
+    /// point) — a graceful alternative to the abrupt `handle.abort()`. The
+    /// handle is retained only as a hard backstop if cooperative cancellation
+    /// does not land within a grace period.
+    pub(super) cancel_token: tokio_util::sync::CancellationToken,
+}
+
+/// A concurrent, forked (`QueueMode::Parallel`) turn on a thread. Tracked in a
+/// separate lane keyed by `request_id` so any number can run alongside the
+/// primary in-flight turn without participating in interrupt/steer/queue
+/// semantics. Carries `thread_id` so cancellation-by-thread can find it.
+#[derive(Debug)]
+pub(super) struct ParallelEntry {
+    pub(super) thread_id: String,
+    pub(super) handle: tokio::task::JoinHandle<()>,
+    pub(super) cancel_token: tokio_util::sync::CancellationToken,
 }
 
 #[derive(Debug, Clone)]

src/openhuman/channels/providers/web_tests.rs

@@ -2,14 +2,18 @@ use super::{
     all_web_channel_controller_schemas, all_web_channel_registered_controllers, cancel_chat,
     classify_inference_error, compose_system_prompt_suffix, event_session_id_for,
     extract_provider_error_detail, generic_inference_error_user_message,
-    inference_budget_exceeded_user_message, is_inference_budget_exceeded_error, json_output,
-    key_for, locale_reply_directive, normalize_model_override, optional_bool, optional_f64,
-    optional_string, optional_u64, provider_role_for_model_override, required_string, schemas,
-    set_test_forced_run_chat_task_error, start_chat, subscribe_web_channel_events,
-    ChatRequestMetadata, ClassifiedError, WebChatParams,
+    in_flight_entries_for_test, inference_budget_exceeded_user_message,
+    is_inference_budget_exceeded_error, json_output, key_for, locale_reply_directive,
+    normalize_model_override, optional_bool, optional_f64, optional_string, optional_u64,
+    parallel_in_flight_entries_for_test, provider_role_for_model_override, required_string,
+    schemas, set_test_forced_run_chat_task_error, set_test_run_chat_task_block, start_chat,
+    subscribe_web_channel_events, ChatRequestMetadata, ClassifiedError, TestRunChatTaskBlock,
+    WebChatParams,
 };
 use crate::core::TypeSchema;
 use once_cell::sync::Lazy;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
 use tokio::sync::Mutex as TokioMutex;
 use tokio::time::{timeout, Duration};
 
@@ -1691,3 +1695,217 @@ fn web_chat_params_deserialize_with_all_ptt_fields_present() {
     assert_eq!(parsed.source.as_deref(), Some("ptt"));
     assert_eq!(parsed.session_id, Some(42));
 }
+
+/// Helper: poll the global in-flight table until `pred` holds (or time out).
+async fn wait_for_in_flight<F: Fn(&[(String, String)]) -> bool>(pred: F) -> Vec<(String, String)> {
+    timeout(Duration::from_secs(5), async {
+        loop {
+            let entries = in_flight_entries_for_test().await;
+            if pred(&entries) {
+                return entries;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("in-flight condition not met before timeout")
+}
+
+/// Helper: poll an `AtomicBool` until it is `true` (or time out).
+async fn wait_for_flag(flag: &Arc<AtomicBool>, what: &str) {
+    timeout(Duration::from_secs(5), async {
+        while !flag.load(Ordering::SeqCst) {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap_or_else(|_| panic!("flag '{what}' was not set before timeout"));
+}
+
+fn make_block() -> TestRunChatTaskBlock {
+    TestRunChatTaskBlock {
+        started: Arc::new(AtomicBool::new(false)),
+        dropped: Arc::new(AtomicBool::new(false)),
+    }
+}
+
+/// Two turns on DISTINCT threads must be in-flight at the same time — the core
+/// invariant behind cross-thread parallel inference.
+#[tokio::test]
+async fn start_chat_runs_distinct_threads_concurrently() {
+    let _serial = FORCED_ERROR_TEST_LOCK.lock().await;
+    let block = make_block();
+    set_test_run_chat_task_block(Some(block.clone())).await;
+
+    let thread_a = "concurrent-thread-a";
+    let thread_b = "concurrent-thread-b";
+
+    start_chat(
+        "client-a",
+        thread_a,
+        "hello a",
+        None,
+        None,
+        None,
+        None,
+        None,
+        ChatRequestMetadata::default(),
+    )
+    .await
+    .expect("thread A should start");
+    start_chat(
+        "client-b",
+        thread_b,
+        "hello b",
+        None,
+        None,
+        None,
+        None,
+        None,
+        ChatRequestMetadata::default(),
+    )
+    .await
+    .expect("thread B should start");
+
+    // Both threads' turns must be parked in-flight simultaneously.
+    let entries = wait_for_in_flight(|e| {
+        let keys: Vec<&str> = e.iter().map(|(k, _)| k.as_str()).collect();
+        keys.contains(&thread_a) && keys.contains(&thread_b)
+    })
+    .await;
+    assert!(
+        entries.iter().any(|(k, _)| k == thread_a) && entries.iter().any(|(k, _)| k == thread_b),
+        "expected both threads in-flight concurrently, got {entries:?}"
+    );
+
+    // Cleanup: cancel both and clear the test hook.
+    let _ = cancel_chat("client-a", thread_a).await;
+    let _ = cancel_chat("client-b", thread_b).await;
+    set_test_run_chat_task_block(None).await;
+}
+
+/// `cancel_chat` must cooperatively tear down the in-flight turn (drop its
+/// future at the next await point) rather than leave it sleeping — proven by
+/// the parked future's `Drop` guard firing well before its 30s sleep elapses.
+#[tokio::test]
+async fn cancel_chat_cooperatively_stops_in_flight_turn() {
+    let _serial = FORCED_ERROR_TEST_LOCK.lock().await;
+    let block = make_block();
+    set_test_run_chat_task_block(Some(block.clone())).await;
+
+    let thread_id = "cancel-coop-thread";
+    let request_id = start_chat(
+        "cancel-client",
+        thread_id,
+        "park me",
+        None,
+        None,
+        None,
+        None,
+        None,
+        ChatRequestMetadata::default(),
+    )
+    .await
+    .expect("turn should start");
+
+    // Wait until the turn future has actually parked (guard created) — only then
+    // is a cooperative cancel meaningful.
+    wait_for_flag(&block.started, "turn started").await;
+    assert!(
+        !block.dropped.load(Ordering::SeqCst),
+        "turn should still be parked, not yet dropped"
+    );
+
+    let cancelled = cancel_chat("cancel-client", thread_id)
+        .await
+        .expect("cancel_chat should succeed");
+    assert_eq!(
+        cancelled.as_deref(),
+        Some(request_id.as_str()),
+        "cancel_chat should report the cancelled request id"
+    );
+
+    // The in-flight entry is removed and the parked future is dropped promptly
+    // (cooperative cancel), long before the 30s test sleep would elapse.
+    wait_for_in_flight(|e| !e.iter().any(|(k, _)| k == thread_id)).await;
+    wait_for_flag(&block.dropped, "turn future dropped by cooperative cancel").await;
+
+    set_test_run_chat_task_block(None).await;
+}
+
+/// Helper: poll the parallel in-flight lane until `pred` holds (or time out).
+async fn wait_for_parallel<F: Fn(&[(String, String)]) -> bool>(pred: F) -> Vec<(String, String)> {
+    timeout(Duration::from_secs(5), async {
+        loop {
+            let entries = parallel_in_flight_entries_for_test().await;
+            if pred(&entries) {
+                return entries;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("parallel in-flight condition not met before timeout")
+}
+
+/// A `parallel`-mode turn runs CONCURRENTLY with the primary turn on the SAME
+/// thread (it does not interrupt it), and a thread-level cancel tears down both.
+#[tokio::test]
+async fn parallel_turn_runs_concurrently_with_primary_on_same_thread() {
+    let _serial = FORCED_ERROR_TEST_LOCK.lock().await;
+    let block = make_block();
+    set_test_run_chat_task_block(Some(block.clone())).await;
+
+    let thread_id = "parallel-same-thread";
+
+    // Primary turn (default interrupt mode) parks in IN_FLIGHT.
+    start_chat(
+        "pp-client",
+        thread_id,
+        "primary",
+        None,
+        None,
+        None,
+        None,
+        None,
+        ChatRequestMetadata::default(),
+    )
+    .await
+    .expect("primary turn should start");
+    wait_for_in_flight(|e| e.iter().any(|(k, _)| k == thread_id)).await;
+
+    // Parallel turn on the SAME thread must NOT interrupt the primary — it
+    // lives in the parallel lane while the primary stays in-flight.
+    start_chat(
+        "pp-client",
+        thread_id,
+        "branch",
+        None,
+        None,
+        None,
+        None,
+        Some("parallel".to_string()),
+        ChatRequestMetadata::default(),
+    )
+    .await
+    .expect("parallel turn should start");
+
+    wait_for_parallel(|e| e.iter().any(|(_, t)| t == thread_id)).await;
+    // Primary is still in-flight — the parallel send did not interrupt it.
+    assert!(
+        in_flight_entries_for_test()
+            .await
+            .iter()
+            .any(|(k, _)| k == thread_id),
+        "primary turn must remain in-flight alongside the parallel turn"
+    );
+
+    // A thread-level cancel tears down BOTH the primary and the parallel turn.
+    cancel_chat("pp-client", thread_id)
+        .await
+        .expect("cancel should succeed");
+    wait_for_in_flight(|e| !e.iter().any(|(k, _)| k == thread_id)).await;
+    wait_for_parallel(|e| !e.iter().any(|(_, t)| t == thread_id)).await;
+
+    set_test_run_chat_task_block(None).await;
+}