fix(streaming): add delta_only mode so pure-delta chunks are never dropped (fixes #9)

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))

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):

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]