fix(chat): refresh history when non-text streamed content changes

apps/webclaw/src/screens/chat/chat-screen.tsx

@@ -267,7 +267,7 @@ export function ChatScreen({
         sessionKey,
         friendlyId,
         message: body,
-        thinking: 'low',
+        thinking: 'high',
         idempotencyKey: crypto.randomUUID(),
         attachments: attachmentsPayload,
       }),

apps/webclaw/src/screens/chat/hooks/use-chat-history.ts

@@ -93,7 +93,8 @@ export function useChatHistory({
     const lastId = typeof last?.id === 'string' ? last.id : ''
     const lastRole = typeof last?.role === 'string' ? last.role : ''
     const lastText = last ? textFromMessage(last) : ''
-    const signature = `${messages.length}:${lastRole}:${lastId}:${lastText.slice(-32)}`
+    const lastContentSignature = last ? contentSignatureFromMessage(last) : ''
+    const signature = `${messages.length}:${lastRole}:${lastId}:${lastText.slice(-32)}:${lastContentSignature}`
     if (signature === stableHistorySignatureRef.current) {
       return stableHistoryMessagesRef.current
     }
@@ -125,6 +126,24 @@ export function useChatHistory({
   }
 }
 
+function contentSignatureFromMessage(message: GatewayMessage): string {
+  const content = Array.isArray(message.content) ? message.content : []
+  return content
+    .map((part) => {
+      if (part.type === 'text') {
+        return `text:${String(part.text ?? '').length}`
+      }
+      if (part.type === 'thinking') {
+        return `thinking:${String(part.thinking ?? '').length}`
+      }
+      const id = 'id' in part ? String(part.id ?? '') : ''
+      const name = 'name' in part ? String(part.name ?? '') : ''
+      const partialJson = 'partialJson' in part ? String(part.partialJson ?? '') : ''
+      return `toolCall:${id}:${name}:${partialJson.length}`
+    })
+    .join('|')
+}
+
 function mergeStreamingHistoryMessages(
   serverMessages: Array<GatewayMessage>,
   streamingMessages: Array<GatewayMessage>,
@@ -139,20 +158,31 @@ function mergeStreamingHistoryMessages(
     const hasMatch = merged.some((serverMessage) => {
       const serverRunId = (serverMessage as { __streamRunId?: unknown })
         .__streamRunId
+
+      if (serverMessage.role !== streamingMessage.role) return false
+      const streamingTime = getMessageTimestamp(streamingMessage)
+      const serverTime = getMessageTimestamp(serverMessage)
+      if (Math.abs(streamingTime - serverTime) > 15000) return false
+
       if (
         typeof serverRunId === 'string' &&
         serverRunId.trim().length > 0 &&
         serverRunId === runId
       ) {
-        return true
+        return messageCoversStreamingMessage(serverMessage, streamingMessage)
       }
-      if (serverMessage.role !== streamingMessage.role) return false
+
       const streamingText = textFromMessage(streamingMessage)
-      if (!streamingText) return false
-      if (streamingText !== textFromMessage(serverMessage)) return false
-      const streamingTime = getMessageTimestamp(streamingMessage)
-      const serverTime = getMessageTimestamp(serverMessage)
-      return Math.abs(streamingTime - serverTime) <= 15000
+      const serverText = textFromMessage(serverMessage)
+      if (
+        streamingText &&
+        streamingText !== serverText &&
+        !serverText.startsWith(streamingText)
+      ) {
+        return false
+      }
+
+      return messageCoversStreamingMessage(serverMessage, streamingMessage)
     })
 
     if (!hasMatch) {
@@ -163,6 +193,37 @@ function mergeStreamingHistoryMessages(
   return merged
 }
 
+function messageCoversStreamingMessage(
+  serverMessage: GatewayMessage,
+  streamingMessage: GatewayMessage,
+): boolean {
+  const serverSignatures = nonTextPartSignatures(serverMessage)
+  const streamingSignatures = nonTextPartSignatures(streamingMessage)
+  if (streamingSignatures.size === 0) return true
+
+  for (const signature of streamingSignatures) {
+    if (!serverSignatures.has(signature)) {
+      return false
+    }
+  }
+
+  return true
+}
+
+function nonTextPartSignatures(message: GatewayMessage): Set<string> {
+  const signatures = new Set<string>()
+  const parts = Array.isArray(message.content) ? message.content : []
+  for (const part of parts) {
+    if (part.type === 'text') continue
+    try {
+      signatures.add(`${part.type}:${JSON.stringify(part)}`)
+    } catch {
+      signatures.add(`${part.type}:unserializable`)
+    }
+  }
+  return signatures
+}
+
 function mergeOptimisticHistoryMessages(
   serverMessages: Array<GatewayMessage>,
   optimisticMessages: Array<GatewayMessage>,

apps/webclaw/src/screens/chat/hooks/use-chat-stream.ts

@@ -7,7 +7,7 @@ import {
   updateSessionLastMessage,
 } from '../chat-queries'
 import type { QueryClient } from '@tanstack/react-query'
-import type { GatewayMessage } from '../types'
+import type { GatewayMessage, MessageContent } from '../types'
 
 type UseChatStreamInput = {
   activeFriendlyId: string
@@ -25,6 +25,14 @@ type UseChatStreamInput = {
   }) => void
 }
 
+type StreamChatPayload = {
+  runId?: string
+  sessionKey?: string
+  state?: string
+  message?: GatewayMessage
+  seq?: number
+}
+
 export function useChatStream({
   activeFriendlyId,
   isNewChat,
@@ -37,10 +45,14 @@ export function useChatStream({
 }: UseChatStreamInput) {
   const streamSourceRef = useRef<EventSource | null>(null)
   const streamReconnectTimer = useRef<number | null>(null)
+  const streamAgentRefreshTimer = useRef<number | null>(null)
+  const streamHistoryPollTimer = useRef<number | null>(null)
+  const streamActiveRunsRef = useRef(new Set<string>())
   const streamReconnectAttempt = useRef(0)
-  const streamRunTextRef = useRef(new Map<string, string>())
   const streamRunSeqRef = useRef(new Map<string, number>())
   const streamRunStateVersionRef = useRef(new Map<string, number>())
+  const streamRunSourceRef = useRef(new Map<string, 'agent' | 'chat'>())
+  const streamSeenEventKeysRef = useRef(new Set<string>())
   const refreshHistoryRef = useRef(refreshHistory)
 
   useEffect(() => {
@@ -56,9 +68,59 @@ export function useChatStream({
       streamSourceRef.current.close()
       streamSourceRef.current = null
     }
-    streamRunTextRef.current.clear()
+    if (streamAgentRefreshTimer.current) {
+      window.clearTimeout(streamAgentRefreshTimer.current)
+      streamAgentRefreshTimer.current = null
+    }
+    if (streamHistoryPollTimer.current) {
+      window.clearInterval(streamHistoryPollTimer.current)
+      streamHistoryPollTimer.current = null
+    }
+    streamActiveRunsRef.current.clear()
     streamRunSeqRef.current.clear()
     streamRunStateVersionRef.current.clear()
+    streamRunSourceRef.current.clear()
+    streamSeenEventKeysRef.current.clear()
+  }, [])
+
+  const scheduleHistoryRefresh = useCallback(() => {
+    if (streamAgentRefreshTimer.current) return
+    streamAgentRefreshTimer.current = window.setTimeout(() => {
+      streamAgentRefreshTimer.current = null
+      refreshHistoryRef.current()
+    }, 500)
+  }, [])
+
+  const ensureHistoryPolling = useCallback(() => {
+    if (streamHistoryPollTimer.current) return
+    streamHistoryPollTimer.current = window.setInterval(() => {
+      if (streamActiveRunsRef.current.size === 0) {
+        if (streamHistoryPollTimer.current) {
+          window.clearInterval(streamHistoryPollTimer.current)
+          streamHistoryPollTimer.current = null
+        }
+        return
+      }
+      refreshHistoryRef.current()
+    }, 800)
+  }, [])
+
+  const markRunActive = useCallback(
+    (runId: string) => {
+      if (!runId) return
+      streamActiveRunsRef.current.add(runId)
+      ensureHistoryPolling()
+    },
+    [ensureHistoryPolling],
+  )
+
+  const markRunDone = useCallback((runId: string) => {
+    if (!runId) return
+    streamActiveRunsRef.current.delete(runId)
+    if (streamActiveRunsRef.current.size === 0 && streamHistoryPollTimer.current) {
+      window.clearInterval(streamHistoryPollTimer.current)
+      streamHistoryPollTimer.current = null
+    }
   }, [])
 
   useEffect(() => {
@@ -101,26 +163,64 @@ export function useChatStream({
             return
           }
           if (!parsed.event) return
-          if (parsed.event === 'chat') {
-            const payload = parsed.payload as
-              | {
-                  runId?: string
-                  sessionKey?: string
-                  state?: string
-                  message?: GatewayMessage
+          if (parsed.event === 'chat' || parsed.event === 'agent') {
+            const payloads: Array<StreamChatPayload | null> =
+              parsed.event === 'chat'
+                ? [parsed.payload as StreamChatPayload | null]
+                : extractChatPayloadsFromAgentPayload(parsed.payload)
+
+            if (parsed.event === 'agent' && payloads.length === 0) {
+              scheduleHistoryRefresh()
+              return
+            }
+
+            for (const payload of payloads) {
+              if (!payload) continue
+              const streamRunId =
+                typeof payload.runId === 'string' ? payload.runId : ''
+              const payloadSource: 'agent' | 'chat' =
+                parsed.event === 'agent' ? 'agent' : 'chat'
+              if (streamRunId) {
+                const currentSource = streamRunSourceRef.current.get(streamRunId)
+                if (
+                  payloadSource === 'chat' &&
+                  currentSource === 'agent' &&
+                  payload.state === 'delta'
+                ) {
+                  continue
                 }
-              | null
-            const streamRunId =
-              typeof payload?.runId === 'string' ? payload.runId : ''
-            const eventSeq =
-              typeof parsed.seq === 'number' && Number.isFinite(parsed.seq)
-                ? parsed.seq
-                : undefined
-            const eventStateVersion =
-              typeof parsed.stateVersion === 'number' &&
-              Number.isFinite(parsed.stateVersion)
-                ? parsed.stateVersion
-                : undefined
+                if (payloadSource === 'agent' || !currentSource) {
+                  streamRunSourceRef.current.set(streamRunId, payloadSource)
+                }
+              }
+
+              const payloadSeq =
+                typeof payload.seq === 'number' && Number.isFinite(payload.seq)
+                  ? payload.seq
+                  : undefined
+              const eventSeq =
+                payloadSource === 'agent'
+                  ? payloadSeq
+                  : typeof parsed.seq === 'number' && Number.isFinite(parsed.seq)
+                    ? parsed.seq
+                    : undefined
+              const eventStateVersion =
+                typeof parsed.stateVersion === 'number' &&
+                Number.isFinite(parsed.stateVersion)
+                  ? parsed.stateVersion
+                  : undefined
+
+              if (
+                shouldSkipDuplicateEvent(
+                  streamSeenEventKeysRef.current,
+                  payloadSource,
+                  streamRunId,
+                  payload.state,
+                  eventSeq,
+                )
+              ) {
+                continue
+              }
 
             if (
               shouldSkipStaleRunEvent(
@@ -131,13 +231,23 @@ export function useChatStream({
                 streamRunStateVersionRef.current,
               )
             ) {
-              return
+              continue
             }
 
-            if (payload) {
-              onChatEvent?.(payload)
-            }
-            if (payload?.message && typeof payload.message === 'object') {
+            onChatEvent?.(payload)
+            const payloadState =
+              typeof payload.state === 'string' ? payload.state : ''
+              if (
+                payloadState === 'final' ||
+                payloadState === 'error' ||
+                payloadState === 'aborted'
+              ) {
+                markRunDone(streamRunId)
+                refreshHistoryRef.current()
+              } else if (payloadState === 'delta') {
+                markRunActive(streamRunId)
+              }
+            if (payload.message && typeof payload.message === 'object') {
               const payloadSessionKey = payload.sessionKey
               if (
                 payloadSessionKey &&
@@ -145,51 +255,23 @@ export function useChatStream({
                 payloadSessionKey !== resolvedSessionKey &&
                 payloadSessionKey !== sessionKeyForHistory
               ) {
-                return
+                continue
               }
               const state = typeof payload.state === 'string' ? payload.state : ''
-              let nextMessage: GatewayMessage = {
+              const nextMessage: GatewayMessage = {
                 ...payload.message,
                 __streamRunId: streamRunId || undefined,
               }
 
-              if (streamRunId && state === 'delta') {
-                const deltaText = rawTextFromMessage(nextMessage)
-                const previousText = streamRunTextRef.current.get(streamRunId) ?? ''
-                const cumulativeText = mergeDeltaText(previousText, deltaText)
-                if (cumulativeText.length > 0) {
-                  streamRunTextRef.current.set(streamRunId, cumulativeText)
-                  nextMessage = {
-                    ...nextMessage,
-                    content: [{ type: 'text', text: cumulativeText }],
-                  }
+                if (
+                  streamRunId &&
+                  (state === 'final' || state === 'error' || state === 'aborted')
+                ) {
+                  markRunDone(streamRunId)
+                  streamRunSeqRef.current.delete(streamRunId)
+                  streamRunStateVersionRef.current.delete(streamRunId)
+                  streamRunSourceRef.current.delete(streamRunId)
                 }
-              }
-
-              if (streamRunId && state === 'final') {
-                const finalText = rawTextFromMessage(nextMessage)
-                if (!finalText) {
-                  const bufferedText = streamRunTextRef.current.get(streamRunId)
-                  if (bufferedText) {
-                    nextMessage = {
-                      ...nextMessage,
-                      content: [{ type: 'text', text: bufferedText }],
-                    }
-                  }
-                }
-                streamRunTextRef.current.delete(streamRunId)
-                streamRunSeqRef.current.delete(streamRunId)
-                streamRunStateVersionRef.current.delete(streamRunId)
-              }
-
-              if (
-                streamRunId &&
-                (state === 'error' || state === 'aborted')
-              ) {
-                streamRunTextRef.current.delete(streamRunId)
-                streamRunSeqRef.current.delete(streamRunId)
-                streamRunStateVersionRef.current.delete(streamRunId)
-              }
 
               function upsert(messages: Array<GatewayMessage>) {
                 const lastUserIndex = [...messages]
@@ -199,15 +281,15 @@ export function useChatStream({
                   lastUserIndex >= 0 ? messages.length - 1 - lastUserIndex : -1
 
                 if (streamRunId) {
-                  const index = messages.findIndex(
-                    (message) =>
-                      (message as { __streamRunId?: string }).__streamRunId ===
-                      streamRunId,
+                  const index = findStreamMessageIndex(
+                    messages,
+                    nextMessage,
+                    streamRunId,
                   )
                   if (index >= 0) {
                     if (index > resolvedLastUserIndex) {
                       const next = [...messages]
-                      next[index] = nextMessage
+                      next[index] = mergeStreamMessage(messages[index], nextMessage)
                       return next
                     }
                     return [...messages, nextMessage]
@@ -224,7 +306,7 @@ export function useChatStream({
                       const targetTime = getMessageTimestamp(messages[target])
                       if (Math.abs(nextTime - targetTime) <= 15000) {
                         const next = [...messages]
-                        next[target] = nextMessage
+                        next[target] = mergeStreamMessage(messages[target], nextMessage)
                         return next
                       }
                     }
@@ -261,6 +343,7 @@ export function useChatStream({
                 )
               }
             }
+            }
             return
           }
           if (!parsed.event.startsWith('chat.')) {
@@ -314,29 +397,74 @@ export function useChatStream({
   return { stopStream }
 }
 
-function rawTextFromMessage(message: GatewayMessage): string {
-  const parts = Array.isArray(message.content) ? message.content : []
-  return parts
-    .map((part) => (part.type === 'text' ? String(part.text ?? '') : ''))
-    .join('')
-}
+function mergeStreamMessage(
+  previousMessage: GatewayMessage,
+  nextMessage: GatewayMessage,
+): GatewayMessage {
+  const previousContent = Array.isArray(previousMessage.content)
+    ? previousMessage.content
+    : []
+  const nextContent = Array.isArray(nextMessage.content) ? nextMessage.content : []
 
-function mergeDeltaText(previousText: string, nextText: string): string {
-  if (!previousText) return nextText
-  if (!nextText) return previousText
-  if (nextText.startsWith(previousText)) return nextText
-  if (previousText.endsWith(nextText)) return previousText
-
-  const maxOverlap = Math.min(previousText.length, nextText.length)
-  for (let overlap = maxOverlap; overlap > 0; overlap -= 1) {
-    const previousSuffix = previousText.slice(-overlap)
-    const nextPrefix = nextText.slice(0, overlap)
-    if (previousSuffix === nextPrefix) {
-      return `${previousText}${nextText.slice(overlap)}`
-    }
+  if (previousContent.length === 0) {
+    return nextMessage
   }
 
-  return `${previousText}${nextText}`
+  if (nextContent.length === 0) {
+    return { ...previousMessage, ...nextMessage }
+  }
+
+  return {
+    ...previousMessage,
+    ...nextMessage,
+    content: mergeMessageContent(previousContent, nextContent),
+  }
+}
+
+function mergeMessageContent(
+  previousContent: Array<MessageContent>,
+  nextContent: Array<MessageContent>,
+): Array<MessageContent> {
+  const mergedByIdentity = new Map<string, MessageContent>()
+  const orderedKeys: Array<string> = []
+
+  function upsertPart(part: MessageContent) {
+    const identity = partIdentity(part)
+    if (!mergedByIdentity.has(identity)) {
+      orderedKeys.push(identity)
+    }
+    mergedByIdentity.set(identity, part)
+  }
+
+  for (const part of previousContent) {
+    upsertPart(part)
+  }
+  for (const part of nextContent) {
+    upsertPart(part)
+  }
+
+  return orderedKeys
+    .map((key) => mergedByIdentity.get(key))
+    .filter((part): part is MessageContent => Boolean(part))
+}
+
+function partIdentity(part: MessageContent): string {
+  switch (part.type) {
+    case 'text':
+      return 'text'
+    case 'thinking':
+      return 'thinking'
+    case 'toolCall': {
+      const toolCallId = normalizeString((part as { id?: unknown }).id)
+      const toolName = normalizeString((part as { name?: unknown }).name)
+      if (toolCallId || toolName) {
+        return `toolCall:${toolCallId}:${toolName}`
+      }
+      return `toolCall:${JSON.stringify(part)}`
+    }
+    default:
+      return 'unknown'
+  }
 }
 
 function shouldSkipStaleRunEvent(
@@ -369,3 +497,197 @@ function shouldSkipStaleRunEvent(
 
   return false
 }
+
+function findStreamMessageIndex(
+  messages: Array<GatewayMessage>,
+  targetMessage: GatewayMessage,
+  streamRunId: string,
+): number {
+  const targetId = getMessageId(targetMessage)
+  if (targetId) {
+    const byId = messages.findIndex((message) => getMessageId(message) === targetId)
+    if (byId >= 0) return byId
+  }
+
+  const targetRole = normalizeString(targetMessage.role)
+  const targetToolCallId = normalizeString(targetMessage.toolCallId)
+  let index = -1
+  messages.forEach((message, currentIndex) => {
+    const runId = normalizeString((message as { __streamRunId?: unknown }).__streamRunId)
+    if (!runId || runId !== streamRunId) return
+    if (normalizeString(message.role) !== targetRole) return
+    const messageToolCallId = normalizeString(message.toolCallId)
+    if (targetToolCallId || messageToolCallId) {
+      if (targetToolCallId !== messageToolCallId) return
+    }
+    index = currentIndex
+  })
+  return index
+}
+
+function getMessageId(message: GatewayMessage): string {
+  return normalizeString((message as { id?: unknown }).id)
+}
+
+function normalizeString(value: unknown): string {
+  return typeof value === 'string' ? value.trim() : ''
+}
+
+function shouldSkipDuplicateEvent(
+  seen: Set<string>,
+  source: 'agent' | 'chat',
+  runId: string,
+  state: string | undefined,
+  seq: number | undefined,
+): boolean {
+  if (!runId || typeof seq !== 'number') return false
+  const key = `${source}:${runId}:${state ?? ''}:${seq}`
+  if (seen.has(key)) return true
+  seen.add(key)
+  if (seen.size > 4000) {
+    seen.clear()
+  }
+  return false
+}
+
+function extractChatPayloadsFromAgentPayload(
+  payload: unknown,
+): Array<StreamChatPayload | null> {
+  if (!payload || typeof payload !== 'object') return []
+  const value = payload as Record<string, unknown>
+  const runId = normalizeString(value.runId)
+  const sessionKey = normalizeString(value.sessionKey)
+  const stream = normalizeString(value.stream)
+  const seq =
+    typeof value.seq === 'number' && Number.isFinite(value.seq)
+      ? value.seq
+      : undefined
+  const data =
+    value.data && typeof value.data === 'object'
+      ? (value.data as Record<string, unknown>)
+      : null
+
+  if (stream === 'assistant') {
+    const text =
+      normalizeString(data?.text) || normalizeString(data?.delta) || ''
+    if (!text) return []
+    return [
+      {
+        runId,
+        sessionKey,
+        state: 'delta',
+        seq,
+        message: {
+          role: 'assistant',
+          content: [{ type: 'text', text }],
+        },
+      },
+    ]
+  }
+
+  if (stream === 'thinking') {
+    const thinking =
+      normalizeString(data?.thinking) || normalizeString(data?.text) || ''
+    if (!thinking) return []
+    return [
+      {
+        runId,
+        sessionKey,
+        state: 'delta',
+        seq,
+        message: {
+          role: 'assistant',
+          content: [{ type: 'thinking', thinking }],
+        },
+      },
+    ]
+  }
+
+  if (stream === 'lifecycle') {
+    const phase = normalizeString(data?.phase)
+    if (phase === 'end') {
+      return [
+        {
+          runId,
+          sessionKey,
+          state: 'final',
+          seq,
+        },
+      ]
+    }
+    return []
+  }
+
+  if (stream.includes('tool')) {
+    const toolCallId =
+      normalizeString(data?.toolCallId) ||
+      normalizeString(data?.id) ||
+      normalizeString(data?.callId)
+    const toolName =
+      normalizeString(data?.toolName) || normalizeString(data?.name)
+
+    if (stream.includes('call')) {
+      const partialJson = normalizeString(data?.partialJson)
+      const input =
+        data && typeof data.input === 'object'
+          ? (data.input as Record<string, unknown>)
+          : data && typeof data.arguments === 'object'
+            ? (data.arguments as Record<string, unknown>)
+            : undefined
+
+      return [
+        {
+          runId,
+          sessionKey,
+          state: 'delta',
+          seq,
+          message: {
+            role: 'assistant',
+            content: [
+              {
+                type: 'toolCall',
+                id: toolCallId || undefined,
+                name: toolName || undefined,
+                partialJson: partialJson || undefined,
+                arguments: input,
+              },
+            ],
+          },
+        },
+      ]
+    }
+
+    if (stream.includes('result') || stream.includes('output')) {
+      const output = data?.output
+      const errorText = normalizeString(data?.error)
+      return [
+        {
+          runId,
+          sessionKey,
+          state: 'delta',
+          seq,
+          message: {
+            role: 'toolResult',
+            toolCallId: toolCallId || undefined,
+            toolName: toolName || undefined,
+            details:
+              output && typeof output === 'object'
+                ? (output as Record<string, unknown>)
+                : undefined,
+            isError: Boolean(errorText),
+            content: [
+              {
+                type: 'text',
+                text: errorText || (typeof output === 'string' ? output : ''),
+              },
+            ],
+          },
+        },
+      ]
+    }
+
+    return []
+  }
+
+  return []
+}