fix(streaming): drop silent delta-discard in _merge_stream_text (#4085)

The dual-mode _merge_stream_text used short-circuits chunk==existing
and existing.endswith(chunk) that silently dropped legitimate delta
chunks in messages-tuple mode:

- CJK reduplication: '谢谢' tokenized as ['谢','谢'] -> buffer stays '谢'
- Repeated tokens: 'gogo' as ['go','go'] -> buffer stays 'go'
- Suffix-matching tails: 'hello' as ['hel','l','o'] -> drops 'l'

Delta payloads must always append; only strictly-longer cumulative
snapshots should replace. The fix gates the cumulative-replace branch
with len(chunk) > len(existing) and removes the dropped-shape guards.

Both copies are fixed:
- app/channels/manager.py (IM streaming to Feishu/Telegram/etc.)
- deerflow/tui/view_state.py (TUI client rendering)

The TUI reduce() caller now handles exact re-sends (values snapshot
re-emitting history) before the merge, and the len>1 guard prevents
single-char CJK deltas from being mistaken for re-sends.

Co-authored-by: Claude <noreply@anthropic.com>

backend/app/channels/manager.py

@@ -370,12 +370,17 @@ def _merge_stream_text(existing: str, chunk: str) -> str:
     """Merge either delta text or cumulative text into a single snapshot."""
     if not chunk:
         return existing
-    if not existing or chunk == existing:
-        return chunk or existing
-    if chunk.startswith(existing):
+    if not existing:
         return chunk
-    if existing.endswith(chunk):
-        return existing
+    # Cumulative re-delivery: strictly longer and starts with existing.
+    if len(chunk) > len(existing) and chunk.startswith(existing):
+        return chunk
+    # Everything else is a delta — always append, even when the delta
+    # happens to match the buffer suffix (e.g. 'hel' + 'l') or equals
+    # the buffer (CJK reduplication: '谢' + '谢' = '谢谢'). Channels feed
+    # only delta ('messages-tuple') events to this function; 'values'
+    # snapshots are consumed via a separate branch, so a same-content
+    # delta (chunk == existing) still represents a fresh token to keep.
     return existing + chunk
 
 

backend/packages/harness/deerflow/tui/view_state.py

@@ -222,11 +222,13 @@ def _apply_assistant_delta(state: ViewState, action: AssistantDelta) -> ViewStat
         # match here anyway — the guard is belt-and-suspenders to keep an error
         # row from being merged into if a future change ever gives it an id.
         if isinstance(row, AssistantRow) and row.id == action.id and not row.error:
-            merged = _merge_stream_text(row.text, action.text)
-            if merged == row.text:
-                # No-op re-send (e.g. a values snapshot re-emitting history) —
-                # don't mark this as the actively-streaming message.
+            # Exact re-send of the same full text (e.g. a values snapshot
+            # re-emitting history after reconnection): no-op.  Only multi-char
+            # matches are treated as re-sends so single-char deltas that happen
+            # to equal the buffer (CJK reduplication) are NOT mistaken for no-ops.
+            if row.text == action.text and len(action.text) > 1:
                 return state
+            merged = _merge_stream_text(row.text, action.text)
             rows[i] = replace(row, text=merged)
             return _mark_streaming(replace(state, rows=tuple(rows)), action.id)
     return _mark_streaming(_append(state, AssistantRow(text=action.text, id=action.id)), action.id)
@@ -242,10 +244,14 @@ def _mark_streaming(state: ViewState, message_id: str) -> ViewState:
 def _merge_stream_text(existing: str, incoming: str) -> str:
     if not existing:
         return incoming
-    if incoming.startswith(existing):
-        return incoming  # cumulative snapshot or exact full re-send
-    if existing.startswith(incoming):
-        return existing  # shorter/stale re-send
+    # Cumulative re-delivery: incoming strictly extends existing.
+    if len(incoming) > len(existing) and incoming.startswith(existing):
+        return incoming
+    # Stale/shorter re-send: existing already contains incoming as a prefix
+    # (e.g. a values snapshot re-emitting history that has already been
+    # accumulated from deltas). Only treat as stale when strictly shorter.
+    if len(existing) > len(incoming) and existing.startswith(incoming):
+        return existing
     return existing + incoming  # genuine incremental delta
 
 

backend/tests/test_channels.py

@@ -7306,3 +7306,62 @@ class TestHandleGoalCommand:
             assert reply == "Failed to set goal."
 
         _run(go())
+
+
+# ---------------------------------------------------------------------------
+# _merge_stream_text regression: CJK reduplication, repeated tokens, suffix
+# matching tails.  Proves that the fixed function does not drop legitimate
+# deltas that happen to match the accumulated buffer or its suffix.
+# Import is deferred because app.channels.manager pulls in fastapi.
+# ---------------------------------------------------------------------------
+
+
+def _get_merge_stream_text():
+    from app.channels.manager import _merge_stream_text
+
+    return _merge_stream_text
+
+
+def test_merge_stream_text_cjk_reduplication():
+    """Two identical CJK tokens ('谢','谢') -> '谢谢', not '谢'."""
+    _merge = _get_merge_stream_text()
+    assert _merge("谢", "谢") == "谢谢"
+
+
+def test_merge_stream_text_repeated_token_append():
+    """Identical repeated tokens ('go','go') -> 'gogo', not 'go'."""
+    _merge = _get_merge_stream_text()
+    assert _merge("go", "go") == "gogo"
+
+
+def test_merge_stream_text_suffix_tail_not_dropped():
+    """Delta equal to buffer suffix ('l' after 'hel') -> 'hell', not 'hel'."""
+    _merge = _get_merge_stream_text()
+    assert _merge("hel", "l") == "hell"
+
+
+def test_merge_stream_text_cumulative_strictly_longer_replaces():
+    """A strictly longer cumulative snapshot that starts with existing replaces it."""
+    _merge = _get_merge_stream_text()
+    assert _merge("Hel", "Hel lo world") == "Hel lo world"
+
+
+def test_merge_stream_text_empty_chunk_noop():
+    _merge = _get_merge_stream_text()
+    assert _merge("Hello", "") == "Hello"
+
+
+def test_merge_stream_text_empty_existing_returns_chunk():
+    _merge = _get_merge_stream_text()
+    assert _merge("", "Hello") == "Hello"
+
+
+def test_merge_stream_text_newline_split():
+    """'\\n\\n' split across two '\\n' deltas accumulates to two newlines."""
+    _merge = _get_merge_stream_text()
+    assert _merge("\n", "\n") == "\n\n"
+
+
+def test_merge_stream_text_normal_append():
+    _merge = _get_merge_stream_text()
+    assert _merge("Hello ", "world") == "Hello world"

backend/tests/test_tui_view_state.py

@@ -14,6 +14,7 @@ from deerflow.tui.view_state import (
     ToolResult,
     ToolStarted,
     UserSubmitted,
+    _merge_stream_text,
     initial_state,
     reduce,
 )
@@ -197,3 +198,46 @@ def test_reduce_is_pure_does_not_mutate_input_state():
     # Reducing again must not mutate the previous state object.
     _ = reduce(state, UserSubmitted("second"))
     assert len(state.rows) == before_len
+
+
+# ---------------------------------------------------------------------------
+# _merge_stream_text regression: CJK reduplication and repeated-token deltas
+# ---------------------------------------------------------------------------
+
+
+def test_merge_stream_text_cjk_reduplication_not_dropped():
+    """Two identical CJK tokens must both accumulate, not collapse to one."""
+    assert _merge_stream_text("谢", "谢") == "谢谢"
+
+
+def test_merge_stream_text_repeated_token_not_dropped():
+    """Repeated tokens (e.g. 'go' + 'go') must accumulate."""
+    assert _merge_stream_text("go", "go") == "gogo"
+
+
+def test_merge_stream_text_suffix_matching_tail_not_dropped():
+    """A delta equal to the buffer suffix must append, not be dropped."""
+    assert _merge_stream_text("hel", "l") == "hell"
+
+
+def test_merge_stream_text_cumulative_longer_snapshot_still_works():
+    """A strictly longer chunk starting with existing is a cumulative re-delivery."""
+    assert _merge_stream_text("Hel", "Hel lo world") == "Hel lo world"
+
+
+def test_merge_stream_text_empty_existing_returns_incoming():
+    assert _merge_stream_text("", "Hello") == "Hello"
+
+
+def test_merge_stream_text_empty_incoming_returns_existing():
+    assert _merge_stream_text("Hello", "") == "Hello"
+
+
+def test_merge_stream_text_newline_split_across_chunks():
+    """'\\n\\n' split into two '\\n' deltas must accumulate."""
+    assert _merge_stream_text("\n", "\n") == "\n\n"
+
+
+def test_merge_stream_text_genuine_delta_append():
+    """Normal deltas that don't overlap still append."""
+    assert _merge_stream_text("Hello ", "world") == "Hello world"