From 5475c5070f3e14da7d3cb9a749652b7bac293274 Mon Sep 17 00:00:00 2001 From: Davit Date: Wed, 27 May 2026 14:49:10 +0400 Subject: [PATCH] fixed duplicate messages and polling --- api/src/routes/message/controller.ts | 101 +++++++++------ api/src/services/openclaw/index.ts | 1 + api/src/services/openclaw/jsonlParser.ts | 120 ++++++++++++------ .../message/send/model/useSendMessage.ts | 101 ++++++++++++++- client/src/widgets/chat/model/useChat.ts | 63 ++++++++- client/src/widgets/chat/ui/MessageList.tsx | 8 ++ package.json | 2 +- 7 files changed, 308 insertions(+), 88 deletions(-) diff --git a/api/src/routes/message/controller.ts b/api/src/routes/message/controller.ts index c794316..3faa688 100644 --- a/api/src/routes/message/controller.ts +++ b/api/src/routes/message/controller.ts @@ -46,18 +46,6 @@ const apiPublicUrl = (req: { return envOverride || 'http://localhost:18802'; }; -function stripWrapperTags(text: string): string { - return text - .replace( - /<(?:think|thinking|redacted_thinking)>[\s\S]*?<\/(?:think|thinking|redacted_thinking)>/gi, - '' - ) - .replace(/^<(?:final|output|think|thinking|redacted_thinking)\b[^>]*>/i, '') - .replace(/<\/(?:final|output|think|thinking|redacted_thinking)\s*>\s*$/i, '') - .replace(/<\/[a-z]*\s*$/i, '') - .trim(); -} - const uploadsDir = path.join(__dirname, '../../public/uploads'); if (!fs.existsSync(uploadsDir)) { fs.mkdirSync(uploadsDir, { recursive: true }); @@ -407,33 +395,68 @@ const poll: RequestHandler<{ conversationId: string }, unknown, never, { after?: : []; const actualInserts: typeof toInsert = []; - for (const m of toInsert) { - if (m.role === 'assistant' && m.text) { - const existing = recentAssistants.find( - (r) => - (r.text && m.text.startsWith(r.text)) || - (r.text && r.text.startsWith(m.text)) || - r.text === m.text - ); - if (existing) { - // Update the existing row with the latest text / externalId - const patch: Record = { externalId: m.externalId }; - if (m.text.length >= (existing.text?.length || 0)) patch.text = m.text; - if (m.thinking) patch.thinking = m.thinking; - if (m.toolSteps && m.toolSteps.length > 0) patch.toolSteps = m.toolSteps; - await msgRepo.update( - existing._id, - patch as unknown as Parameters[1] - ); - // Update the row in recentAssistants so subsequent candidates - // can also match against it with the new text. - existing.text = m.text.length >= (existing.text?.length || 0) ? m.text : existing.text; - existing.externalId = m.externalId!; - synced++; - continue; + /* Collect DB writes during the loop and dispatch them concurrently + * after — keeps each iteration sync (no `await`-in-loop lint) and + * lets TypeORM batch the round-trips. The in-memory mutations to + * `recentAssistants` still happen synchronously inside the loop so + * subsequent iterations consult the updated text/externalId; the + * pending DB update for that same row is independent and safe to + * resolve in parallel. + * + * Index-style loop, not `for…of`: the eslint preset (Airbnb) bans + * `for…of` because it expands to a generator under transpile. */ + const pendingUpdates: { id: number; patch: Record }[] = []; + for (let i = 0; i < toInsert.length; i += 1) { + const m = toInsert[i]; + /* `existing` is only resolved for assistant rows with non-empty + * text — user rows never coalesce, and an empty assistant slot + * has nothing to match against. The ternary keeps the lookup + * gated so we can fall through to a single insert/update branch + * below and avoid `continue` (banned by the eslint preset). */ + const existing = + m.role === 'assistant' && m.text + ? recentAssistants.find( + (r) => + (r.text && m.text.startsWith(r.text)) || + (r.text && r.text.startsWith(m.text)) || + r.text === m.text + ) + : undefined; + + if (existing) { + const patch: Record = { externalId: m.externalId }; + /* Two scenarios overwrite the stored text: + * 1. The candidate is longer — normal "in-flight stream + * grew" case, prefer the more complete reply. + * 2. The existing row is exactly K copies of the candidate + * — legacy row saved before the JSONL parser learned to + * collapse gateway-v4 self-repeats. Without this, rows + * polluted by the K-copy bug stay corrupted forever + * because the K×N stored text "wins" the longer-of test + * every poll. Prefer the canonical 1× form. */ + const existingText = existing.text || ''; + if (m.text.length >= existingText.length) { + patch.text = m.text; + } else if (ocService.isSelfRepeatOf(existingText, m.text)) { + patch.text = m.text; } + if (m.thinking) patch.thinking = m.thinking; + if (m.toolSteps && m.toolSteps.length > 0) patch.toolSteps = m.toolSteps; + pendingUpdates.push({ id: existing._id, patch }); + if (patch.text) existing.text = m.text; + existing.externalId = m.externalId!; + synced += 1; + } else { + actualInserts.push(m); } - actualInserts.push(m); + } + + if (pendingUpdates.length) { + await Promise.all( + pendingUpdates.map((u) => + msgRepo.update(u.id, u.patch as unknown as Parameters[1]) + ) + ); } if (actualInserts.length) { @@ -467,9 +490,7 @@ const poll: RequestHandler<{ conversationId: string }, unknown, never, { after?: * overwrite the stale doubled value so the UI heals on next poll. * * We compare canonical JSON to skip no-op writes. */ - const liveAssistants = jsonlMessages.filter( - (m) => m.role === 'assistant' && m.externalId - ); + const liveAssistants = jsonlMessages.filter((m) => m.role === 'assistant' && m.externalId); if (liveAssistants.length) { const liveIds = liveAssistants.map((m) => m.externalId!); const existing = await msgRepo.find({ diff --git a/api/src/services/openclaw/index.ts b/api/src/services/openclaw/index.ts index 7a498fc..d2c83d9 100644 --- a/api/src/services/openclaw/index.ts +++ b/api/src/services/openclaw/index.ts @@ -5,6 +5,7 @@ export { readFirstUserMessage, extractUserText, extractAssistantText, + isSelfRepeatOf, } from './jsonlParser'; export { listSessions, diff --git a/api/src/services/openclaw/jsonlParser.ts b/api/src/services/openclaw/jsonlParser.ts index 4a25604..07cdba7 100644 --- a/api/src/services/openclaw/jsonlParser.ts +++ b/api/src/services/openclaw/jsonlParser.ts @@ -144,60 +144,76 @@ export function extractUserText(raw: string): string { } /** - * Detect and remove self-repeated text. Gateway v4 can write the same - * assistant text 2-4x concatenated within a single JSONL entry. - * Only deduplicates exact N-copy repeats to avoid false positives. + * Detect and remove self-repeated text. Gateway v4 keeps re-writing the + * same assistant turn into a single JSONL entry across multi-pass + * playback, so a reply that started as N chars ends up stored as K×N for + * some K ≥ 2 (K observed in the wild: 2, 3, 4 … 7+). + * + * The previous heuristic only tried K = 2..6 by trial division, so a + * 7-copy entry slipped through unchanged and the UI rendered the body + * seven times (see openclaw_client #?, screenshot 2026-05-26). It's also + * the same shape the upstream OpenClaw default UI shows when reading the + * raw JSONL — i.e. the duplication is baked into the file, our parser is + * the only line of defence. + * + * Strategy: + * 1. KMP failure function gives us the minimal period of the string in + * O(N). If the string is exactly K copies of a prefix p, then + * n % period === 0 with period < n, and we return p — no matter what + * K is. This handles the 7-copy case (and any larger K future + * gateway versions might emit). + * 2. We keep the existing whitespace-tolerant fuzzy heuristic as a + * fallback for the rare case where copies are separated by a stray + * space/newline that breaks strict periodicity, and extend it to + * K = 8 to match the new strict-mode ceiling. + * + * The 20-char floor on `period` is kept so legitimately repetitive short + * prose ("ha ha ha …", "ok ok ok …") isn't collapsed. */ function deduplicateSelfRepeat(text: string): string { if (!text || text.length < 40) return text; - // Normalize: collapse runs of newlines to single newline for matching, - // but return the original first segment (untouched) on match. - const normalized = text.replace(/\n{2,}/g, '\n'); - for (let n = 2; n <= 6; n++) { - // Try exact division on normalized text - if (normalized.length % n === 0) { - const segLen = normalized.length / n; - const seg = normalized.slice(0, segLen); - let isRepeat = true; - for (let i = 1; i < n; i++) { - if (normalized.slice(i * segLen, (i + 1) * segLen) !== seg) { - isRepeat = false; - break; - } - } - if (isRepeat) { - // Return the original (un-normalized) first segment - // Find where the first copy ends in the original text - const firstCopyEnd = text.indexOf(seg.slice(-20)) + 20; - // Safer: just split by the segment and return first match - return text.slice(0, text.length / n).trim(); - } - } - // Also try with flexible boundaries: check if the first ~1/n of the - // text repeats by searching for it later in the string - const approxLen = Math.floor(text.length / n); - for (let fuzz = -2; fuzz <= 2; fuzz++) { + const n = text.length; + + /* KMP failure function — failure[i] = length of the longest proper + * prefix of text[0..i] that is also a suffix. Standard textbook impl. */ + const failure = new Array(n).fill(0); + for (let i = 1; i < n; i += 1) { + let j = failure[i - 1]; + while (j > 0 && text[i] !== text[j]) j = failure[j - 1]; + if (text[i] === text[j]) j += 1; + failure[i] = j; + } + const period = n - failure[n - 1]; + if (period >= 20 && period < n && n % period === 0) { + return text.slice(0, period).trim(); + } + + /* Whitespace-tolerant fallback: KMP requires strict equality, so a + * stray space between copies defeats it. The brute-force candidate + * scan below tolerates inter-copy whitespace. K = 2..8 covers every + * gateway-v4 case we've observed; legitimate non-repeated text just + * exits the loop without a match. */ + for (let copies = 2; copies <= 8; copies += 1) { + const approxLen = Math.floor(n / copies); + for (let fuzz = -2; fuzz <= 2; fuzz += 1) { const tryLen = approxLen + fuzz; - if (tryLen < 20 || tryLen >= text.length) continue; + if (tryLen < 20 || tryLen >= n) continue; const candidate = text.slice(0, tryLen).trim(); if (!candidate) continue; - // Check if the rest of the text is just repeats of candidate (with whitespace flex) let pos = tryLen; - let copies = 1; - while (pos < text.length) { - // Skip whitespace between copies - while (pos < text.length && /\s/.test(text[pos])) pos++; - if (pos >= text.length) break; + let count = 1; + while (pos < n) { + while (pos < n && /\s/.test(text[pos])) pos += 1; + if (pos >= n) break; if (text.startsWith(candidate, pos)) { - copies++; + count += 1; pos += candidate.length; } else { break; } } - // Allow trailing whitespace const remaining = text.slice(pos).trim(); - if (copies === n && remaining.length === 0) { + if (count === copies && remaining.length === 0) { return candidate; } } @@ -205,6 +221,30 @@ function deduplicateSelfRepeat(text: string): string { return text; } +/** + * True iff `longer` is exactly K ≥ 2 consecutive copies of `shorter`. + * + * Used by the poll handler to recognise legacy DB rows that were saved + * before {@link deduplicateSelfRepeat} learned to collapse K-copy + * gateway-v4 self-repeats. Without this, the row's stored text stays + * stuck at K×N for that conversation forever, because the prefix-match + * dedup in `controller.poll` prefers the longer of (existing, candidate) + * — which is the corrupted one. + */ +export function isSelfRepeatOf(longer: string, shorter: string): boolean { + if (!shorter || !longer) return false; + if (longer.length <= shorter.length) return false; + if (longer.length % shorter.length !== 0) return false; + const k = longer.length / shorter.length; + if (k < 2) return false; + for (let i = 0; i < k; i += 1) { + if (longer.slice(i * shorter.length, (i + 1) * shorter.length) !== shorter) { + return false; + } + } + return true; +} + export function extractAssistantText(raw: string): string { const cleaned = raw .replace(/<\/?final>/gi, '') diff --git a/client/src/features/message/send/model/useSendMessage.ts b/client/src/features/message/send/model/useSendMessage.ts index 2f29390..291161c 100644 --- a/client/src/features/message/send/model/useSendMessage.ts +++ b/client/src/features/message/send/model/useSendMessage.ts @@ -1,12 +1,24 @@ import { useCallback, useRef, useState } from 'react'; import { useAppDispatch } from '../../../../app/store/hooks'; import { API_BASE_URL, baseApi } from '../../../../shared/api'; -import { useGetMessagesQuery, type MessageFile } from '../../../../entities/message'; +import { + messagesApi, + useGetMessagesQuery, + type Message, + type MessageFile, +} from '../../../../entities/message'; interface UseSendMessageArgs { conversationId: string | undefined; refetch: ReturnType['refetch']; hasMessages: boolean; + /** `createdAt` of the newest message in cache when the send begins. The + * post-stream `pollMessages` call uses this as its `after` filter so we + * only pull in genuinely new rows. Without it the server returns the + * full conversation tail (up to 200 rows) and the cache merge silently + * surfaces messages the user had never loaded — they're real DB content, + * just older than the initial `GET /message/conversation/:id` page. */ + lastMessageTs?: string; } export interface SendMessageState { @@ -29,6 +41,7 @@ export function useSendMessage({ conversationId, refetch, hasMessages, + lastMessageTs, }: UseSendMessageArgs): SendMessageState { const [streamingText, setStreamingText] = useState(''); const [streamingThinking, setStreamingThinking] = useState(''); @@ -38,6 +51,11 @@ export function useSendMessage({ const [pendingFilesPreviews, setPendingFilesPreviews] = useState([]); const abortRef = useRef(null); + /* Mirror `lastMessageTs` into a ref so the `send` callback can read the + * latest value without re-creating itself every time a new message lands + * (which would otherwise re-trigger the parent's memoized props chain). */ + const lastMessageTsRef = useRef(lastMessageTs); + lastMessageTsRef.current = lastMessageTs; const dispatch = useAppDispatch(); const abort = useCallback(() => { @@ -136,7 +154,86 @@ export function useSendMessage({ parts.forEach(processLine); } - await refetch(); + /* Pull the assistant turn into the messages cache BEFORE clearing + * the streaming bubble. + * + * Why not `refetch()` anymore: `GET /message/conversation/:id` reads + * the DB only, and assistant persistence was moved out of the chat + * handler into the poll endpoint (dcc4f94) to dedupe gateway-v4 + * multi-pass JSONL writes. So when the SSE stream ends, the DB + * still has the user row only — `refetch()` returns a list without + * the assistant, the streaming bubble unmounts in `finally`, and + * the user sees the message vanish for one full 5 s polling cycle + * until the next `usePollMessagesQuery` tick re-syncs JSONL → DB. + * + * Calling `pollMessages` here does the JSONL → DB sync server-side + * and returns the new rows in the same round-trip; we merge them + * into the `getMessages` cache (same _id dedup as `useChat`'s + * periodic merge) before `finally` clears the streaming UI, so the + * persisted bubble takes over the exact frame the streaming one + * leaves. + * + * One short retry covers the (rare) case where the gateway hasn't + * flushed JSONL by the time `[DONE]` reaches us; the periodic 5 s + * poll remains as the last-resort backstop. */ + const mergePollItems = (items: Message[]) => { + if (items.length === 0) return 0; + let added = 0; + dispatch( + messagesApi.util.updateQueryData( + 'getMessages', + { conversationId, before: undefined }, + (draft) => { + const existing = new Set(draft.items.map((m) => m._id)); + const additions = items.filter((m) => !existing.has(m._id)); + if (additions.length === 0) return; + draft.items = [...draft.items, ...additions]; + draft.total = draft.items.length; + added = additions.length; + } + ) + ); + return added; + }; + + /* `after = lastMessageTs` keeps the server response bounded to rows + * the cache hasn't seen yet. `after: undefined` would return up to + * 200 historical rows; merging them into the cache silently + * surfaces ancient messages that were below the initial 50-row + * fold (e.g. legacy NO_REPLY turns), which the user perceives as + * "previous messages turned into NO_REPLY after sending". */ + const pollAfter = lastMessageTsRef.current; + const fetchPoll = async () => { + try { + const result = await dispatch( + messagesApi.endpoints.pollMessages.initiate( + { conversationId, after: pollAfter }, + { forceRefetch: true } + ) + ).unwrap(); + return result.items; + } catch { + return null; + } + }; + + let assistantSynced = false; + for (let attempt = 0; attempt < 2 && !assistantSynced; attempt += 1) { + if (attempt > 0) await new Promise((r) => setTimeout(r, 400)); + const items = await fetchPoll(); + if (!items) break; + const added = mergePollItems(items); + assistantSynced = + added > 0 && items.some((m) => m.role === 'assistant'); + } + + if (!assistantSynced) { + /* JSONL hadn't caught up — fall back to the legacy refetch so the + * user message at least appears immediately. The next periodic + * poll (≤ 5 s) will fill in the assistant row. */ + await refetch(); + } + if (!hasMessages) { dispatch(baseApi.util.invalidateTags(['Conversation'])); } diff --git a/client/src/widgets/chat/model/useChat.ts b/client/src/widgets/chat/model/useChat.ts index 49f5f26..0d6c1da 100644 --- a/client/src/widgets/chat/model/useChat.ts +++ b/client/src/widgets/chat/model/useChat.ts @@ -1,4 +1,11 @@ -import { useCallback, useEffect, useMemo, useRef, useState } from 'react'; +import { + useCallback, + useEffect, + useLayoutEffect, + useMemo, + useRef, + useState, +} from 'react'; import { useAppDispatch } from '../../../app/store/hooks'; import { messagesApi, @@ -39,6 +46,10 @@ export function useChat(conversationId: string | undefined): ChatState { const messages = useMemo(() => data?.items ?? [], [data?.items]); const hasMore = (data as MessagesResponse | undefined)?.hasMore ?? false; + // Polling: only fetch messages newer than the latest one we have. + // Skip while streaming so SSE flow owns the update. + const lastMessageTs = messages.length > 0 ? messages[messages.length - 1].createdAt : undefined; + const { isStreaming, streamingText, @@ -53,12 +64,9 @@ export function useChat(conversationId: string | undefined): ChatState { conversationId, refetch, hasMessages: messages.length > 0, + lastMessageTs, }); - // Polling: only fetch messages newer than the latest one we have. - // Skip while streaming so SSE flow owns the update. - const lastMessageTs = messages.length > 0 ? messages[messages.length - 1].createdAt : undefined; - const { data: pollData } = usePollMessagesQuery( { conversationId: conversationId!, after: lastMessageTs }, { @@ -139,6 +147,51 @@ export function useChat(conversationId: string | undefined): ChatState { messagesEndRef.current?.scrollIntoView({ behavior: 'smooth' }); }, [streamingText, streamingThinking]); + /* Snap to the bottom when streaming transitions true → false. + * + * Background: with the post-stream `pollMessages` merge in + * `useSendMessage`, the `dispatch(updateQueryData(...))` that appends the + * assistant row is synchronous and gets auto-batched by React 18 with the + * `set*('')` clears in `useSendMessage`'s `finally`. The messages effect + * above therefore runs exactly once, with `pendingUserText` already + * cleared, and its scroll-to-bottom branch never fires. Worse, the + * persisted bubble's height usually differs from the streaming bubble's + * (Thought-Process collapses, markdown re-renders, tool blocks appear), + * so the viewport visibly shifts. + * + * Two layout phases to handle: + * 1. SYNCHRONOUS swap. The streaming bubble unmounts and the persisted + * bubble mounts in the same React commit. `useLayoutEffect` lets us + * adjust scrollTop in that same commit, BEFORE the browser paints — + * so the user never sees the intermediate "shorter content at the + * bottom" frame. `behavior: 'auto'` is mandatory here (smooth would + * reintroduce the visible animation we are trying to hide). + * 2. ASYNC reflows after first mount. On the FIRST send after a hard + * refresh, the persisted bubble's deps (markdown renderer, syntax + * highlighter, image decoders) resolve a frame or two later and + * grow the bubble. Subsequent sends in the same session never hit + * this because those deps are cached — which is exactly the + * "happens once after refresh, then stops" symptom. Re-scrolling at + * 0 / 120 / 400 ms catches all three reflow generations we have + * observed; the cleanup tears them down if the user navigates away + * mid-window. */ + const prevIsStreaming = useRef(isStreaming); + useLayoutEffect(() => { + const wasStreaming = prevIsStreaming.current; + prevIsStreaming.current = isStreaming; + if (!wasStreaming || isStreaming) return undefined; + const snap = () => { + messagesEndRef.current?.scrollIntoView({ behavior: 'auto' }); + }; + snap(); + const t1 = window.setTimeout(snap, 120); + const t2 = window.setTimeout(snap, 400); + return () => { + window.clearTimeout(t1); + window.clearTimeout(t2); + }; + }, [isStreaming]); + const [prevConvId, setPrevConvId] = useState(conversationId); if (prevConvId !== conversationId) { setPrevConvId(conversationId); diff --git a/client/src/widgets/chat/ui/MessageList.tsx b/client/src/widgets/chat/ui/MessageList.tsx index f53332e..7f0caa6 100644 --- a/client/src/widgets/chat/ui/MessageList.tsx +++ b/client/src/widgets/chat/ui/MessageList.tsx @@ -76,6 +76,14 @@ export default function MessageList({ chat }: MessageListProps) { minWidth: 0, overflowY: 'auto', overflowX: 'hidden', + /* Disable browser scroll-anchoring. When the streaming MessageBubble + * unmounts at end-of-turn and the persisted one mounts in its place + * the layout height changes (Thought-Process collapses, markdown + * re-renders, tool-step blocks appear). With anchoring on, the + * browser shifts scrollTop to keep an upper element pinned, which + * the user sees as the chat "jumping up". Anchoring off lets our + * explicit scroll-to-bottom triggers stay authoritative. */ + overflowAnchor: 'none', px: { xs: 2, sm: 2, md: 3 }, py: 2, }} diff --git a/package.json b/package.json index e1f0692..5692a0b 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "openclaw-client", - "version": "2.5.6", + "version": "2.5.7", "description": "Web-based chat interface for OpenClaw AI agents", "private": true, "type": "module",