fix(runtime): stop subgraph stream frames impersonating root frames (#4407)

* fix(runtime): stop subgraph stream frames impersonating root frames

The web frontend always requested stream_subgraphs, and since delegated
subagent graphs inherit the parent checkpoint namespace (#4215), their
values snapshots and token chunks ride the parent stream. The worker's
_unpack_stream_item dropped the namespace and published every subgraph
frame under a bare event name, so a subagent's values snapshot replaced
the whole thread view in SDK clients (#4399), its token chunks flooded
the parent message stream, and a subagent's LLM error fallback could be
mistaken for the parent run's.

Publish subgraph frames under namespace-qualified SSE event names
(mode|ns1|ns2, LangGraph Platform style) and keep root-only consumers
(file-tool chunk batcher, subagent event persistence, error-fallback
detection) on root frames only. Drop streamSubgraphs from the frontend
submit paths: subtask progress arrives via root-namespace task_* custom
events, so the flag only exposed the leak.

* test(runtime): add production-shaped subgraph stream regression tests

Address review: the namespace tests validated the publishing helpers
with hand-fed namespaces, while the #4399 regression lived in the
integration between LangGraph's delegation routing and the worker's
stream loop. Add TestWorkerSubgraphStreamIntegration: a real parent
graph delegates through the real SubagentExecutor and streams through
run_agent into a real MemoryStreamBridge, locking both stream_subgraphs
modes -- delegated frames arrive namespaced (never bare), a delegated
error fallback cannot mark the parent run as errored, and without the
flag delegated frames stay out while task_* custom events remain.

backend/AGENTS.md

@@ -15,6 +15,7 @@ DeerFlow is a LangGraph-based AI super agent system with a full-stack architectu
 **Runtime**:
 - `make dev`, Docker dev, and production all run the agent runtime in Gateway via `RunManager` + `run_agent()` + `StreamBridge` (`packages/harness/deerflow/runtime/`). Nginx exposes that runtime at `/api/langgraph/*` and rewrites it to Gateway's native `/api/*` routers.
 - Gateway streams `write_file` and `str_replace` argument deltas in bounded batches when clients also subscribe to `values`; messages-only consumers retain the original per-chunk contract, while `values` preserves the complete tool call.
+- With `stream_subgraphs`, subgraph frames keep their namespace in the SSE event name (`values|<ns>`, LangGraph Platform style) instead of impersonating root frames — a delegated subagent inherits the parent checkpoint namespace, so publishing its `values` snapshot as bare `values` replaces the whole thread view in SDK clients (#4399). Root-only consumers (file-tool chunk batcher, subagent event persistence, LLM error-fallback detection) ignore namespaced frames. The web frontend does not request subgraph streaming; subtask progress rides root-namespace `task_*` custom events.
 - Scheduled-task executions must reuse that same Gateway run lifecycle. The scheduler may decide *when* work runs, but it must dispatch through the existing run path rather than introducing a parallel execution stack.
 
 **Project Structure**:

backend/packages/harness/deerflow/runtime/runs/worker.py

@@ -687,21 +687,24 @@ async def run_agent(
                             logger.info("Run %s abort requested — stopping", run_id)
                             break
 
-                        mode, chunk = _unpack_stream_item(item, lg_modes, stream_subgraphs)
+                        mode, chunk, namespace = _unpack_stream_item(item, lg_modes, stream_subgraphs)
                         if mode is None:
                             continue
 
-                        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(mode)
-                        if file_tool_chunk_batcher is not None and mode != "messages":
-                            pending_chunks = file_tool_chunk_batcher.finish() if mode == "values" else file_tool_chunk_batcher.flush()
-                            for publish_chunk in pending_chunks:
-                                await bridge.publish(run_id, "messages", serialize(publish_chunk, mode="messages"))
-                        chunks_to_publish = file_tool_chunk_batcher.push(chunk) if mode == "messages" and file_tool_chunk_batcher is not None else [chunk]
-                        for publish_chunk in chunks_to_publish:
-                            await bridge.publish(run_id, sse_event, serialize(publish_chunk, mode=mode))
-                        if mode == "custom":
-                            await subagent_events.add(chunk)
+                        if not namespace:
+                            # 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.
+                            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,
+                            run_id=run_id,
+                            mode=mode,
+                            chunk=chunk,
+                            namespace=namespace,
+                            file_tool_chunk_batcher=file_tool_chunk_batcher,
+                            subagent_events=subagent_events,
+                        )
             finally:
                 stream_error = sys.exception()
                 if file_tool_chunk_batcher is not None:
@@ -1776,23 +1779,77 @@ def _unpack_stream_item(
     item: Any,
     lg_modes: list[str],
     stream_subgraphs: bool,
-) -> tuple[str | None, Any]:
-    """Unpack a multi-mode or subgraph stream item into (mode, chunk).
+) -> tuple[str | None, Any, tuple[str, ...]]:
+    """Unpack a multi-mode or subgraph stream item into (mode, chunk, namespace).
 
-    Returns ``(None, None)`` if the item cannot be parsed.
+    ``namespace`` is the subgraph namespace tuple LangGraph prefixes onto each
+    frame when ``subgraphs=True``; it is empty for root-graph frames. Delegated
+    subagent graphs inherit the parent's checkpoint namespace (see
+    ``subagents/executor.py``), so their frames arrive here with a non-empty
+    namespace and must not be mistaken for root frames.
+
+    Returns ``(None, None, ())`` if the item cannot be parsed.
     """
     if stream_subgraphs:
         if isinstance(item, tuple) and len(item) == 3:
-            _ns, mode, chunk = item
-            return str(mode), chunk
+            ns, mode, chunk = item
+            namespace = tuple(str(part) for part in ns) if isinstance(ns, (list, tuple)) else (str(ns),)
+            return str(mode), chunk, namespace
         if isinstance(item, tuple) and len(item) == 2:
             mode, chunk = item
-            return str(mode), chunk
-        return None, None
+            return str(mode), chunk, ()
+        return None, None, ()
 
     if isinstance(item, tuple) and len(item) == 2:
         mode, chunk = item
-        return str(mode), chunk
+        return str(mode), chunk, ()
 
     # Fallback: single-element output from first mode
-    return lg_modes[0] if lg_modes else None, item
+    return lg_modes[0] if lg_modes else None, item, ()
+
+
+def _compose_sse_event(sse_event: str, namespace: tuple[str, ...]) -> str:
+    """Namespace-qualified SSE event name, LangGraph Platform style.
+
+    Root frames keep the bare event name; subgraph frames become
+    ``mode|ns1|ns2`` so clients can tell them apart. The LangGraph SDK parses
+    exactly this shape (``event.split("|").slice(1)``) and routes
+    subagent-namespaced values away from the thread view.
+    """
+    if not namespace:
+        return sse_event
+    return "|".join((sse_event, *namespace))
+
+
+async def _publish_stream_item(
+    *,
+    bridge: Any,
+    run_id: str,
+    mode: str,
+    chunk: Any,
+    namespace: tuple[str, ...],
+    file_tool_chunk_batcher: Any,
+    subagent_events: Any,
+) -> None:
+    """Publish one stream frame, preserving the subgraph namespace.
+
+    A subgraph frame published under a bare event name impersonates the root
+    graph: a delegated subagent's ``values`` snapshot then replaces the whole
+    thread view in SDK clients and its token chunks flood the parent message
+    stream (#4399). Subgraph frames therefore keep their namespace in the event
+    name and bypass the root-only consumers (file-tool chunk batcher, subagent
+    event persistence — task_* lifecycle events are root frames already).
+    """
+    sse_event = _compose_sse_event(_lg_mode_to_sse_event(mode), namespace)
+    if namespace:
+        await bridge.publish(run_id, sse_event, serialize(chunk, mode=mode))
+        return
+    if file_tool_chunk_batcher is not None and mode != "messages":
+        pending_chunks = file_tool_chunk_batcher.finish() if mode == "values" else file_tool_chunk_batcher.flush()
+        for publish_chunk in pending_chunks:
+            await bridge.publish(run_id, "messages", serialize(publish_chunk, mode="messages"))
+    chunks_to_publish = file_tool_chunk_batcher.push(chunk) if mode == "messages" and file_tool_chunk_batcher is not None else [chunk]
+    for publish_chunk in chunks_to_publish:
+        await bridge.publish(run_id, sse_event, serialize(publish_chunk, mode=mode))
+    if mode == "custom":
+        await subagent_events.add(chunk)

backend/tests/test_worker_stream_subgraph_namespace.py

@@ -0,0 +1,556 @@
+"""Subgraph stream frames must not impersonate root-graph frames (#4399).
+
+The gateway worker drives ``agent.astream(subgraphs=...)`` and publishes each
+frame to the StreamBridge. Delegated subagent graphs inherit the parent's
+checkpoint namespace (``subagents/executor.py``), so with ``subgraphs=True``
+their values snapshots and token chunks arrive interleaved with root frames.
+Publishing them under bare event names lets a subagent's values snapshot
+replace the whole thread view in SDK clients and floods the parent message
+stream with the subagent's token chunks. The namespace must ride the SSE event
+name (LangGraph Platform style ``mode|ns1|ns2``) and namespaced frames must
+bypass the root-only consumers (file-tool chunk batcher, subagent event
+persistence).
+"""
+
+import asyncio
+import importlib
+import sys
+from importlib.metadata import version as package_version
+from types import SimpleNamespace
+
+import pytest
+from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
+from packaging.version import Version
+
+from deerflow.runtime.runs import worker
+from deerflow.runtime.runs.manager import RunRecord
+from deerflow.runtime.runs.schemas import DisconnectMode, RunStatus
+from deerflow.runtime.runs.worker import (
+    _compose_sse_event,
+    _publish_stream_item,
+    _unpack_stream_item,
+)
+from deerflow.runtime.stream_bridge.memory import MemoryStreamBridge
+
+SUBAGENT_NS = ("tools:call_subagent_1",)
+
+# Delegated graphs inherit the parent checkpoint namespace (and therefore
+# stream as subgraphs) only on LangGraph >= 1.2.6 — same gate as
+# tests/test_subagent_executor.py::TestSubagentCheckpointLineage.
+_LANGGRAPH_INHERITS_SUBGRAPH_NAMESPACE = Version(package_version("langgraph")) >= Version("1.2.6")
+
+
+class _FakeBridge:
+    def __init__(self) -> None:
+        self.published: list[tuple[str, str, object]] = []
+
+    async def publish(self, run_id: str, event: str, payload: object) -> None:
+        self.published.append((run_id, event, payload))
+
+
+class _FakeSubagentEvents:
+    def __init__(self) -> None:
+        self.added: list[object] = []
+
+    async def add(self, chunk: object) -> None:
+        self.added.append(chunk)
+
+
+class _SpyBatcher:
+    """Observable stand-in for _LargeFileToolChunkBatcher."""
+
+    def __init__(self) -> None:
+        self.pushed: list[object] = []
+        self.finish_calls = 0
+        self.flush_calls = 0
+
+    def push(self, chunk: object) -> list[object]:
+        self.pushed.append(chunk)
+        return [chunk]
+
+    def finish(self) -> list[object]:
+        self.finish_calls += 1
+        return []
+
+    def flush(self) -> list[object]:
+        self.flush_calls += 1
+        return []
+
+
+class TestUnpackStreamItem:
+    def test_root_frame_with_subgraphs_has_empty_namespace(self):
+        mode, chunk, namespace = _unpack_stream_item(((), "values", {"messages": []}), ["values"], True)
+        assert mode == "values"
+        assert namespace == ()
+
+    def test_subgraph_frame_preserves_namespace(self):
+        mode, chunk, namespace = _unpack_stream_item((SUBAGENT_NS, "values", {"messages": []}), ["values"], True)
+        assert mode == "values"
+        assert namespace == SUBAGENT_NS
+
+    def test_nested_subgraph_namespace_is_preserved_in_order(self):
+        ns = ("tools:call_a", "model_request:xyz")
+        _mode, _chunk, namespace = _unpack_stream_item((ns, "messages", object()), ["messages"], True)
+        assert namespace == ns
+
+    def test_two_tuple_under_subgraphs_is_root(self):
+        mode, _chunk, namespace = _unpack_stream_item(("custom", {"type": "task_started"}), ["custom"], True)
+        assert mode == "custom"
+        assert namespace == ()
+
+    def test_without_subgraphs_frames_are_root(self):
+        mode, _chunk, namespace = _unpack_stream_item(("values", {}), ["values"], False)
+        assert mode == "values"
+        assert namespace == ()
+
+    def test_single_mode_fallback_is_root(self):
+        mode, chunk, namespace = _unpack_stream_item({"messages": []}, ["values"], False)
+        assert mode == "values"
+        assert chunk == {"messages": []}
+        assert namespace == ()
+
+    def test_unparsable_item_under_subgraphs(self):
+        mode, chunk, namespace = _unpack_stream_item("garbage", ["values"], True)
+        assert mode is None
+        assert chunk is None
+        assert namespace == ()
+
+
+class TestComposeSseEvent:
+    def test_root_frame_keeps_bare_event_name(self):
+        assert _compose_sse_event("values", ()) == "values"
+
+    def test_subgraph_frame_gets_namespace_qualified_name(self):
+        assert _compose_sse_event("values", SUBAGENT_NS) == "values|tools:call_subagent_1"
+
+    def test_nested_namespace_joins_all_segments(self):
+        assert _compose_sse_event("messages", ("tools:call_a", "model_request:xyz")) == "messages|tools:call_a|model_request:xyz"
+
+
+class TestPublishStreamItem:
+    @pytest.mark.asyncio
+    async def test_subagent_values_snapshot_is_never_published_as_bare_values(self):
+        # The #4399 regression: a delegated subagent's values snapshot published
+        # as bare "values" replaces the whole thread view in SDK clients.
+        bridge = _FakeBridge()
+        await _publish_stream_item(
+            bridge=bridge,
+            run_id="run-1",
+            mode="values",
+            chunk={"messages": [{"type": "human", "content": "subagent task prompt"}]},
+            namespace=SUBAGENT_NS,
+            file_tool_chunk_batcher=None,
+            subagent_events=_FakeSubagentEvents(),
+        )
+        assert [event for _run, event, _payload in bridge.published] == ["values|tools:call_subagent_1"]
+
+    @pytest.mark.asyncio
+    async def test_root_values_snapshot_keeps_bare_event_name(self):
+        bridge = _FakeBridge()
+        await _publish_stream_item(
+            bridge=bridge,
+            run_id="run-1",
+            mode="values",
+            chunk={"messages": []},
+            namespace=(),
+            file_tool_chunk_batcher=None,
+            subagent_events=_FakeSubagentEvents(),
+        )
+        assert [event for _run, event, _payload in bridge.published] == ["values"]
+
+    @pytest.mark.asyncio
+    async def test_subagent_message_chunks_are_namespaced(self):
+        bridge = _FakeBridge()
+        await _publish_stream_item(
+            bridge=bridge,
+            run_id="run-1",
+            mode="messages",
+            chunk=({"content": "token"}, {"langgraph_node": "model"}),
+            namespace=SUBAGENT_NS,
+            file_tool_chunk_batcher=_SpyBatcher(),
+            subagent_events=_FakeSubagentEvents(),
+        )
+        assert [event for _run, event, _payload in bridge.published] == ["messages|tools:call_subagent_1"]
+
+    @pytest.mark.asyncio
+    async def test_root_custom_event_is_persisted_for_subagent_history(self):
+        bridge = _FakeBridge()
+        subagent_events = _FakeSubagentEvents()
+        chunk = {"type": "task_started", "task_id": "call_1"}
+        await _publish_stream_item(
+            bridge=bridge,
+            run_id="run-1",
+            mode="custom",
+            chunk=chunk,
+            namespace=(),
+            file_tool_chunk_batcher=None,
+            subagent_events=subagent_events,
+        )
+        assert [event for _run, event, _payload in bridge.published] == ["custom"]
+        assert subagent_events.added == [chunk]
+
+    @pytest.mark.asyncio
+    async def test_subgraph_custom_event_is_not_persisted(self):
+        bridge = _FakeBridge()
+        subagent_events = _FakeSubagentEvents()
+        await _publish_stream_item(
+            bridge=bridge,
+            run_id="run-1",
+            mode="custom",
+            chunk={"type": "noise"},
+            namespace=SUBAGENT_NS,
+            file_tool_chunk_batcher=None,
+            subagent_events=subagent_events,
+        )
+        assert [event for _run, event, _payload in bridge.published] == ["custom|tools:call_subagent_1"]
+        assert subagent_events.added == []
+
+    @pytest.mark.asyncio
+    async def test_only_root_frames_drive_the_file_tool_batcher(self):
+        bridge = _FakeBridge()
+        batcher = _SpyBatcher()
+        # A subagent values frame must not finish() a pending root batch...
+        await _publish_stream_item(
+            bridge=bridge,
+            run_id="run-1",
+            mode="values",
+            chunk={"messages": []},
+            namespace=SUBAGENT_NS,
+            file_tool_chunk_batcher=batcher,
+            subagent_events=_FakeSubagentEvents(),
+        )
+        assert batcher.finish_calls == 0
+        assert batcher.pushed == []
+        # ...while a root values frame does.
+        await _publish_stream_item(
+            bridge=bridge,
+            run_id="run-1",
+            mode="values",
+            chunk={"messages": []},
+            namespace=(),
+            file_tool_chunk_batcher=batcher,
+            subagent_events=_FakeSubagentEvents(),
+        )
+        assert batcher.finish_calls == 1
+
+    @pytest.mark.asyncio
+    async def test_root_message_chunks_go_through_the_batcher(self):
+        bridge = _FakeBridge()
+        batcher = _SpyBatcher()
+        chunk = ({"content": "token"}, {"langgraph_node": "model"})
+        await _publish_stream_item(
+            bridge=bridge,
+            run_id="run-1",
+            mode="messages",
+            chunk=chunk,
+            namespace=(),
+            file_tool_chunk_batcher=batcher,
+            subagent_events=_FakeSubagentEvents(),
+        )
+        assert batcher.pushed == [chunk]
+        assert [event for _run, event, _payload in bridge.published] == ["messages"]
+
+
+# ---------------------------------------------------------------------------
+# Production-shaped integration: SubagentExecutor -> astream(subgraphs=...)
+# -> run_agent stream loop -> StreamBridge. The namespace must originate from
+# LangGraph's own delegation routing (checkpoint-namespace inheritance), not
+# be hand-fed to the publishing helper — the #4399 regression lived in that
+# interaction, not in any helper in isolation.
+# ---------------------------------------------------------------------------
+
+_CHILD_MESSAGE_IDS = frozenset(
+    {
+        "child-task-sentinel",
+        "child-ai-sentinel",
+        "child-tool-sentinel",
+        "child-final-sentinel",
+    }
+)
+_PARENT_FINAL_ID = "parent-final-sentinel"
+_THREAD_ID = "thread-subgraph-stream-integration"
+
+
+@pytest.fixture
+def real_executor_module():
+    """Swap the conftest MagicMock for the real subagent executor module.
+
+    conftest.py mocks ``deerflow.subagents.executor`` to break a package-init
+    import cycle; by the time this fixture runs every other deerflow module is
+    already imported, so a fresh import of the real module is safe.
+    """
+    original = sys.modules.get("deerflow.subagents.executor")
+    sys.modules.pop("deerflow.subagents.executor", None)
+    subagents_pkg = sys.modules.get("deerflow.subagents")
+    if subagents_pkg is not None and hasattr(subagents_pkg, "executor"):
+        delattr(subagents_pkg, "executor")
+
+    module = importlib.import_module("deerflow.subagents.executor")
+    # Hermetic in CI (no config.yaml) — same defaults as test_subagent_executor.
+    module.get_app_config = lambda: SimpleNamespace(tool_search=SimpleNamespace(enabled=False))
+    module.build_tracing_callbacks = lambda: []
+    yield module
+
+    if original is not None:
+        sys.modules["deerflow.subagents.executor"] = original
+    else:
+        sys.modules.pop("deerflow.subagents.executor", None)
+    subagents_pkg = sys.modules.get("deerflow.subagents")
+    if subagents_pkg is not None and hasattr(subagents_pkg, "executor"):
+        delattr(subagents_pkg, "executor")
+
+
+class _RecordingStreamBridge(MemoryStreamBridge):
+    """Real in-memory bridge that also records (event, payload) pairs."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.published: list[tuple[str, object]] = []
+
+    async def publish(self, run_id: str, event: str, payload: object) -> None:
+        self.published.append((event, payload))
+        await super().publish(run_id, event, payload)
+
+
+class _IntegrationRunManager:
+    def __init__(self, record: RunRecord) -> None:
+        self._record = record
+
+    async def wait_for_prior_finalizing(self, *_args, **_kwargs):
+        return None
+
+    async def set_status(self, _run_id, status, **_kwargs):
+        self._record.status = status
+
+    async def update_model_name(self, *_args, **_kwargs):
+        return None
+
+    async def update_run_completion(self, *_args, **_kwargs):
+        return None
+
+    async def has_later_started_run(self, *_args, **_kwargs):
+        return False
+
+    async def set_finalizing(self, *_args, **_kwargs):
+        return None
+
+
+def _collect_ids(payload: object) -> set[str]:
+    """All string ``id`` values anywhere in a serialized stream payload."""
+    ids: set[str] = set()
+
+    def walk(node: object) -> None:
+        if isinstance(node, dict):
+            node_id = node.get("id")
+            if isinstance(node_id, str):
+                ids.add(node_id)
+            for value in node.values():
+                walk(value)
+        elif isinstance(node, (list, tuple)):
+            for value in node:
+                walk(value)
+
+    walk(payload)
+    return ids
+
+
+def _build_delegating_parent_graph(executor_module, monkeypatch, *, child_emits_error_fallback: 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``).
+
+    With ``child_emits_error_fallback`` the child stream contains an assistant
+    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).
+    """
+    from langgraph.config import get_stream_writer
+    from langgraph.graph import END, START, MessagesState, StateGraph
+
+    from deerflow.subagents.config import SubagentConfig
+
+    child_builder = StateGraph(MessagesState)
+    child_builder.add_node(
+        "child_model",
+        lambda _state: {
+            "messages": [
+                AIMessage(
+                    content="",
+                    id="child-ai-sentinel",
+                    tool_calls=[{"name": "child_tool", "args": {}, "id": "child-tool-call", "type": "tool_call"}],
+                )
+            ]
+        },
+    )
+    child_builder.add_node(
+        "child_tool",
+        lambda _state: {"messages": [ToolMessage(content="child tool output", name="child_tool", tool_call_id="child-tool-call", id="child-tool-sentinel")]},
+    )
+    child_builder.add_node(
+        "child_fallback",
+        lambda _state: {
+            "messages": [
+                AIMessage(
+                    content="child provider failed after retries",
+                    id="child-fallback-sentinel",
+                    additional_kwargs={"deerflow_error_fallback": True},
+                )
+            ]
+        },
+    )
+    child_builder.add_node(
+        "child_final",
+        lambda _state: {"messages": [AIMessage(content="child final answer", id="child-final-sentinel")]},
+    )
+    child_builder.add_edge(START, "child_model")
+    child_builder.add_edge("child_model", "child_tool")
+    if child_emits_error_fallback:
+        child_builder.add_edge("child_tool", "child_fallback")
+        child_builder.add_edge("child_fallback", "child_final")
+    else:
+        child_builder.add_edge("child_tool", "child_final")
+    child_builder.add_edge("child_final", END)
+    child_graph = child_builder.compile(checkpointer=False)
+
+    executor = executor_module.SubagentExecutor(
+        config=SubagentConfig(
+            name="general-purpose",
+            description="Namespace integration test agent",
+            system_prompt="You are a namespace integration test agent.",
+            max_turns=5,
+            timeout_seconds=30,
+        ),
+        tools=[],
+        parent_model="test-model",
+        thread_id=_THREAD_ID,
+        trace_id="trace-namespace-integration",
+    )
+
+    async def build_initial_state(task):
+        return ({"messages": [HumanMessage(content=task, id="child-task-sentinel")]}, [], None)
+
+    monkeypatch.setattr(executor, "_build_initial_state", build_initial_state)
+    monkeypatch.setattr(executor, "_create_agent", lambda *_args, **_kwargs: child_graph)
+
+    async def delegate(_state):
+        writer = get_stream_writer()
+        task_id = executor.execute_async("run the delegated child graph")
+        writer({"type": "task_started", "task_id": task_id})
+        try:
+            deadline = asyncio.get_running_loop().time() + 10
+            while True:
+                result = executor_module.get_background_task_result(task_id)
+                if result is not None and result.status.is_terminal:
+                    break
+                if asyncio.get_running_loop().time() >= deadline:
+                    pytest.fail("delegated subagent did not complete")
+                await asyncio.sleep(0.001)
+            assert result.status.value == "completed", f"delegation failed: {result.error}"
+        finally:
+            executor_module.cleanup_background_task(task_id)
+        writer({"type": "task_completed", "task_id": task_id})
+        return {"messages": [AIMessage(content="parent final answer", id=_PARENT_FINAL_ID)]}
+
+    parent_builder = StateGraph(MessagesState)
+    parent_builder.add_node("delegate", delegate)
+    parent_builder.add_edge(START, "delegate")
+    parent_builder.add_edge("delegate", END)
+    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]:
+    from langgraph.checkpoint.memory import InMemorySaver
+
+    parent_graph = _build_delegating_parent_graph(executor_module, monkeypatch, child_emits_error_fallback=child_emits_error_fallback)
+    bridge = _RecordingStreamBridge()
+    record = RunRecord(
+        run_id=f"run-ns-int-{int(stream_subgraphs)}",
+        thread_id=_THREAD_ID,
+        assistant_id="lead-agent",
+        status=RunStatus.pending,
+        on_disconnect=DisconnectMode.cancel,
+        model_name=None,
+    )
+    record.abort_event = asyncio.Event()
+
+    await worker.run_agent(
+        bridge,
+        _IntegrationRunManager(record),
+        record,
+        ctx=worker.RunContext(checkpointer=InMemorySaver()),
+        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_subgraphs=stream_subgraphs,
+    )
+    return record, bridge
+
+
+@pytest.mark.skipif(
+    not _LANGGRAPH_INHERITS_SUBGRAPH_NAMESPACE,
+    reason="delegated graphs stream as namespaced subgraphs only on LangGraph >= 1.2.6",
+)
+class TestWorkerSubgraphStreamIntegration:
+    @pytest.mark.asyncio
+    async def test_stream_subgraphs_publishes_delegated_frames_namespaced_never_bare(self, real_executor_module, monkeypatch):
+        record, bridge = await _run_delegation_through_worker(real_executor_module, monkeypatch, stream_subgraphs=True)
+        assert record.status == RunStatus.success, f"run failed: {bridge.published}"
+
+        events = bridge.published
+        bare_values = [payload for event, payload in events if event == "values"]
+        bare_messages = [payload for event, payload in events if event == "messages"]
+        namespaced_values = [(event, payload) for event, payload in events if event.startswith("values|")]
+        namespaced_messages = [(event, payload) for event, payload in events if event.startswith("messages|")]
+
+        # The #4399 takeover: a delegated values snapshot must never be
+        # published as bare "values" (SDK clients replace the thread view).
+        for payload in bare_values:
+            assert not (_collect_ids(payload) & _CHILD_MESSAGE_IDS), f"delegated messages leaked into a bare values frame: {payload}"
+        for payload in bare_messages:
+            assert not (_collect_ids(payload) & _CHILD_MESSAGE_IDS), f"delegated message chunk leaked into the bare messages stream: {payload}"
+
+        # The delegated frames must actually arrive — namespaced by LangGraph,
+        # not silently dropped (guards against a vacuous pass).
+        assert any(_collect_ids(payload) & _CHILD_MESSAGE_IDS for _event, payload in namespaced_values), f"expected namespaced delegated values frames, got events: {[event for event, _ in events]}"
+        assert any(_collect_ids(payload) & _CHILD_MESSAGE_IDS for _event, payload in namespaced_messages), f"expected namespaced delegated message chunks, got events: {[event for event, _ in events]}"
+        for event, _payload in namespaced_values + namespaced_messages:
+            segments = event.split("|")[1:]
+            assert segments and all(segments), f"namespaced event name has empty namespace segments: {event}"
+
+        # Root frames stay bare and intact.
+        assert any(_PARENT_FINAL_ID in _collect_ids(payload) for payload in bare_values)
+        custom_types = [payload.get("type") for event, payload in events if event == "custom" and isinstance(payload, dict)]
+        assert "task_started" in custom_types and "task_completed" in custom_types
+
+    @pytest.mark.asyncio
+    async def test_delegated_error_fallback_does_not_mark_the_parent_run_as_error(self, real_executor_module, monkeypatch):
+        record, bridge = await _run_delegation_through_worker(real_executor_module, monkeypatch, stream_subgraphs=True, child_emits_error_fallback=True)
+
+        # A delegated subagent's LLM error fallback is the executor's to map
+        # (task_failed); it must not decide the parent run's status.
+        assert record.status == RunStatus.success, "delegated error fallback 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 the stream —
+        # namespaced, where the root-only fallback detector must ignore it.
+        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
+    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)
+        assert record.status == RunStatus.success, f"run failed: {bridge.published}"
+
+        events = bridge.published
+        # No delegated frame of any mode reaches the parent stream...
+        for event, payload in events:
+            assert not (_collect_ids(payload) & _CHILD_MESSAGE_IDS), f"delegated messages leaked into event {event!r}: {payload}"
+        assert not [event for event, _payload in events if "|" in event]
+
+        # ...while the parent's own frames and the task_* progress contract
+        # (what the web frontend relies on instead of the flag) still hold.
+        bare_values = [payload for event, payload in events if event == "values"]
+        assert any(_PARENT_FINAL_ID in _collect_ids(payload) for payload in bare_values)
+        custom_types = [payload.get("type") for event, payload in events if event == "custom" and isinstance(payload, dict)]
+        assert "task_started" in custom_types and "task_completed" in custom_types

frontend/src/core/threads/hooks.ts

@@ -1537,7 +1537,9 @@ export function useThreadStream({
           },
           {
             threadId: threadId,
-            streamSubgraphs: true,
+            // No streamSubgraphs: subtask progress arrives via root-namespace
+            // custom events, while subgraph frames would leak a delegated
+            // subagent's values/messages into the thread view (#4399).
             streamResumable: true,
             config: {
               recursion_limit: 1000,
@@ -1643,7 +1645,7 @@ export function useThreadStream({
           threadId,
           checkpoint: prepared.checkpoint,
           metadata: prepared.metadata,
-          streamSubgraphs: true,
+          // No streamSubgraphs — same contract as the main submit path (#4399).
           streamResumable: true,
           config: {
             recursion_limit: 1000,