perf(stream): dedupe snapshot serialization

src/main/agent/acp/compatibility/adapters.ts

@@ -20,6 +20,7 @@ import {
 import { createAcpPromptTerminalEvents } from '@/agent/acp/runtime/acpContentMapper'
 import {
   createState,
+  markStreamChanged,
   type DeepChatEventPublisher,
   type DeepChatSessionUpdatePublisher,
   type IoParams,
@@ -159,7 +160,7 @@ export class AcpCompatibilityProjectionAdapter implements AcpCompatibilityProjec
         ? block.extra.permissionType
         : 'all'
     markStreamingProviderPermissionResolved(block, granted, permissionType)
-    state.stream.dirty = true
+    markStreamChanged(state.stream)
     this.flushIfDirty(state)
   }
 

src/main/agent/deepchat/runtime/accumulator.ts

@@ -4,7 +4,7 @@ import type {
   ProviderUrlSourcePayload
 } from '@shared/types/core/llm-events'
 import type { ChatMessageProviderOptions } from '@shared/types/core/chat-message'
-import type { StreamState } from './types'
+import { markStreamChanged, type StreamState } from './types'
 
 const MAX_VISIBLE_SEARCH_PAGES = 6
 
@@ -144,7 +144,7 @@ export function accumulate(state: StreamState, event: LLMCoreStreamEvent): void
       if (state.firstTokenTime === null) state.firstTokenTime = Date.now()
       const block = getCurrentBlock(state.blocks, 'content', event.provider_options)
       block.content += event.content
-      state.dirty = true
+      markStreamChanged(state)
       break
     }
     case 'reasoning': {
@@ -167,12 +167,12 @@ export function accumulate(state: StreamState, event: LLMCoreStreamEvent): void
       }
       const reasoningTime = block.reasoning_time as { start: number; end: number }
       updateReasoningMetadata(state, reasoningTime.start, reasoningTime.end)
-      state.dirty = true
+      markStreamChanged(state)
       break
     }
     case 'plan': {
       if (finalizeTrailingPendingNarrativeBlocks(state.blocks)) {
-        state.dirty = true
+        markStreamChanged(state)
       }
       const revision = event.revision ?? (state.latestAgentPlanSnapshot?.revision ?? 0) + 1
       state.latestAgentPlanSnapshot = {
@@ -209,7 +209,7 @@ export function accumulate(state: StreamState, event: LLMCoreStreamEvent): void
         executionOwner: event.tool_call_execution_owner ?? 'deepchat',
         providerOptions: event.provider_options
       })
-      state.dirty = true
+      markStreamChanged(state)
       break
     }
     case 'tool_call_chunk': {
@@ -229,7 +229,7 @@ export function accumulate(state: StreamState, event: LLMCoreStreamEvent): void
             }
           }
         }
-        state.dirty = true
+        markStreamChanged(state)
       }
       break
     }
@@ -265,7 +265,7 @@ export function accumulate(state: StreamState, event: LLMCoreStreamEvent): void
           })
         }
         state.pendingToolCalls.delete(event.tool_call_id)
-        state.dirty = true
+        markStreamChanged(state)
       }
       break
     }
@@ -296,12 +296,12 @@ export function accumulate(state: StreamState, event: LLMCoreStreamEvent): void
         }
       }
       state.blocks.push(block)
-      state.dirty = true
+      markStreamChanged(state)
       break
     }
     case 'provider_url_source': {
       if (appendProviderUrlSource(state.blocks, event.provider_url_source)) {
-        state.dirty = true
+        markStreamChanged(state)
       }
       break
     }
@@ -318,7 +318,7 @@ export function accumulate(state: StreamState, event: LLMCoreStreamEvent): void
         }
       }
       state.blocks.push(block)
-      state.dirty = true
+      markStreamChanged(state)
       break
     }
     case 'usage': {
@@ -354,7 +354,7 @@ export function accumulate(state: StreamState, event: LLMCoreStreamEvent): void
         if (block.status === 'pending') block.status = 'error'
       }
       state.stopReason = 'error'
-      state.dirty = true
+      markStreamChanged(state)
       break
     }
     default:

src/main/agent/deepchat/runtime/deepChatLoopRunner.ts

