perf(subagents): dedup streamed AI messages via a seen-id set (O(n^2) -> O(n)) (#3687)

_aexecute collects AI messages from agent.astream(stream_mode="values"),
which re-yields the full state every super-step. The duplicate check rescanned
the append-only ai_messages list on every chunk -- any(m["id"] == message_id) --
so a run with M messages did O(M^2) work, and M reaches max_turns=150 for the
general-purpose / deep-research subagent.

Track an id-keyed set alongside ai_messages: id-bearing messages become O(1)
set lookups, and the id-less full-dict-compare fallback is preserved. Behavior
is unchanged.

Co-authored-by: ly-wang19 <ly-wang19@users.noreply.github.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/packages/harness/deerflow/subagents/executor.py

@@ -527,6 +527,12 @@ class SubagentExecutor:
         if ai_messages is None:
             ai_messages = []
             result.ai_messages = ai_messages
+        # O(1) duplicate detection for streamed AI messages. ``stream_mode="values"``
+        # re-yields the full state every super-step, so the same trailing message is
+        # re-examined on each chunk; an id-keyed set keeps that check O(1) instead of
+        # rescanning the append-only ``ai_messages`` list (O(n) per chunk -> O(n^2)
+        # over a run, which reaches max_turns=150 for deep-research subagents).
+        seen_message_ids: set[str] = {mid for msg in ai_messages if (mid := msg.get("id"))}
 
         collector: SubagentTokenCollector | None = None
         try:
@@ -633,14 +639,16 @@ class SubagentExecutor:
                         # Only add if it's not already in the list (avoid duplicates)
                         # Check by comparing message IDs if available, otherwise compare full dict
                         message_id = message_dict.get("id")
-                        is_duplicate = False
                         if message_id:
-                            is_duplicate = any(msg.get("id") == message_id for msg in ai_messages)
+                            is_duplicate = message_id in seen_message_ids
                         else:
+                            # id-less messages can't be keyed; fall back to a full-dict compare
                             is_duplicate = message_dict in ai_messages
 
                         if not is_duplicate:
                             ai_messages.append(message_dict)
+                            if message_id:
+                                seen_message_ids.add(message_id)
                             logger.info(f"[trace={self.trace_id}] Subagent {self.config.name} captured AI message #{len(ai_messages)}")
 
             logger.info(f"[trace={self.trace_id}] Subagent {self.config.name} completed async execution")

backend/tests/test_subagent_executor.py

@@ -758,6 +758,53 @@ class TestAsyncExecutionPath:
 
         assert len(result.ai_messages) == 1
 
+    @pytest.mark.anyio
+    async def test_aexecute_dedup_scales_over_repeated_chunks(self, classes, base_config, mock_agent, msg):
+        """``stream_mode="values"`` re-yields the same trailing message across many
+        snapshots before the next one appears. Dedup must collapse the repeats and
+        still capture each distinct message exactly once, in arrival order."""
+        SubagentExecutor = classes["SubagentExecutor"]
+
+        m1 = msg.ai("first", "msg-1")
+        m2 = msg.ai("second", "msg-2")
+        m3 = msg.ai("third", "msg-3")
+        # m1 is re-yielded as the trailing message several times before m2/m3 arrive.
+        chunks = [
+            {"messages": [msg.human("Task"), m1]},
+            {"messages": [msg.human("Task"), m1]},
+            {"messages": [msg.human("Task"), m1]},
+            {"messages": [msg.human("Task"), m1, m2]},
+            {"messages": [msg.human("Task"), m1, m2]},
+            {"messages": [msg.human("Task"), m1, m2, m3]},
+        ]
+        mock_agent.astream = lambda *args, **kwargs: async_iterator(chunks)
+
+        executor = SubagentExecutor(config=base_config, tools=[], thread_id="test-thread")
+        with patch.object(executor, "_create_agent", return_value=mock_agent):
+            result = await executor._aexecute("Task")
+
+        assert [m["id"] for m in result.ai_messages] == ["msg-1", "msg-2", "msg-3"]
+
+    @pytest.mark.anyio
+    async def test_aexecute_dedup_idless_messages_fall_back_to_content(self, classes, base_config, mock_agent, msg):
+        """Messages without an id can't be keyed by the seen-id set, so dedup must
+        fall back to a full content compare: identical content collapses, distinct
+        content is kept."""
+        SubagentExecutor = classes["SubagentExecutor"]
+
+        chunks = [
+            {"messages": [msg.human("Task"), msg.ai("same")]},  # id-less
+            {"messages": [msg.human("Task"), msg.ai("same")]},  # id-less, identical content -> dropped
+            {"messages": [msg.human("Task"), msg.ai("different")]},  # id-less, distinct -> kept
+        ]
+        mock_agent.astream = lambda *args, **kwargs: async_iterator(chunks)
+
+        executor = SubagentExecutor(config=base_config, tools=[], thread_id="test-thread")
+        with patch.object(executor, "_create_agent", return_value=mock_agent):
+            result = await executor._aexecute("Task")
+
+        assert [m["content"] for m in result.ai_messages] == ["same", "different"]
+
     @pytest.mark.anyio
     async def test_aexecute_handles_list_content(self, classes, base_config, mock_agent, msg):
         """Test handling of list-type content in AIMessage."""