Repository navigation
feat(chat): parallel agent inference (cross-thread + within-thread forks) - #3633
Conversation
…llelism Replace the global single `activeThreadId` lock with a per-thread `activeThreadIds` set so turns on different threads run concurrently — matching Codex's model where each conversation is the unit of concurrency and the core already runs per-thread turns in isolated tokio tasks. - threadSlice: `activeThreadId: string|null` -> `activeThreadIds: Record<string,true>`; add markThreadInferenceActive / clearThreadInferenceActive; setActiveThread kept as compat (string marks active, null clears all). - Conversations: composer/voice/autocomplete gates now key on the *selected* thread only; convert single-turn refs (pending send, silence timer, status baseline) to per-thread maps/sets so concurrent sends each track their own state. Per-thread reference signatures keep the silence-timer rearm scoped to a thread's own activity. - ChatRuntimeProvider: clear inference markers per thread on done/error; disconnect reconcile clears all active threads. - socketService: re-subscribe to every active thread room on reconnect. - Tests updated + new multi-active-thread coverage. Phase 1 of parallel inference. Core hardening + within-thread fork model follow.
Replace the abrupt `handle.abort()` interrupt path with a cooperative `CancellationToken` threaded into each turn's `tokio::select!`, with a timed hard-abort backstop. Cancelling now tears the turn future down at a safe await point — cancelling the in-flight LLM request and releasing locks cleanly — instead of killing the task mid-write, which matters for the append-only transcripts Phase 3 will rely on. - InFlightEntry gains `cancel_token`; start_chat wraps the turn in a biased select on `token.cancelled()` with a quiet cooperative-cancel branch (the interrupting/cancelling side already emits chat_error). - cancel_in_flight_gracefully(): cancel() then a detached 5s abort backstop. Used by both the interrupt-on-new-message path and cancel_chat. - Tests: prove two distinct threads run turns concurrently, and that cancel_chat cooperatively drops the in-flight turn future (Drop guard fires well before the parked sleep would elapse). New test-only block hook with started/dropped signals. Cross-thread parallelism was already safe in the core (per-thread keying, no lock held across the LLM call); this hardens the interrupt path. Phase 2 of parallel inference.
…thread Adds within-thread parallel inference via the snapshot+append fork model. A 'parallel' queue_mode spawns an independent turn that snapshots the thread's history-at-start and runs ALONGSIDE the primary in-flight turn instead of interrupting or queueing it. - run_queue: new QueueMode::Parallel. - run_chat_task gains a `fork` flag: a forked turn never reads or writes the shared THREAD_SESSIONS agent cache (always builds fresh from the history snapshot), so concurrent same-thread turns can't clobber each other's cached agent. - ops: separate PARALLEL_IN_FLIGHT lane keyed by request_id; spawn_parallel_turn runs the fork with cooperative cancellation and emits the same per-request stream events as a primary turn (so the UI can render interleaved branches). cancel_chat now tears down the primary turn AND every parallel turn on the thread. - Test: a parallel turn runs concurrently with the primary on the same thread (does not interrupt it), and a thread cancel tears down both. Phase 3 (core) of parallel inference. Frontend trigger + interleaved bubble rendering follow.
Exposes QueueMode::Parallel in the UI. A second turn on an already-active thread can be sent as a concurrent forked 'branch' (Cmd/Ctrl+Enter), rendered as its own interleaved bubble instead of clobbering the primary stream. Mirrors the core's separate parallel lane rather than re-keying the whole chatRuntime slice (which would destabilise the primary streaming path): - chatRuntimeSlice: additive parallelStreamsByThread (threadId->requestId->stream) + parallelRequestThreads map; registerParallelRequest / setParallelStream / clearParallelRequest; wired into clearRuntimeForThread / clearAllChatRuntime. - ChatRuntimeProvider: onTextDelta/onThinkingDelta route forked-turn deltas to the parallel lane; onDone/onError resolve only the forked lane, leaving the primary turn's stream/status/lifecycle/active marker intact. - chatService: QueueMode gains 'parallel'; chatSend returns the request_id. - Conversations: handleSendParallel (own path, never touches the primary lifecycle) + Cmd/Ctrl+Enter trigger; renders parallel branch bubbles. - ChatComposer: allowParallelSend keeps the textarea editable while the thread is active; branch hint placeholder. - i18n: chat.parallelBranchHint / chat.parallelBranchLabel in all 14 locales. - Tests: parallel-lane reducers + provider routing; reset session usage in the provider test harness for isolation. Phase 3c of parallel inference — feature complete (cross-thread + within-thread).
# Conflicts: # app/src/components/chat/ChatComposer.tsx
|
Warning Review limit reached
More reviews will be available in 3 hours, 34 minutes, and 31 seconds. Learn how PR review limits work. Your organization has run out of usage credits. Purchase more credits in the billing tab to continue. ⌛ How to resolve this issue?After more reviews become available, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans include higher PR review limits than trial, open-source, and free plans. In all cases, reviews become available again over time. During sustained high-volume PR review activity, CodeRabbit may temporarily slow when the next review becomes available. Please see our Fair Usage Limits Policy for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (36)
Comment |
After the upstream tinyhumansai#3633 (parallel inference) merge, run_chat_task's future is very large. Adding the per-profile with_source_scope wrapper inlined the giant run_single future into an extra ambient-scope layer, pushing the run_chat_task frame past the default test-thread stack — every web-channel turn coverage test aborted with 'stack overflow' under cargo-llvm-cov (and plain cargo test). Box the run_single future so both ambient-scope wrappers (thread_id + source scope) hold a pointer instead of inlining it. No behaviour change. Verified the previously-aborting lib + channels_web_* e2e targets now pass.
Summary
activeThreadIdlock becomes a per-threadactiveThreadIdsset.QueueMode::Parallelspawns an independent forked turn (snapshot-history-at-start, run alongside, commit append-only) instead of interrupting; Cmd/Ctrl+Enter sends it, rendered as an interleaved "branch" bubble.CancellationTokeninterruption (with timed hard-abort backstop) replaces the abrupthandle.abort().Problem
The desktop app serialized agent inference: the Rust core was already structurally per-thread (each web turn is its own
tokio::spawn, no lock held across the LLM call, conversation store is append-only under a global lock), but the frontend's single globalactiveThreadIdflag blocked the composer whenever any thread had a turn in flight. There was also no way to run two turns on the same conversation, and interruption used a hard task abort that could tear a turn down mid-write.Solution
threadSlice.activeThreadId→activeThreadIdsset; composer/voice/autocomplete gate on the selected thread only; the single-turn refs inConversations.tsx(pending-send, 120s silence timer, status baseline) become per-thread maps keyed by reference-identity signatures.InFlightEntrycarries aCancellationToken; the turn runs undertokio::select!;cancel_in_flight_gracefully()cancels cooperatively then hard-aborts after a 5s grace.QueueMode::Parallel+run_chat_task(fork: bool)(a fork never touches the sharedTHREAD_SESSIONSagent cache) + a separatePARALLEL_IN_FLIGHTlane keyed byrequest_id;cancel_chattears down the primary turn and all parallel turns.parallelStreamsByThreadlane (mirrors the core's separate lane rather than re-keying the whole slice, keeping the battle-tested primary stream path untouched);ChatRuntimeProviderroutes forked-turn deltas/done/error to the lane;chatSendreturns therequest_id;handleSendParallel+ a typeable-while-active composer render branch bubbles.Submission Checklist
diff-cover) meet the gate enforced by.github/workflows/pr-ci.yml. Runpnpm test:coverageandpnpm test:rustlocally; PRs below 80% on changed lines will not merge.docs/TEST-COVERAGE-MATRIX.md## Relateddocs/RELEASE-MANUAL-SMOKE.md)Impact
Related
AI Authored PR Metadata (required for Codex/Linear PRs)
Linear Issue
Commit & Branch
Validation Run
pnpm --filter openhuman-app format:checkpnpm typecheckcargo fmt --check+pnpm rust:check(pre-push hook) greenapp/src-tauriRust changed (verified by pre-pushrust:check)Validation Blocked
command:N/Aerror:N/Aimpact:N/ABehavior Changes
parallelqueue mode forks a concurrent same-thread turn.Parity Contract
interrupt; the primary streaming path andstreamingAssistantByThreadare untouched (additive parallel lane).QueueMode::Paralleladded to all match arms; forked turns isolated from theTHREAD_SESSIONScache; narrow test stores defended viaactiveThreadIds ?? {}.Duplicate / Superseded PR Handling