@@ -661,6 +661,7 @@ export function buildTapeViewSelection(
 
 export class DeepChatLoopRunner {
   private readonly toolSurfaceAdapterHistory = new ToolSurfaceAdapterHistory()
+  private rateLimitRevision = 0
 
   constructor(private readonly ports: DeepChatLoopRunnerPorts) {}
 
@@ -2726,6 +2727,7 @@ export class DeepChatLoopRunner {
       sessionId,
       messageId,
       updatedAt: Date.now(),
+      revision: ++this.rateLimitRevision,
       blocks: cloneBlocksForRenderer([block])
     })
   }
@@ -2737,6 +2739,7 @@ export class DeepChatLoopRunner {
       sessionId,
       messageId,
       updatedAt: Date.now(),
+      revision: ++this.rateLimitRevision,
       blocks: []
     })
   }

src/main/agent/deepchat/runtime/dispatch.ts

@@ -32,6 +32,7 @@ import type {
   ToolCallResult,
   ToolDispatchCollaborators
 } from './types'
+import { markStreamChanged } from './types'
 import type {
   ChatMessage,
   ChatMessageProviderOptions,
@@ -1253,7 +1254,7 @@ function finalizePendingNarrativeBeforeToolSettlement(state: StreamState): void
   }
 
   finalizeTrailingPendingNarrativeBlocks(state.blocks)
-  state.dirty = true
+  markStreamChanged(state)
 }
 
 function applyFinalizedToolResults(params: {
@@ -1387,7 +1388,7 @@ function applyFinalizedToolResults(params: {
     }
   }
 
-  state.dirty = true
+  markStreamChanged(state)
   return interactions
 }
 
@@ -1673,7 +1674,7 @@ async function reviewAutoApproveAction(params: {
   }
 
   if (setToolCallAutoApproveReviewing(batchToolCallBlocks, execution.completedToolCall.id, true)) {
-    state.dirty = true
+    markStreamChanged(state)
     rendererFlushHandle.flush()
   }
   try {
@@ -1705,7 +1706,7 @@ async function reviewAutoApproveAction(params: {
     if (
       setToolCallAutoApproveReviewing(batchToolCallBlocks, execution.completedToolCall.id, false)
     ) {
-      state.dirty = true
+      markStreamChanged(state)
       rendererFlushHandle.flush()
     }
   }
@@ -1772,7 +1773,7 @@ function appendPermissionActionBlock(
       ...(permission.rememberable === false ? { rememberable: false } : {})
     }
   })
-  state.dirty = true
+  markStreamChanged(state)
   return {
     type: 'permission',
     origin,
@@ -1830,7 +1831,7 @@ function appendQuestionActionBlock(
       ...extra
     }
   })
-  state.dirty = true
+  markStreamChanged(state)
   return {
     type: 'question',
     origin,
@@ -1896,8 +1897,8 @@ function appendSkillDraftQuestionActionBlock(
   )
 }
 
-function flushBlocksToRenderer(io: IoParams, blocks: AssistantMessageBlock[]): void {
-  const renderedBlocks = cloneBlocksForRenderer(blocks)
+function flushBlocksToRenderer(io: IoParams, state: StreamState): void {
+  const renderedBlocks = cloneBlocksForRenderer(state.blocks)
   io.publishEvent('chat.stream.updated', {
     kind: 'snapshot',
     requestId: io.requestId,
@@ -1906,6 +1907,7 @@ function flushBlocksToRenderer(io: IoParams, blocks: AssistantMessageBlock[]): v
     providerId: io.providerId,
     modelId: io.modelId,
     updatedAt: Date.now(),
+    revision: state.blocksRevision,
     blocks: renderedBlocks
   })
 
@@ -1914,10 +1916,10 @@ function flushBlocksToRenderer(io: IoParams, blocks: AssistantMessageBlock[]): v
     kind: 'blocks',
     updatedAt: Date.now(),
     messageId: io.messageId,
-    previewMarkdown: buildAssistantPreviewMarkdown(blocks),
-    responseMarkdown: buildAssistantResponseMarkdown(blocks),
-    deliverySegments: buildAssistantDeliverySegments(io.messageId, blocks),
-    waitingInteraction: extractWaitingInteraction(blocks, io.messageId)
+    previewMarkdown: buildAssistantPreviewMarkdown(state.blocks),
+    responseMarkdown: buildAssistantResponseMarkdown(state.blocks),
+    deliverySegments: buildAssistantDeliverySegments(io.messageId, state.blocks),
+    waitingInteraction: extractWaitingInteraction(state.blocks, io.messageId)
   })
 }
 
