fix(client): preserve text appended before the first state snapshot (#6910)

* fix(client): preserve text appended before the first state snapshot

* test(client): exercise first-snapshot reconciliation with stable IDs

---------

Co-authored-by: BaiHeng Peng <191328900+Aldenshile@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -2786,6 +2786,9 @@ are checked recursively as well.
 
 ## Embedded Python Client
 
+`chat()` and `stream()` preserve text appended by model middleware before the
+first state snapshot, emitting only the suffix after the streamed text.
+
 For `DeerFlowClient(agent_name="researcher")`, the named agent's `mcp_plugins`
 selection applies to both the lead agent and its `task` / `batch_task`
 delegations: `null` inherits all enabled MCP plugins, `[]` selects none, and

backend/packages/harness/deerflow/AGENTS.md

@@ -67,6 +67,7 @@ drift.
 
 **Agent Conversation**:
 - `chat(message, thread_id)` — synchronous; returns final AI text.
+- The first `values` snapshot reconciles any suffix appended to already-streamed AI text by a model wrapper; unchanged text and non-prefix replacements are not re-emitted. Later snapshots retain the same append-only rule.
 - `stream(message, thread_id)` — subscribes to LangGraph `stream_mode=["values", "messages", "custom"]` and yields `StreamEvent`:
   - `"values"` — state snapshot (title, messages, artifacts, summary_text). Always forward `summary_text` (current summary or `None`), including unchanged values/resets. Never re-emit AI text delivered via `messages`; serialized `ToolMessage` entries retain non-`None` native `artifact`
   - `"messages-tuple"` — current-turn AI text **deltas** by `id` and each tool call/result once; excludes resumed history and preserves result `artifact`

backend/packages/harness/deerflow/client.py

@@ -1237,13 +1237,19 @@ class DeerFlowClient:
                 if msg_id:
                     seen_messages[msg_id] = msg
 
-                # Already streamed via ``messages`` mode; only (defensively)
-                # capture usage here and skip re-synthesizing the event.
+                # Already streamed via ``messages`` mode; emit only text added
+                # before this first snapshot, plus unsent tool calls/metadata.
                 if msg_id and msg_id in streamed_ids:
                     if isinstance(msg, AIMessage):
                         _account_usage(msg_id, getattr(msg, "usage_metadata", None))
                         additional_kwargs = self._serialize_additional_kwargs(msg)
                         additional_kwargs_delta = _unsent_additional_kwargs(msg_id, additional_kwargs)
+                        text = self._extract_text(msg.content)
+                        sent_text = sent_text_by_id.get(msg_id, "")
+                        if len(text) > len(sent_text) and text.startswith(sent_text):
+                            sent_text_by_id[msg_id] = text
+                            yield self._ai_text_event(msg_id, text[len(sent_text) :], None, additional_kwargs_delta)
+                            additional_kwargs_delta = None
                         if msg_id in pending_tool_call_ids and msg.tool_calls:
                             pending_tool_call_ids.discard(msg_id)
                             yield self._ai_tool_calls_event(msg_id, msg.tool_calls, additional_kwargs_delta)

backend/tests/test_client.py

@@ -811,6 +811,28 @@ class TestStream:
 
         assert [event.data["content"] for event in _ai_events(events)] == ["Checking again.", "\n\n[FORCED STOP] Repeated tool calls exceeded the safety limit."]
 
+    @pytest.mark.parametrize("snapshot_text, expected_delta", [("Hello\n\n附加说明", "\n\n附加说明"), ("Hello", ""), ("Rewritten answer", "")])
+    def test_first_streamed_snapshot_preserves_text_and_metadata_once(self, client, snapshot_text, expected_delta):
+        call = {"name": "bash", "args": {"command": "ls"}, "id": "call-1"}
+        message = AIMessage(content=snapshot_text, id="ai-1", tool_calls=[call], additional_kwargs={"note": "added"})
+        agent = _make_agent_mock(
+            [
+                ("messages", (AIMessageChunk(content="Hello", id="ai-1"), {})),
+                ("messages", (AIMessageChunk(content="", id="ai-1", tool_call_chunks=[{"name": "bash", "args": '{"command":"ls"}', "id": "call-1", "index": 0}]), {})),
+                ("values", {"messages": [message]}),
+                ("values", {"messages": [message.model_copy()]}),
+            ]
+        )
+        with patch.object(client, "_ensure_agent"), patch.object(client, "_agent", agent):
+            events = list(client.stream("hi", thread_id="t-first-streamed-snapshot"))
+
+        ai_events = [event.data for event in events if event.type == "messages-tuple" and event.data.get("type") == "ai"]
+        assert "".join(event["content"] for event in ai_events) == "Hello" + expected_delta
+        first_snapshot_index = next(index for index, event in enumerate(events) if event.type == "values")
+        assert "".join(event.data.get("content", "") for event in events[:first_snapshot_index]) == "Hello" + expected_delta
+        assert [event["additional_kwargs"] for event in ai_events if event.get("additional_kwargs")] == [{"note": "added"}]
+        assert [event["tool_calls"] for event in ai_events if event.get("tool_calls")] == [[call]]
+
     def test_stream_does_not_resend_a_replacement_that_does_not_extend_the_sent_text(self, client):
         """Only appended text is sent; a rewrite of what was already sent would duplicate output."""
         sent = AIMessage(content="Let me look.", id="ai-1")
@@ -1260,6 +1282,26 @@ class TestStream:
 
 
 class TestChat:
+    @pytest.mark.parametrize("suffix", ["", "\n\n附加说明"])
+    def test_returns_text_appended_before_the_first_values_snapshot(self, client, suffix):
+        """A model wrapper can extend streamed text before the model node commits."""
+        from langchain.agents import create_agent
+        from langchain.agents.middleware import AgentMiddleware, ModelResponse
+        from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
+
+        class AppendTextMiddleware(AgentMiddleware):
+            def wrap_model_call(self, request, handler):
+                response = handler(request)
+                message = response.result[0]
+                return ModelResponse(result=[message.model_copy(update={"content": message.content + suffix})])
+
+        # A stable ID connects streamed text to the first committed snapshot;
+        # id-less responses would bypass reconciliation and mask a regression.
+        model = FakeMessagesListChatModel(responses=[AIMessage(content="Hello", id="ai-1")])
+        graph = create_agent(model=model, middleware=[AppendTextMiddleware()], state_schema=ThreadState)
+        with patch.object(client, "_ensure_agent"), patch.object(client, "_agent", graph):
+            assert client.chat("hi", thread_id="t-first-snapshot") == "Hello" + suffix
+
     @pytest.mark.parametrize(
         ("events", "expected"),
         [