diff --git a/backend/packages/harness/deerflow/subagents/executor.py b/backend/packages/harness/deerflow/subagents/executor.py
index 60fb02778..cfdf21fa9 100644
--- a/backend/packages/harness/deerflow/subagents/executor.py
+++ b/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")
diff --git a/backend/tests/test_subagent_executor.py b/backend/tests/test_subagent_executor.py
index aa1368bd5..87f80cc52 100644
--- a/backend/tests/test_subagent_executor.py
+++ b/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."""