fix(sdk): keep private state out of subagent propagation (#3542)

This change makes Deep Agents derive private state keys directly from
middleware state schemas and filters them out when state is copied into
subagents.

It fixes two leak paths:
- parent state fields marked with `PrivateStateAttr` no longer propagate
into child agents
- private fields written by one sibling subagent no longer leak into
another sibling through shared runtime state

The test coverage now exercises the declarative `SubAgent` path with
fake chat models and verifies both parent-to-child and
sibling-to-sibling isolation.

---------

Co-authored-by: Mason Daugherty <mason@langchain.dev>
Co-authored-by: Sydney Runkle <sydneymarierunkle@gmail.com>

libs/deepagents/deepagents/graph.py

@@ -38,6 +38,7 @@ from deepagents._version import __version__
 from deepagents.backends import StateBackend
 from deepagents.backends.protocol import BackendFactory, BackendProtocol
 from deepagents.middleware._fs_interrupt import _build_interrupt_on_from_permissions
+from deepagents.middleware._state import private_state_field_names
 from deepagents.middleware._tool_exclusion import _ToolExclusionMiddleware
 from deepagents.middleware.async_subagents import AsyncSubAgent, AsyncSubAgentMiddleware
 from deepagents.middleware.filesystem import FilesystemMiddleware, FilesystemPermission
@@ -817,6 +818,9 @@ def create_deep_agent(  # noqa: C901, PLR0912, PLR0915  # Complex graph assembly
         matched_classes=_main_matched_classes,
         matched_names=_main_matched_names,
     )
+    private_state_keys = private_state_field_names(*(mw.state_schema for mw in deepagent_middleware if getattr(mw, "state_schema", None) is not None))
+    if sub_agent_middleware is not None:
+        sub_agent_middleware.private_state_keys = private_state_keys
     # Verify every main-profile exclusion matched at least one middleware in
     # either the main agent stack or the GP subagent stack. An entry that
     # matched nothing across both is almost certainly a typo or a stale

libs/deepagents/deepagents/middleware/_state.py

@@ -0,0 +1,30 @@
+"""Helpers for working with Deep Agents state schemas."""
+
+from __future__ import annotations
+
+import contextlib
+from typing import Annotated, get_args, get_origin, get_type_hints
+
+from langchain.agents.middleware.types import PrivateStateAttr
+
+
+def private_state_field_names(*state_schemas: type[object]) -> frozenset[str]:
+    """Return fields annotated with `PrivateStateAttr` across state schemas."""
+    names: set[str] = set()
+    for state_schema in state_schemas:
+        with contextlib.suppress(Exception):
+            hints = get_type_hints(state_schema, include_extras=True)
+            for name, annotation in hints.items():
+                if _has_marker(annotation, PrivateStateAttr):
+                    names.add(name)
+    return frozenset(names)
+
+
+def _has_marker(annotation: object, marker: object) -> bool:
+    origin = get_origin(annotation)
+    if origin is Annotated:
+        args = get_args(annotation)
+        return any(meta is marker for meta in args[1:])
+    if origin is not None:
+        return any(_has_marker(arg, marker) for arg in get_args(annotation))
+    return False

libs/deepagents/deepagents/middleware/subagents.py

@@ -241,9 +241,6 @@ _EXCLUDED_STATE_KEYS = {
     "messages",
     "todos",
     "structured_response",
-    "skills_metadata",
-    "skills_load_errors",
-    "memory_contents",
 }
 """State keys that are excluded when passing state to subagents and when
 returning updates from subagents.
@@ -255,12 +252,8 @@ When returning updates:
 2. The todos and `structured_response` keys are excluded as they do not have
     a defined reducer and no clear meaning for returning them from a subagent
     to the main agent.
-3. The `skills_metadata`, `skills_load_errors`, and `memory_contents` keys are
-    automatically excluded from subagent output via `PrivateStateAttr`
-    annotations on their respective state schemas. However, they must ALSO
-    be explicitly filtered from runtime.state when invoking a subagent to
-    prevent parent state from leaking to child agents (e.g., the general-purpose
-    subagent loads its own skills via `SkillsMiddleware`).
+3. Agent-private fields on middleware state schemas are excluded from both
+    subagent output and subagent inputs.
 """
 
 
@@ -467,6 +460,8 @@ def _subagent_tracing_context() -> Generator[None, None, None]:
 def _build_task_tool(  # noqa: C901, PLR0915
     subagents: list[_SubagentSpec],
     task_description: str | None = None,
+    *,
+    private_state_keys: frozenset[str] = frozenset(),
 ) -> BaseTool:
     """Create a task tool from pre-built subagent graphs.
 
@@ -474,6 +469,8 @@ def _build_task_tool(  # noqa: C901, PLR0915
         subagents: List of subagent specs containing name, description, and runnable.
         task_description: Custom description for the task tool. If `None`,
             uses default template. Supports `{available_agents}` placeholder.
+        private_state_keys: State keys marked with `PrivateStateAttr` that
+            should be stripped from parent state before invoking subagents.
 
     Returns:
         A StructuredTool that can invoke subagents by type.
@@ -536,6 +533,7 @@ def _build_task_tool(  # noqa: C901, PLR0915
         subagent = subagent_graphs[subagent_type]
         # Create a new state dict to avoid mutating the original
         subagent_state = {k: v for k, v in runtime.state.items() if k not in _EXCLUDED_STATE_KEYS}
+        subagent_state = {k: v for k, v in subagent_state.items() if k not in private_state_keys}
         subagent_state["messages"] = [HumanMessage(content=description)]
         return subagent, subagent_state
 
@@ -661,6 +659,7 @@ class SubAgentMiddleware(AgentMiddleware[Any, ContextT, ResponseT]):
         subagents: Sequence[SubAgent | CompiledSubAgent],
         system_prompt: str | None = TASK_SYSTEM_PROMPT,
         task_description: str | None = None,
+        private_state_keys: frozenset[str] | None = None,
         state_schema: type | None = None,
     ) -> None:
         """Initialize the `SubAgentMiddleware`."""
@@ -671,13 +670,20 @@ class SubAgentMiddleware(AgentMiddleware[Any, ContextT, ResponseT]):
             raise ValueError(msg)
         self._backend = backend
         self._subagents = subagents
+        self._private_state_keys = private_state_keys or frozenset()
+        self._task_description = task_description
         self._state_schema = state_schema
         subagent_specs = self._get_subagents()
+        self._subagent_specs = subagent_specs
         self.subagent_names: frozenset[str] = frozenset(spec["name"] for spec in subagent_specs)
         """Declared subagent names. Public so streamers can discover them
         without introspecting the `task` tool's closure."""
 
-        task_tool = _build_task_tool(subagent_specs, task_description)
+        task_tool = _build_task_tool(
+            subagent_specs,
+            task_description,
+            private_state_keys=self._private_state_keys,
+        )
 
         # Build system prompt with available agents
         if system_prompt and subagent_specs:
@@ -688,6 +694,21 @@ class SubAgentMiddleware(AgentMiddleware[Any, ContextT, ResponseT]):
 
         self.tools = [task_tool]
 
+    @property
+    def private_state_keys(self) -> frozenset[str]:
+        """State keys stripped from parent state before invoking subagents."""
+        return self._private_state_keys
+
+    @private_state_keys.setter
+    def private_state_keys(self, value: frozenset[str]) -> None:
+        self._private_state_keys = value
+        task_tool = _build_task_tool(
+            self._subagent_specs,
+            task_description=self._task_description,
+            private_state_keys=value,
+        )
+        self.tools = [task_tool]
+
     def _get_subagents(self) -> list[_SubagentSpec]:
         """Create runnable agents from specs.
 

libs/deepagents/tests/unit_tests/test_subagents.py

@@ -10,12 +10,13 @@ import json
 import uuid
 from collections.abc import Callable, Sequence
 from pathlib import Path
-from typing import Any, TypedDict
+from typing import Annotated, Any, TypedDict
 from unittest.mock import MagicMock
 
 import pytest
 from langchain.agents import create_agent
 from langchain.agents.middleware import TodoListMiddleware
+from langchain.agents.middleware.types import AgentMiddleware, AgentState, PrivateStateAttr
 from langchain.agents.structured_output import ToolStrategy
 from langchain.tools import ToolRuntime
 from langchain_core.callbacks import BaseCallbackHandler, CallbackManagerForLLMRun
@@ -327,6 +328,168 @@ class TestSubAgents:
             f"Multiplication subagent should return exact message, got: {multiplication_tool_message.content}"
         )
 
+    def test_private_state_does_not_propagate_between_sibling_subagents(self) -> None:
+        """A private state field should not propagate from one sibling subagent to another."""
+
+        class _LocalPrivateState(AgentState):
+            shared_value: Annotated[str | None, PrivateStateAttr]
+
+        class _LocalPrivateMiddleware(AgentMiddleware[_LocalPrivateState, Any, Any]):
+            state_schema = _LocalPrivateState
+
+            def before_agent(self, state: _LocalPrivateState, runtime: object) -> dict[str, Any] | None:
+                if "shared_value" in state:
+                    return None
+                return {"shared_value": "seeded"}
+
+        parent_chat_model = GenericFakeChatModel(
+            messages=iter(
+                [
+                    AIMessage(
+                        content="",
+                        tool_calls=[
+                            {
+                                "name": "task",
+                                "args": {
+                                    "description": "Seed the interpreter state",
+                                    "subagent_type": "writer",
+                                },
+                                "id": "call_writer",
+                                "type": "tool_call",
+                            }
+                        ],
+                    ),
+                    AIMessage(
+                        content="",
+                        tool_calls=[
+                            {
+                                "name": "task",
+                                "args": {
+                                    "description": "Read the interpreter state",
+                                    "subagent_type": "reader",
+                                },
+                                "id": "call_reader",
+                                "type": "tool_call",
+                            }
+                        ],
+                    ),
+                    AIMessage(content="done"),
+                ]
+            )
+        )
+
+        writer_model = GenericFakeChatModel(messages=iter([AIMessage(content="writer saw seeded")]))
+        reader_model = GenericFakeChatModel(messages=iter([AIMessage(content="reader saw missing")]))
+
+        parent_agent = create_deep_agent(
+            model=parent_chat_model,
+            checkpointer=InMemorySaver(),
+            subagents=[
+                SubAgent(
+                    name="writer",
+                    description="Writes state.",
+                    system_prompt="Write the seeded private state value and report completion.",
+                    model=writer_model,
+                    middleware=[_LocalPrivateMiddleware()],
+                ),
+                SubAgent(
+                    name="reader",
+                    description="Reads state.",
+                    system_prompt="Read the private state value and report what you received.",
+                    model=reader_model,
+                ),
+            ],
+        )
+
+        result = parent_agent.invoke(
+            {"messages": [HumanMessage(content="run the two subagents")]},
+            config={"configurable": {"thread_id": "test_shared_quickjs_subagents"}},
+        )
+
+        tool_messages = [msg for msg in result["messages"] if msg.type == "tool"]
+        assert len(tool_messages) == 2
+        assert "seeded" in tool_messages[0].content
+        assert "missing" in tool_messages[1].content
+
+    def test_private_state_does_not_propagate_from_parent_to_subagent(self) -> None:
+        """A private state field on the parent should not be visible to a child subagent."""
+
+        class _ParentPrivateState(AgentState):
+            shared_value: Annotated[str | None, PrivateStateAttr]
+
+        class _ChildCaptureState(AgentState):
+            shared_value: Annotated[str | None, PrivateStateAttr]
+
+        captured_child_states: list[dict[str, Any]] = []
+
+        class _ChildCaptureMiddleware(AgentMiddleware[_ChildCaptureState, Any, Any]):
+            state_schema = _ChildCaptureState
+
+            def before_agent(self, state: _ChildCaptureState, runtime: object) -> dict[str, Any] | None:
+                captured_child_states.append(dict(state))
+                return None
+
+        class _ParentSeedMiddleware(AgentMiddleware[_ParentPrivateState, Any, Any]):
+            state_schema = _ParentPrivateState
+
+            def before_agent(self, state: _ParentPrivateState, runtime: object) -> dict[str, Any] | None:
+                if "shared_value" in state:
+                    return None
+                return {"shared_value": "parent-secret"}
+
+        parent_chat_model = GenericFakeChatModel(
+            messages=iter(
+                [
+                    AIMessage(
+                        content="",
+                        tool_calls=[
+                            {
+                                "name": "task",
+                                "args": {
+                                    "description": "Run the child subagent",
+                                    "subagent_type": "child",
+                                },
+                                "id": "call_child",
+                                "type": "tool_call",
+                            }
+                        ],
+                    ),
+                    AIMessage(content="done"),
+                ]
+            )
+        )
+
+        child_model = GenericFakeChatModel(messages=iter([AIMessage(content="child done")]))
+
+        parent_agent = create_deep_agent(
+            model=parent_chat_model,
+            checkpointer=InMemorySaver(),
+            middleware=[_ParentSeedMiddleware()],
+            subagents=[
+                SubAgent(
+                    name="child",
+                    description="Captures its incoming state.",
+                    system_prompt="Capture the incoming state and complete the task.",
+                    model=child_model,
+                    middleware=[_ChildCaptureMiddleware()],
+                ),
+            ],
+        )
+
+        result = parent_agent.invoke(
+            {"messages": [HumanMessage(content="run the child subagent")]},
+            config={
+                "configurable": {"thread_id": "test_private_state_parent_to_child"},
+                "metadata": {"shared_value": "parent-secret"},
+            },
+        )
+
+        tool_messages = [msg for msg in result["messages"] if msg.type == "tool"]
+        assert len(tool_messages) == 1
+        assert "child done" in tool_messages[0].content
+        assert captured_child_states, "Child subagent should have received state"
+        assert "shared_value" not in captured_child_states[0]
+
     def test_agent_with_structured_output_tool_strategy(self) -> None:
         """Test that an agent with ToolStrategy properly generates structured output.