test_parallel_write_todos_calls_rejected_async() — langchain Function Reference

Source Code

libs/langchain_v1/tests/unit_tests/agents/middleware/implementations/test_todo.py lines 652–702

async def test_parallel_write_todos_calls_rejected_async() -> None:
    """Test async version - parallel write_todos calls are rejected with error messages."""
    middleware = TodoListMiddleware()

    # Create an AI message with two write_todos tool calls
    ai_message = AIMessage(
        content="I'll update the todos",
        tool_calls=[
            {
                "name": "write_todos",
                "args": {"todos": [{"content": "Task 1", "status": "pending"}]},
                "id": "call_1",
                "type": "tool_call",
            },
            {
                "name": "write_todos",
                "args": {"todos": [{"content": "Task 2", "status": "pending"}]},
                "id": "call_2",
                "type": "tool_call",
            },
        ],
    )

    state: PlanningState = {"messages": [HumanMessage(content="Hello"), ai_message]}

    # Call aafter_model hook
    result = await middleware.aafter_model(state, _fake_runtime())

    # Should return error messages
    assert result == {
        "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="call_1",
                status="error",
            ),
            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="call_2",
                status="error",
            ),
        ]
    }