@@ -2130,7 +2132,7 @@ async function runToolCall(params: {
         }
         state.latestAgentPlanSnapshot = snapshot
         publishPlanUpdated(io, snapshot)
-        state.dirty = true
+        markStreamChanged(state)
         scheduleRendererFlush(state, rendererFlushHandle)
         return
       }
@@ -2149,7 +2151,7 @@ async function runToolCall(params: {
         update.responseMarkdown,
         update.progressJson
       )
-      state.dirty = true
+      markStreamChanged(state)
       scheduleRendererFlush(state, rendererFlushHandle)
     }
 
@@ -3204,7 +3206,7 @@ export async function settleToolBatch(
             content: errorText
           })
           updateToolCallBlock(batchToolCallBlocks, tc.id, errorText, true)
-          state.dirty = true
+          markStreamChanged(state)
           batchState.committedResultCallIds.add(tc.id)
           executed += 1
           persistToolExecutionState(io, state, rendererFlushHandle)
@@ -3534,7 +3536,7 @@ export function finalizePaused(state: StreamState, io: IoParams): void {
   stampGenerationTiming(state)
 
   io.messageStore.updateAssistantContent(io.messageId, state.blocks, JSON.stringify(state.metadata))
-  flushBlocksToRenderer(io, state.blocks)
+  flushBlocksToRenderer(io, state)
   io.publishEvent('chat.stream.completed', {
     requestId: io.requestId,
     sessionId: io.sessionId,
@@ -3547,6 +3549,7 @@ export function finalize(state: StreamState, io: IoParams): void {
   for (const block of state.blocks) {
     if (block.status === 'pending') block.status = 'success'
   }
+  markStreamChanged(state)
   stampPlanTerminalIfOpen(state, io, state.planTerminalReason)
 
   stampGenerationTiming(state)
@@ -3556,7 +3559,7 @@ export function finalize(state: StreamState, io: IoParams): void {
     state.blocks,
     JSON.stringify(state.metadata)
   )
-  flushBlocksToRenderer(io, state.blocks)
+  flushBlocksToRenderer(io, state)
   io.publishEvent('chat.stream.completed', {
     requestId: io.requestId,
     sessionId: io.sessionId,
@@ -3568,6 +3571,7 @@ export function finalize(state: StreamState, io: IoParams): void {
 export function finalizeError(state: StreamState, io: IoParams, error: unknown): void {
   const errorMessage = error instanceof Error ? error.message : String(error)
   state.blocks = buildTerminalErrorBlocks(state.blocks, errorMessage)
+  markStreamChanged(state)
   stampPlanTerminalIfOpen(
     state,
     io,
@@ -3577,7 +3581,7 @@ export function finalizeError(state: StreamState, io: IoParams, error: unknown):
   stampGenerationTiming(state)
 
   io.messageStore.setMessageError(io.messageId, state.blocks, JSON.stringify(state.metadata))
-  flushBlocksToRenderer(io, state.blocks)
+  flushBlocksToRenderer(io, state)
   io.publishEvent('chat.stream.failed', {
     requestId: io.requestId,
     sessionId: io.sessionId,
@@ -3596,5 +3600,5 @@ export function persistAbortExceptionPlanState(state: StreamState, io: IoParams)
   }
 
   io.messageStore.updateAssistantContent(io.messageId, state.blocks)
-  flushBlocksToRenderer(io, state.blocks)
+  flushBlocksToRenderer(io, state)
 }

src/main/agent/deepchat/runtime/echo.ts

@@ -23,6 +23,7 @@ export function startEcho(state: StreamState, io: IoParams): EchoHandle {
       providerId: io.providerId,
       modelId: io.modelId,
       updatedAt: Date.now(),
+      revision: state.blocksRevision,
       blocks: renderedBlocks
     })
   }