From e7d96fe05110c4aac4e03242c16ba1a65ef8842f Mon Sep 17 00:00:00 2001 From: Hyeonsang Cho Date: Fri, 23 Oct 2026 19:29:08 +0900 Subject: [PATCH] fix(tui): append streamed assistant deltas verbatim (#7325) * fix(tui): append streamed assistant deltas verbatim The reducer merged each delta by comparing it with the row text: an equal delta was a no-op, a delta starting with the row replaced it, and a delta the row started with was dropped. Those rules absorbed re-sends, replayed history and cumulative snapshots, none of which DeerFlowClient.stream emits any more, so they only deleted real tokens ("ha" + "ha" -> "ha", "1" + "10" -> "10") and every standalone Markdown marker an answer opened with (closing **, list bullets, table pipes, closing code fences). Append every delta on both the id-keyed and anonymous paths, and drop _merge_stream_text. Rewrite the tests that pinned the re-send premise as append contracts, run the merge cases through reduce() on both paths, and drive the real client stream into the reducer. * docs(changelog): note TUI streamed-delta fix (#7325) * docs(changelog): drop trailing spaces after the #7325 entries --------- Co-authored-by: Willem Jiang --- CHANGELOG.md | 7 ++ CHANGELOG_zh.md | 6 + .../packages/harness/deerflow/tui/AGENTS.md | 2 +- .../harness/deerflow/tui/view_state.py | 50 ++------ backend/tests/test_tui_render.py | 8 +- backend/tests/test_tui_runtime.py | 40 +++++- backend/tests/test_tui_view_state.py | 115 +++++++----------- 7 files changed, 112 insertions(+), 116 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5e567af2df..d69007c87d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1397,6 +1397,12 @@ As of 2026-10-21, the milestone includes **829 merged pull requests**. ### Fixed +- **tui:** The streamed transcript no longer drops tokens that repeat or extend + the text already shown. Deltas were merged by prefix, so `ha` + `ha` rendered + as `ha`, `1` + `10` as `10`, and an answer that opened with a Markdown marker + lost every later standalone copy of it (closing `**`, list bullets, table + pipes, closing code fences). Each delta is now appended as sent. ([#7325]) + - **sandbox:** Release an E2B VM's ownership lease when acquire finds it gone (a dead cached client or a warm-pool VM reaped while parked). The lease was kept and renewed until Gateway shutdown, so every reaped VM left a tracked @@ -13020,3 +13026,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#7255]: https://github.com/bytedance/deer-flow/pull/7255 [#7258]: https://github.com/bytedance/deer-flow/pull/7258 [#7277]: https://github.com/bytedance/deer-flow/pull/7277 +[#7325]: https://github.com/bytedance/deer-flow/pull/7325 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index bce6271462..20e122a09a 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -1112,6 +1112,11 @@ ### 修复 +- **TUI：** 流式对话记录不再丢弃与已显示文本重复或在其基础上延伸的 token。此前增量 + 按前缀合并，`ha` + `ha` 显示为 `ha`，`1` + `10` 显示为 `10`，以 Markdown 标记开头的 + 回答还会丢失之后单独出现的同一标记（闭合的 `**`、列表符号、表格竖线、闭合代码块）。 + 现在每个增量都按原样追加。([#7325]) + - **沙箱：** acquire 发现 E2B VM 已不存在（缓存客户端已失效，或暖池中的 VM 在停放期间被 回收）时，释放该 VM 的所有权租约。此前租约会一直保留并续约到 Gateway 关闭，每个被回收的 VM 都会留下一个跟踪 ID 和一个持续续约的所有权键。([#7277]) @@ -10393,3 +10398,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#7255]: https://github.com/bytedance/deer-flow/pull/7255 [#7258]: https://github.com/bytedance/deer-flow/pull/7258 [#7277]: https://github.com/bytedance/deer-flow/pull/7277 +[#7325]: https://github.com/bytedance/deer-flow/pull/7325 diff --git a/backend/packages/harness/deerflow/tui/AGENTS.md b/backend/packages/harness/deerflow/tui/AGENTS.md index f5c24a969c..4cb37674a8 100644 --- a/backend/packages/harness/deerflow/tui/AGENTS.md +++ b/backend/packages/harness/deerflow/tui/AGENTS.md @@ -4,7 +4,7 @@ A terminal-native UI over the embedded harness, exposed as the `deerflow` consol **Module layout** (all layers except `app.py` and `widgets/` are pure / Textual-free and unit-tested directly): - `cli.py` — `plan_launch()` (pure launch-mode decision) + headless `--print` / `--json` + `main()` entry point. Headless modes exit `1` when the final AI message carries `deerflow_error_fallback` (provider failures arrive as flagged messages, not exceptions). `_RunOutcome` and `chat()` share `_AIMessageAccumulator`: deltas accumulate per message id and the last content-bearing AI id supplies the text; metadata-only events preserve that id. Fallback writers must supply text or flag an id whose text was already emitted. Keep the client import lazy so help and launch planning do not load the agent graph. TTY → TUI, else headless help. `--tui-transparent` / `DEER_FLOW_TUI_TRANSPARENT` opt into terminal-default backgrounds without changing the solid-theme default. Uses an **absolute** `from deerflow.tui.app import run_tui` so the `app.py` module name doesn't trip `test_harness_boundary.py` (which records relative import module names verbatim). -- `view_state.py` — `ViewState` + `reduce(state, action)`, the testable heart. Rows: user / assistant / tool / system. Title captured from `values` events. +- `view_state.py` — `ViewState` + `reduce(state, action)`, the testable heart. Rows: user / assistant / tool / system. Title captured from `values` events. Assistant deltas append verbatim: the client sends only incremental text per id, so a delta that repeats or extends the row (`ha` + `ha`, a closing `**`) is new text, not a re-send. - `runtime.py` — `translate(StreamEvent) -> [Action]` (pure) + `stream_actions()` which brackets a run with `RunStarted`/`RunEnded` and turns model errors into an `AssistantError` row. - `message_format.py` / `command_registry.py` / `input_history.py` / `render.py` / `theme.py` — pure helpers (tool summaries, slash registry + `resolve()`, ↑/↓ history, Rich renderers). The command registry must exclude every shared `RESERVED_SLASH_SKILL_NAMES` entry that the agent runtime rejects from both its picker and resolver; the `context` skill is the exception for ordinary task text, while the exact composer-only `/context compact` alias remains unavailable as a skill activation. - `app.py` — Textual `App`. Runs `DeerFlowClient.stream()` (sync) on a worker thread and marshals actions to the UI thread via `call_from_thread`. Slash palette with `/goal` management + model/thread modal pickers; routes idle display-only `/clear` through `ClearRows` without replacing the active thread, and blocks state-resetting local commands like `/new` and `/clear` with the standard "Still working" message during an active run; priority key bindings gated by `check_action` so they never steal keys from overlays or the composer. Application-level PageUp/PageDown bindings scroll the transcript while preserving composer focus; streaming follows output only while the transcript remains at the bottom. diff --git a/backend/packages/harness/deerflow/tui/view_state.py b/backend/packages/harness/deerflow/tui/view_state.py index 22aa0850a8..d23846e471 100644 --- a/backend/packages/harness/deerflow/tui/view_state.py +++ b/backend/packages/harness/deerflow/tui/view_state.py @@ -171,8 +171,7 @@ def reduce(state: ViewState, action: Action) -> ViewState: return _append(state, UserRow(text=action.text)) if isinstance(action, RunStarted): - # New turn: no message is actively streaming yet (the client re-emits - # prior messages first; those must not be treated as the active one). + # New turn: no message is actively streaming until its first delta. return replace(state, streaming=True, streaming_id=None, streaming_anonymous_row_index=None) if isinstance(action, RunEnded): @@ -212,10 +211,8 @@ def _apply_assistant_delta(state: ViewState, action: AssistantDelta) -> ViewStat """Update the assistant row for this delta, or start a new one. A genuine (non-empty) id is matched anywhere in the transcript, not just - the most recent assistant row: on a thread with history, the client - re-emits every prior message on each new turn (its dedup is per-turn), and - a re-emitted *older* message can arrive after a newer one has started — so - matching only the tail row would duplicate prior answers. + the most recent assistant row: every delta for one message belongs to that + message's row, even after other rows have been appended behind it. An empty id ("" — some providers/paths never stamp per-chunk ids, see ``runtime._as_str``) is NOT a reliable key for that same scan: unlike a @@ -226,12 +223,11 @@ def _apply_assistant_delta(state: ViewState, action: AssistantDelta) -> ViewStat deltas are therefore routed to `_apply_assistant_delta_anonymous`, which tracks "this turn's" row by position instead of by id. - Updates also merge by content rather than blindly concatenating, to absorb - full re-sends / cumulative snapshots vs. genuine incremental deltas: - - * new text == accumulated, or starts with it -> cumulative/re-send: replace - * accumulated starts with new text -> stale/shorter re-send: keep - * otherwise -> a real delta: append + Every update is appended verbatim. ``DeerFlowClient.stream`` emits only + incremental text per message id: it never re-sends delivered text, + replays prior turns, or sends a cumulative snapshot. A delta that repeats + or extends the accumulated text is therefore still new text (``"ha"`` + + ``"ha"``, or a closing ``"**"`` after an opening one), never a re-send. """ if not action.id: return _apply_assistant_delta_anonymous(state, action) @@ -242,14 +238,7 @@ 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: - # 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) + rows[i] = replace(row, text=row.text + action.text) return _mark_streaming(replace(state, rows=tuple(rows)), action.id) return _mark_streaming(_append(state, AssistantRow(text=action.text, id=action.id)), action.id) @@ -282,12 +271,9 @@ def _apply_assistant_delta_anonymous(state: ViewState, action: AssistantDelta) - if index is not None and index == len(state.rows) - 1: row = state.rows[index] if isinstance(row, AssistantRow) and not row.error: - # Same no-op / merge semantics as the id-keyed path above. - if row.text == action.text and len(action.text) > 1: - return state + # Same append-only semantics as the id-keyed path above. rows = list(state.rows) - merged = _merge_stream_text(row.text, action.text) - rows[index] = replace(row, text=merged) + rows[index] = replace(row, text=row.text + action.text) return _mark_streaming_anonymous(replace(state, rows=tuple(rows)), index) new_state = _append(state, AssistantRow(text=action.text, id=action.id)) @@ -319,20 +305,6 @@ def _mark_streaming_anonymous(state: ViewState, index: int) -> ViewState: return state -def _merge_stream_text(existing: str, incoming: str) -> str: - if not existing: - return incoming - # 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 - - def _apply_tool_started(state: ViewState, action: ToolStarted) -> ViewState: """Create or update a tool card, de-duplicated by ``tool_call_id``. diff --git a/backend/tests/test_tui_render.py b/backend/tests/test_tui_render.py index f17b8be905..fe45191a7a 100644 --- a/backend/tests/test_tui_render.py +++ b/backend/tests/test_tui_render.py @@ -68,15 +68,13 @@ def test_actively_streaming_assistant_stays_plain(): def test_prior_message_stays_markdown_when_a_followup_run_starts(): # Regression: sending a follow-up must NOT revert a finalized markdown answer - # to raw text. Between RunStarted and the new answer's first delta (and during - # the client's re-emit of prior messages), the previous answer is the last - # assistant row — it must still render as Markdown. + # to raw text. Between RunStarted and the new answer's first delta, the + # previous answer is the last assistant row — it must still render as Markdown. state = initial_state() state = reduce(state, AssistantDelta(id="m1", text="**bold answer**")) state = reduce(state, RunEnded()) state = reduce(state, UserSubmitted("follow up")) - state = reduce(state, RunStarted()) - state = reduce(state, AssistantDelta(id="m1", text="**bold answer**")) # re-emit, no new answer yet + state = reduce(state, RunStarted()) # no new answer yet out = _render_to_text(render_transcript(state)) assert "**bold answer**" not in out # still Markdown-rendered diff --git a/backend/tests/test_tui_runtime.py b/backend/tests/test_tui_runtime.py index 1b607db1fc..3178eadcc8 100644 --- a/backend/tests/test_tui_runtime.py +++ b/backend/tests/test_tui_runtime.py @@ -4,7 +4,13 @@ The translation layer is pure and is exercised here against real ``StreamEvent`` objects plus a fake client, with no Textual involved. """ -from deerflow.client import StreamEvent +from unittest.mock import MagicMock, patch + +from langchain_core.messages import AIMessage, AIMessageChunk, HumanMessage + +from deerflow.client import DeerFlowClient, StreamEvent +from deerflow.config.authorization_config import AuthorizationConfig +from deerflow.config.database_config import DatabaseConfig from deerflow.tui.runtime import stream_actions, translate from deerflow.tui.view_state import ( AssistantDelta, @@ -176,3 +182,35 @@ def test_stream_actions_two_turns_with_none_ids_produce_separate_rows(): assert len(assistants) == 2 assert assistants[0].text == "First turn answer." assert assistants[1].text == "Second turn answer." + + +def test_real_client_stream_renders_the_streamed_answer_verbatim(): + """Drive the real ``DeerFlowClient.stream`` into the reducer: its token + deltas, including ones that repeat or extend earlier text, plus the + closing values snapshot, must render exactly the model's answer.""" + chunks = ["**", "Note", "**", ": ", "ha", "ha", " ", "1", "10", " apples"] + answer = "".join(chunks) + + def stream_turn(_state, *, context, **_kwargs): + user = HumanMessage(content="q", id="h-1", additional_kwargs={"run_id": context["run_id"]}) + yield ("values", {"messages": [user]}) + for chunk in chunks: + yield ("messages", (AIMessageChunk(content=chunk, id="ai-1"), {})) + yield ("values", {"messages": [user, AIMessage(content=answer, id="ai-1")]}) + + config = MagicMock() + config.models = [] + config.token_usage.enabled = False + config.database = DatabaseConfig(backend="memory") + config.authorization = AuthorizationConfig(enabled=False) + with patch("deerflow.client.get_app_config", return_value=config): + client = DeerFlowClient() + agent = MagicMock() + agent.stream.side_effect = stream_turn + + state = initial_state() + with patch.object(client, "_ensure_agent"), patch.object(client, "_agent", agent): + for action in stream_actions(client, "q", thread_id="t-1"): + state = reduce(state, action) + + assert [(r.kind, r.text) for r in state.rows] == [("assistant", answer)] diff --git a/backend/tests/test_tui_view_state.py b/backend/tests/test_tui_view_state.py index 333f044159..78a95a9c16 100644 --- a/backend/tests/test_tui_view_state.py +++ b/backend/tests/test_tui_view_state.py @@ -4,6 +4,8 @@ The reducer is the testable heart of the TUI: a pure function mapping (state, action) -> state, with no Textual / rendering dependency. """ +import pytest + from deerflow.tui.view_state import ( AssistantDelta, AssistantError, @@ -15,7 +17,6 @@ from deerflow.tui.view_state import ( ToolResult, ToolStarted, UserSubmitted, - _merge_stream_text, initial_state, reduce, ) @@ -96,27 +97,28 @@ def test_tool_result_without_call_id_is_ignored(): assert state.rows == () -# --- streaming robustness: the client can re-emit the same id (values -# re-synthesis) and stream partial tool-call chunks. The reducer must not -# duplicate text or tool cards. --- +# --- streaming: DeerFlowClient.stream emits only incremental text per id +# (no re-sends, replayed turns or cumulative snapshots), and partial +# tool-call chunks. The reducer appends text and de-duplicates tool cards. --- -def test_assistant_delta_skips_full_resend_of_same_id(): +def test_assistant_delta_repeating_the_accumulated_text_appends(): state = initial_state() - state = reduce(state, AssistantDelta(id="m1", text="Hey there!\nWhat's up?")) - state = reduce(state, AssistantDelta(id="m1", text="Hey there!\nWhat's up?")) + state = reduce(state, AssistantDelta(id="m1", text="ha")) + state = reduce(state, AssistantDelta(id="m1", text="ha")) assert [r.kind for r in state.rows] == ["assistant"] - assert state.rows[0].text == "Hey there!\nWhat's up?" + assert state.rows[0].text == "haha" -def test_assistant_delta_treats_cumulative_snapshot_as_replace(): +def test_assistant_delta_extending_the_accumulated_text_appends(): state = initial_state() - state = reduce(state, AssistantDelta(id="m1", text="Hel")) - state = reduce(state, AssistantDelta(id="m1", text="Hel lo world")) - assert state.rows[0].text == "Hel lo world" + state = reduce(state, AssistantDelta(id="m1", text="1")) + state = reduce(state, AssistantDelta(id="m1", text="10")) + state = reduce(state, AssistantDelta(id="m1", text=" apples")) + assert state.rows[0].text == "110 apples" -def test_streaming_id_tracks_active_message_not_reemitted_history(): +def test_streaming_id_tracks_the_message_receiving_deltas_this_turn(): state = initial_state() state = reduce(state, AssistantDelta(id="m1", text="answer one")) state = reduce(state, RunEnded()) @@ -124,28 +126,22 @@ def test_streaming_id_tracks_active_message_not_reemitted_history(): state = reduce(state, RunStarted()) assert state.streaming_id is None # new turn: nothing actively streaming yet - state = reduce(state, AssistantDelta(id="m1", text="answer one")) # re-emit (no-op) - assert state.streaming_id is None # re-emit of history must not mark active - state = reduce(state, AssistantDelta(id="m2", text="new content")) # real new content + state = reduce(state, AssistantDelta(id="m2", text="new content")) assert state.streaming_id == "m2" state = reduce(state, RunEnded()) assert state.streaming_id is None -def test_assistant_resend_of_older_message_updates_in_place_not_duplicated(): - # On a thread with history, the client re-emits every prior message on each - # new turn. A values snapshot can re-emit an OLDER message's full text AFTER - # a newer message has already started — the reducer must update the old row - # by id, not append a verbatim duplicate at the end. +def test_assistant_delta_for_an_older_id_updates_that_row_in_place(): state = initial_state() state = reduce(state, AssistantDelta(id="m1", text="First answer.")) state = reduce(state, UserSubmitted("second question")) state = reduce(state, AssistantDelta(id="m2", text="Second answer.")) - state = reduce(state, AssistantDelta(id="m1", text="First answer.")) # re-emit of old m1 + state = reduce(state, AssistantDelta(id="m1", text=" (stopped)")) assistants = [r for r in state.rows if r.kind == "assistant"] - assert [a.text for a in assistants] == ["First answer.", "Second answer."] + assert [a.text for a in assistants] == ["First answer. (stopped)", "Second answer."] def test_tool_started_dedupes_by_call_id(): @@ -216,46 +212,36 @@ def test_reduce_is_pure_does_not_mutate_input_state(): # --------------------------------------------------------------------------- -# _merge_stream_text regression: CJK reduplication and repeated-token deltas +# Deltas are appended verbatim on both the id-keyed and anonymous paths, even +# when one repeats, extends or prefixes the accumulated text. # --------------------------------------------------------------------------- -def test_merge_stream_text_cjk_reduplication_not_dropped(): - """Two identical CJK tokens must both accumulate, not collapse to one.""" - assert _merge_stream_text("谢", "谢") == "谢谢" +@pytest.mark.parametrize("message_id", ["m1", ""], ids=["id", "anonymous"]) +@pytest.mark.parametrize( + "chunks", + [ + pytest.param(["谢", "谢"], id="cjk-reduplication"), + pytest.param(["go", "go"], id="repeated-token"), + pytest.param(["ha", "ha", "!"], id="repeated-multichar-token"), + pytest.param(["hel", "l", "o"], id="suffix-matching-tail"), + pytest.param(["1", "10", " apples"], id="extends-accumulated"), + pytest.param(["\n", "\n"], id="newline-split"), + pytest.param(["Hello ", "world"], id="plain-append"), + pytest.param(["**", "Note", "**", ": done"], id="markdown-bold-close"), + pytest.param(["-", " a", "\n", "-", " b"], id="markdown-list-bullet"), + pytest.param(["```", "py", "\nx = 1\n", "```"], id="markdown-fence-close"), + pytest.param(["|", " a ", "|", " b ", "|"], id="markdown-table-pipe"), + ], +) +def test_assistant_deltas_accumulate_verbatim(message_id, chunks): + state = reduce(initial_state(), RunStarted()) + for chunk in chunks: + state = reduce(state, AssistantDelta(id=message_id, text=chunk)) + state = reduce(state, RunEnded()) - -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" + assistants = [r for r in state.rows if r.kind == "assistant"] + assert [a.text for a in assistants] == ["".join(chunks)] # --------------------------------------------------------------------------- @@ -365,17 +351,6 @@ def test_assistant_delta_empty_id_does_not_disturb_legitimate_id_sequence(): assert [a.text for a in assistants] == ["anonymous turn", "Hello"] -def test_assistant_delta_empty_id_resend_within_turn_is_noop(): - """Same multi-char no-op re-send semantics apply to the anonymous path.""" - state = initial_state() - state = reduce(state, RunStarted()) - state = reduce(state, AssistantDelta(id="", text="Hey there!")) - state = reduce(state, AssistantDelta(id="", text="Hey there!")) - assistants = [r for r in state.rows if r.kind == "assistant"] - assert len(assistants) == 1 - assert assistants[0].text == "Hey there!" - - def test_clear_rows_resets_anonymous_streaming_index(): """A stale anonymous-row index must not resurrect after ClearRows.""" state = initial_state()