diff --git a/changelogs/fragments/fix-copy-plugin-code-bug-in-directory.yml b/changelogs/fragments/fix-copy-plugin-code-bug-in-directory.yml
new file mode 100644
index 00000000000..252233d3773
--- /dev/null
+++ b/changelogs/fragments/fix-copy-plugin-code-bug-in-directory.yml
@@ -0,0 +1,2 @@
+bugfixes:
+  - fix bug in copy plugin where directory may be repeatedly using the file module to create
diff --git a/lib/ansible/plugins/action/copy.py b/lib/ansible/plugins/action/copy.py
index 2047671b47c..f3abf229858 100644
--- a/lib/ansible/plugins/action/copy.py
+++ b/lib/ansible/plugins/action/copy.py
@@ -525,11 +525,13 @@ class ActionModule(ActionBase):
                 result.update(module_return)
                 return self._ensure_invocation(result)
 
-            paths = os.path.split(source_rel)
+            paths = source_rel.split(os.path.sep)
             dir_path = ''
-            for dir_component in paths:
-                os.path.join(dir_path, dir_component)
+            # skip last file name
+            for dir_component in paths[:-1]:
+                dir_path = os.path.join(dir_path, dir_component)
                 implicit_directories.add(dir_path)
+
             if 'diff' in result and not result['diff']:
                 del result['diff']
             module_executed = True
diff --git a/test/units/plugins/action/test_copy.py b/test/units/plugins/action/test_copy.py
new file mode 100644
index 00000000000..2ace87239a2
--- /dev/null
+++ b/test/units/plugins/action/test_copy.py
@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+import os
+
+import unittest
+
+
+class TestCopyDirectoryData(unittest.TestCase):
+    
+    def test_copy_directory(self):
+        """Verify code from copy plugin"""
+        source_files = {
+            'files': [
+                ('/home/store/test/b/bb/test_bb.txt', 'b/bb/test_bb.txt'), 
+                ('/home/store/test/a/test2.txt', 'a/test2.txt'), 
+                ('/home/store/test/a/test1.txt', 'a/test1.txt'), 
+                ('/home/store/test/e/eb/test_e.txt', 'e/eb/test_e.txt')
+            ], 
+            'directories': [
+                ('/home/store/test/f', 'f'), 
+                ('/home/store/test/d', 'd'), 
+                ('/home/store/test/b', 'b'), 
+                ('/home/store/test/a', 'a'), 
+                ('/home/store/test/e', 'e'), 
+                ('/home/store/test/c', 'c'), 
+                ('/home/store/test/b/bc', 'b/bc'), 
+                ('/home/store/test/b/bb', 'b/bb'), 
+                ('/home/store/test/e/ec', 'e/ec'), 
+                ('/home/store/test/e/ea', 'e/ea'), 
+                ('/home/store/test/e/eb', 'e/eb')
+            ], 
+            'symlinks': []
+        }
+
+        implicit_directories = set()
+        for source_full, source_rel in source_files['files']:
+            paths = source_rel.split(os.path.sep)
+            dir_path = ''
+            # skip last file name
+            for dir_component in paths[:-1]:   
+                dir_path = os.path.join(dir_path, dir_component)
+                print(dir_path)
+                implicit_directories.add(dir_path)
+        
+        self.assertSetEqual(implicit_directories, {'a', 'b', 'e', 'b/bb', 'e/eb'})
+
+        leaves = set()
+        for src, dest_path in source_files['directories']:
+            if dest_path in implicit_directories:
+                continue
+            leaves.add(dest_path)
+
+        self.assertSetEqual(
+            leaves, 
+            {'b/bc', 'c', 'd', 'e/ea', 'e/ec', 'f'}
+        )