fix(langchain): handle parallel usage of the todo tool in planning middleware (#34637)

The agent should only make a single call to update the todo list at a
time. A parallel call doesn't make sense, but also cannot work as
there's no obvious reducer to use.

On parallel calls of the todo tool, we return ToolMessage containing to
guide the LLM to not call the tool in parallel.

---------

Co-authored-by: Eugene Yurtsev <[email protected]>

Author: hwchase17

libs/langchain_v1/langchain/agents/middleware/todo.py

@@ -2,12 +2,14 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Annotated, Literal, cast
+from typing import TYPE_CHECKING, Annotated, Any, Literal, cast
 
 if TYPE_CHECKING:
     from collections.abc import Awaitable, Callable
 
-from langchain_core.messages import SystemMessage, ToolMessage
+    from langgraph.runtime import Runtime
+
+from langchain_core.messages import AIMessage, SystemMessage, ToolMessage
 from langchain_core.tools import tool
 from langgraph.types import Command
 from typing_extensions import NotRequired, TypedDict
@@ -135,7 +137,9 @@ class TodoListMiddleware(AgentMiddleware):
     into task completion status.
 
     The middleware automatically injects system prompts that guide the agent on when
-    and how to use the todo functionality effectively.
+    and how to use the todo functionality effectively. It also enforces that the
+    `write_todos` tool is called at most once per model turn, since the tool replaces
+    the entire todo list and parallel calls would create ambiguity about precedence.
 
     Example:
         ```python
@@ -222,3 +226,79 @@ async def awrap_model_call(
             content=cast("list[str | dict[str, str]]", new_system_content)
         )
         return await handler(request.override(system_message=new_system_message))
+
+    def after_model(
+        self,
+        state: AgentState,
+        runtime: Runtime,  # noqa: ARG002
+    ) -> dict[str, Any] | None:
+        """Check for parallel write_todos tool calls and return errors if detected.
+
+        The todo list is designed to be updated at most once per model turn. Since
+        the `write_todos` tool replaces the entire todo list with each call, making
+        multiple parallel calls would create ambiguity about which update should take
+        precedence. This method prevents such conflicts by rejecting any response that
+        contains multiple write_todos tool calls.
+
+        Args:
+            state: The current agent state containing messages.
+            runtime: The LangGraph runtime instance.
+
+        Returns:
+            A dict containing error ToolMessages for each write_todos call if multiple
+            parallel calls are detected, otherwise None to allow normal execution.
+        """
+        messages = state["messages"]
+        if not messages:
+            return None
+
+        last_ai_msg = next((msg for msg in reversed(messages) if isinstance(msg, AIMessage)), None)
+        if not last_ai_msg or not last_ai_msg.tool_calls:
+            return None
+
+        # Count write_todos tool calls
+        write_todos_calls = [tc for tc in last_ai_msg.tool_calls if tc["name"] == "write_todos"]
+
+        if len(write_todos_calls) > 1:
+            # Create error tool messages for all write_todos calls
+            error_messages = [
+                ToolMessage(
+                    content=(
+                        "Error: The `write_todos` tool should never be called multiple times "
+                        "in parallel. Please call it only once per model invocation to update "
+                        "the todo list."
+                    ),
+                    tool_call_id=tc["id"],
+                    status="error",
+                )
+                for tc in write_todos_calls
+            ]
+
+            # Keep the tool calls in the AI message but return error messages
+            # This follows the same pattern as HumanInTheLoopMiddleware
+            return {"messages": error_messages}
+
+        return None
+
+    async def aafter_model(
+        self,
+        state: AgentState,
+        runtime: Runtime,
+    ) -> dict[str, Any] | None:
+        """Check for parallel write_todos tool calls and return errors if detected.
+
+        Async version of `after_model`. The todo list is designed to be updated at
+        most once per model turn. Since the `write_todos` tool replaces the entire
+        todo list with each call, making multiple parallel calls would create ambiguity
+        about which update should take precedence. This method prevents such conflicts
+        by rejecting any response that contains multiple write_todos tool calls.
+
+        Args:
+            state: The current agent state containing messages.
+            runtime: The LangGraph runtime instance.
+
+        Returns:
+            A dict containing error ToolMessages for each write_todos call if multiple
+            parallel calls are detected, otherwise None to allow normal execution.
+        """
+        return self.after_model(state, runtime)