github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-09-16 01:28:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(worker): don't fail the parent run on a subagent error carried by task_running (#5407)

Browse Source 
* fix(worker): don't fail the parent run on a subagent error carried by task_running

The worker scans root frames for the LLM error-fallback marker to decide the
parent run's status, and custom frames are root frames. task_running events
carry each delegated subagent message, including a subagent's marked error
fallback, so a subagent that failed after retries marked the whole parent run
as error (and an edit-and-rerun rolled back) even though the executor already
reported task_failed and the lead answered. Skip custom frames in both the
single-mode and multi-mode stream paths.

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

* test(worker): cover the single-mode stream loop for task_running fallbacks

Run the task_running regression test with stream_modes=["custom"] too,
which takes the worker's single-mode loop. Without the single-mode
guard only that case fails.

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

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
alanhuangyoo 2026-09-14 07:17:51 +08:00 committed by GitHub
parent b1dad2480a
commit 5a3725d1b8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 56 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
backend/packages/harness/deerflow/runtime/runs/worker.py
View File

@ -1205,7 +1205,10 @@ async def run_agent(
broke_on_abort = True
logger.info("Run %s abort requested — stopping", run_id)
break
llm_error_fallback_message = llm_error_fallback_message or _extract_llm_error_fallback_message(chunk, pre_existing_message_ids)
if single_mode != "custom":
# Custom frames carry task_* events whose payload can hold a delegated
# subagent's messages; see the multi-mode branch below.
llm_error_fallback_message = llm_error_fallback_message or _extract_llm_error_fallback_message(chunk, pre_existing_message_ids)
sse_event = _lg_mode_to_sse_event(single_mode)
single_payload = serialize(chunk, mode=single_mode)
if single_mode == "values" and seq_stamper is not None:
@ -1245,10 +1248,12 @@ async def run_agent(
if mode is None:
continue
if not namespace:
if not namespace and mode != "custom":
# Only root-graph frames may decide the parent run's error
# fallback: a delegated subagent's marked fallback is the
# executor's to map (task_failed), not this run's.
# executor's to map (task_failed), not this run's. That
# includes the child messages task_running custom events
# carry, which are root frames too.
llm_error_fallback_message = llm_error_fallback_message or _extract_llm_error_fallback_message(chunk, pre_existing_message_ids)
await _publish_stream_item(
bridge=bridge,

52
backend/tests/test_worker_stream_subgraph_namespace.py
View File

@ -365,7 +365,7 @@ def _collect_ids(payload: object) -> set[str]:
return ids
def _build_delegating_parent_graph(executor_module, monkeypatch, *, child_emits_error_fallback: bool = False):
def _build_delegating_parent_graph(executor_module, monkeypatch, *, child_emits_error_fallback: bool = False, emit_task_running: bool = False):
"""Real parent graph whose node delegates a scripted child through the
real ``SubagentExecutor`` and emits ``task_*`` custom events the way the
production task tool does (root-graph ``get_stream_writer``).
@ -374,6 +374,10 @@ def _build_delegating_parent_graph(executor_module, monkeypatch, *, child_emits_
message carrying the ``deerflow_error_fallback`` marker (not as its final
message, so the delegation itself still completes) — the shape whose leak
would mark the *parent* run as errored (#4399).
With ``emit_task_running`` the delegate also emits one ``task_running``
event per captured child step, carrying the message dict the way the
production task tool does.
"""
from langgraph.config import get_stream_writer
from langgraph.graph import END, START, MessagesState, StateGraph
@ -457,6 +461,9 @@ def _build_delegating_parent_graph(executor_module, monkeypatch, *, child_emits_
pytest.fail("delegated subagent did not complete")
await asyncio.sleep(0.001)
assert result.status.value == "completed", f"delegation failed: {result.error}"
if emit_task_running:
for index, message in enumerate(result.ai_messages or [], start=1):
writer({"type": "task_running", "task_id": task_id, "message": message, "message_index": index})
finally:
executor_module.cleanup_background_task(task_id)
writer({"type": "task_completed", "task_id": task_id})
@ -469,10 +476,18 @@ def _build_delegating_parent_graph(executor_module, monkeypatch, *, child_emits_
return parent_builder.compile()
async def _run_delegation_through_worker(executor_module, monkeypatch, *, stream_subgraphs: bool, child_emits_error_fallback: bool = False) -> tuple[RunRecord, _RecordingStreamBridge]:
async def _run_delegation_through_worker(
executor_module,
monkeypatch,
*,
stream_subgraphs: bool,
child_emits_error_fallback: bool = False,
emit_task_running: bool = False,
stream_modes: list[str] | None = None,
) -> tuple[RunRecord, _RecordingStreamBridge]:
from langgraph.checkpoint.memory import InMemorySaver
parent_graph = _build_delegating_parent_graph(executor_module, monkeypatch, child_emits_error_fallback=child_emits_error_fallback)
parent_graph = _build_delegating_parent_graph(executor_module, monkeypatch, child_emits_error_fallback=child_emits_error_fallback, emit_task_running=emit_task_running)
bridge = _RecordingStreamBridge()
record = RunRecord(
run_id=f"run-ns-int-{int(stream_subgraphs)}",
@ -492,7 +507,7 @@ async def _run_delegation_through_worker(executor_module, monkeypatch, *, stream
agent_factory=lambda config: parent_graph,
graph_input={"messages": [HumanMessage(content="delegate to the subagent")]},
config={"configurable": {"thread_id": _THREAD_ID}},
stream_modes=["values", "messages-tuple", "custom"],
stream_modes=stream_modes or ["values", "messages-tuple", "custom"],
stream_subgraphs=stream_subgraphs,
)
return record, bridge
@ -548,6 +563,35 @@ class TestWorkerSubgraphStreamIntegration:
namespaced_payloads = [payload for event, payload in bridge.published if event.startswith(("values|", "messages|"))]
assert any("child-fallback-sentinel" in _collect_ids(payload) for payload in namespaced_payloads)
@pytest.mark.asyncio
@pytest.mark.parametrize(
("stream_modes", "stream_subgraphs"),
[
(["values", "messages-tuple", "custom"], True),
(["values", "messages-tuple", "custom"], False),
# A single requested mode takes the worker's single-mode stream loop.
(["custom"], False),
],
)
async def test_delegated_error_fallback_in_task_running_event_does_not_mark_the_parent_run_as_error(self, real_executor_module, monkeypatch, stream_modes, stream_subgraphs):
record, bridge = await _run_delegation_through_worker(
real_executor_module,
monkeypatch,
stream_subgraphs=stream_subgraphs,
child_emits_error_fallback=True,
emit_task_running=True,
stream_modes=stream_modes,
)
# task_running is a root-level custom frame, but the message it carries is
# the subagent's; its fallback marker must not decide the parent run's status.
assert record.status == RunStatus.success, "delegated error fallback in a task_running event leaked into the parent run status"
assert not [payload for event, payload in bridge.published if event == "error"]
# Non-vacuous: the marked child message really rode a root custom frame.
task_running = [payload for event, payload in bridge.published if event == "custom" and isinstance(payload, dict) and payload.get("type") == "task_running"]
assert any("child-fallback-sentinel" in _collect_ids(payload) for payload in task_running)
@pytest.mark.asyncio
async def test_without_stream_subgraphs_delegated_frames_stay_out_while_task_events_remain(self, real_executor_module, monkeypatch):
record, bridge = await _run_delegation_through_worker(real_executor_module, monkeypatch, stream_subgraphs=False)