From 8e934ed5f2fcd9deb52a14a52de742925bdadc83 Mon Sep 17 00:00:00 2001 From: zhangmo8 Date: Wed, 15 Jul 2026 22:07:12 +0800 Subject: [PATCH 1/2] fix(chat): harden session view ownership --- .../chat-scroll-ownership/spec.md | 13 + .../chat-scroll-ownership/tasks.md | 10 +- docs/issues/chat-session-view-races/spec.md | 114 +++++ src/renderer/src/pages/ChatPage.vue | 204 ++++++--- src/renderer/src/stores/ui/message.ts | 200 +++++++-- src/renderer/src/stores/ui/messageIpc.ts | 132 +++++- src/renderer/src/stores/ui/pendingInput.ts | 64 ++- src/renderer/src/stores/ui/session.ts | 30 +- src/renderer/src/stores/ui/stream.ts | 21 +- test/renderer/components/ChatPage.test.ts | 124 +++++- .../stores/messageStore.reactivity.test.ts | 1 + test/renderer/stores/messageStore.test.ts | 399 ++++++++++++++++++ .../renderer/stores/pendingInputStore.test.ts | 43 ++ test/renderer/stores/sessionStore.test.ts | 45 +- 14 files changed, 1249 insertions(+), 151 deletions(-) create mode 100644 docs/issues/chat-session-view-races/spec.md diff --git a/docs/architecture/chat-scroll-ownership/spec.md b/docs/architecture/chat-scroll-ownership/spec.md index 697f45d590..b4595b0e61 100644 --- a/docs/architecture/chat-scroll-ownership/spec.md +++ b/docs/architecture/chat-scroll-ownership/spec.md @@ -23,6 +23,12 @@ GitHub issue sync was not requested and is not part of this work. prepared records atomically, and uses a renderer-only five-session/count-and-memory-bounded LRU. - Pending inputs load as secondary state and no longer block first message commit. Recent parsed messages and message measurements remain bounded and reusable across cached switches. +- The message store is the single readiness owner. Load, history, cache, submit, stream, pending + input, and session hydration commits use request generations so session-ID ABA cycles cannot + revive stale work. +- Same-session loads are fenced by live mutation revisions. Streams remain session-scoped before + commit and are ordered and settled by request identity; stale terminals cannot clear newer + streams and duplicate terminals cannot start duplicate refreshes. - Unit, component, architecture-guard, and opt-in Electron smoke coverage protect the new contract. ## Validation record @@ -240,6 +246,8 @@ secondary path viewport anchors. The cache is ephemeral, keyed by session ID plus message revision, and never persisted. - A prepared session result commits only when its session epoch is still current. +- Selection changes advance both message-load and history generations, including A-B-A cycles; + cache misses do not advance either generation. - A failed or superseded preparation does not expose the target view or enable its composer unless a matching cached view was already committed safely. - Message data and the latest viewport are the critical path. Pending inputs, plan state, status @@ -257,6 +265,11 @@ secondary path reading mode afterward. - Token revisions do not independently write scroll position. They invalidate one coalesced auto-follow request. +- Stream snapshots are monotonic within a request. Each request settles once, terminal events are + ignored when a newer same-session request owns the view, and post-terminal snapshots cannot + resurrect streaming state. +- Async submit, queue, steer, and compaction continuations must still own the same page generation + before mutating composer or optimistic view state. - When the user scrolls away, streaming continues without changing the viewport. - Coalesced search or Spotlight requests preserve the ownership state captured by the first active navigation transaction until that transaction completes. diff --git a/docs/architecture/chat-scroll-ownership/tasks.md b/docs/architecture/chat-scroll-ownership/tasks.md index 48a87281d8..d8be2fdb7f 100644 --- a/docs/architecture/chat-scroll-ownership/tasks.md +++ b/docs/architecture/chat-scroll-ownership/tasks.md @@ -26,9 +26,9 @@ - [x] Add an immutable prepared-session-view type with stale-epoch rejection. - [x] Replace visible clear-then-load behavior with one atomic target-session commit. - [x] Allow message readiness to unblock paint before pending inputs, plans, and metadata. -- [ ] Add a bounded renderer-only LRU for recent session views, measurements, and anchors. -- [ ] Add cache revision invalidation and memory/count eviction tests. -- [ ] Add cold initial load, cached switch, uncached switch, and rapid A/B/C race tests. +- [x] Add a bounded renderer-only LRU for recent session views, measurements, and anchors. +- [x] Add cache revision invalidation and memory/count eviction tests. +- [x] Add cold initial load, cached switch, uncached switch, and rapid A/B/C race tests. ## Phase 3: Controller integration on current DOM @@ -100,3 +100,7 @@ - [x] Persist recent measurement snapshots across the keyed `ChatPage` remount lifecycle. - [x] Replace timing-only Electron search assertions with observable completion state. - [x] Add regressions for every review-hardening invariant and rerun the full quality gates. +- [x] Make committed message readiness store-owned and fence same-session refreshes against live + message mutations. +- [x] Fence stream terminal identity, pending-input writes, history overlap, submit continuations, + and A-B-A session hydration. diff --git a/docs/issues/chat-session-view-races/spec.md b/docs/issues/chat-session-view-races/spec.md new file mode 100644 index 0000000000..1e18af18bd --- /dev/null +++ b/docs/issues/chat-session-view-races/spec.md @@ -0,0 +1,114 @@ +# Chat Session View Races + +## Issue + +PR #1974 introduced atomic chat-session view commits and a recent-session cache, but the page and +message store retain separate readiness state. A first-turn stream completion can supersede the +page-owned restore request, commit the correct messages through another caller, and leave +`ChatPage` permanently guarded. Cached or already-visible sessions have the inverse race: the +composer is enabled while a background restore can still replace optimistic or streaming records +with a snapshot that started before those records existed. + +## Impact + +- The first submit can appear to do nothing even though the backend accepted it. +- Switching away and back can make the missing messages appear. +- Cached session refreshes can temporarily remove optimistic user messages or streaming assistant + content. +- A cache miss can cancel an unrelated in-flight restore before the replacement restore starts. +- Known inactive-session updates do not invalidate cached message views. +- Late stream terminals, snapshots, pending-input mutations, and session hydration can pass a + session-ID-only guard after an A-B-A navigation cycle. + +No persisted messages are lost. The failures are renderer ownership, readiness, and stale-view +presentation problems. + +## Root Cause + +1. `ChatPage` owns `committedMessageSessionId` and `isSessionViewPreparing` separately from + `messageStore.committedSessionId`. Only the page's own restore promise releases its guard. +2. `messageStore.loadMessages()` rejects superseded requests but does not protect a current request + from optimistic, streaming, or history mutations that occur after the request starts. +3. `setCurrentSessionId()` treats an empty selected session as committed before a prepared message + view exists so early stream records can mutate the shared visible cache. +4. `activateRecentSessionView()` invalidates active loads before it knows whether a cache entry + exists, and cached revisions are stored without being validated or invalidated by session + events. +5. `ChatPage` clears streaming state on mount even when the stream belongs to the newly selected + session. +6. Stream terminal handling is scoped only by session ID, so an old completion can clear a newer + same-session stream, duplicate terminals can launch duplicate refreshes, and a post-terminal + snapshot can resurrect settled state. +7. Pending-input loads and session detail hydration compare only session IDs. A rapid A-B-A cycle + makes an old A response look current again, while pending-input mutations write returned arrays + without checking which view owns them. +8. History pagination writes records into the cache before it removes duplicate IDs, allowing an + overlapping old page to replace a newer in-memory record. + +## Required Invariants + +- The message store is the sole owner of the selected and committed session-view identities. +- `committedSessionId` means a complete message view, including a valid empty view; selection alone + never commits it. +- Any successful matching store commit releases the page loading boundary, regardless of which + caller initiated the load. +- A persisted refresh never replaces same-session message mutations that occurred after the + refresh started. +- Target-session streams received before the first commit remain staged and become visible after + the base view commits without mixing with a previous session. +- Cache misses do not cancel loads. Known message/session mutations invalidate matching cached + views before activation. +- Session switches, failed sends, and late async submit preparation cannot write records into a + different committed view. +- Stream updates are monotonic per request, a request settles once, and a terminal event cannot + settle a different request in the same session. +- Selection and async submit guards use request generations, not session equality alone, so A-B-A + cannot revive an earlier operation. +- Pending-input mutations refresh only their matching active view, and overlapping history pages + never replace records already owned by the committed view. +- Message loading continues in parallel with pending-input and other secondary-state loading. + +## Fix Plan + +- Derive ChatPage readiness directly from the store's committed session identity and handle the + first matching commit as a reactive transition. +- Give message loads a same-session mutation fence. Discard a prepared snapshot when the visible + view changed after request start; allow a later persisted refresh to reconcile optimistic records. +- Stage pre-commit stream state outside the visible message cache, then hydrate the current stream + into the newly committed view. +- Validate cached view revisions, make cache activation non-destructive on misses, and invalidate + cache entries on known session updates/status changes. +- Scope streaming-derived UI state to the active session and preserve a matching stream during page + mount. +- Add defensive session ownership checks around optimistic mutations and post-await submit paths. +- Fence stream snapshots and terminals by request identity and monotonic event time, including + duplicate terminal suppression and inactive-session cache invalidation. +- Fence pending-input operations, session hydration, history pagination, and submit continuations + against stale or ABA ownership. + +## Tasks + +- [x] Make committed view ownership and readiness store-driven. +- [x] Add mutation-fenced refresh commits and pre-commit stream staging. +- [x] Make recent-session cache activation and invalidation race-safe. +- [x] Update ChatPage submit, stream, and restore integration. +- [x] Fence stream terminals, pending-input mutations, history overlap, and session hydration. +- [x] Add cold-start, first-turn completion, cached-refresh, cache-miss, and rapid-switch regressions. +- [x] Update the retained chat scroll ownership contract. +- [x] Run focused and full repository quality gates. + +## Validation + +- Renderer store and ChatPage Vitest suites cover deterministic deferred-promise interleavings, + including A-B-A selection, submit, pending-input, and hydration cycles. +- Full `pnpm test` completed with 5,516 passing and 198 skipped tests. Its only three failures were + Feishu loopback callback tests blocked from listening on `127.0.0.1` by the sandbox; the complete + callback suite passed 22/22 when rerun with local-listen permission. +- Web/Node typecheck, format, i18n, lint, and the production build pass. The build used existing + provider and ACP registry snapshots because the validation environment could not reach their + remote refresh endpoints. +- The final diff contains no database, persisted-data, preload, or shared IPC contract changes. + +## GitHub + +No issue is linked. The requested delivery artifact is a pull request against `dev`. diff --git a/src/renderer/src/pages/ChatPage.vue b/src/renderer/src/pages/ChatPage.vue index 28b7ff47e0..edc7d82852 100644 --- a/src/renderer/src/pages/ChatPage.vue +++ b/src/renderer/src/pages/ChatPage.vue @@ -341,14 +341,23 @@ const modelClient = createModelClient() const sessionClient = createSessionClient() const { t } = useI18n() const { toast } = useToast() -const committedMessageSessionId = ref(null) -const isSessionViewPreparing = ref(false) +const isSessionViewCommitted = computed( + () => + messageStore.currentSessionId === props.sessionId && + messageStore.committedSessionId === props.sessionId +) +const isSessionViewPreparing = computed(() => + Boolean(props.sessionId && !isSessionViewCommitted.value) +) +const isCurrentSessionStreaming = computed( + () => messageStore.isStreaming && messageStore.currentStreamSessionId === props.sessionId +) const sessionTitle = computed(() => sessionStore.activeSession?.title ?? t('common.newChat')) const sessionProject = computed(() => sessionStore.activeSession?.projectDir ?? '') const isReadOnlySession = computed(() => sessionStore.activeSession?.sessionKind === 'subagent') const isGenerating = computed( - () => sessionStore.activeSession?.status === 'working' || messageStore.isStreaming + () => sessionStore.activeSession?.status === 'working' || isCurrentSessionStreaming.value ) const RATE_LIMIT_STREAM_MESSAGE_PREFIX = '__rate_limit__:' const INITIAL_MESSAGE_RESTORE_COUNT = 100 @@ -392,35 +401,14 @@ async function restoreSessionMessages(id: string, requestId: number) { return } - if (restoredSession === null || messageStore.committedSessionId !== id) { - if (committedMessageSessionId.value !== id) { - isSessionViewPreparing.value = true - } - return + if (restoredSession !== null) { + applyRestoredSessionSummary(restoredSession) } - - markChatSessionPerformance('messages-prepared', id, chatScrollSessionEpoch) - committedMessageSessionId.value = id - isSessionViewPreparing.value = false - markChatSessionPerformance('messages-committed', id, chatScrollSessionEpoch) - applyRestoredSessionSummary(restoredSession) void pendingInputsPromise.then(() => { if (requestId === sessionRestoreRequestId) { markChatSessionPerformance('secondary-state-ready', id, chatScrollSessionEpoch) } }) - - await nextTick() - window.requestAnimationFrame(() => { - if (requestId === sessionRestoreRequestId) { - markChatSessionPerformance('first-message-paint', id, chatScrollSessionEpoch) - } - }) - if (spotlightStore.pendingMessageJump?.sessionId === id) { - void focusPendingSpotlightMessageJump() - return - } - requestChatScroll('session-restore', { kind: 'bottom' }) } // --- Auto-scroll --- @@ -523,6 +511,8 @@ let restoreMessageWindowMeasurements = (snapshot: MessageMeasurementSnapshot) => } let measurementSessionId = '' let sessionRestoreRequestId = 0 +let isChatPageActive = true +let handledCommittedSessionId: string | null = null type LogicalViewportAnchor = { messageId: string layoutTop: number @@ -1044,6 +1034,7 @@ function cacheCurrentMessageMeasurements(): void { watch( () => props.sessionId, async (id) => { + handledCommittedSessionId = null if (measurementSessionId && measurementSessionId !== id) { cacheCurrentMessageMeasurements() } @@ -1068,23 +1059,9 @@ watch( if (cachedMeasurements) { restoreMessageWindowMeasurements(cachedMeasurements) } - const existingMessagesMatch = Boolean( - id && - messageStore.messages.length > 0 && - messageStore.messages.every((message) => message.sessionId === id) - ) - committedMessageSessionId.value = activatedFromCache || existingMessagesMatch ? id : null - isSessionViewPreparing.value = Boolean(id && !activatedFromCache && !existingMessagesMatch) - messageStore.clearStreamingState() + messageStore.clearStreamingStateForOtherSession(id) if (activatedFromCache) { markChatSessionPerformance('cache-committed', id, chatScrollSessionEpoch) - void nextTick(() => { - window.requestAnimationFrame(() => { - if (id === props.sessionId) { - markChatSessionPerformance('first-message-paint', id, chatScrollSessionEpoch) - } - }) - }) } pendingInputStore.clear() if (id) { @@ -1098,12 +1075,51 @@ watch( } return } - committedMessageSessionId.value = null - isSessionViewPreparing.value = false }, { immediate: true } ) +watch( + [() => props.sessionId, () => messageStore.committedSessionId], + async ([id, committedSessionId]) => { + if (!id || committedSessionId !== id || handledCommittedSessionId === id) { + return + } + + handledCommittedSessionId = id + markChatSessionPerformance('messages-prepared', id, chatScrollSessionEpoch) + markChatSessionPerformance('messages-committed', id, chatScrollSessionEpoch) + if (messageStore.committedSession?.id === id) { + applyRestoredSessionSummary(messageStore.committedSession) + } + + await nextTick() + if ( + props.sessionId !== id || + messageStore.currentSessionId !== id || + messageStore.committedSessionId !== id + ) { + return + } + + window.requestAnimationFrame(() => { + if ( + props.sessionId === id && + messageStore.currentSessionId === id && + messageStore.committedSessionId === id + ) { + markChatSessionPerformance('first-message-paint', id, chatScrollSessionEpoch) + } + }) + if (spotlightStore.pendingMessageJump?.sessionId === id) { + void focusPendingSpotlightMessageJump() + return + } + requestChatScroll('session-restore', { kind: 'bottom' }) + }, + { immediate: true, flush: 'post' } +) + function resolveAssistantModelName(modelId: string): string { if (!modelId) { return 'Assistant' @@ -1243,9 +1259,10 @@ function shouldRenderDisplayMessage(message: DisplayMessage): boolean { } const hasInlineStreamingTarget = computed(() => { + if (!isCurrentSessionStreaming.value) return false const messageId = messageStore.currentStreamMessageId if (!messageId) return false - return messageStore.messageCache.has(messageId) + return messageStore.messageCache.get(messageId)?.sessionId === props.sessionId }) const hasFirstStreamingContent = computed( () => messageStore.streamingBlocks.length > 0 && hasInlineStreamingTarget.value @@ -1254,7 +1271,7 @@ const hasFirstStreamingContent = computed( const ephemeralRateLimitMessageId = computed(() => { const messageId = messageStore.currentStreamMessageId if ( - !messageStore.isStreaming || + !isCurrentSessionStreaming.value || !messageId || !messageId.startsWith(RATE_LIMIT_STREAM_MESSAGE_PREFIX) ) { @@ -1402,11 +1419,11 @@ watch( */ const stableDisplayMessages = computed(() => { void messageStore.lastPersistedRevision - if (committedMessageSessionId.value !== props.sessionId) { + if (!isSessionViewCommitted.value) { return [] } const streamId = - messageStore.isStreaming && hasInlineStreamingTarget.value + isCurrentSessionStreaming.value && hasInlineStreamingTarget.value ? messageStore.currentStreamMessageId : null @@ -1439,14 +1456,14 @@ const stableDisplayMessages = computed(() => { /** High-frequency streaming row + pending placeholder only. */ const streamingDisplayTail = computed(() => { void messageStore.streamRevision - if (committedMessageSessionId.value !== props.sessionId) { + if (!isSessionViewCommitted.value) { return [] } const msgs: DisplayMessage[] = [] // Single-track: stream blocks are folded into the message record in place, so the // generating message is the same id/node through completion (no flash). - if (messageStore.isStreaming && hasInlineStreamingTarget.value) { + if (isCurrentSessionStreaming.value && hasInlineStreamingTarget.value) { const streamId = messageStore.currentStreamMessageId const record = streamId ? messageStore.messageCache.get(streamId) : undefined if (record) { @@ -1456,7 +1473,7 @@ const streamingDisplayTail = computed(() => { } } } else if ( - messageStore.isStreaming && + isCurrentSessionStreaming.value && messageStore.streamingBlocks.length > 0 && !hasInlineStreamingTarget.value && !ephemeralRateLimitBlock.value @@ -2112,7 +2129,7 @@ function hasPendingInteractionForSession(sessionId: string): boolean { } function isSessionPlanActive(sessionId: string): boolean { - if (sessionId === props.sessionId && messageStore.isStreaming) { + if (sessionId === props.sessionId && isCurrentSessionStreaming.value) { return true } @@ -2260,7 +2277,9 @@ const sessionStatusLifecycleKey = computed(() => { if (active) { entries.push(`${active.id}:${active.status}`) } - entries.push(`${props.sessionId}:${messageStore.isStreaming ? 'streaming' : 'not-streaming'}`) + entries.push( + `${props.sessionId}:${isCurrentSessionStreaming.value ? 'streaming' : 'not-streaming'}` + ) return entries.sort().join('|') }) @@ -2393,8 +2412,20 @@ const withMessageSkills = (text: string, files: MessageFile[]) => { } } +function canWriteSessionView(sessionId: string, restoreRequestId: number): boolean { + return ( + isChatPageActive && + sessionRestoreRequestId === restoreRequestId && + props.sessionId === sessionId && + messageStore.currentSessionId === sessionId && + messageStore.committedSessionId === sessionId + ) +} + function beginOutgoingTurnFeedback(sessionId: string, payload: SendMessageInput) { const optimisticUserMessageId = messageStore.addOptimisticUserMessage(sessionId, payload) + if (!optimisticUserMessageId) return null + const pendingAssistantPlaceholderId = createPendingAssistantPlaceholder(sessionId) beginPlanTurn(sessionId) return { optimisticUserMessageId, pendingAssistantPlaceholderId } @@ -2403,13 +2434,13 @@ function beginOutgoingTurnFeedback(sessionId: string, payload: SendMessageInput) async function sendMessageWithOutgoingTurnFeedback( sessionId: string, payload: SendMessageInput, - feedback: ReturnType + feedback: NonNullable> ) { try { await chatClient.sendMessage(sessionId, payload) } catch (error) { clearPendingAssistantPlaceholder(feedback.pendingAssistantPlaceholderId) - messageStore.removeOptimisticMessage(feedback.optimisticUserMessageId) + messageStore.removeOptimisticMessage(feedback.optimisticUserMessageId, sessionId) console.error('[ChatPage] send message failed:', error) } } @@ -2419,10 +2450,15 @@ async function onSubmit() { if (isSessionViewPreparing.value) return if (isAcpWorkdirMissing.value) return if (activePendingInteraction.value || isHandlingInteraction.value) return + const sessionId = props.sessionId + const restoreRequestId = sessionRestoreRequestId const text = message.value.trim() const files = (await prepareFilesForCurrentModel([...attachedFiles.value])).map((f) => toRaw(f)) + if (!canWriteSessionView(sessionId, restoreRequestId)) return if (!text && files.length === 0) return - if (await handleManualCompactionCommand(text)) { + const handledCompaction = await handleManualCompactionCommand(text, sessionId, restoreRequestId) + if (!canWriteSessionView(sessionId, restoreRequestId)) return + if (handledCompaction) { if (!isGenerating.value) { message.value = '' } @@ -2430,14 +2466,15 @@ async function onSubmit() { } const payload = withMessageSkills(text, files) if (isGenerating.value) { - await pendingInputStore.queueInput(props.sessionId, payload) + await pendingInputStore.queueInput(sessionId, payload) + if (!canWriteSessionView(sessionId, restoreRequestId)) return message.value = '' attachedFiles.value = [] clearComposerSkills() schedulePostSubmitScrollToBottom() } else { - const sessionId = props.sessionId const feedback = beginOutgoingTurnFeedback(sessionId, payload) + if (!feedback) return message.value = '' attachedFiles.value = [] clearComposerSkills() @@ -2451,31 +2488,41 @@ async function onCommandSubmit(command: string) { if (isSessionViewPreparing.value) return if (isAcpWorkdirMissing.value) return if (activePendingInteraction.value || isHandlingInteraction.value) return + const sessionId = props.sessionId + const restoreRequestId = sessionRestoreRequestId const text = command.trim() if (!text) return - if (await handleManualCompactionCommand(text)) { + const handledCompaction = await handleManualCompactionCommand(text, sessionId, restoreRequestId) + if (!canWriteSessionView(sessionId, restoreRequestId)) return + if (handledCompaction) { return } const files = await prepareFilesForCurrentModel([...attachedFiles.value]) + if (!canWriteSessionView(sessionId, restoreRequestId)) return const payload = withMessageSkills(text, files) if (isGenerating.value) { - await pendingInputStore.queueInput(props.sessionId, payload) + await pendingInputStore.queueInput(sessionId, payload) + if (!canWriteSessionView(sessionId, restoreRequestId)) return attachedFiles.value = [] clearComposerSkills() schedulePostSubmitScrollToBottom() return } - const sessionId = props.sessionId const feedback = beginOutgoingTurnFeedback(sessionId, payload) + if (!feedback) return attachedFiles.value = [] clearComposerSkills() schedulePostSubmitScrollToBottom() await sendMessageWithOutgoingTurnFeedback(sessionId, payload, feedback) } -async function handleManualCompactionCommand(text: string): Promise { +async function handleManualCompactionCommand( + text: string, + sessionId: string, + restoreRequestId: number +): Promise { if (!isManualCompactionCommand(text)) { return false } @@ -2485,10 +2532,16 @@ async function handleManualCompactionCommand(text: string): Promise { if (isGenerating.value) { return true } + if (!canWriteSessionView(sessionId, restoreRequestId)) { + return true + } try { - const result = await sessionClient.compactSession(props.sessionId) - applyRestoredSessionSummary(await loadMessagesForSession(props.sessionId)) + const result = await sessionClient.compactSession(sessionId) + if (!canWriteSessionView(sessionId, restoreRequestId)) return true + const restoredSession = await loadMessagesForSession(sessionId) + if (!canWriteSessionView(sessionId, restoreRequestId)) return true + applyRestoredSessionSummary(restoredSession) if (!result.compacted) { toast({ title: t('chat.compaction.noopTitle'), @@ -2511,13 +2564,19 @@ async function onQueueSubmit() { if (isSessionViewPreparing.value) return if (isAcpWorkdirMissing.value) return if (activePendingInteraction.value || isHandlingInteraction.value) return + const sessionId = props.sessionId + const restoreRequestId = sessionRestoreRequestId const text = message.value.trim() const files = (await prepareFilesForCurrentModel([...attachedFiles.value])).map((f) => toRaw(f)) + if (!canWriteSessionView(sessionId, restoreRequestId)) return if (!text && files.length === 0) return - if (await handleManualCompactionCommand(text)) { + const handledCompaction = await handleManualCompactionCommand(text, sessionId, restoreRequestId) + if (!canWriteSessionView(sessionId, restoreRequestId)) return + if (handledCompaction) { return } - await pendingInputStore.queueInput(props.sessionId, withMessageSkills(text, files)) + await pendingInputStore.queueInput(sessionId, withMessageSkills(text, files)) + if (!canWriteSessionView(sessionId, restoreRequestId)) return message.value = '' attachedFiles.value = [] clearComposerSkills() @@ -2528,14 +2587,20 @@ async function onSteer() { if (isSessionViewPreparing.value) return if (isAcpWorkdirMissing.value) return if (activePendingInteraction.value || isHandlingInteraction.value) return + const sessionId = props.sessionId + const restoreRequestId = sessionRestoreRequestId const text = message.value.trim() const files = (await prepareFilesForCurrentModel([...attachedFiles.value])).map((f) => toRaw(f)) + if (!canWriteSessionView(sessionId, restoreRequestId)) return if (!text && files.length === 0) return - if (await handleManualCompactionCommand(text)) { + const handledCompaction = await handleManualCompactionCommand(text, sessionId, restoreRequestId) + if (!canWriteSessionView(sessionId, restoreRequestId)) return + if (handledCompaction) { return } - beginPlanTurn(props.sessionId) - await chatClient.steerActiveTurn(props.sessionId, withMessageSkills(text, files)) + beginPlanTurn(sessionId) + await chatClient.steerActiveTurn(sessionId, withMessageSkills(text, files)) + if (!canWriteSessionView(sessionId, restoreRequestId)) return message.value = '' attachedFiles.value = [] clearComposerSkills() @@ -2760,6 +2825,9 @@ onMounted(() => { }) onUnmounted(() => { + isChatPageActive = false + sessionRestoreRequestId += 1 + attachmentFilterToken += 1 cacheCurrentMessageMeasurements() removeModelConfigChangedListener() cancelAllPlanSnapshotClearTimers() diff --git a/src/renderer/src/stores/ui/message.ts b/src/renderer/src/stores/ui/message.ts index 7900593bab..dacc61c7a4 100644 --- a/src/renderer/src/stores/ui/message.ts +++ b/src/renderer/src/stores/ui/message.ts @@ -1,5 +1,14 @@ import { defineStore } from 'pinia' -import { ref, computed, onScopeDispose, getCurrentScope, isRef, toRef, type Ref } from 'vue' +import { + ref, + shallowRef, + computed, + onScopeDispose, + getCurrentScope, + isRef, + toRef, + type Ref +} from 'vue' import { createSessionClient } from '../../../api/SessionClient' import type { DisplayAssistantMessageBlock, @@ -43,7 +52,10 @@ export const useMessageStore = defineStore('message', () => { const streamStateStore = useStreamStateStore() const isStreaming = toStoreStateRef(streamStateStore, 'isStreaming') const streamingBlocks = toStoreStateRef(streamStateStore, 'streamingBlocks') + const currentStreamSessionId = toStoreStateRef(streamStateStore, 'currentStreamSessionId') + const currentStreamRequestId = toStoreStateRef(streamStateStore, 'currentStreamRequestId') const currentStreamMessageId = toStoreStateRef(streamStateStore, 'currentStreamMessageId') + const currentStreamMetadata = toStoreStateRef(streamStateStore, 'currentStreamMetadata') const streamRevision = toStoreStateRef(streamStateStore, 'streamRevision') // --- State --- @@ -59,13 +71,16 @@ export const useMessageStore = defineStore('message', () => { const isLoadingHistory = ref(false) const parsedMessageCache = new Map() const recentSessionViews = new RecentMessageViewCache() + const messageMutationRevisions = new Map() + const recentViewInvalidationRevisions = new Map() + const dirtyRecentSessionViews = new Set() // Stream message ids currently being hydrated into the cache as a placeholder // record (before the backend persists them). Prevents re-entrant duplicate inserts. const hydratingStreamMessageIds = new Set() let latestLoadRequestId = 0 let latestHistoryRequestId = 0 let latestLoadSessionId: string | null = null - let currentSessionSummary: SessionWithState | null = null + const committedSession = shallowRef(null) // --- Getters --- const messages = computed(() => { @@ -427,10 +442,40 @@ export const useMessageStore = defineStore('message', () => { } function setCurrentSessionId(sessionId: string | null): void { + if (currentSessionId.value === sessionId) return + + latestLoadRequestId += 1 + latestHistoryRequestId += 1 + latestLoadSessionId = null + isLoadingHistory.value = false currentSessionId.value = sessionId - if (sessionId && committedSessionId.value === null && messageIds.value.length === 0) { - committedSessionId.value = sessionId - } + } + + function getMessageMutationRevision(sessionId: string): number { + return messageMutationRevisions.get(sessionId) ?? 0 + } + + function markMessageViewMutation(sessionId: string): number { + const revision = getMessageMutationRevision(sessionId) + 1 + messageMutationRevisions.set(sessionId, revision) + recentSessionViews.delete(sessionId) + return revision + } + + function markLiveMessageViewMutation(sessionId: string): number { + const revision = markMessageViewMutation(sessionId) + invalidateRecentSessionView(sessionId) + return revision + } + + function getRecentViewInvalidationRevision(sessionId: string): number { + return recentViewInvalidationRevisions.get(sessionId) ?? 0 + } + + function invalidateRecentSessionView(sessionId: string): void { + recentViewInvalidationRevisions.set(sessionId, getRecentViewInvalidationRevision(sessionId) + 1) + dirtyRecentSessionViews.add(sessionId) + recentSessionViews.delete(sessionId) } function isCurrentLoadRequest(requestId: number, sessionId: string): boolean { @@ -451,27 +496,30 @@ export const useMessageStore = defineStore('message', () => { function cacheCurrentSessionView(): void { const sessionId = committedSessionId.value - if (!sessionId) return + if (!sessionId || dirtyRecentSessionViews.has(sessionId)) return recentSessionViews.set({ sessionId, - session: currentSessionSummary, + session: committedSession.value, messageIds: messageIds.value, messageCache: messageCache.value, nextCursor: nextCursor.value, hasMoreHistory: hasMoreHistory.value, - revision: lastPersistedRevision.value + revision: getMessageMutationRevision(sessionId) }) } - function commitSessionView(view: RecentMessageView): void { + function commitSessionView( + view: RecentMessageView, + options: { clearRecentViewDirty?: boolean } = {} + ): void { if (committedSessionId.value && committedSessionId.value !== view.sessionId) { cacheCurrentSessionView() } recentSessionViews.delete(view.sessionId) - currentSessionSummary = view.session - setCurrentSessionId(view.sessionId) + committedSession.value = view.session committedSessionId.value = view.sessionId + messageMutationRevisions.set(view.sessionId, view.revision) hydratingStreamMessageIds.clear() messageCache.value = new Map(view.messageCache) messageIds.value = [...view.messageIds] @@ -479,15 +527,41 @@ export const useMessageStore = defineStore('message', () => { hasMoreHistory.value = view.hasMoreHistory isLoadingHistory.value = false lastPersistedRevision.value += 1 + if (options.clearRecentViewDirty) { + dirtyRecentSessionViews.delete(view.sessionId) + } + + const streamMessageId = currentStreamMessageId.value + if ( + isStreaming.value && + currentStreamSessionId.value === view.sessionId && + streamMessageId && + !isEphemeralStreamMessageId(streamMessageId) + ) { + applyStreamingBlocksToMessage( + streamMessageId, + view.sessionId, + streamingBlocks.value as AssistantMessageBlock[], + currentStreamMetadata.value ?? undefined + ) + } } function activateRecentSessionView(sessionId: string): boolean { + if (currentSessionId.value !== sessionId) return false + if (dirtyRecentSessionViews.has(sessionId)) return false + + const cachedView = recentSessionViews.get(sessionId) + if (!cachedView) return false + if (cachedView.revision !== getMessageMutationRevision(sessionId)) { + recentSessionViews.delete(sessionId) + return false + } + latestLoadRequestId += 1 latestHistoryRequestId += 1 latestLoadSessionId = null isLoadingHistory.value = false - const cachedView = recentSessionViews.get(sessionId) - if (!cachedView) return false commitSessionView(cachedView) return true } @@ -553,14 +627,19 @@ export const useMessageStore = defineStore('message', () => { sessionId: string, desiredCountOverride?: number ): Promise { + if (currentSessionId.value !== sessionId) { + return null + } + const desiredCount = desiredCountOverride ?? (committedSessionId.value === sessionId ? Math.max(messageIds.value.length, 100) : 100) const requestId = ++latestLoadRequestId latestHistoryRequestId += 1 latestLoadSessionId = sessionId - setCurrentSessionId(sessionId) isLoadingHistory.value = false + const mutationRevisionAtStart = getMessageMutationRevision(sessionId) + const cacheInvalidationRevisionAtStart = getRecentViewInvalidationRevision(sessionId) try { const restored = await restoreMessageWindow(sessionId, desiredCount, requestId) if (!restored) { @@ -578,21 +657,40 @@ export const useMessageStore = defineStore('message', () => { nextMessageIds.push(msg.id) } - commitSessionView({ - sessionId, - session: restored.session, - messageIds: nextMessageIds, - messageCache: nextMessageCache, - nextCursor: restored.nextCursor, - hasMoreHistory: restored.hasMore, - revision: lastPersistedRevision.value + 1 - }) + if ( + committedSessionId.value === sessionId && + getMessageMutationRevision(sessionId) !== mutationRevisionAtStart + ) { + latestLoadSessionId = null + return restored.session + } + + const committedRevision = markMessageViewMutation(sessionId) + + commitSessionView( + { + sessionId, + session: restored.session, + messageIds: nextMessageIds, + messageCache: nextMessageCache, + nextCursor: restored.nextCursor, + hasMoreHistory: restored.hasMore, + revision: committedRevision + }, + { + clearRecentViewDirty: + getRecentViewInvalidationRevision(sessionId) === cacheInvalidationRevisionAtStart + } + ) if (isCurrentLoadRequest(requestId, sessionId)) { latestLoadSessionId = null } return restored.session } catch (e) { console.error('Failed to load messages:', e) + if (isCurrentLoadRequest(requestId, sessionId)) { + latestLoadSessionId = null + } return null } } @@ -620,16 +718,17 @@ export const useMessageStore = defineStore('message', () => { } const incomingIds: string[] = [] for (const msg of page.messages) { + if (messageCache.value.has(msg.id)) { + continue + } messageCache.value.set(msg.id, msg) incomingIds.push(msg.id) } + markMessageViewMutation(sessionId) + if (incomingIds.length > 0) { - const existingIds = new Set(messageIds.value) - messageIds.value = [ - ...incomingIds.filter((id) => !existingIds.has(id)), - ...messageIds.value - ] + messageIds.value = [...incomingIds, ...messageIds.value] } nextCursor.value = page.nextCursor @@ -665,7 +764,11 @@ export const useMessageStore = defineStore('message', () => { sessionId: string, input: string | SendMessageInput, files: MessageFile[] = [] - ): string { + ): string | null { + if (currentSessionId.value !== sessionId || committedSessionId.value !== sessionId) { + return null + } + const normalizedInput = typeof input === 'string' ? { text: input, files } : input const id = `__optimistic_user_${Date.now()}_${messageIds.value.length + 1}` const record: ChatMessageRecord = { @@ -689,13 +792,22 @@ export const useMessageStore = defineStore('message', () => { createdAt: Date.now(), updatedAt: Date.now() } + markLiveMessageViewMutation(sessionId) messageCache.value.set(id, record) messageIds.value.push(id) return id } - function removeOptimisticMessage(id: string): void { + function removeOptimisticMessage(id: string, sessionId?: string): void { if (!id.startsWith('__optimistic_')) return + const targetSessionId = sessionId ?? messageCache.value.get(id)?.sessionId + if (targetSessionId && committedSessionId.value !== targetSessionId) { + invalidateRecentSessionView(targetSessionId) + return + } + if (!targetSessionId || !messageCache.value.has(id)) return + + markLiveMessageViewMutation(targetSessionId) messageCache.value.delete(id) messageIds.value = messageIds.value.filter((messageId) => messageId !== id) parsedMessageCache.delete(id) @@ -707,7 +819,7 @@ export const useMessageStore = defineStore('message', () => { latestLoadSessionId = null setCurrentSessionId(null) committedSessionId.value = null - currentSessionSummary = null + committedSession.value = null messageIds.value = [] messageCache.value.clear() nextCursor.value = null @@ -716,6 +828,9 @@ export const useMessageStore = defineStore('message', () => { parsedMessageCache.clear() hydratingStreamMessageIds.clear() recentSessionViews.clear() + messageMutationRevisions.clear() + recentViewInvalidationRevisions.clear() + dirtyRecentSessionViews.clear() clearStreamingState() } @@ -723,6 +838,13 @@ export const useMessageStore = defineStore('message', () => { streamStateStore.clearStreamingState() } + function clearStreamingStateForOtherSession(sessionId: string): void { + const streamSessionId = currentStreamSessionId.value + if (isStreaming.value && streamSessionId !== sessionId) { + clearStreamingState() + } + } + function isEphemeralStreamMessageId(messageId: string): boolean { return EPHEMERAL_STREAM_MESSAGE_PREFIXES.some((prefix) => messageId.startsWith(prefix)) } @@ -756,6 +878,7 @@ export const useMessageStore = defineStore('message', () => { ) { return } + markLiveMessageViewMutation(conversationId) upsertMessageRecord({ ...existing, content: serializedBlocks, @@ -768,6 +891,7 @@ export const useMessageStore = defineStore('message', () => { if (hydratingStreamMessageIds.has(messageId)) return hydratingStreamMessageIds.add(messageId) + markLiveMessageViewMutation(conversationId) upsertMessageRecord({ id: messageId, sessionId: conversationId, @@ -786,11 +910,16 @@ export const useMessageStore = defineStore('message', () => { const cleanupIpcBindings = bindMessageStoreIpc({ getActiveSessionId: () => currentSessionId.value, - setStreamingState: ({ sessionId, messageId, blocks }) => { - streamStateStore.setStream(sessionId, blocks, messageId) + getCurrentStreamIdentity: () => ({ + sessionId: currentStreamSessionId.value, + requestId: currentStreamRequestId.value + }), + setStreamingState: ({ sessionId, requestId, messageId, updatedAt, blocks, metadata }) => { + streamStateStore.setStream(sessionId, blocks, messageId, metadata, requestId, updatedAt) }, clearStreamingState, loadMessages, + invalidateRecentSessionView, applyStreamingBlocksToMessage, isEphemeralStreamMessageId }) @@ -801,11 +930,14 @@ export const useMessageStore = defineStore('message', () => { messageCache, isStreaming, streamingBlocks, + currentStreamSessionId, + currentStreamRequestId, currentStreamMessageId, streamRevision, lastPersistedRevision, currentSessionId, committedSessionId, + committedSession, nextCursor, hasMoreHistory, isLoadingHistory, @@ -814,6 +946,7 @@ export const useMessageStore = defineStore('message', () => { getUserMessageContent, getMessageMetadata, setCurrentSessionId, + invalidateRecentSessionView, activateRecentSessionView, loadMessages, loadOlderMessages, @@ -821,6 +954,7 @@ export const useMessageStore = defineStore('message', () => { addOptimisticUserMessage, removeOptimisticMessage, clearStreamingState, + clearStreamingStateForOtherSession, clear } }) diff --git a/src/renderer/src/stores/ui/messageIpc.ts b/src/renderer/src/stores/ui/messageIpc.ts index 7f77fc47e9..fe21edce04 100644 --- a/src/renderer/src/stores/ui/messageIpc.ts +++ b/src/renderer/src/stores/ui/messageIpc.ts @@ -3,13 +3,21 @@ import type { AssistantMessageBlock } from '@shared/types/agent-interface' interface BindMessageStoreIpcOptions { getActiveSessionId: () => string | null + getCurrentStreamIdentity: () => { + sessionId: string | null + requestId: string | null + } setStreamingState: (payload: { sessionId: string + requestId: string messageId?: string + updatedAt: number blocks: AssistantMessageBlock[] + metadata?: { providerId?: string; modelId?: string } }) => void clearStreamingState: () => void loadMessages: (sessionId: string) => void | Promise + invalidateRecentSessionView: (sessionId: string) => void applyStreamingBlocksToMessage?: ( messageId: string, sessionId: string, @@ -19,20 +27,114 @@ interface BindMessageStoreIpcOptions { isEphemeralStreamMessageId: (messageId: string) => boolean } +type StreamCursor = { + requestId: string + updatedAt: number +} + +const MAX_SETTLED_STREAMS = 128 + export function bindMessageStoreIpc(options: BindMessageStoreIpcOptions): () => void { const chatClient = createChatClient() - const reloadPersistedMessages = (sessionId: string) => { + const latestStreamBySession = new Map() + const settledStreams = new Set() + + const streamKey = (sessionId: string, requestId: string) => `${sessionId}\u0000${requestId}` + + const markStreamSettled = (sessionId: string, requestId: string): boolean => { + const key = streamKey(sessionId, requestId) + if (settledStreams.has(key)) return false + + settledStreams.add(key) + while (settledStreams.size > MAX_SETTLED_STREAMS) { + const oldestKey = settledStreams.keys().next().value + if (!oldestKey) break + settledStreams.delete(oldestKey) + } + return true + } + + const acceptStreamUpdate = (payload: { + sessionId: string + requestId: string + updatedAt: number + }): boolean => { + if (settledStreams.has(streamKey(payload.sessionId, payload.requestId))) { + return false + } + + const latest = latestStreamBySession.get(payload.sessionId) + if (latest) { + if (latest.requestId === payload.requestId && payload.updatedAt < latest.updatedAt) { + return false + } + if (latest.requestId !== payload.requestId && payload.updatedAt <= latest.updatedAt) { + return false + } + } + + latestStreamBySession.set(payload.sessionId, { + requestId: payload.requestId, + updatedAt: payload.updatedAt + }) + return true + } + + const reloadPersistedMessages = (sessionId: string, clearCurrentStream: boolean) => { // Streaming blocks were folded into the message record in place during // generation (applyStreamingBlocksToMessage), so the record already exists and // stays mounted. Clearing the stream flag first just stops the high-frequency // mutation; loadMessages then swaps the same id to its persisted copy. Same // node throughout — no blank, no remount. - options.clearStreamingState() + if (clearCurrentStream) { + options.clearStreamingState() + } void options.loadMessages(sessionId) } + const settleStream = (payload: { sessionId: string; requestId: string }) => { + // requestId is the turn identity; messageId may move from an ephemeral + // rate-limit row to the persisted assistant row within the same turn. + if (!markStreamSettled(payload.sessionId, payload.requestId)) { + return + } + + options.invalidateRecentSessionView(payload.sessionId) + + const latest = latestStreamBySession.get(payload.sessionId) + if (latest?.requestId === payload.requestId) { + latestStreamBySession.delete(payload.sessionId) + } else if (latest) { + return + } + + if (payload.sessionId !== options.getActiveSessionId()) { + return + } + + const currentStream = options.getCurrentStreamIdentity() + if ( + currentStream.sessionId === payload.sessionId && + currentStream.requestId && + currentStream.requestId !== payload.requestId + ) { + return + } + + reloadPersistedMessages( + payload.sessionId, + currentStream.sessionId === payload.sessionId && + (!currentStream.requestId || currentStream.requestId === payload.requestId) + ) + } + const cleanups = [ chatClient.onStreamUpdated((payload) => { + if (!acceptStreamUpdate(payload)) { + return + } + + options.invalidateRecentSessionView(payload.sessionId) const blocks = payload.blocks as AssistantMessageBlock[] if (payload.sessionId !== options.getActiveSessionId()) { return @@ -41,8 +143,14 @@ export function bindMessageStoreIpc(options: BindMessageStoreIpcOptions): () => const streamMessageId = payload.messageId ?? payload.requestId options.setStreamingState({ sessionId: payload.sessionId, + requestId: payload.requestId, messageId: streamMessageId, - blocks + updatedAt: payload.updatedAt, + blocks, + metadata: { + providerId: payload.providerId, + modelId: payload.modelId + } }) if ( @@ -57,18 +165,16 @@ export function bindMessageStoreIpc(options: BindMessageStoreIpcOptions): () => } }), chatClient.onStreamCompleted((payload) => { - if (payload.sessionId !== options.getActiveSessionId()) { - return - } - - reloadPersistedMessages(payload.sessionId) + settleStream({ + sessionId: payload.sessionId, + requestId: payload.requestId + }) }), chatClient.onStreamFailed((payload) => { - if (payload.sessionId !== options.getActiveSessionId()) { - return - } - - reloadPersistedMessages(payload.sessionId) + settleStream({ + sessionId: payload.sessionId, + requestId: payload.requestId + }) }) ] diff --git a/src/renderer/src/stores/ui/pendingInput.ts b/src/renderer/src/stores/ui/pendingInput.ts index c48287ca23..5b88eff0e5 100644 --- a/src/renderer/src/stores/ui/pendingInput.ts +++ b/src/renderer/src/stores/ui/pendingInput.ts @@ -12,6 +12,7 @@ export const usePendingInputStore = defineStore('pendingInput', () => { const items = ref([]) const loading = ref(false) const error = ref(null) + let latestLoadRequestId = 0 const steerItems = computed(() => items.value.filter((item) => item.mode === 'steer')) const queueItems = computed(() => @@ -24,36 +25,41 @@ export const usePendingInputStore = defineStore('pendingInput', () => { async function loadPendingInputs(sessionId: string): Promise { const requestedId = sessionId + const requestId = ++latestLoadRequestId currentSessionId.value = requestedId loading.value = true error.value = null try { const loadedItems = await sessionClient.listPendingInputs(requestedId) - if (requestedId !== currentSessionId.value) { + if (requestId !== latestLoadRequestId || requestedId !== currentSessionId.value) { return } items.value = loadedItems } catch (e) { - if (requestedId !== currentSessionId.value) { + if (requestId !== latestLoadRequestId || requestedId !== currentSessionId.value) { return } error.value = `Failed to load pending inputs: ${e}` } finally { - if (requestedId === currentSessionId.value) { + if (requestId === latestLoadRequestId && requestedId === currentSessionId.value) { loading.value = false } } } async function queueInput(sessionId: string, input: string | SendMessageInput): Promise { - error.value = null + if (currentSessionId.value === sessionId) { + error.value = null + } try { await sessionClient.queuePendingInput(sessionId, input) if (currentSessionId.value === sessionId) { await loadPendingInputs(sessionId) } } catch (e) { - error.value = `Failed to queue message: ${e}` + if (currentSessionId.value === sessionId) { + error.value = `Failed to queue message: ${e}` + } throw e } } @@ -63,55 +69,75 @@ export const usePendingInputStore = defineStore('pendingInput', () => { itemId: string, input: string | SendMessageInput ): Promise { - error.value = null + if (currentSessionId.value === sessionId) { + error.value = null + } try { - const updated = await sessionClient.updateQueuedInput(sessionId, itemId, input) - items.value = items.value.map((item) => (item.id === updated.id ? updated : item)) + await sessionClient.updateQueuedInput(sessionId, itemId, input) if (currentSessionId.value === sessionId) { await loadPendingInputs(sessionId) } } catch (e) { - error.value = `Failed to update queued message: ${e}` + if (currentSessionId.value === sessionId) { + error.value = `Failed to update queued message: ${e}` + } throw e } } async function moveQueueInput(sessionId: string, itemId: string, toIndex: number): Promise { - error.value = null + if (currentSessionId.value === sessionId) { + error.value = null + } try { - items.value = await sessionClient.moveQueuedInput(sessionId, itemId, toIndex) + await sessionClient.moveQueuedInput(sessionId, itemId, toIndex) + if (currentSessionId.value === sessionId) { + await loadPendingInputs(sessionId) + } } catch (e) { - error.value = `Failed to reorder queued message: ${e}` + if (currentSessionId.value === sessionId) { + error.value = `Failed to reorder queued message: ${e}` + } throw e } } async function steerPendingInput(sessionId: string, itemId: string): Promise { - error.value = null + if (currentSessionId.value === sessionId) { + error.value = null + } try { - const updated = await sessionClient.steerPendingInput(sessionId, itemId) - items.value = items.value.map((item) => (item.id === updated.id ? updated : item)) + await sessionClient.steerPendingInput(sessionId, itemId) if (currentSessionId.value === sessionId) { await loadPendingInputs(sessionId) } } catch (e) { - error.value = `Failed to steer queued message: ${e}` + if (currentSessionId.value === sessionId) { + error.value = `Failed to steer queued message: ${e}` + } throw e } } async function deleteInput(sessionId: string, itemId: string): Promise { - error.value = null + if (currentSessionId.value === sessionId) { + error.value = null + } try { await sessionClient.deletePendingInput(sessionId, itemId) - items.value = items.value.filter((item) => item.id !== itemId) + if (currentSessionId.value === sessionId) { + await loadPendingInputs(sessionId) + } } catch (e) { - error.value = `Failed to delete queued message: ${e}` + if (currentSessionId.value === sessionId) { + error.value = `Failed to delete queued message: ${e}` + } throw e } } function clear(): void { + latestLoadRequestId += 1 currentSessionId.value = null items.value = [] loading.value = false diff --git a/src/renderer/src/stores/ui/session.ts b/src/renderer/src/stores/ui/session.ts index 08d37fb43d..febaad0b7e 100644 --- a/src/renderer/src/stores/ui/session.ts +++ b/src/renderer/src/stores/ui/session.ts @@ -383,6 +383,7 @@ export const useSessionStore = defineStore('session', () => { sessions.value = sessions.value.filter((session) => !targetIds.has(session.id)) for (const sessionId of targetIds) { agentPlanStore.purge(sessionId) + messageStore.invalidateRecentSessionView(sessionId) } if (bootstrapActiveSession.value && targetIds.has(bootstrapActiveSession.value.id)) { @@ -462,6 +463,7 @@ export const useSessionStore = defineStore('session', () => { } const applySessionStatus = (sessionId: string, status: string): void => { + messageStore.invalidateRecentSessionView(sessionId) const nextStatus = mapSessionStatus(status) const index = sessions.value.findIndex((session) => session.id === sessionId) if (index >= 0 && sessions.value[index].status !== nextStatus) { @@ -500,19 +502,25 @@ export const useSessionStore = defineStore('session', () => { return } - activeSessionSummary.value = mapToUIActiveSessionSummary(session) const lightweightSession = mapToUISession(session) upsertSessions([lightweightSession]) - if (activeSessionId.value === session.id) { - bootstrapActiveSession.value = lightweightSession - syncSelectedAgentToSession(session.id) - } + if (activeSessionId.value !== session.id) return + + activeSessionSummary.value = mapToUIActiveSessionSummary(session) + bootstrapActiveSession.value = lightweightSession + syncSelectedAgentToSession(session.id) } - const hydrateActiveSessionSummary = async (sessionId: string): Promise => { + const hydrateActiveSessionSummary = async ( + sessionId: string, + activationRequestId: number + ): Promise => { try { const active = await sessionClient.getActive() - if (active.session?.id === sessionId) { + if ( + isCurrentActivationNavigation(activationRequestId, sessionId) && + active.session?.id === sessionId + ) { applyRestoredSession(active.session) } } catch (restoreError) { @@ -654,6 +662,10 @@ export const useSessionStore = defineStore('session', () => { return } + for (const sessionId of normalizedIds) { + messageStore.invalidateRecentSessionView(sessionId) + } + error.value = null try { const items = await sessionClient.getLightweightByIds(normalizedIds) @@ -713,7 +725,7 @@ export const useSessionStore = defineStore('session', () => { clearActiveSessionSummary() syncSelectedAgentToSession(sessionId) setActiveSessionId(sessionId) - await hydrateActiveSessionSummary(sessionId) + await hydrateActiveSessionSummary(sessionId, requestId) if (!isCurrentActivationNavigation(requestId, sessionId)) { return } @@ -1060,7 +1072,7 @@ export const useSessionStore = defineStore('session', () => { } syncSelectedAgentToSession(sessionId) setActiveSessionId(sessionId) - await hydrateActiveSessionSummary(sessionId) + await hydrateActiveSessionSummary(sessionId, requestId) if (!isCurrentActivationNavigation(requestId, sessionId)) { return } diff --git a/src/renderer/src/stores/ui/stream.ts b/src/renderer/src/stores/ui/stream.ts index 7bee0eac48..7f5b315a43 100644 --- a/src/renderer/src/stores/ui/stream.ts +++ b/src/renderer/src/stores/ui/stream.ts @@ -6,13 +6,26 @@ export const useStreamStateStore = defineStore('streamState', () => { const isStreaming = ref(false) const streamingBlocks = ref([]) const currentStreamSessionId = ref(null) + const currentStreamRequestId = ref(null) const currentStreamMessageId = ref(null) + const currentStreamUpdatedAt = ref(0) + const currentStreamMetadata = ref<{ providerId?: string; modelId?: string } | null>(null) const streamRevision = ref(0) - function setStream(sessionId: string, blocks: AssistantMessageBlock[], messageId?: string): void { + function setStream( + sessionId: string, + blocks: AssistantMessageBlock[], + messageId?: string, + metadata?: { providerId?: string; modelId?: string }, + requestId?: string, + updatedAt?: number + ): void { isStreaming.value = true currentStreamSessionId.value = sessionId + currentStreamRequestId.value = requestId ?? messageId ?? null currentStreamMessageId.value = messageId ?? null + currentStreamUpdatedAt.value = updatedAt ?? 0 + currentStreamMetadata.value = metadata ?? null streamingBlocks.value = blocks streamRevision.value += 1 } @@ -21,7 +34,10 @@ export const useStreamStateStore = defineStore('streamState', () => { isStreaming.value = false streamingBlocks.value = [] currentStreamSessionId.value = null + currentStreamRequestId.value = null currentStreamMessageId.value = null + currentStreamUpdatedAt.value = 0 + currentStreamMetadata.value = null streamRevision.value += 1 } @@ -29,7 +45,10 @@ export const useStreamStateStore = defineStore('streamState', () => { isStreaming, streamingBlocks, currentStreamSessionId, + currentStreamRequestId, currentStreamMessageId, + currentStreamUpdatedAt, + currentStreamMetadata, streamRevision, setStream, clearStreamingState diff --git a/test/renderer/components/ChatPage.test.ts b/test/renderer/components/ChatPage.test.ts index 1a198aa9fa..4c3ef2fec2 100644 --- a/test/renderer/components/ChatPage.test.ts +++ b/test/renderer/components/ChatPage.test.ts @@ -49,6 +49,9 @@ type SetupOptions = { isStreaming?: boolean streamingBlocks?: unknown[] currentStreamMessageId?: string | null + currentStreamSessionId?: string | null + currentSessionId?: string | null + committedSessionId?: string | null pendingInputStorePatch?: Record sessionKind?: 'regular' | 'subagent' activeSessionPatch?: Record @@ -105,7 +108,21 @@ const setup = async (options: SetupOptions = {}) => { currentStreamMessageId: options.currentStreamMessageId ?? null, streamRevision: 0, lastPersistedRevision: 0, - committedSessionId: 's1', + currentSessionId: options.currentSessionId === undefined ? 's1' : options.currentSessionId, + committedSessionId: + options.committedSessionId === undefined ? 's1' : options.committedSessionId, + committedSession: + options.committedSessionId === null + ? null + : { + id: options.committedSessionId ?? 's1' + }, + currentStreamSessionId: + options.currentStreamSessionId === undefined + ? options.isStreaming + ? 's1' + : null + : options.currentStreamSessionId, hasMoreHistory: false, isLoadingHistory: false, messageIds: ( @@ -140,13 +157,17 @@ const setup = async (options: SetupOptions = {}) => { loadMessages: vi.fn(), loadOlderMessages: vi.fn().mockResolvedValue(0), activateRecentSessionView: vi.fn().mockReturnValue(false), + invalidateRecentSessionView: vi.fn(), clear: vi.fn(), clearStreamingState: vi.fn(), + clearStreamingStateForOtherSession: vi.fn(), addOptimisticUserMessage: vi.fn().mockReturnValue('__optimistic_user_1'), removeOptimisticMessage: vi.fn() }) messageStore.loadMessages.mockImplementation(async (sessionId: string) => { + messageStore.currentSessionId = sessionId messageStore.committedSessionId = sessionId + messageStore.committedSession = { id: sessionId } return { id: sessionId } }) @@ -950,7 +971,8 @@ describe('ChatPage', () => { }) expect(messageStore.clear).not.toHaveBeenCalled() - expect(messageStore.clearStreamingState).toHaveBeenCalledTimes(1) + expect(messageStore.clearStreamingState).not.toHaveBeenCalled() + expect(messageStore.clearStreamingStateForOtherSession).toHaveBeenCalledWith('s1') expect(pendingInputStore.clear).toHaveBeenCalledTimes(1) expect(messageStore.loadMessages).not.toHaveBeenCalled() expect(pendingInputStore.loadPendingInputs).not.toHaveBeenCalled() @@ -961,12 +983,52 @@ describe('ChatPage', () => { expect(pendingInputStore.loadPendingInputs).toHaveBeenCalledWith('s1') }) + it('releases a cold-session composer when another load commits the selected view', async () => { + const pageRestore = createDeferred() + const { wrapper, messageStore, chatClient, flushStartupDeferredTasks } = await setup({ + messages: [], + currentSessionId: 's1', + committedSessionId: null, + deferStartupTasks: true + }) + messageStore.loadMessages.mockReturnValueOnce(pageRestore.promise) + const input = wrapper.findComponent({ name: 'ChatInputBox' }) + + input.vm.$emit('update:modelValue', 'first turn') + await flushPromises() + expect(wrapper.find('[data-testid="chat-session-loading-overlay"]').exists()).toBe(true) + expect(input.props('submitDisabled')).toBe(true) + + const drainingStartupTasks = flushStartupDeferredTasks() + await flushPromises() + expect(messageStore.loadMessages).toHaveBeenCalledWith('s1', 100) + + messageStore.committedSessionId = 's1' + messageStore.committedSession = { id: 's1' } + await flushPromises() + + expect(wrapper.find('[data-testid="chat-session-loading-overlay"]').exists()).toBe(false) + expect(input.props('submitDisabled')).toBe(false) + input.vm.$emit('submit') + await flushPromises() + expect(chatClient.sendMessage).toHaveBeenCalledWith('s1', { + text: 'first turn', + files: [] + }) + + pageRestore.resolve(null) + await drainingStartupTasks + wrapper.unmount() + }) + it('hides the previous session behind a fixed viewport overlay until atomic commit', async () => { const targetLoad = createDeferred() const { wrapper, messageStore } = await setup() messageStore.loadMessages.mockReturnValueOnce( targetLoad.promise.then(() => { + messageStore.currentSessionId = 's2' messageStore.committedSessionId = 's2' + messageStore.committedSession = { id: 's2' } return { id: 's2' } }) ) @@ -1025,7 +1087,9 @@ describe('ChatPage', () => { messageStore.messages = [cachedMessage] messageStore.messageIds = [cachedMessage.id] messageStore.messageCache = new Map([[cachedMessage.id, cachedMessage]]) + messageStore.currentSessionId = 's2' messageStore.committedSessionId = 's2' + messageStore.committedSession = { id: 's2' } return true }) messageStore.loadMessages.mockReturnValueOnce(refresh.promise) @@ -1087,7 +1151,9 @@ describe('ChatPage', () => { messageStore.loadMessages.mockReturnValueOnce( deferredSessionLoad.promise.then((session) => { + messageStore.currentSessionId = 's2' messageStore.committedSessionId = 's2' + messageStore.committedSession = { id: 's2' } return session }) ) @@ -1429,7 +1495,9 @@ describe('ChatPage', () => { } messageStore.loadMessages.mockImplementation(async (sessionId: 's1' | 's2') => { messageStore.messages = messagesBySession[sessionId] + messageStore.currentSessionId = sessionId messageStore.committedSessionId = sessionId + messageStore.committedSession = { id: sessionId } return { id: sessionId } }) @@ -1672,6 +1740,20 @@ describe('ChatPage', () => { expect(chatInputClearPendingSkills).toHaveBeenCalled() }) + it('does not submit after another navigation takes message-store ownership', async () => { + const { wrapper, messageStore, chatClient } = await setup() + const input = wrapper.findComponent({ name: 'ChatInputBox' }) + + input.vm.$emit('update:modelValue', 'stale submit') + await flushPromises() + input.vm.$emit('submit') + messageStore.currentSessionId = 's2' + await flushPromises() + + expect(messageStore.addOptimisticUserMessage).not.toHaveBeenCalled() + expect(chatClient.sendMessage).not.toHaveBeenCalled() + }) + it('shows a pending assistant row immediately after submitting before stream starts', async () => { const deferredSend = createDeferred<{ accepted: true; requestId: null; messageId: null }>() const { wrapper, chatClient, messageStore } = await setup() @@ -1692,6 +1774,7 @@ describe('ChatPage', () => { expect(messages.some((message) => message.id.startsWith('__pending_assistant_'))).toBe(true) messageStore.isStreaming = true + messageStore.currentStreamSessionId = 's1' await flushPromises() const streamingMessages = messageList.props('messages') as Array<{ id: string; role: string }> @@ -1937,7 +2020,7 @@ describe('ChatPage', () => { const messageList = wrapper.findComponent({ name: 'MessageList' }) const messages = messageList.props('messages') as Array<{ id: string }> expect(messages.some((message) => message.id.startsWith('__pending_assistant_'))).toBe(false) - expect(messageStore.removeOptimisticMessage).toHaveBeenCalledWith('__optimistic_user_1') + expect(messageStore.removeOptimisticMessage).toHaveBeenCalledWith('__optimistic_user_1', 's1') } finally { errorSpy.mockRestore() } @@ -1956,7 +2039,7 @@ describe('ChatPage', () => { const messageList = wrapper.findComponent({ name: 'MessageList' }) const messages = messageList.props('messages') as Array<{ id: string }> expect(messages.some((message) => message.id.startsWith('__pending_assistant_'))).toBe(false) - expect(messageStore.removeOptimisticMessage).toHaveBeenCalledWith('__optimistic_user_1') + expect(messageStore.removeOptimisticMessage).toHaveBeenCalledWith('__optimistic_user_1', 's1') } finally { errorSpy.mockRestore() } @@ -2555,6 +2638,39 @@ describe('ChatPage', () => { expect(chatClient.sendMessage).not.toHaveBeenCalled() }) + it('does not clear a new draft when an old A-B-A queue request resolves', async () => { + const queued = createDeferred() + const { wrapper, pendingInputStore } = await setup({ + isStreaming: true, + pendingInputStorePatch: { + queueInput: vi.fn().mockReturnValue(queued.promise) + } + }) + const input = wrapper.findComponent({ name: 'ChatInputBox' }) + + input.vm.$emit('update:modelValue', 'old draft') + await flushPromises() + input.vm.$emit('submit') + await flushPromises() + expect(pendingInputStore.queueInput).toHaveBeenCalledWith('s1', { + text: 'old draft', + files: [] + }) + + await wrapper.setProps({ sessionId: 's2' }) + await flushPromises() + await wrapper.setProps({ sessionId: 's1' }) + await flushPromises() + input.vm.$emit('update:modelValue', 'new draft') + await flushPromises() + expect(wrapper.findComponent({ name: 'ChatInputToolbar' }).props('hasInput')).toBe(true) + + queued.resolve() + await flushPromises() + + expect(wrapper.findComponent({ name: 'ChatInputToolbar' }).props('hasInput')).toBe(true) + }) + it('disables queue submit when the waiting queue is full but keeps steer button available', async () => { const { wrapper } = await setup({ isStreaming: true, diff --git a/test/renderer/stores/messageStore.reactivity.test.ts b/test/renderer/stores/messageStore.reactivity.test.ts index e3a579adf5..f586a039fa 100644 --- a/test/renderer/stores/messageStore.reactivity.test.ts +++ b/test/renderer/stores/messageStore.reactivity.test.ts @@ -56,6 +56,7 @@ describe('messageStore reactivity', () => { const { useMessageStore } = await import('@/stores/ui/message') const store = useMessageStore() + store.setCurrentSessionId('s1') await store.loadMessages('s1') const responseHandler = streamListeners.updated[0] diff --git a/test/renderer/stores/messageStore.test.ts b/test/renderer/stores/messageStore.test.ts index d0f479112d..6c755efb20 100644 --- a/test/renderer/stores/messageStore.test.ts +++ b/test/renderer/stores/messageStore.test.ts @@ -95,6 +95,7 @@ const setupStore = async () => { } const { useMessageStore } = await import('@/stores/ui/message') const store = useMessageStore() + store.setCurrentSessionId('s1') return { store, sessionClient, streamListeners, ipcListeners } } @@ -125,6 +126,12 @@ describe('messageStore', () => { expect(store.isStreaming.value).toBe(true) expect(store.currentStreamMessageId.value).toBe('m1') + expect(store.committedSessionId.value).toBeNull() + expect(store.messages.value).toHaveLength(0) + + await store.loadMessages('s1') + + expect(store.committedSessionId.value).toBe('s1') expect(store.messages.value).toHaveLength(1) expect(store.messages.value[0]?.id).toBe('m1') expect(store.messages.value[0]?.metadata).toBe('{"provider":"acp","model":"dimcode"}') @@ -162,6 +169,7 @@ describe('messageStore', () => { it('can remove optimistic messages by id', async () => { const { store } = await setupStore() + await store.loadMessages('s1') const optimisticId = store.addOptimisticUserMessage('s1', { text: 'hello', @@ -272,6 +280,252 @@ describe('messageStore', () => { expect(store.messages.value[0]?.id).toBe('m2') }) + it('does not let a refresh overwrite an optimistic message added after request start', async () => { + const { store, sessionClient } = await setupStore() + const persisted = buildUserMessage('m1', 's1', 1, 'persisted') + sessionClient.restore.mockResolvedValueOnce({ + session: { id: 's1' }, + nextCursor: null, + hasMore: false, + messages: [persisted] + }) + await store.loadMessages('s1') + + const refresh = createDeferred[]>() + sessionClient.restore.mockReturnValueOnce( + refresh.promise.then((messages) => ({ + session: { id: 's1' }, + nextCursor: null, + hasMore: false, + messages + })) + ) + const pendingRefresh = store.loadMessages('s1') + const optimisticId = store.addOptimisticUserMessage('s1', 'optimistic') + + refresh.resolve([persisted]) + await pendingRefresh + + expect(optimisticId).not.toBeNull() + expect(store.messageIds.value).toEqual(['m1', optimisticId]) + }) + + it('does not let a refresh overwrite streaming records added after request start', async () => { + const { store, sessionClient, streamListeners } = await setupStore() + const persisted = buildUserMessage('m1', 's1', 1, 'persisted') + sessionClient.restore.mockResolvedValueOnce({ + session: { id: 's1' }, + nextCursor: null, + hasMore: false, + messages: [persisted] + }) + await store.loadMessages('s1') + + const refresh = createDeferred[]>() + sessionClient.restore.mockReturnValueOnce( + refresh.promise.then((messages) => ({ + session: { id: 's1' }, + nextCursor: null, + hasMore: false, + messages + })) + ) + const pendingRefresh = store.loadMessages('s1') + streamListeners.updated[0]({ + sessionId: 's1', + requestId: 'assistant-stream', + messageId: 'assistant-stream', + updatedAt: 2, + blocks: [{ type: 'content', content: 'live', status: 'pending', timestamp: 2 }] + }) + + refresh.resolve([persisted]) + await pendingRefresh + + expect(store.messageIds.value).toEqual(['m1', 'assistant-stream']) + expect(store.messageCache.value.get('assistant-stream')?.content).toContain('live') + }) + + it('does not cancel an in-flight load when recent-session cache lookup misses', async () => { + const { store, sessionClient } = await setupStore() + const deferredLoad = createDeferred[]>() + sessionClient.restore.mockReturnValueOnce( + deferredLoad.promise.then((messages) => ({ + session: { id: 's1' }, + nextCursor: null, + hasMore: false, + messages + })) + ) + + const pendingLoad = store.loadMessages('s1') + expect(store.activateRecentSessionView('s1')).toBe(false) + deferredLoad.resolve([buildUserMessage('m1', 's1', 1, 'loaded')]) + await pendingLoad + + expect(store.committedSessionId.value).toBe('s1') + expect(store.messageIds.value).toEqual(['m1']) + }) + + it('rejects a cached view after a known session invalidation', async () => { + const { store, sessionClient } = await setupStore() + sessionClient.restore + .mockResolvedValueOnce({ + session: { id: 's1' }, + nextCursor: null, + hasMore: false, + messages: [buildUserMessage('s1-message', 's1', 1, 'session one')] + }) + .mockResolvedValueOnce({ + session: { id: 's2' }, + nextCursor: null, + hasMore: false, + messages: [buildUserMessage('s2-message', 's2', 1, 'session two')] + }) + + await store.loadMessages('s1') + store.setCurrentSessionId('s2') + await store.loadMessages('s2') + store.invalidateRecentSessionView('s1') + store.setCurrentSessionId('s1') + + expect(store.activateRecentSessionView('s1')).toBe(false) + expect(store.committedSessionId.value).toBe('s2') + }) + + it('invalidates a cached view when its inactive session receives a stream update', async () => { + const { store, sessionClient, streamListeners } = await setupStore() + sessionClient.restore + .mockResolvedValueOnce({ + session: { id: 's1' }, + nextCursor: null, + hasMore: false, + messages: [buildUserMessage('s1-message', 's1', 1, 'session one')] + }) + .mockResolvedValueOnce({ + session: { id: 's2' }, + nextCursor: null, + hasMore: false, + messages: [buildUserMessage('s2-message', 's2', 1, 'session two')] + }) + + await store.loadMessages('s1') + store.setCurrentSessionId('s2') + await store.loadMessages('s2') + streamListeners.updated[0]({ + kind: 'snapshot', + sessionId: 's1', + requestId: 'background-request', + messageId: 'background-message', + updatedAt: 2, + blocks: [{ type: 'content', content: 'background', status: 'pending', timestamp: 2 }] + }) + store.setCurrentSessionId('s1') + + expect(store.activateRecentSessionView('s1')).toBe(false) + expect(store.committedSessionId.value).toBe('s2') + }) + + it('does not recache an active view invalidated before a session switch', async () => { + const { store, sessionClient } = await setupStore() + sessionClient.restore + .mockResolvedValueOnce({ + session: { id: 's1' }, + nextCursor: null, + hasMore: false, + messages: [buildUserMessage('s1-message', 's1', 1, 'session one')] + }) + .mockResolvedValueOnce({ + session: { id: 's2' }, + nextCursor: null, + hasMore: false, + messages: [buildUserMessage('s2-message', 's2', 1, 'session two')] + }) + + await store.loadMessages('s1') + store.invalidateRecentSessionView('s1') + store.setCurrentSessionId('s2') + await store.loadMessages('s2') + store.setCurrentSessionId('s1') + + expect(store.activateRecentSessionView('s1')).toBe(false) + }) + + it('keeps a view uncacheable when invalidation crosses its persisted refresh', async () => { + const { store, sessionClient } = await setupStore() + const sessionOne = buildUserMessage('s1-message', 's1', 1, 'session one') + sessionClient.restore.mockResolvedValueOnce({ + session: { id: 's1' }, + nextCursor: null, + hasMore: false, + messages: [sessionOne] + }) + await store.loadMessages('s1') + + store.invalidateRecentSessionView('s1') + const refresh = createDeferred[]>() + sessionClient.restore.mockReturnValueOnce( + refresh.promise.then((messages) => ({ + session: { id: 's1' }, + nextCursor: null, + hasMore: false, + messages + })) + ) + const pendingRefresh = store.loadMessages('s1') + store.invalidateRecentSessionView('s1') + refresh.resolve([sessionOne]) + await pendingRefresh + + sessionClient.restore.mockResolvedValueOnce({ + session: { id: 's2' }, + nextCursor: null, + hasMore: false, + messages: [buildUserMessage('s2-message', 's2', 1, 'session two')] + }) + store.setCurrentSessionId('s2') + await store.loadMessages('s2') + store.setCurrentSessionId('s1') + + expect(store.activateRecentSessionView('s1')).toBe(false) + }) + + it('does not let a late load select a session owned by another navigation', async () => { + const { store, sessionClient } = await setupStore() + await store.loadMessages('s1') + store.setCurrentSessionId('s2') + + const restoreCallsBeforeLateLoad = sessionClient.restore.mock.calls.length + expect(await store.loadMessages('s1')).toBeNull() + + expect(store.currentSessionId.value).toBe('s2') + expect(store.committedSessionId.value).toBe('s1') + expect(sessionClient.restore).toHaveBeenCalledTimes(restoreCallsBeforeLateLoad) + }) + + it('rejects an old session load after a rapid A-B-A selection cycle', async () => { + const { store, sessionClient } = await setupStore() + const staleLoad = createDeferred[]>() + sessionClient.restore.mockReturnValueOnce( + staleLoad.promise.then((messages) => ({ + session: { id: 's1' }, + nextCursor: null, + hasMore: false, + messages + })) + ) + + const pendingStaleLoad = store.loadMessages('s1') + store.setCurrentSessionId('s2') + store.setCurrentSessionId('s1') + staleLoad.resolve([buildUserMessage('stale', 's1', 1, 'stale')]) + await pendingStaleLoad + + expect(store.currentSessionId.value).toBe('s1') + expect(store.committedSessionId.value).toBeNull() + expect(store.messageIds.value).toEqual([]) + }) + it('keeps the committed session view intact until an uncached target is ready', async () => { const { store, sessionClient } = await setupStore() const secondLoad = createDeferred[]>() @@ -292,6 +546,7 @@ describe('messageStore', () => { ) await store.loadMessages('s1') + store.setCurrentSessionId('s2') const pendingSwitch = store.loadMessages('s2') expect(store.currentSessionId.value).toBe('s2') @@ -323,8 +578,10 @@ describe('messageStore', () => { }) await store.loadMessages('s1') + store.setCurrentSessionId('s2') await store.loadMessages('s2') + store.setCurrentSessionId('s1') expect(store.activateRecentSessionView('s1')).toBe(true) expect(store.currentSessionId.value).toBe('s1') expect(store.messages.value.map((message) => message.id)).toEqual(['s1-message']) @@ -472,6 +729,7 @@ describe('messageStore', () => { void store.loadOlderMessages() store.clear() + store.setCurrentSessionId('s2') await store.loadMessages('s2') olderPage.resolve({ @@ -487,6 +745,32 @@ describe('messageStore', () => { expect(store.isLoadingHistory.value).toBe(false) }) + it('does not let an overlapping history page replace an existing record', async () => { + const { store, sessionClient } = await setupStore() + const current = buildUserMessage('m2', 's1', 2, 'current') + sessionClient.restore.mockResolvedValueOnce({ + session: { id: 's1' }, + messages: [current], + nextCursor: { orderSeq: 2, id: 'm2' }, + hasMore: true + }) + sessionClient.listMessagesPage.mockResolvedValueOnce({ + messages: [ + buildUserMessage('m1', 's1', 1, 'older'), + buildUserMessage('m2', 's1', 2, 'stale duplicate') + ], + nextCursor: null, + hasMore: false + }) + + await store.loadMessages('s1', 1) + const loadedCount = await store.loadOlderMessages() + + expect(loadedCount).toBe(1) + expect(store.messageIds.value).toEqual(['m1', 'm2']) + expect(store.messageCache.value.get('m2')?.content).toBe(current.content) + }) + it('keeps rate-limit stream messages ephemeral and skips message hydration', async () => { const { store, streamListeners } = await setupStore() await store.loadMessages('s1') @@ -611,6 +895,121 @@ describe('messageStore', () => { expect(store.messages.value[0]?.id).toBe('user-1') }) + it('does not let an old terminal event settle a newer stream in the same session', async () => { + const { store, sessionClient, streamListeners } = await setupStore() + await store.loadMessages('s1') + + streamListeners.updated[0]({ + kind: 'snapshot', + sessionId: 's1', + requestId: 'request-old', + messageId: 'message-old', + updatedAt: 1, + blocks: [{ type: 'content', content: 'old', status: 'pending', timestamp: 1 }] + }) + streamListeners.updated[0]({ + kind: 'snapshot', + sessionId: 's1', + requestId: 'request-new', + messageId: 'message-new', + updatedAt: 2, + blocks: [{ type: 'content', content: 'new', status: 'pending', timestamp: 2 }] + }) + + streamListeners.completed[0]({ + sessionId: 's1', + requestId: 'request-old', + messageId: 'message-old', + completedAt: 3 + }) + await new Promise((resolve) => setTimeout(resolve, 0)) + + expect(sessionClient.restore).toHaveBeenCalledTimes(1) + expect(store.isStreaming.value).toBe(true) + expect(store.currentStreamRequestId.value).toBe('request-new') + expect(store.currentStreamMessageId.value).toBe('message-new') + expect(store.messageCache.value.get('message-new')?.content).toContain('new') + }) + + it('ignores stream snapshots that arrive after their terminal event', async () => { + const { store, sessionClient, streamListeners } = await setupStore() + sessionClient.restore + .mockResolvedValueOnce({ + session: { id: 's1' }, + messages: [], + nextCursor: null, + hasMore: false + }) + .mockResolvedValueOnce({ + session: { id: 's1' }, + messages: [buildUserMessage('persisted', 's1', 1, 'persisted')], + nextCursor: null, + hasMore: false + }) + await store.loadMessages('s1') + + streamListeners.updated[0]({ + kind: 'snapshot', + sessionId: 's1', + requestId: 'request-1', + messageId: 'message-1', + updatedAt: 1, + blocks: [{ type: 'content', content: 'live', status: 'pending', timestamp: 1 }] + }) + streamListeners.completed[0]({ + sessionId: 's1', + requestId: 'request-1', + messageId: 'message-1', + completedAt: 2 + }) + streamListeners.updated[0]({ + kind: 'snapshot', + sessionId: 's1', + requestId: 'request-1', + messageId: 'message-1', + updatedAt: 3, + blocks: [{ type: 'content', content: 'late', status: 'pending', timestamp: 3 }] + }) + streamListeners.failed[0]({ + sessionId: 's1', + requestId: 'request-1', + messageId: 'message-1', + failedAt: 4, + error: 'duplicate terminal' + }) + await new Promise((resolve) => setTimeout(resolve, 0)) + + expect(sessionClient.restore).toHaveBeenCalledTimes(2) + expect(store.isStreaming.value).toBe(false) + expect(store.currentStreamRequestId.value).toBeNull() + expect(store.messageIds.value).toEqual(['persisted']) + }) + + it('rejects an older snapshot for the current stream request', async () => { + const { store, streamListeners } = await setupStore() + await store.loadMessages('s1') + + streamListeners.updated[0]({ + kind: 'snapshot', + sessionId: 's1', + requestId: 'request-1', + messageId: 'message-1', + updatedAt: 2, + blocks: [{ type: 'content', content: 'newer', status: 'pending', timestamp: 2 }] + }) + streamListeners.updated[0]({ + kind: 'snapshot', + sessionId: 's1', + requestId: 'request-1', + messageId: 'message-1', + updatedAt: 1, + blocks: [{ type: 'content', content: 'older', status: 'pending', timestamp: 1 }] + }) + + expect(store.streamingBlocks.value[0]).toMatchObject({ content: 'newer' }) + expect(store.messageCache.value.get('message-1')?.content).toContain('newer') + }) + it('reloads persisted messages when a typed stream failure arrives', async () => { const { store, sessionClient, streamListeners, ipcListeners } = await setupStore() sessionClient.restore diff --git a/test/renderer/stores/pendingInputStore.test.ts b/test/renderer/stores/pendingInputStore.test.ts index 2230e98579..3068fd0f85 100644 --- a/test/renderer/stores/pendingInputStore.test.ts +++ b/test/renderer/stores/pendingInputStore.test.ts @@ -86,6 +86,49 @@ describe('pendingInput store', () => { expect(store.error).toBeNull() }) + it('rejects an old load after a rapid A-B-A session cycle', async () => { + const { store, sessionClient } = await setupStore() + const firstA = createDeferred[]>() + const sessionB = createDeferred[]>() + const secondA = createDeferred[]>() + sessionClient.listPendingInputs + .mockReturnValueOnce(firstA.promise) + .mockReturnValueOnce(sessionB.promise) + .mockReturnValueOnce(secondA.promise) + + const firstAPromise = store.loadPendingInputs('s1') + const sessionBPromise = store.loadPendingInputs('s2') + const secondAPromise = store.loadPendingInputs('s1') + + secondA.resolve([createPendingItem('a-latest', 's1')]) + await secondAPromise + firstA.resolve([createPendingItem('a-stale', 's1')]) + sessionB.resolve([createPendingItem('b-stale', 's2')]) + await Promise.all([firstAPromise, sessionBPromise]) + + expect(store.currentSessionId).toBe('s1') + expect(store.items).toEqual([createPendingItem('a-latest', 's1')]) + expect(store.loading).toBe(false) + }) + + it('does not apply a completed reorder to another session view', async () => { + const { store, sessionClient } = await setupStore() + const reorderedSessionOne = createDeferred[]>() + sessionClient.listPendingInputs + .mockResolvedValueOnce([createPendingItem('a-1', 's1')]) + .mockResolvedValueOnce([createPendingItem('b-1', 's2')]) + sessionClient.moveQueuedInput.mockReturnValueOnce(reorderedSessionOne.promise) + + await store.loadPendingInputs('s1') + const pendingReorder = store.moveQueueInput('s1', 'a-1', 0) + await store.loadPendingInputs('s2') + reorderedSessionOne.resolve([createPendingItem('a-reordered', 's1')]) + await pendingReorder + + expect(store.currentSessionId).toBe('s2') + expect(store.items).toEqual([createPendingItem('b-1', 's2')]) + }) + it('preserves clear state when an in-flight load later fails', async () => { const { store, sessionClient } = await setupStore() const load = createDeferred[]>() diff --git a/test/renderer/stores/sessionStore.test.ts b/test/renderer/stores/sessionStore.test.ts index 745d0a33de..413b420143 100644 --- a/test/renderer/stores/sessionStore.test.ts +++ b/test/renderer/stores/sessionStore.test.ts @@ -30,6 +30,14 @@ type SetupStoreOptions = { const SIDEBAR_GROUP_MODE_KEY = 'sidebar_group_mode' +function createDeferred() { + let resolve!: (value: T) => void + const promise = new Promise((innerResolve) => { + resolve = innerResolve + }) + return { promise, resolve } +} + afterEach(() => { window.sessionStorage.removeItem(GUIDED_ONBOARDING_RESUME_STORAGE_KEY) }) @@ -295,9 +303,11 @@ const setupStore = async (options: SetupStoreOptions = {}) => { })) const clearStreamingState = vi.fn() const setCurrentSessionId = vi.fn() + const invalidateRecentSessionView = vi.fn() vi.doMock('@/stores/ui/message', () => ({ useMessageStore: () => ({ clearStreamingState, + invalidateRecentSessionView, loadMessages: vi.fn(), setCurrentSessionId }) @@ -334,6 +344,7 @@ const setupStore = async (options: SetupStoreOptions = {}) => { settings, configClient, clearStreamingState, + invalidateRecentSessionView, setCurrentSessionId, sessionClient, chatClient, @@ -1211,8 +1222,39 @@ describe('sessionStore streaming cleanup', () => { expect(pageRouter.goToChat).not.toHaveBeenCalledWith('session-a') }) + it('rejects stale session hydration after an A-B-A activation cycle', async () => { + const { store, sessionClient } = await setupStore() + store.sessions.value = [ + createSession({ id: 'session-a', title: 'Session A' }), + createSession({ id: 'session-b', title: 'Session B' }) + ] + const staleSessionA = createDeferred<{ session: ReturnType }>() + sessionClient.getActive + .mockReturnValueOnce(staleSessionA.promise) + .mockResolvedValueOnce({ + session: createSession({ id: 'session-b', title: 'Session B hydrated' }) + }) + .mockResolvedValueOnce({ + session: createSession({ id: 'session-a', title: 'Session A latest' }) + }) + + const firstSelection = store.selectSession('session-a') + await Promise.resolve() + await store.selectSession('session-b') + await store.selectSession('session-a') + + expect(store.activeSession.value?.title).toBe('Session A latest') + + staleSessionA.resolve({ + session: createSession({ id: 'session-a', title: 'Session A stale' }) + }) + await firstSelection + + expect(store.activeSession.value?.title).toBe('Session A latest') + }) + it('updates the local session status immediately from the session status event', async () => { - const { store, emitSessionStatusChange } = await setupStore() + const { store, emitSessionStatusChange, invalidateRecentSessionView } = await setupStore() store.sessions.value = [createSession({ id: 'session-status', status: 'none' })] store.activeSessionId.value = 'session-status' @@ -1222,6 +1264,7 @@ describe('sessionStore streaming cleanup', () => { }) expect(store.activeSession.value?.status).toBe('working') + expect(invalidateRecentSessionView).toHaveBeenCalledWith('session-status') emitSessionStatusChange({ sessionId: 'session-status', From 3a11f01cd112b7fba60a71b867544cac0c341893 Mon Sep 17 00:00:00 2001 From: zhangmo8 Date: Wed, 15 Jul 2026 23:23:04 +0800 Subject: [PATCH 2/2] fix(chat): address race review findings --- docs/issues/chat-session-view-races/spec.md | 10 ++ src/renderer/src/pages/ChatPage.vue | 2 +- src/renderer/src/stores/ui/message.ts | 13 ++- src/renderer/src/stores/ui/messageIpc.ts | 115 +++++++++++++------- src/renderer/src/stores/ui/pendingInput.ts | 94 ++++++---------- src/renderer/src/stores/ui/session.ts | 1 + test/renderer/components/ChatPage.test.ts | 22 ++++ test/renderer/stores/messageStore.test.ts | 79 ++++++++++++++ test/renderer/stores/sessionStore.test.ts | 18 +++ 9 files changed, 253 insertions(+), 101 deletions(-) diff --git a/docs/issues/chat-session-view-races/spec.md b/docs/issues/chat-session-view-races/spec.md index 1e18af18bd..fccd5e59d1 100644 --- a/docs/issues/chat-session-view-races/spec.md +++ b/docs/issues/chat-session-view-races/spec.md @@ -44,6 +44,9 @@ presentation problems. without checking which view owns them. 8. History pagination writes records into the cache before it removes duplicate IDs, allowing an overlapping old page to replace a newer in-memory record. +9. Stream request ordering retains only the current request and a bounded settled-key set. A known + superseded request can reclaim the stream with a later timestamp, and an evicted settled request + can be resurrected by a late snapshot. ## Required Invariants @@ -62,6 +65,8 @@ presentation problems. different committed view. - Stream updates are monotonic per request, a request settles once, and a terminal event cannot settle a different request in the same session. +- Superseded and settled stream request identities remain tombstoned for the tracked session + lifetime and are purged only when that session is permanently removed or the binding is disposed. - Selection and async submit guards use request generations, not session equality alone, so A-B-A cannot revive an earlier operation. - Pending-input mutations refresh only their matching active view, and overlapping history pages @@ -83,6 +88,8 @@ presentation problems. - Add defensive session ownership checks around optimistic mutations and post-await submit paths. - Fence stream snapshots and terminals by request identity and monotonic event time, including duplicate terminal suppression and inactive-session cache invalidation. +- Assign stream requests a per-session local generation so later updates from known older requests + cannot reclaim ownership, and purge generations when sessions are deleted. - Fence pending-input operations, session hydration, history pagination, and submit continuations against stale or ABA ownership. @@ -93,6 +100,7 @@ presentation problems. - [x] Make recent-session cache activation and invalidation race-safe. - [x] Update ChatPage submit, stream, and restore integration. - [x] Fence stream terminals, pending-input mutations, history overlap, and session hydration. +- [x] Preserve superseded and settled stream tombstones until permanent session removal. - [x] Add cold-start, first-turn completion, cached-refresh, cache-miss, and rapid-switch regressions. - [x] Update the retained chat scroll ownership contract. - [x] Run focused and full repository quality gates. @@ -101,6 +109,8 @@ presentation problems. - Renderer store and ChatPage Vitest suites cover deterministic deferred-promise interleavings, including A-B-A selection, submit, pending-input, and hydration cycles. +- Review hardening adds null compaction restore, superseded request reclamation, durable settled + tombstone, and permanent-session purge regressions; the four focused suites pass 171/171 tests. - Full `pnpm test` completed with 5,516 passing and 198 skipped tests. Its only three failures were Feishu loopback callback tests blocked from listening on `127.0.0.1` by the sandbox; the complete callback suite passed 22/22 when rerun with local-listen permission. diff --git a/src/renderer/src/pages/ChatPage.vue b/src/renderer/src/pages/ChatPage.vue index edc7d82852..216d90b4e9 100644 --- a/src/renderer/src/pages/ChatPage.vue +++ b/src/renderer/src/pages/ChatPage.vue @@ -2540,7 +2540,7 @@ async function handleManualCompactionCommand( const result = await sessionClient.compactSession(sessionId) if (!canWriteSessionView(sessionId, restoreRequestId)) return true const restoredSession = await loadMessagesForSession(sessionId) - if (!canWriteSessionView(sessionId, restoreRequestId)) return true + if (!canWriteSessionView(sessionId, restoreRequestId) || restoredSession === null) return true applyRestoredSessionSummary(restoredSession) if (!result.compacted) { toast({ diff --git a/src/renderer/src/stores/ui/message.ts b/src/renderer/src/stores/ui/message.ts index dacc61c7a4..80f2d0c019 100644 --- a/src/renderer/src/stores/ui/message.ts +++ b/src/renderer/src/stores/ui/message.ts @@ -908,7 +908,7 @@ export const useMessageStore = defineStore('message', () => { hydratingStreamMessageIds.delete(messageId) } - const cleanupIpcBindings = bindMessageStoreIpc({ + const messageIpcBinding = bindMessageStoreIpc({ getActiveSessionId: () => currentSessionId.value, getCurrentStreamIdentity: () => ({ sessionId: currentStreamSessionId.value, @@ -923,7 +923,15 @@ export const useMessageStore = defineStore('message', () => { applyStreamingBlocksToMessage, isEphemeralStreamMessageId }) - registerStoreCleanup(cleanupIpcBindings) + registerStoreCleanup(messageIpcBinding.cleanup) + + function purgeSessionTracking(sessionId: string): void { + recentSessionViews.delete(sessionId) + messageMutationRevisions.delete(sessionId) + recentViewInvalidationRevisions.delete(sessionId) + dirtyRecentSessionViews.delete(sessionId) + messageIpcBinding.purgeSessionTracking(sessionId) + } return { messageIds, @@ -947,6 +955,7 @@ export const useMessageStore = defineStore('message', () => { getMessageMetadata, setCurrentSessionId, invalidateRecentSessionView, + purgeSessionTracking, activateRecentSessionView, loadMessages, loadOlderMessages, diff --git a/src/renderer/src/stores/ui/messageIpc.ts b/src/renderer/src/stores/ui/messageIpc.ts index fe21edce04..ae29458925 100644 --- a/src/renderer/src/stores/ui/messageIpc.ts +++ b/src/renderer/src/stores/ui/messageIpc.ts @@ -30,28 +30,60 @@ interface BindMessageStoreIpcOptions { type StreamCursor = { requestId: string updatedAt: number + generation: number } -const MAX_SETTLED_STREAMS = 128 +type StreamRequestState = { + generation: number + settled: boolean +} -export function bindMessageStoreIpc(options: BindMessageStoreIpcOptions): () => void { - const chatClient = createChatClient() - const latestStreamBySession = new Map() - const settledStreams = new Set() +type StreamSessionState = { + current: StreamCursor | null + latestAcceptedGeneration: number + nextGeneration: number + requests: Map +} - const streamKey = (sessionId: string, requestId: string) => `${sessionId}\u0000${requestId}` +type MessageStoreIpcBinding = { + cleanup: () => void + purgeSessionTracking: (sessionId: string) => void +} - const markStreamSettled = (sessionId: string, requestId: string): boolean => { - const key = streamKey(sessionId, requestId) - if (settledStreams.has(key)) return false +export function bindMessageStoreIpc(options: BindMessageStoreIpcOptions): MessageStoreIpcBinding { + const chatClient = createChatClient() + // Request entries are tombstones. Keep them until permanent session removal so + // a known older request cannot become current again after a newer request arrives. + const streamSessions = new Map() + + const getStreamSessionState = (sessionId: string): StreamSessionState => { + const existing = streamSessions.get(sessionId) + if (existing) return existing + + const created: StreamSessionState = { + current: null, + latestAcceptedGeneration: 0, + nextGeneration: 0, + requests: new Map() + } + streamSessions.set(sessionId, created) + return created + } - settledStreams.add(key) - while (settledStreams.size > MAX_SETTLED_STREAMS) { - const oldestKey = settledStreams.keys().next().value - if (!oldestKey) break - settledStreams.delete(oldestKey) + const getStreamRequestState = ( + session: StreamSessionState, + requestId: string + ): StreamRequestState => { + const existing = session.requests.get(requestId) + if (existing) return existing + + session.nextGeneration += 1 + const created: StreamRequestState = { + generation: session.nextGeneration, + settled: false } - return true + session.requests.set(requestId, created) + return created } const acceptStreamUpdate = (payload: { @@ -59,24 +91,21 @@ export function bindMessageStoreIpc(options: BindMessageStoreIpcOptions): () => requestId: string updatedAt: number }): boolean => { - if (settledStreams.has(streamKey(payload.sessionId, payload.requestId))) { - return false - } + const session = getStreamSessionState(payload.sessionId) + const request = getStreamRequestState(session, payload.requestId) + if (request.settled || request.generation < session.latestAcceptedGeneration) return false - const latest = latestStreamBySession.get(payload.sessionId) - if (latest) { - if (latest.requestId === payload.requestId && payload.updatedAt < latest.updatedAt) { - return false - } - if (latest.requestId !== payload.requestId && payload.updatedAt <= latest.updatedAt) { - return false - } + if (request.generation === session.latestAcceptedGeneration) { + if (!session.current || session.current.requestId !== payload.requestId) return false + if (payload.updatedAt < session.current.updatedAt) return false } - latestStreamBySession.set(payload.sessionId, { + session.latestAcceptedGeneration = request.generation + session.current = { requestId: payload.requestId, - updatedAt: payload.updatedAt - }) + updatedAt: payload.updatedAt, + generation: request.generation + } return true } @@ -95,17 +124,19 @@ export function bindMessageStoreIpc(options: BindMessageStoreIpcOptions): () => const settleStream = (payload: { sessionId: string; requestId: string }) => { // requestId is the turn identity; messageId may move from an ephemeral // rate-limit row to the persisted assistant row within the same turn. - if (!markStreamSettled(payload.sessionId, payload.requestId)) { - return - } + const session = getStreamSessionState(payload.sessionId) + const request = getStreamRequestState(session, payload.requestId) + if (request.settled) return + request.settled = true options.invalidateRecentSessionView(payload.sessionId) - const latest = latestStreamBySession.get(payload.sessionId) - if (latest?.requestId === payload.requestId) { - latestStreamBySession.delete(payload.sessionId) - } else if (latest) { + if (session.current?.requestId === payload.requestId) { + session.current = null + } else if (session.current || request.generation < session.latestAcceptedGeneration) { return + } else { + session.latestAcceptedGeneration = request.generation } if (payload.sessionId !== options.getActiveSessionId()) { @@ -178,9 +209,15 @@ export function bindMessageStoreIpc(options: BindMessageStoreIpcOptions): () => }) ] - return () => { - for (const cleanup of cleanups) { - cleanup() + return { + cleanup: () => { + for (const cleanup of cleanups) { + cleanup() + } + streamSessions.clear() + }, + purgeSessionTracking: (sessionId: string) => { + streamSessions.delete(sessionId) } } } diff --git a/src/renderer/src/stores/ui/pendingInput.ts b/src/renderer/src/stores/ui/pendingInput.ts index 5b88eff0e5..3a7fcd2479 100644 --- a/src/renderer/src/stores/ui/pendingInput.ts +++ b/src/renderer/src/stores/ui/pendingInput.ts @@ -47,93 +47,69 @@ export const usePendingInputStore = defineStore('pendingInput', () => { } } - async function queueInput(sessionId: string, input: string | SendMessageInput): Promise { + async function runSessionScopedMutation( + sessionId: string, + operation: () => Promise, + failureMessage: string + ): Promise { if (currentSessionId.value === sessionId) { error.value = null } try { - await sessionClient.queuePendingInput(sessionId, input) + await operation() if (currentSessionId.value === sessionId) { await loadPendingInputs(sessionId) } } catch (e) { if (currentSessionId.value === sessionId) { - error.value = `Failed to queue message: ${e}` + error.value = `${failureMessage}: ${e}` } throw e } } + async function queueInput(sessionId: string, input: string | SendMessageInput): Promise { + return runSessionScopedMutation( + sessionId, + () => sessionClient.queuePendingInput(sessionId, input), + 'Failed to queue message' + ) + } + async function updateQueueInput( sessionId: string, itemId: string, input: string | SendMessageInput ): Promise { - if (currentSessionId.value === sessionId) { - error.value = null - } - try { - await sessionClient.updateQueuedInput(sessionId, itemId, input) - if (currentSessionId.value === sessionId) { - await loadPendingInputs(sessionId) - } - } catch (e) { - if (currentSessionId.value === sessionId) { - error.value = `Failed to update queued message: ${e}` - } - throw e - } + return runSessionScopedMutation( + sessionId, + () => sessionClient.updateQueuedInput(sessionId, itemId, input), + 'Failed to update queued message' + ) } async function moveQueueInput(sessionId: string, itemId: string, toIndex: number): Promise { - if (currentSessionId.value === sessionId) { - error.value = null - } - try { - await sessionClient.moveQueuedInput(sessionId, itemId, toIndex) - if (currentSessionId.value === sessionId) { - await loadPendingInputs(sessionId) - } - } catch (e) { - if (currentSessionId.value === sessionId) { - error.value = `Failed to reorder queued message: ${e}` - } - throw e - } + return runSessionScopedMutation( + sessionId, + () => sessionClient.moveQueuedInput(sessionId, itemId, toIndex), + 'Failed to reorder queued message' + ) } async function steerPendingInput(sessionId: string, itemId: string): Promise { - if (currentSessionId.value === sessionId) { - error.value = null - } - try { - await sessionClient.steerPendingInput(sessionId, itemId) - if (currentSessionId.value === sessionId) { - await loadPendingInputs(sessionId) - } - } catch (e) { - if (currentSessionId.value === sessionId) { - error.value = `Failed to steer queued message: ${e}` - } - throw e - } + return runSessionScopedMutation( + sessionId, + () => sessionClient.steerPendingInput(sessionId, itemId), + 'Failed to steer queued message' + ) } async function deleteInput(sessionId: string, itemId: string): Promise { - if (currentSessionId.value === sessionId) { - error.value = null - } - try { - await sessionClient.deletePendingInput(sessionId, itemId) - if (currentSessionId.value === sessionId) { - await loadPendingInputs(sessionId) - } - } catch (e) { - if (currentSessionId.value === sessionId) { - error.value = `Failed to delete queued message: ${e}` - } - throw e - } + return runSessionScopedMutation( + sessionId, + () => sessionClient.deletePendingInput(sessionId, itemId), + 'Failed to delete queued message' + ) } function clear(): void { diff --git a/src/renderer/src/stores/ui/session.ts b/src/renderer/src/stores/ui/session.ts index febaad0b7e..4f76a08b9f 100644 --- a/src/renderer/src/stores/ui/session.ts +++ b/src/renderer/src/stores/ui/session.ts @@ -384,6 +384,7 @@ export const useSessionStore = defineStore('session', () => { for (const sessionId of targetIds) { agentPlanStore.purge(sessionId) messageStore.invalidateRecentSessionView(sessionId) + messageStore.purgeSessionTracking(sessionId) } if (bootstrapActiveSession.value && targetIds.has(bootstrapActiveSession.value.id)) { diff --git a/test/renderer/components/ChatPage.test.ts b/test/renderer/components/ChatPage.test.ts index 4c3ef2fec2..1172d891d4 100644 --- a/test/renderer/components/ChatPage.test.ts +++ b/test/renderer/components/ChatPage.test.ts @@ -1624,6 +1624,28 @@ describe('ChatPage', () => { ]) }) + it('does not apply a null result from a superseded compaction restore', async () => { + const { wrapper, messageStore, sessionStore } = await setup({ + activeSessionPatch: { + providerId: 'openai', + modelId: 'gpt-4' + } + }) + const applyRestoredSession = vi.fn() + ;( + sessionStore as typeof sessionStore & { + applyRestoredSession: (session: unknown) => void + } + ).applyRestoredSession = applyRestoredSession + messageStore.loadMessages.mockResolvedValueOnce(null) + + wrapper.findComponent({ name: 'ChatInputBox' }).vm.$emit('command-submit', '/compact') + await flushPromises() + + expect(messageStore.loadMessages).toHaveBeenCalledWith('s1', 100) + expect(applyRestoredSession).not.toHaveBeenCalled() + }) + it('shows a no-op notice when manual compaction has no eligible history', async () => { const { wrapper, sessionClient, toast, messageStore } = await setup({ activeSessionPatch: { diff --git a/test/renderer/stores/messageStore.test.ts b/test/renderer/stores/messageStore.test.ts index 6c755efb20..bc1e42325c 100644 --- a/test/renderer/stores/messageStore.test.ts +++ b/test/renderer/stores/messageStore.test.ts @@ -931,6 +931,85 @@ describe('messageStore', () => { expect(store.messageCache.value.get('message-new')?.content).toContain('new') }) + it('does not let a superseded request reclaim the stream with a later snapshot', async () => { + const { store, streamListeners } = await setupStore() + await store.loadMessages('s1') + + streamListeners.updated[0]({ + kind: 'snapshot', + sessionId: 's1', + requestId: 'request-old', + messageId: 'message-old', + updatedAt: 1, + blocks: [{ type: 'content', content: 'old', status: 'pending', timestamp: 1 }] + }) + streamListeners.updated[0]({ + kind: 'snapshot', + sessionId: 's1', + requestId: 'request-new', + messageId: 'message-new', + updatedAt: 2, + blocks: [{ type: 'content', content: 'new', status: 'pending', timestamp: 2 }] + }) + streamListeners.updated[0]({ + kind: 'snapshot', + sessionId: 's1', + requestId: 'request-old', + messageId: 'message-old', + updatedAt: 3, + blocks: [{ type: 'content', content: 'late old', status: 'pending', timestamp: 3 }] + }) + + expect(store.currentStreamRequestId.value).toBe('request-new') + expect(store.currentStreamMessageId.value).toBe('message-new') + expect(store.streamingBlocks.value[0]).toMatchObject({ content: 'new' }) + expect(store.messageCache.value.get('message-old')?.content).not.toContain('late old') + }) + + it('keeps settled request tombstones after many later terminals', async () => { + const { store, sessionClient, streamListeners } = await setupStore() + await store.loadMessages('s1') + + streamListeners.updated[0]({ + kind: 'snapshot', + sessionId: 's1', + requestId: 'request-settled', + messageId: 'message-settled', + updatedAt: 1, + blocks: [{ type: 'content', content: 'live', status: 'pending', timestamp: 1 }] + }) + streamListeners.completed[0]({ + sessionId: 's1', + requestId: 'request-settled', + messageId: 'message-settled', + completedAt: 2 + }) + await new Promise((resolve) => setTimeout(resolve, 0)) + + for (let index = 0; index < 129; index += 1) { + streamListeners.completed[0]({ + sessionId: `inactive-${index}`, + requestId: `request-${index}`, + messageId: `message-${index}`, + completedAt: index + 3 + }) + } + + streamListeners.updated[0]({ + kind: 'snapshot', + sessionId: 's1', + requestId: 'request-settled', + messageId: 'message-settled', + updatedAt: 500, + blocks: [{ type: 'content', content: 'resurrected', status: 'pending', timestamp: 500 }] + }) + + expect(sessionClient.restore).toHaveBeenCalledTimes(2) + expect(store.isStreaming.value).toBe(false) + expect(store.currentStreamRequestId.value).toBeNull() + expect(store.streamingBlocks.value).toEqual([]) + }) + it('ignores stream snapshots that arrive after their terminal event', async () => { const { store, sessionClient, streamListeners } = await setupStore() sessionClient.restore diff --git a/test/renderer/stores/sessionStore.test.ts b/test/renderer/stores/sessionStore.test.ts index 413b420143..e4090d7953 100644 --- a/test/renderer/stores/sessionStore.test.ts +++ b/test/renderer/stores/sessionStore.test.ts @@ -304,10 +304,12 @@ const setupStore = async (options: SetupStoreOptions = {}) => { const clearStreamingState = vi.fn() const setCurrentSessionId = vi.fn() const invalidateRecentSessionView = vi.fn() + const purgeSessionTracking = vi.fn() vi.doMock('@/stores/ui/message', () => ({ useMessageStore: () => ({ clearStreamingState, invalidateRecentSessionView, + purgeSessionTracking, loadMessages: vi.fn(), setCurrentSessionId }) @@ -345,6 +347,7 @@ const setupStore = async (options: SetupStoreOptions = {}) => { configClient, clearStreamingState, invalidateRecentSessionView, + purgeSessionTracking, setCurrentSessionId, sessionClient, chatClient, @@ -1273,6 +1276,21 @@ describe('sessionStore streaming cleanup', () => { expect(store.activeSession.value?.status).toBe('none') }) + + it('purges message tracking when a session is permanently removed', async () => { + const { store, emitSessionUpdate, invalidateRecentSessionView, purgeSessionTracking } = + await setupStore() + store.sessions.value = [createSession({ id: 'session-removed' })] + + emitSessionUpdate({ + reason: 'deleted', + sessionIds: ['session-removed'] + }) + + expect(invalidateRecentSessionView).toHaveBeenCalledWith('session-removed') + expect(purgeSessionTracking).toHaveBeenCalledWith('session-removed') + expect(store.sessions.value).toEqual([]) + }) }) describe('sessionStore pagination', () => {