diff --git a/backend/packages/harness/deerflow/runtime/runs/worker.py b/backend/packages/harness/deerflow/runtime/runs/worker.py index b6037d95e..1f6e0ec0c 100644 --- a/backend/packages/harness/deerflow/runtime/runs/worker.py +++ b/backend/packages/harness/deerflow/runtime/runs/worker.py @@ -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, diff --git a/backend/tests/test_worker_stream_subgraph_namespace.py b/backend/tests/test_worker_stream_subgraph_namespace.py index 2010a3629..d7fadaca2 100644 --- a/backend/tests/test_worker_stream_subgraph_namespace.py +++ b/backend/tests/test_worker_stream_subgraph_namespace.py @@ -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)