fix(subagents): classify recursion-capped LLM error fallbacks as failed (#4056)

The GraphRecursionError except-block in SubagentExecutor._aexecute derives
usable_partial from the last AIMessage's raw non-empty text, without
checking _extract_llm_error_fallback (#4042) first. A handled provider
failure (LLMErrorHandlingMiddleware's deerflow_error_fallback marker)
always carries non-empty user-facing text, so when it lands on the same
turn that trips max_turns, it is indistinguishable from genuine partial
output and gets misclassified as a completed task instead of the failed
provider error it is.

Consult _extract_llm_error_fallback in this except-block too, same as the
normal-completion path above it, and classify FAILED with
stop_reason=turn_capped when it detects the marker.

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

@@ -872,30 +872,46 @@ class SubagentExecutor:
             # consistent and pops the reason so it is not orphaned in the dict.
             max_turns = self.config.max_turns
             logger.warning(f"[trace={self.trace_id}] Subagent {self.config.name} reached max_turns={max_turns} (GraphRecursionError); recovering partial result")
-            messages = (final_state or {}).get("messages", [])
-            usable_partial: str | None = None
-            for m in reversed(messages):
-                if isinstance(m, AIMessage):
-                    text = message_content_to_text(m.content).strip()
-                    if text:
-                        usable_partial = text
-                    break
             records = collector.snapshot_records() if collector is not None else None
             stop_reason = self._consume_guard_stop_reason() or "turn_capped"
-            if usable_partial is not None:
+
+            # A handled LLM provider failure (#4042) carries non-empty
+            # user-facing text on its terminal ``AIMessage`` just like genuine
+            # partial output, so it must be checked here too or it is
+            # indistinguishable from the raw-text scan below and gets
+            # misclassified as a completed task. Consult the same marker the
+            # normal-completion path above uses, before falling back to that scan.
+            llm_error = _extract_llm_error_fallback(final_state)
+            if llm_error is not None:
                 result.try_set_terminal(
-                    SubagentStatus.COMPLETED,
-                    result=usable_partial,
+                    SubagentStatus.FAILED,
+                    error=llm_error,
                     stop_reason=stop_reason,
                     token_usage_records=records,
                 )
             else:
-                result.try_set_terminal(
-                    SubagentStatus.FAILED,
-                    error=f"Reached max_turns={max_turns}",
-                    stop_reason=stop_reason,
-                    token_usage_records=records,
-                )
+                messages = (final_state or {}).get("messages", [])
+                usable_partial: str | None = None
+                for m in reversed(messages):
+                    if isinstance(m, AIMessage):
+                        text = message_content_to_text(m.content).strip()
+                        if text:
+                            usable_partial = text
+                        break
+                if usable_partial is not None:
+                    result.try_set_terminal(
+                        SubagentStatus.COMPLETED,
+                        result=usable_partial,
+                        stop_reason=stop_reason,
+                        token_usage_records=records,
+                    )
+                else:
+                    result.try_set_terminal(
+                        SubagentStatus.FAILED,
+                        error=f"Reached max_turns={max_turns}",
+                        stop_reason=stop_reason,
+                        token_usage_records=records,
+                    )
 
         except Exception as e:
             logger.exception(f"[trace={self.trace_id}] Subagent {self.config.name} async execution failed")

backend/tests/test_subagent_executor.py

@@ -1203,6 +1203,56 @@ class TestAsyncExecutionPath:
         assert str(base_config.max_turns) in (result.error or "")
         assert result.completed_at is not None
 
+    @pytest.mark.anyio
+    async def test_aexecute_recursion_error_with_llm_error_fallback_surfaces_failed(self, classes, base_config, mock_agent, msg):
+        """A structured LLM error fallback that coincides with hitting
+        ``max_turns`` must still classify as ``failed``, not ``completed``.
+
+        ``_extract_llm_error_fallback`` (#4042) marks a terminal ``AIMessage``
+        as a handled provider failure via
+        ``additional_kwargs.deerflow_error_fallback``, and the
+        normal-completion branch above already consults it before falling
+        back to ``_extract_final_result``. This except-block must apply the
+        same check before recovering ``usable_partial`` from raw non-empty
+        ``AIMessage`` text: a fallback message always carries non-empty
+        user-facing text, so without checking the marker first it is
+        indistinguishable from genuine partial output and gets misclassified
+        as a completed task rather than the failed provider error it is.
+        """
+        from langgraph.errors import GraphRecursionError
+
+        AIMessage = classes["AIMessage"]
+        SubagentExecutor = classes["SubagentExecutor"]
+        SubagentStatus = classes["SubagentStatus"]
+
+        fallback_text = "LLM request failed: provider rejected the request"
+        fallback_message = AIMessage(
+            content=fallback_text,
+            additional_kwargs={
+                "deerflow_error_fallback": True,
+                "error_type": "BadRequestError",
+                "error_reason": "generic",
+                "error_detail": "Error code: 400 - InvalidParameter",
+            },
+        )
+        fallback_state = {"messages": [msg.human("Task"), fallback_message]}
+
+        async def mock_astream(*args, **kwargs):
+            yield fallback_state
+            raise GraphRecursionError("Recursion limit reached right after the LLM error fallback")
+
+        mock_agent.astream = mock_astream
+
+        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 result.status == SubagentStatus.FAILED
+        assert result.error == fallback_text
+        assert result.result is None
+        assert result.stop_reason == "turn_capped"
+
     @pytest.mark.anyio
     async def test_aexecute_token_capped_surfaces_completed_token_capped(self, classes, base_config, mock_agent, msg):
         """#3875 Phase 2: the token-budget hard-stop does not raise — it strips