fix(client): emit text a later node appends to an AI message already sent (#5479)

* fix(client): emit text a later node appends to an AI message already sent

Loop detection, the token budget, safety termination, subagent limits and the
terminal-response fallback rewrite the last AI message under the same id in
their own after_model node. stream() skipped every id it had seen, so chat(),
the TUI and --print kept the text from before the rewrite: no [FORCED STOP]
notice, no fallback error, and token_usage_attribution added after the model
node never arrived.

Look again at a known id when a snapshot holds a different message object,
emit only the text appended to what was already sent, and send new
additional_kwargs as the existing metadata-only follow-up.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* test(client): pin that a non-extending replacement is not re-sent

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

backend/packages/harness/deerflow/client.py

@@ -851,7 +851,10 @@ class DeerFlowClient:
         message.  ``values`` events continue to carry full state snapshots
         after each graph node finishes; AI text already delivered via the
         ``messages`` stream is **not** re-synthesized from the snapshot to
-        avoid duplicate deliveries.
+        avoid duplicate deliveries. When a later node replaces a delivered AI
+        message under the same id and appends to its text (a guard's stop
+        notice), only the appended text is emitted, as one more delta for that
+        id; new ``additional_kwargs`` arrive as a metadata-only follow-up.
 
         Why not reuse Gateway's ``run_agent``?
         ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
@@ -969,7 +972,13 @@ class DeerFlowClient:
         if self._agent_name:
             context["agent_name"] = self._agent_name
 
-        seen_ids: set[str] = set()
+        # Last message object seen per id in ``values`` snapshots. A later node,
+        # such as a guard's ``after_model``, can replace a message under the same
+        # id, so a different object for a known id is looked at again.
+        seen_messages: dict[str, Any] = {}
+        # AI text already emitted per id. A replacement that appends to it (a
+        # guard's stop notice) only needs the part that was added.
+        sent_text_by_id: dict[str, str] = {}
         # Cross-mode handoff: ids already streamed via LangGraph ``messages``
         # mode so the ``values`` path skips re-synthesis of the same message.
         streamed_ids: set[str] = set()
@@ -1060,6 +1069,7 @@ class DeerFlowClient:
                     if text:
                         if msg_id:
                             streamed_ids.add(msg_id)
+                            sent_text_by_id[msg_id] = sent_text_by_id.get(msg_id, "") + text
                         additional_kwargs_delta = _unsent_additional_kwargs(msg_id, additional_kwargs)
                         yield self._ai_text_event(
                             msg_id,
@@ -1095,10 +1105,24 @@ class DeerFlowClient:
 
             for msg in messages:
                 msg_id = getattr(msg, "id", None)
-                if msg_id and msg_id in seen_ids:
+                if msg_id and msg_id in seen_messages:
+                    if seen_messages[msg_id] is msg:
+                        continue
+                    seen_messages[msg_id] = msg
+                    if isinstance(msg, AIMessage):
+                        # Replaced after it was sent. Emit text appended to what
+                        # was already sent, plus any new metadata.
+                        text = self._extract_text(msg.content)
+                        sent_text = sent_text_by_id.get(msg_id, "")
+                        additional_kwargs_delta = _unsent_additional_kwargs(msg_id, self._serialize_additional_kwargs(msg))
+                        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)
+                        elif additional_kwargs_delta:
+                            yield self._ai_text_event(msg_id, "", None, additional_kwargs_delta)
                     continue
                 if msg_id:
-                    seen_ids.add(msg_id)
+                    seen_messages[msg_id] = msg
 
                 # Already streamed via ``messages`` mode; only (defensively)
                 # capture usage here and skip re-synthesizing the event.
@@ -1134,6 +1158,8 @@ class DeerFlowClient:
 
                     text = self._extract_text(msg.content)
                     if text:
+                        if msg_id:
+                            sent_text_by_id[msg_id] = text
                         additional_kwargs_delta = None if sent_additional_kwargs else _unsent_additional_kwargs(msg_id, additional_kwargs)
                         yield self._ai_text_event(
                             msg_id,

backend/tests/test_client.py

@@ -704,6 +704,68 @@ class TestStream:
         assert any(event.data.get("content") == "Hello!" for event in ai_events)
         assert any(event.data.get("additional_kwargs", {}).get("token_usage_attribution", {}).get("kind") == "final_answer" for event in ai_events)
 
+    @pytest.mark.parametrize("streamed", [True, False])
+    def test_stream_emits_text_a_later_node_appends_to_a_sent_ai_message(self, client, streamed):
+        """A guard's ``after_model`` replaces the message under the same id after it was sent."""
+        call = {"name": "bash", "args": {"command": "ls"}, "id": "call-1"}
+        sent = AIMessage(content="Checking again.", id="ai-1", tool_calls=[call])
+        stopped = AIMessage(content="Checking again.\n\n[FORCED STOP] Repeated tool calls exceeded the safety limit.", id="ai-1")
+        chunks = [
+            ("values", {"messages": [HumanMessage(content="hi", id="h-1"), sent]}),
+            ("values", {"messages": [HumanMessage(content="hi", id="h-1"), stopped]}),
+            ("values", {"messages": [HumanMessage(content="hi", id="h-1"), stopped]}),
+        ]
+        if streamed:
+            chunks.insert(0, ("messages", (AIMessageChunk(content="Checking again.", id="ai-1"), {})))
+        agent = _make_agent_mock(chunks)
+
+        with (
+            patch.object(client, "_ensure_agent"),
+            patch.object(client, "_agent", agent),
+        ):
+            events = list(client.stream("hi", thread_id="t-stream-replaced"))
+
+        assert [event.data["content"] for event in _ai_events(events)] == ["Checking again.", "\n\n[FORCED STOP] Repeated tool calls exceeded the safety limit."]
+
+    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")
+        rewritten = AIMessage(content="The model returned no final response.", id="ai-1")
+        agent = _make_agent_mock(
+            [
+                ("values", {"messages": [HumanMessage(content="hi", id="h-1"), sent]}),
+                ("values", {"messages": [HumanMessage(content="hi", id="h-1"), rewritten]}),
+            ]
+        )
+
+        with (
+            patch.object(client, "_ensure_agent"),
+            patch.object(client, "_agent", agent),
+        ):
+            events = list(client.stream("hi", thread_id="t-stream-rewritten"))
+
+        assert [event.data["content"] for event in _ai_events(events)] == ["Let me look."]
+
+    def test_stream_emits_metadata_a_later_node_adds_to_a_sent_ai_message(self, client):
+        attribution = {"version": 1, "kind": "final_answer", "shared_attribution": False, "actions": []}
+        sent = AIMessage(content="Hello!", id="ai-1")
+        attributed = AIMessage(content="Hello!", id="ai-1", additional_kwargs={"token_usage_attribution": attribution})
+        agent = _make_agent_mock(
+            [
+                ("values", {"messages": [HumanMessage(content="hi", id="h-1"), sent]}),
+                ("values", {"messages": [HumanMessage(content="hi", id="h-1"), attributed]}),
+            ]
+        )
+
+        with (
+            patch.object(client, "_ensure_agent"),
+            patch.object(client, "_agent", agent),
+        ):
+            events = list(client.stream("hi", thread_id="t-stream-attributed"))
+
+        ai_events = [event for event in events if event.type == "messages-tuple" and event.data.get("type") == "ai"]
+        assert [(event.data["content"], event.data.get("additional_kwargs")) for event in ai_events] == [("Hello!", None), ("", {"token_usage_attribution": attribution})]
+
     def test_stream_emits_new_additional_kwargs_after_prior_metadata(self, client):
         """stream() emits later attribution metadata even after earlier kwargs for the same id."""
         attribution = {
@@ -1132,6 +1194,40 @@ class TestChat:
 
         assert result == "final answer"
 
+    def test_returns_the_loop_detection_stop_notice(self, client):
+        """Real graph: the hard stop rewrites the last AI message after it was sent."""
+        from langchain.agents import create_agent
+        from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
+        from langchain_core.tools import tool
+
+        from deerflow.agents.middlewares.loop_detection_middleware import LoopDetectionMiddleware
+
+        class _ToolCallingFakeModel(FakeMessagesListChatModel):
+            def bind_tools(self, tools, **kwargs):
+                return self
+
+        @tool
+        def bash(command: str) -> str:
+            """Run a command."""
+            return "ok"
+
+        responses = [AIMessage(content="Checking again.", id=f"ai-{i}", tool_calls=[{"name": "bash", "args": {"command": "ls"}, "id": f"call-{i}"}]) for i in range(3)]
+        graph = create_agent(
+            model=_ToolCallingFakeModel(responses=[*responses, AIMessage(content="unreachable", id="ai-end")]),
+            tools=[bash],
+            middleware=[LoopDetectionMiddleware(warn_threshold=10, hard_limit=3)],
+            state_schema=ThreadState,
+        )
+
+        with (
+            patch.object(client, "_ensure_agent"),
+            patch.object(client, "_agent", graph),
+        ):
+            result = client.chat("q", thread_id="t-loop-stop")
+
+        assert result.startswith("Checking again.")
+        assert "[FORCED STOP]" in result
+
     def test_empty_response(self, client):
         """chat() returns empty string if no AI message produced."""
         chunks = [{"messages": []}]