From d87a1c11934bbc3a26d650411d812d6565a24b94 Mon Sep 17 00:00:00 2001 From: alanhuangyoo Date: Thu, 17 Sep 2026 08:45:29 +0800 Subject: [PATCH] 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 * test(client): pin that a non-extending replacement is not re-sent Co-Authored-By: Claude Opus 5 --------- Co-authored-by: Claude Opus 5 --- backend/packages/harness/deerflow/client.py | 34 +++++++- backend/tests/test_client.py | 96 +++++++++++++++++++++ 2 files changed, 126 insertions(+), 4 deletions(-) diff --git a/backend/packages/harness/deerflow/client.py b/backend/packages/harness/deerflow/client.py index 7b377ac6c..09efb3881 100644 --- a/backend/packages/harness/deerflow/client.py +++ b/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, diff --git a/backend/tests/test_client.py b/backend/tests/test_client.py index fc0cd5416..38b9fb162 100644 --- a/backend/tests/test_client.py +++ b/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": []}]