From fbbd2d0426cef050d8fa0c52505faa190edfd6d0 Mon Sep 17 00:00:00 2001 From: Xuxchloris <7482714452@qq.com> Date: Fri, 14 Aug 2026 19:04:26 +0000 Subject: [PATCH] fix(streaming): add delta_only mode so pure-delta chunks are never dropped (fixes #9) --- .../outbound/streaming/markdown_stream.py | 7 ++- .../channel/outbound/streaming/merge_text.py | 12 +++++- .../tests/test_streaming_primitives.py | 43 ++++++++++++++++++- 3 files changed, 59 insertions(+), 3 deletions(-) diff --git a/lark_channel/channel/outbound/streaming/markdown_stream.py b/lark_channel/channel/outbound/streaming/markdown_stream.py index ff72570..1552a1a 100644 --- a/lark_channel/channel/outbound/streaming/markdown_stream.py +++ b/lark_channel/channel/outbound/streaming/markdown_stream.py @@ -54,6 +54,7 @@ def __init__( min_chars: int = 50, initial_text: str = INITIAL_TEXT, element_id: str = ELEMENT_ID, + delta_only: bool = False, ) -> None: self._to = to self._rit = receive_id_type @@ -66,6 +67,10 @@ def __init__( self._finish_streaming_card = finish_streaming_card self._initial_text = initial_text self._element_id = element_id + # Pure-delta producers (each chunk is brand-new text) must not have + # their chunks reinterpreted as rewinds/overlaps — that silently drops + # characters (issue #9). Opt in via delta_only=True. + self._delta_only = delta_only self._card_id: Optional[str] = None self._message_id: str = "" @@ -92,7 +97,7 @@ async def append(self, chunk: str) -> None: if not chunk: return await self._ensure_started() - new_content = merge_streaming_text(self._content, chunk) + new_content = merge_streaming_text(self._content, chunk, delta_only=self._delta_only) delta = len(new_content) - len(self._content) self._content = new_content self._throttle.note(max(1, delta)) diff --git a/lark_channel/channel/outbound/streaming/merge_text.py b/lark_channel/channel/outbound/streaming/merge_text.py index 161d44a..36be5dc 100644 --- a/lark_channel/channel/outbound/streaming/merge_text.py +++ b/lark_channel/channel/outbound/streaming/merge_text.py @@ -14,14 +14,24 @@ return prev # else: compute the largest suffix of prev that is also a prefix of next, # drop that prefix from next, and concatenate. + +For **pure-delta** producers none of the heuristics apply — a chunk that +happens to be a prefix of the accumulated text (e.g. the next digit `4` after +`40`, or a repeated trailing `0`) is brand-new content, not a rewind or an +overlap. Passing ``delta_only=True`` makes the merge a plain concatenation so +characters are never dropped (issue #9). """ -def merge_streaming_text(prev: str, chunk: str) -> str: +def merge_streaming_text(prev: str, chunk: str, *, delta_only: bool = False) -> str: if not chunk: return prev if not prev: return chunk + if delta_only: + # Pure-delta producers: every chunk is brand-new text. Treating it + # as a rewind or an overlap would silently drop characters. + return prev + chunk if chunk.startswith(prev): return chunk if prev.startswith(chunk): diff --git a/lark_channel/channel/tests/test_streaming_primitives.py b/lark_channel/channel/tests/test_streaming_primitives.py index cce0e0b..5a2334f 100644 --- a/lark_channel/channel/tests/test_streaming_primitives.py +++ b/lark_channel/channel/tests/test_streaming_primitives.py @@ -43,6 +43,25 @@ def test_merge_empty_inputs(): assert merge_streaming_text("hi", "") == "hi" +def test_merge_delta_only_never_drops_characters(): + # Pure-delta producers (issue #9): a chunk that is a prefix of the + # accumulated text (or overlaps its tail) is brand-new content and must + # be concatenated, never treated as a rewind or overlap. + assert merge_streaming_text("40", "4", delta_only=True) == "404" + assert merge_streaming_text("210", "0", delta_only=True) == "2100" + assert merge_streaming_text("2026-07-2", "3", delta_only=True) == "2026-07-23" + assert merge_streaming_text("Hello", " world", delta_only=True) == "Hello world" + assert merge_streaming_text("Hello", "", delta_only=True) == "Hello" + assert merge_streaming_text("", "hi", delta_only=True) == "hi" + + +def test_merge_default_mode_unchanged_by_delta_only(): + # The default (auto) semantics keep the rewind/overlap heuristics. + assert merge_streaming_text("40", "4") == "40" + assert merge_streaming_text("Hello world", "Hello") == "Hello world" + assert merge_streaming_text("Hello wo", "world!") == "Hello world!" + + # ---- UpdateQueue ------------------------------------------------------------ # # UpdateQueue is coalescing: at most 1 running + 1 pending. A burst of @@ -231,7 +250,7 @@ async def finish_streaming_card(card_id, seq): ) -def _mk_controller(deps, *, to="oc_1", rit="chat_id"): +def _mk_controller(deps, *, to="oc_1", rit="chat_id", delta_only=False): cci, scbr, ucec, fsc = deps return MarkdownStreamController( to=to, receive_id_type=rit, reply_to=None, reply_in_thread=None, @@ -240,6 +259,7 @@ def _mk_controller(deps, *, to="oc_1", rit="chat_id"): update_card_element_content=ucec, finish_streaming_card=fsc, min_ms=10, min_chars=3, + delta_only=delta_only, ) @@ -392,3 +412,24 @@ async def producer(s): await ctl.run(producer) last_elements = patched[-1]["body"]["elements"] assert any("generation interrupted" in (e.get("content") or "") for e in last_elements) + + +@pytest.mark.asyncio +async def test_markdown_stream_delta_only_preserves_character_chunks(): + """Pure-delta producer (issue #9): chunks that are prefixes of the + accumulated text must not be dropped — the final card content keeps every + character.""" + state, deps = _make_cardkit_fakes() + ctl = _mk_controller(deps, delta_only=True) + + async def producer(s): + await s.append("40") + await s.append("4") + await s.append("210") + await s.append("0") + + await ctl.run(producer) + + assert len(state["elem_updates"]) >= 1 + assert "404" in state["elem_updates"][-1][2] + assert "2100" in state["elem_updates"][-1][2]