fixed duplicate messages and polling

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<string, unknown> = { 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<typeof msgRepo.update>[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<string, unknown> }[] = [];
+        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<string, unknown> = { 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<typeof msgRepo.update>[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({

api/src/services/openclaw/index.ts

@@ -5,6 +5,7 @@ export {
   readFirstUserMessage,
   extractUserText,
   extractAssistantText,
+  isSelfRepeatOf,
 } from './jsonlParser';
 export {
   listSessions,

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<number>(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, '')

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<typeof useGetMessagesQuery>['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<MessageFile[]>([]);
 
   const abortRef = useRef<AbortController | null>(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<string | undefined>(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']));
         }

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<Message[]>(() => 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);

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,
       }}

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",