From b650456c6d99526d3e28204b39bb4686d8d95daf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E4=BA=91=E9=BE=99?= <76432572+nankingjing@users.noreply.github.com> Date: Sun, 12 Jul 2026 12:13:02 +0800 Subject: [PATCH] fix(streaming): drop silent delta-discard in _merge_stream_text (#4085) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- backend/app/channels/manager.py | 15 +++-- .../harness/deerflow/tui/view_state.py | 22 ++++--- backend/tests/test_channels.py | 59 +++++++++++++++++++ backend/tests/test_tui_view_state.py | 44 ++++++++++++++ 4 files changed, 127 insertions(+), 13 deletions(-) diff --git a/backend/app/channels/manager.py b/backend/app/channels/manager.py index ba8f47fae..0bbd8ee50 100644 --- a/backend/app/channels/manager.py +++ b/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 diff --git a/backend/packages/harness/deerflow/tui/view_state.py b/backend/packages/harness/deerflow/tui/view_state.py index 178e55b99..cadc17bfe 100644 --- a/backend/packages/harness/deerflow/tui/view_state.py +++ b/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 diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index a9124e9c3..1775044bc 100644 --- a/backend/tests/test_channels.py +++ b/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" diff --git a/backend/tests/test_tui_view_state.py b/backend/tests/test_tui_view_state.py index e06250f78..ade07c43d 100644 --- a/backend/tests/test_tui_view_state.py +++ b/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"