perf(stack-host-copy): write a changed tree in one transfer

When the tree check finds differing entries, pack exactly the differing files into one tar archive in source walk order, transfer it once into the remote tmpdir and let the new copy_batch module write each file through atomic_move with the copy module's directory, mode and unsafe_writes semantics. Mode-only drift and leaf directories settle in the same call; unchanged files are not sent.

Every shape the regular copy treats differently makes the module refuse before it writes: a symlink, directory or other non-regular entry at a destination, a missing path below a non-directory or a symlink loop, and two source paths reaching one remote entry. The regular copy then runs. Check mode, diff mode, unprivileged become, a dest holding $ and an archive above 16 MiB keep the regular copy. A write failure fails the task and names the file.

Executed with ansible-playbook -c local, pipelining on and off: the trees this action leaves equal the builtin copy's, including the symlink fallback; a 6-file changed tree took 2 module calls and 1 transfer instead of 14 and 4. Unit: a flat 15-file tree with every file changed drops from 31 module calls and 15 transfers to 2 and 1. Two dialectic rounds with three independent skeptics; every hole they found is fixed and tested. A third independent review compared end state, changed, check and diff mode against the builtin copy in a real playbook run and found them equal; one deliberate difference remains: a write failing partway reports changed=true with the files already written, where the builtin copy reports changed=false.

| 🎯 Fix | 💥 Break | 🌱 Efficiency | ⚡ Speed | 🛡️ Security | 🏆 Score | Gate |
| --- | --- | --- | --- | --- | --- | --- |
| 90 % | 2 % | 100 % | 15 % (estimated from round trips) | ~100 % | 100 | ⏳ score passes, gate.py not run |

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

library/copy_batch.py

@@ -0,0 +1,332 @@
+"""Write the differing entries of a directory copy in one module call.
+
+Every path is handled like ``ansible.builtin.copy`` handles one entry of a
+recursive copy: a missing parent directory is created and receives
+``directory_mode`` while an existing one stays as found, a file is replaced
+through ``atomic_move`` and receives ``mode``, and a leaf directory is ensured
+like ``ansible.builtin.file`` with ``state: directory``. Files are handled in
+the order of ``tree``, which is the order the copy action walks the source
+in, and leaf directories after them.
+Nothing is written when a file path exists as anything but a regular file, a
+leaf directory path as anything but a directory, a missing path lies below
+anything but a directory, or two paths of ``tree`` name the same remote
+entry; the call then only reports ``refused``.
+
+Args:
+    src: tar archive holding exactly one regular member per entry of ``files``.
+    dest: absolute destination directory.
+    files: list of ``{path, checksum}`` in source walk order, ``path``
+        relative to ``dest`` and ``checksum`` the SHA-1 of its archive member.
+    settled: relative paths of regular files whose content is in place; only
+        ``mode`` is enforced on them.
+    directories: relative paths of leaf directories to ensure.
+    tree: relative path of every file and leaf directory of the source tree,
+        in walk order; holds every path of ``files``, ``settled`` and
+        ``directories``.
+    mode: mode of every path in ``files`` and ``settled``.
+    directory_mode: mode of every directory this call creates and of every
+        path in ``directories``; ``None`` keeps the mode the umask gives.
+    unsafe_writes: the file common argument of the same name, handed to
+        ``atomic_move`` as the copy module does.
+
+Returns:
+    changed: whether any file, mode or directory changed.
+    refused: the reasons nothing was written; present only then.
+"""
+
+import hashlib
+import os
+import shutil
+import stat
+import tarfile
+import tempfile
+from pathlib import Path, PurePosixPath
+
+from ansible.module_utils.basic import FILE_COMMON_ARGUMENTS, AnsibleModule
+
+ARGUMENT_SPEC = {
+    "src": {"type": "path", "required": True},
+    "dest": {"type": "path", "required": True},
+    "files": {
+        "type": "list",
+        "elements": "dict",
+        "required": True,
+        "options": {
+            "path": {"type": "str", "required": True},
+            "checksum": {"type": "str", "required": True},
+        },
+    },
+    "settled": {"type": "list", "elements": "str", "required": True},
+    "directories": {"type": "list", "elements": "str", "required": True},
+    "tree": {"type": "list", "elements": "str", "required": True},
+    "mode": {"type": "raw", "required": True},
+    "directory_mode": {"type": "raw"},
+    "unsafe_writes": FILE_COMMON_ARGUMENTS["unsafe_writes"],
+}
+
+_CHUNK = 1 << 20
+
+
+def is_relative(name):
+    """Tell whether *name* is a normalised path strictly below the destination.
+
+    Args:
+        name: relative path from the module arguments or the archive.
+    """
+    path = PurePosixPath(name)
+    return (
+        bool(path.parts)
+        and str(path) == name
+        and not path.is_absolute()
+        and ".." not in path.parts
+    )
+
+
+def blocker(path, is_kind, kind, may_be_missing):
+    """Name what keeps *path* from taking an entry of the wanted type.
+
+    Args:
+        path: absolute destination path.
+        is_kind: ``stat.S_ISREG`` or ``stat.S_ISDIR``; an existing entry must
+            pass it, so a symlink never does.
+        kind: the wanted type in words.
+        may_be_missing: whether a missing *path* below a directory is fine.
+
+    Returns:
+        ``None`` when *path* can take the entry, else the reason it cannot.
+    """
+    try:
+        info = path.lstat()
+    except FileNotFoundError:
+        if not may_be_missing:
+            return f"{path} is missing"
+        parent = path.parent
+        while not os.path.lexists(parent):
+            parent = parent.parent
+        return None if parent.is_dir() else f"{parent} is not a directory"
+    except OSError as error:
+        return f"{path}: {error.strerror}"
+    return None if is_kind(info.st_mode) else f"{path} is not {kind}"
+
+
+def aliases(dest, tree):
+    """Name the paths of *tree* that reach the same remote entry.
+
+    Args:
+        dest: the destination directory.
+        tree: relative paths of every file and leaf directory of the source.
+
+    Returns:
+        One reason per path that reaches an entry an earlier path reaches.
+    """
+    seen = {}
+    reasons = []
+    for name in tree:
+        path = dest / name
+        try:
+            info = path.stat()
+            key = (info.st_dev, info.st_ino)
+        except OSError:
+            key = str(Path(os.path.realpath(path.parent)) / path.name)
+        if key in seen:
+            reasons.append(f"{dest / seen[key]} and {path} reach the same entry")
+        else:
+            seen[key] = name
+    return reasons
+
+
+def refusals(dest, files, settled, directories, tree):
+    """Collect what keeps any path of the call from taking its entry.
+
+    Args:
+        dest: the destination directory.
+        files: relative paths of the files to write.
+        settled: relative paths of the files to keep.
+        directories: relative paths of the leaf directories to ensure.
+        tree: relative paths of every file and leaf directory of the source.
+
+    Returns:
+        The reasons, empty when every path can take its entry.
+    """
+    checks = [
+        *((name, stat.S_ISREG, "a regular file", True) for name in files),
+        *((name, stat.S_ISREG, "a regular file", False) for name in settled),
+        *((name, stat.S_ISDIR, "a directory", True) for name in directories),
+    ]
+    reasons = [
+        blocker(dest / name, is_kind, kind, may_be_missing)
+        for name, is_kind, kind, may_be_missing in checks
+    ]
+    return [reason for reason in reasons if reason] + aliases(dest, tree)
+
+
+def members(module, archive, expected):
+    """Map every path of *expected* to its archive member, failing on any other.
+
+    Args:
+        module: the running module.
+        archive: the open tar archive.
+        expected: mapping of relative path to the SHA-1 of its member.
+
+    Returns:
+        Mapping of relative path to its tar member.
+    """
+    found = {}
+    for member in archive.getmembers():
+        name = member.name
+        if not is_relative(name):
+            module.fail_json(msg=f"archive member {name!r} leaves the destination")
+        if name not in expected or name in found:
+            module.fail_json(msg=f"archive member {name!r} is unexpected or repeated")
+        if not member.isreg():
+            module.fail_json(msg=f"archive member {name!r} is not a regular file")
+        found[name] = member
+    missing = sorted(set(expected) - set(found))
+    if missing:
+        module.fail_json(msg=f"archive lacks {', '.join(missing)}")
+    return found
+
+
+def sha1(archive, member):
+    """Return the SHA-1 of the content of *member*.
+
+    Args:
+        archive: the open tar archive.
+        member: a regular tar member.
+    """
+    digest = hashlib.sha1(usedforsecurity=False)
+    with archive.extractfile(member) as handle:
+        while chunk := handle.read(_CHUNK):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def attributes(module, path, mode, changed):
+    """Apply *mode* to *path* like the copy and file modules do.
+
+    Args:
+        module: the running module.
+        path: the file or directory.
+        mode: the wanted mode, or ``None`` to leave it.
+        changed: the change state so far.
+
+    Returns:
+        The change state including this call.
+    """
+    file_args = module.load_file_common_arguments({"path": str(path), "mode": mode})
+    return module.set_fs_attributes_if_different(file_args, changed)
+
+
+def make_directories(module, directory, directory_mode):
+    """Create *directory* and its missing parents, each with *directory_mode*.
+
+    Args:
+        module: the running module.
+        directory: the directory that must exist.
+        directory_mode: mode of every directory created, or ``None``.
+
+    Returns:
+        Whether a directory was missing.
+    """
+    missing = []
+    while not os.path.lexists(directory):
+        missing.append(directory)
+        directory = directory.parent
+    for path in reversed(missing):
+        try:
+            path.mkdir()
+        except FileExistsError:
+            if not path.is_dir():
+                raise
+        attributes(module, path, directory_mode, True)
+    return bool(missing)
+
+
+def write(module, archive, member, target, workdir):
+    """Replace *target* atomically with the content of *member*.
+
+    Args:
+        module: the running module, whose ``atomic_move`` keeps the owner,
+            group and SELinux context of a replaced file.
+        archive: the open tar archive.
+        member: the tar member holding the new content.
+        target: destination path.
+        workdir: directory the content is staged in before the move.
+    """
+    descriptor, staged = tempfile.mkstemp(dir=workdir)
+    try:
+        with os.fdopen(descriptor, "wb") as out, archive.extractfile(member) as data:
+            shutil.copyfileobj(data, out)
+        module.atomic_move(
+            staged, str(target), unsafe_writes=module.params["unsafe_writes"]
+        )
+    finally:
+        Path(staged).unlink(missing_ok=True)
+
+
+def main():
+    module = AnsibleModule(argument_spec=ARGUMENT_SPEC)
+    params = module.params
+    dest = Path(params["dest"])
+    files = [entry["path"] for entry in params["files"]]
+    expected = {entry["path"]: entry["checksum"] for entry in params["files"]}
+    settled, directories = params["settled"], params["directories"]
+    kept = set(settled)
+    directory_mode = params["directory_mode"]
+    names = [*files, *settled, *directories, *params["tree"]]
+    unsafe = sorted({name for name in names if not is_relative(name)})
+    if unsafe:
+        module.fail_json(msg=f"paths leave {dest}: {', '.join(unsafe)}")
+    if len(expected) != len(files):
+        module.fail_json(msg="files repeat a path")
+    outside = sorted(set(names) - set(params["tree"]))
+    if outside:
+        module.fail_json(msg=f"paths missing from tree: {', '.join(outside)}")
+    refused = refusals(dest, files, settled, directories, params["tree"])
+    if refused:
+        module.exit_json(changed=False, refused=refused)
+
+    changed = False
+    with tarfile.open(params["src"], mode="r:") as archive:
+        found = members(module, archive, expected)
+        mismatched = [
+            name for name in files if sha1(archive, found[name]) != expected[name]
+        ]
+        if mismatched:
+            module.fail_json(
+                msg=f"archived files do not match their checksum: {', '.join(mismatched)}"
+            )
+        for name in params["tree"]:
+            target = dest / name
+            if name in kept:
+                changed = attributes(module, target, params["mode"], changed)
+            if name not in expected:
+                continue
+            try:
+                changed = (
+                    make_directories(module, target.parent, directory_mode) or changed
+                )
+                write(module, archive, found[name], target, module.tmpdir)
+            except Exception as error:  # noqa: BLE001 - atomic_move raises a bare Exception
+                module.fail_json(
+                    msg=f"Failed to write {target}: {error}",
+                    path=str(target),
+                    changed=changed,
+                )
+            changed = attributes(module, target, params["mode"], True)
+    for name in directories:
+        path = dest / name
+        try:
+            created = make_directories(module, path, directory_mode)
+        except OSError as error:
+            module.fail_json(
+                msg=f"Failed to create {path}: {error}",
+                path=str(path),
+                changed=changed,
+            )
+        changed = created or attributes(module, path, directory_mode, changed)
+    module.exit_json(changed=changed)
+
+
+if __name__ == "__main__":
+    main()

plugins/action/stack_host_copy.py

@@ -1,11 +1,16 @@
 #!/usr/bin/env python3
 from __future__ import annotations
 
+import hashlib
+import io
 import os
 import re
+import tarfile
+import tempfile
 from pathlib import Path, PurePath
 from typing import Any
 
+from ansible import constants
 from ansible.errors import AnsibleError
 from ansible.module_utils.parsing.convert_bool import boolean as _to_bool
 from ansible.plugins.action import copy as _copy_action
@@ -29,6 +34,8 @@ _TREE_CHECK_ARGS = frozenset(
     }
 )
 _OCTAL_MODE = re.compile(r"^[0-7]{3,4}$")
+_ARCHIVE = ".source.tar"
+_ARCHIVE_LIMIT = 16 * 1024 * 1024
 
 
 def _octal(mode: Any) -> int | None:
@@ -83,9 +90,9 @@ class ActionModule(CopyActionModule):
             }
 
         self._task.args.setdefault("mode", "0644")
-        unchanged = self._unchanged_tree(task_vars)
-        if unchanged is not None:
-            return unchanged
+        settled = self._settled_tree(task_vars)
+        if settled is not None:
+            return settled
         walk = _copy_action._walk_dirs
         _copy_action._walk_dirs = _walk_dirs_without_bytecode_caches
         try:
@@ -93,19 +100,22 @@ class ActionModule(CopyActionModule):
         finally:
             _copy_action._walk_dirs = walk
 
-    def _unchanged_tree(self, task_vars: dict[str, Any]) -> dict[str, Any] | None:
-        """Settle a directory copy whose entries all match in one remote call.
+    def _settled_tree(self, task_vars: dict[str, Any]) -> dict[str, Any] | None:
+        """Settle a directory copy in one checksum call and one write call.
 
-        The copy action spends a stat and a file module call per file, and a
-        file module call per directory holding no file, even when nothing
-        differs; any difference, and any argument or source shape not
-        modelled here, leaves the decision to the regular copy.
+        The copy action spends a stat and a file module call per file, plus a
+        transfer and a copy module call per differing file, and a file module
+        call per directory holding no file. Here one remote call compares the
+        whole tree; when entries differ outside check and diff mode, the
+        differing files travel in one archive and one module call writes them.
+        Any argument or source shape not modelled here leaves the decision to
+        the regular copy.
 
         Args:
-            task_vars: Variables of the task, handed to the remote call.
+            task_vars: Variables of the task, handed to the remote calls.
 
         Returns:
-            The unchanged copy result, or ``None`` when the regular copy runs.
+            The copy result, or ``None`` when the regular copy runs.
         """
         args = self._task.args
         src, dest, mode = args.get("src"), args.get("dest"), _octal(args.get("mode"))
@@ -116,6 +126,7 @@ class ActionModule(CopyActionModule):
             or not src.endswith(os.path.sep)
             or not isinstance(dest, str)
             or not dest.startswith("/")
+            or "$" in dest
             or mode is None
             or not _to_bool(args.get("force", True), strict=False)
         ):
@@ -146,7 +157,7 @@ class ActionModule(CopyActionModule):
             return None
 
         dest = shell.join_path(dest, "")
-        directories = [shell.join_path(dest, rel) for rel in leaves]
+        directories = {shell.join_path(dest, rel): rel for rel in leaves}
         decrypt = _to_bool(args.get("decrypt", True), strict=False)
         wanted = {}
         for full, rel in tree["files"]:
@@ -155,7 +166,7 @@ class ActionModule(CopyActionModule):
             except AnsibleError:
                 return None
             try:
-                wanted[shell.join_path(dest, rel)] = checksum(real)
+                wanted[shell.join_path(dest, rel)] = (full, rel, checksum(real))
             finally:
                 self._loader.cleanup_tmp_file(real)
 
@@ -167,21 +178,148 @@ class ActionModule(CopyActionModule):
         states = found.get("states")
         if found.get("failed") or not isinstance(states, dict):
             return None
-        for path, digest in wanted.items():
+        content, settled = {}, []
+        for path, (full, rel, digest) in wanted.items():
             state = states.get(path) or {}
-            if (
-                not state.get("isreg")
-                or state.get("checksum") != digest
-                or _octal(state.get("mode")) != mode
-            ):
-                return None
-        for path in directories:
+            if not state.get("isreg") or state.get("checksum") != digest:
+                content[rel] = full
+            elif _octal(state.get("mode")) != mode:
+                settled.append(rel)
+        drifted = []
+        for path, rel in directories.items():
             state = states.get(path) or {}
             if not state.get("isdir") or (
                 directory_mode is not None
                 and _octal(state.get("mode")) != _octal(directory_mode)
             ):
-                return None
+                drifted.append(rel)
 
+        if not (content or settled or drifted):
+            self._remove_tmp_path(shell.tmpdir)
+            return self._ensure_invocation(
+                {"changed": False, "dest": dest, "src": source}
+            )
+        if self._task.check_mode or self._task.diff or self._is_become_unprivileged():
+            return None
+        return self._write_tree(
+            task_vars,
+            source=source,
+            dest=dest,
+            content=content,
+            settled=settled,
+            directories=drifted,
+            tree=[*(rel for _, rel in tree["files"]), *leaves],
+            decrypt=decrypt,
+        )
+
+    def _write_tree(
+        self,
+        task_vars: dict[str, Any],
+        *,
+        source: str,
+        dest: str,
+        content: dict[str, str],
+        settled: list[str],
+        directories: list[str],
+        tree: list[str],
+        decrypt: bool,
+    ) -> dict[str, Any] | None:
+        """Write the differing entries of a directory copy in one module call.
+
+        Args:
+            task_vars: Variables of the task, handed to the remote call.
+            source: The local source directory, with a trailing slash.
+            dest: The remote destination directory, with a trailing slash.
+            content: Relative path to local path of every file whose remote
+                content differs or is missing, in walk order.
+            settled: Relative paths of the files whose mode alone differs.
+            directories: Relative paths of the leaf directories that are
+                missing or whose mode differs.
+            tree: Relative paths of every file and leaf directory of the
+                source, for the remote check that no two reach one entry.
+            decrypt: Whether vaulted source files are decrypted.
+
+        Returns:
+            The copy result, or ``None`` when the archive would exceed
+            ``_ARCHIVE_LIMIT`` or the module refuses a remote entry, and the
+            regular copy runs.
+        """
+        shell = self._connection._shell
+        descriptor, archive = tempfile.mkstemp(
+            dir=constants.DEFAULT_LOCAL_TMP, prefix="."
+        )
+        try:
+            with os.fdopen(descriptor, "wb") as handle:
+                files = self._pack(handle, content, decrypt)
+            if files is None:
+                return None
+            if shell.tmpdir is None:
+                self._make_tmp_path()
+            remote = self._transfer_file(
+                archive, shell.join_path(shell.tmpdir, _ARCHIVE)
+            )
+        finally:
+            Path(archive).unlink()
+        self._fixup_perms2((shell.tmpdir, remote))
+        written = self._execute_module(
+            module_name="ansible.legacy.copy_batch",
+            module_args={
+                "src": remote,
+                "dest": dest,
+                "files": files,
+                "settled": settled,
+                "directories": directories,
+                "tree": tree,
+                "mode": self._task.args["mode"],
+                "directory_mode": self._task.args.get("directory_mode"),
+            },
+            task_vars=task_vars,
+        )
+        if written.get("refused"):
+            return None
         self._remove_tmp_path(shell.tmpdir)
-        return self._ensure_invocation({"changed": False, "dest": dest, "src": source})
+        if written.get("failed"):
+            return self._ensure_invocation(written)
+        return self._ensure_invocation(
+            {"changed": written["changed"], "dest": dest, "src": source}
+        )
+
+    def _pack(
+        self, handle: Any, content: dict[str, str], decrypt: bool
+    ) -> list[dict[str, str]] | None:
+        """Write the files of *content* as regular members of a tar archive.
+
+        Args:
+            handle: Binary file object the archive is written to.
+            content: Relative path to local path of every file, in walk order.
+            decrypt: Whether vaulted source files are decrypted.
+
+        Returns:
+            ``{path, checksum}`` of every member in walk order, the checksum
+            taken from the bytes packed, or ``None`` once the members exceed
+            ``_ARCHIVE_LIMIT`` bytes.
+        """
+        files, size = [], 0
+        with tarfile.open(fileobj=handle, mode="w") as bundle:
+            for rel, full in content.items():
+                real = self._loader.get_real_file(full, decrypt=decrypt)
+                try:
+                    with Path(real).open("rb") as source:
+                        data = source.read(_ARCHIVE_LIMIT + 1 - size)
+                finally:
+                    self._loader.cleanup_tmp_file(real)
+                size += len(data)
+                if size > _ARCHIVE_LIMIT:
+                    return None
+                member = tarfile.TarInfo(rel)
+                member.size = len(data)
+                bundle.addfile(member, io.BytesIO(data))
+                files.append(
+                    {
+                        "path": rel,
+                        "checksum": hashlib.sha1(
+                            data, usedforsecurity=False
+                        ).hexdigest(),
+                    }
+                )
+        return files

tests/unit/python/library/test_copy_batch.py

@@ -0,0 +1,714 @@
+from __future__ import annotations
+
+import contextlib
+import hashlib
+import importlib.util
+import io
+import json
+import os
+import stat
+import sys
+import tarfile
+import tempfile
+import unittest
+from pathlib import Path
+from typing import TYPE_CHECKING, ClassVar
+from unittest.mock import patch
+
+from ansible.module_utils.basic import AnsibleModule
+from ansible.module_utils.testing import patch_module_args
+from ansible.modules import copy as copy_module
+from ansible.modules import file as file_module
+
+from . import PROJECT_ROOT
+
+if TYPE_CHECKING:
+    from types import ModuleType
+
+sys.path.insert(0, str(PROJECT_ROOT))
+
+MODULE_PATH = PROJECT_ROOT / "library" / "copy_batch.py"
+
+
+def _load_copy_batch() -> ModuleType:
+    spec = importlib.util.spec_from_file_location(
+        "library_copy_batch", str(MODULE_PATH)
+    )
+    if spec is None or spec.loader is None:
+        raise ImportError(f"Could not load: {MODULE_PATH}")
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)  # type: ignore[attr-defined]
+    return mod
+
+
+_MOD = _load_copy_batch()
+
+
+def _sha1(data):
+    return hashlib.sha1(data, usedforsecurity=False).hexdigest()
+
+
+def _umask():
+    current = os.umask(0)
+    os.umask(current)
+    return current
+
+
+def _call(main, args):
+    buffer = io.StringIO()
+    with (
+        patch_module_args(args),
+        contextlib.redirect_stdout(buffer),
+        contextlib.suppress(SystemExit),
+    ):
+        main()
+    return json.loads(buffer.getvalue())
+
+
+class _Tree(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.root = Path(self._tmp.name)
+        self.dest = self.root / "dest"
+        self.dest.mkdir()
+        self.work = self.root / "work"
+        self.work.mkdir()
+        self.archive = self.root / "archive.tar"
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def _pack(self, files, *, names=None, kinds=None):
+        with tarfile.open(self.archive, "w") as bundle:
+            for name, data in files.items():
+                member = tarfile.TarInfo((names or {}).get(name, name))
+                member.type = (kinds or {}).get(name, tarfile.REGTYPE)
+                member.size = len(data) if member.isreg() else 0
+                bundle.addfile(member, io.BytesIO(data) if member.isreg() else None)
+        return {name: _sha1(data) for name, data in files.items()}
+
+    def _run(
+        self,
+        files,
+        *,
+        settled=(),
+        directories=(),
+        mode="0644",
+        directory_mode=None,
+        digests=None,
+        names=None,
+        kinds=None,
+        tree=None,
+    ):
+        packed = self._pack(files, names=names, kinds=kinds)
+        listed = packed if digests is None else digests
+        return _call(
+            _MOD.main,
+            {
+                "src": str(self.archive),
+                "dest": f"{self.dest}/",
+                "files": [
+                    {"path": name, "checksum": digest}
+                    for name, digest in listed.items()
+                ],
+                "settled": list(settled),
+                "directories": list(directories),
+                "tree": [*listed, *settled, *directories] if tree is None else tree,
+                "mode": mode,
+                "directory_mode": directory_mode,
+                "_ansible_tmpdir": str(self.work),
+            },
+        )
+
+    def _write(self, rel, data, mode):
+        path = self.dest / rel
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_bytes(data)
+        path.chmod(mode)
+        return path
+
+    @staticmethod
+    def _mode(path):
+        return stat.S_IMODE(path.lstat().st_mode)
+
+    @staticmethod
+    def _bytes(path):
+        return path.read_bytes()  # nocheck: cache-read
+
+
+class TestWrites(_Tree):
+    def test_replaces_and_creates_files_with_mode(self):
+        self._write("icon.png", b"old", 0o600)
+
+        result = self._run({"icon.png": b"new", "nested/icon.svg": b"<svg/>"})
+
+        self.assertTrue(result["changed"])
+        self.assertNotIn("refused", result)
+        self.assertEqual(self._bytes(self.dest / "icon.png"), b"new")
+        self.assertEqual(self._bytes(self.dest / "nested/icon.svg"), b"<svg/>")
+        self.assertEqual(self._mode(self.dest / "icon.png"), 0o644)
+        self.assertEqual(self._mode(self.dest / "nested/icon.svg"), 0o644)
+
+    def test_created_parents_receive_the_directory_mode(self):
+        self._run({"a/b/icon.svg": b"<svg/>"}, directory_mode="0750")
+
+        self.assertEqual(self._mode(self.dest / "a"), 0o750)
+        self.assertEqual(self._mode(self.dest / "a/b"), 0o750)
+
+    def test_existing_parent_keeps_its_mode(self):
+        (self.dest / "a").mkdir()
+        (self.dest / "a").chmod(0o701)
+
+        self._run({"a/b/icon.svg": b"<svg/>"}, directory_mode="0750")
+
+        self.assertEqual(self._mode(self.dest / "a"), 0o701)
+        self.assertEqual(self._mode(self.dest / "a/b"), 0o750)
+
+    def test_unset_directory_mode_keeps_the_umask_mode(self):
+        self._run({"a/icon.svg": b"<svg/>"}, directory_mode=None)
+
+        self.assertEqual(self._mode(self.dest / "a"), 0o777 & ~_umask())
+
+    def test_existing_leaf_directory_receives_the_directory_mode(self):
+        (self.dest / "leaf").mkdir()
+        (self.dest / "leaf").chmod(0o700)
+
+        result = self._run({}, directories=["leaf"], directory_mode="0750")
+
+        self.assertTrue(result["changed"])
+        self.assertEqual(self._mode(self.dest / "leaf"), 0o750)
+
+    def test_missing_leaf_directory_is_created_with_its_parents(self):
+        result = self._run({}, directories=["a/leaf"], directory_mode="0750")
+
+        self.assertTrue(result["changed"])
+        self.assertEqual(self._mode(self.dest / "a"), 0o750)
+        self.assertEqual(self._mode(self.dest / "a/leaf"), 0o750)
+
+    def test_settled_file_keeps_its_content_and_inode(self):
+        path = self._write("icon.png", b"png", 0o600)
+        inode = path.stat().st_ino
+
+        result = self._run({}, settled=["icon.png"])
+
+        self.assertTrue(result["changed"])
+        self.assertEqual(path.stat().st_ino, inode)
+        self.assertEqual(self._bytes(path), b"png")
+        self.assertEqual(self._mode(path), 0o644)
+
+    def test_settled_file_with_its_mode_reports_no_change(self):
+        self._write("icon.png", b"png", 0o644)
+
+        result = self._run({}, settled=["icon.png"])
+
+        self.assertFalse(result["changed"])
+
+    def test_replaced_file_gets_the_owner_of_the_file_it_replaces(self):
+        path = self._write("icon.png", b"old", 0o600)
+        owner = (path.stat().st_uid, path.stat().st_gid)
+
+        with (
+            patch.object(
+                AnsibleModule,
+                "atomic_move",
+                autospec=True,
+                side_effect=AnsibleModule.atomic_move,
+            ) as move,
+            patch("os.chown", wraps=os.chown) as chown,
+        ):
+            self._run({"icon.png": b"new"})
+
+        self.assertEqual(move.call_args.args[2], str(path))
+        self.assertNotIn("keep_dest_attrs", move.call_args.kwargs)
+        self.assertIn(owner, [call.args[1:] for call in chown.call_args_list])
+        self.assertEqual((path.stat().st_uid, path.stat().st_gid), owner)
+
+    def test_files_outside_the_call_stay(self):
+        other = self._write("other.png", b"other", 0o600)
+
+        self._run({"icon.png": b"new"})
+
+        self.assertEqual(self._bytes(other), b"other")
+        self.assertEqual(self._mode(other), 0o600)
+
+    def test_symlinked_parent_directory_is_written_through(self):
+        real = self.root / "real"
+        real.mkdir()
+        (self.dest / "linked").symlink_to(real)
+
+        result = self._run({"linked/icon.svg": b"<svg/>"})
+
+        self.assertTrue(result["changed"])
+        self.assertEqual(self._bytes(real / "icon.svg"), b"<svg/>")
+
+    def test_staged_copies_leave_no_trace(self):
+        self._run({"icon.png": b"new", "nested/icon.svg": b"<svg/>"})
+
+        self.assertEqual(list(self.work.iterdir()), [])
+        self.assertEqual(
+            sorted(p.name for p in self.dest.iterdir()), ["icon.png", "nested"]
+        )
+
+
+class TestRefusals(_Tree):
+    def _assert_refused(self, result, path):
+        self.assertFalse(result["changed"])
+        self.assertTrue(any(str(path) in reason for reason in result["refused"]))
+        self.assertFalse((self.dest / "fresh.png").exists())
+
+    def test_symlink_in_place_of_a_file_is_refused_and_kept(self):
+        target = self.root / "target.png"
+        target.write_bytes(b"target")
+        (self.dest / "icon.png").symlink_to(target)
+
+        result = self._run({"icon.png": b"new", "fresh.png": b"fresh"})
+
+        self._assert_refused(result, self.dest / "icon.png")
+        self.assertTrue((self.dest / "icon.png").is_symlink())
+        self.assertEqual(self._bytes(target), b"target")
+
+    def test_dangling_symlink_in_place_of_a_file_is_refused(self):
+        (self.dest / "icon.png").symlink_to(self.root / "absent")
+
+        result = self._run({"icon.png": b"new", "fresh.png": b"fresh"})
+
+        self._assert_refused(result, self.dest / "icon.png")
+
+    def test_directory_in_place_of_a_file_is_refused(self):
+        (self.dest / "icon.png").mkdir()
+
+        result = self._run({"icon.png": b"new", "fresh.png": b"fresh"})
+
+        self._assert_refused(result, self.dest / "icon.png")
+        self.assertTrue((self.dest / "icon.png").is_dir())
+
+    def test_fifo_in_place_of_a_file_is_refused(self):
+        os.mkfifo(self.dest / "icon.png")
+
+        result = self._run({"icon.png": b"new", "fresh.png": b"fresh"})
+
+        self._assert_refused(result, self.dest / "icon.png")
+
+    def test_file_in_place_of_a_parent_directory_is_refused(self):
+        self._write("nested", b"file", 0o644)
+
+        result = self._run({"nested/icon.svg": b"<svg/>", "fresh.png": b"fresh"})
+
+        self._assert_refused(result, self.dest / "nested")
+        self.assertEqual(self._bytes(self.dest / "nested"), b"file")
+
+    def test_dangling_symlink_in_place_of_a_parent_directory_is_refused(self):
+        (self.dest / "nested").symlink_to(self.root / "absent")
+
+        result = self._run({"nested/a/icon.svg": b"<svg/>", "fresh.png": b"fresh"})
+
+        self._assert_refused(result, self.dest / "nested")
+        self.assertFalse((self.root / "absent").exists())
+
+    def test_symlink_in_place_of_a_leaf_directory_is_refused(self):
+        (self.root / "real").mkdir()
+        (self.dest / "leaf").symlink_to(self.root / "real")
+
+        result = self._run(
+            {"fresh.png": b"fresh"}, directories=["leaf"], directory_mode="0750"
+        )
+
+        self._assert_refused(result, self.dest / "leaf")
+        self.assertNotEqual(self._mode(self.root / "real"), 0o750)
+
+    def test_file_in_place_of_a_leaf_directory_is_refused(self):
+        self._write("leaf", b"", 0o644)
+
+        result = self._run({"fresh.png": b"fresh"}, directories=["leaf"])
+
+        self._assert_refused(result, self.dest / "leaf")
+
+    def test_symlink_loop_in_place_of_a_parent_directory_is_refused(self):
+        (self.dest / "loop").symlink_to(self.dest / "loop")
+
+        result = self._run({"loop/icon.svg": b"<svg/>", "fresh.png": b"fresh"})
+
+        self._assert_refused(result, self.dest / "loop")
+
+    def test_missing_settled_file_is_refused(self):
+        result = self._run({"fresh.png": b"fresh"}, settled=["icon.png"])
+
+        self._assert_refused(result, self.dest / "icon.png")
+
+
+class TestFailures(_Tree):
+    def _assert_failed(self, result, needle):
+        self.assertTrue(result["failed"])
+        self.assertIn(needle, result["msg"])
+        self.assertFalse((self.dest / "icon.png").exists())
+
+    def test_checksum_mismatch_fails_naming_the_file_and_writes_nothing(self):
+        digests = {"icon.png": _sha1(b"new"), "nested/icon.svg": _sha1(b"other")}
+
+        result = self._run(
+            {"icon.png": b"new", "nested/icon.svg": b"<svg/>"}, digests=digests
+        )
+
+        self._assert_failed(result, "nested/icon.svg")
+        self.assertFalse((self.dest / "nested").exists())
+
+    def test_parent_reference_member_fails(self):
+        result = self._run(
+            {"icon.png": b"new", "evil": b"x"},
+            names={"evil": "../evil"},
+            digests={"icon.png": _sha1(b"new")},
+        )
+
+        self._assert_failed(result, "../evil")
+        self.assertFalse((self.root / "evil").exists())
+
+    def test_absolute_member_fails(self):
+        result = self._run(
+            {"icon.png": b"new", "evil": b"x"},
+            names={"evil": f"{self.root}/evil"},
+            digests={"icon.png": _sha1(b"new")},
+        )
+
+        self._assert_failed(result, "leaves the destination")
+        self.assertFalse((self.root / "evil").exists())
+
+    def test_unexpected_member_fails(self):
+        result = self._run(
+            {"icon.png": b"new", "extra.png": b"x"},
+            digests={"icon.png": _sha1(b"new")},
+        )
+
+        self._assert_failed(result, "extra.png")
+
+    def test_missing_member_fails(self):
+        result = self._run(
+            {"icon.png": b"new"},
+            digests={"icon.png": _sha1(b"new"), "gone.png": _sha1(b"x")},
+        )
+
+        self._assert_failed(result, "gone.png")
+
+    def test_link_member_fails(self):
+        result = self._run(
+            {"icon.png": b"new", "link.png": b""},
+            kinds={"link.png": tarfile.SYMTYPE},
+        )
+
+        self._assert_failed(result, "not a regular file")
+
+    def test_parent_reference_in_the_arguments_fails(self):
+        result = self._run({}, directories=["../outside"])
+
+        self.assertTrue(result["failed"])
+        self.assertIn("../outside", result["msg"])
+        self.assertFalse((self.root / "outside").exists())
+
+    def test_write_failure_names_the_file_and_fails(self):
+        original = AnsibleModule.atomic_move
+
+        def move(module, src, dest, **kwargs):
+            if dest.endswith("nested/icon.svg"):
+                raise Exception("Could not replace")  # noqa: TRY002 - atomic_move raises a bare Exception
+            return original(module, src, dest, **kwargs)
+
+        with patch.object(
+            AnsibleModule, "atomic_move", autospec=True, side_effect=move
+        ):
+            result = self._run({"icon.png": b"new", "nested/icon.svg": b"<svg/>"})
+
+        self.assertTrue(result["failed"])
+        self.assertIn(str(self.dest / "nested/icon.svg"), result["msg"])
+        self.assertTrue(result["changed"])
+
+    def test_failure_after_creating_a_parent_reports_the_change(self):
+        with patch.object(
+            AnsibleModule,
+            "atomic_move",
+            autospec=True,
+            side_effect=Exception("Could not replace"),
+        ):
+            result = self._run({"nested/icon.svg": b"<svg/>"})
+
+        self.assertTrue(result["failed"])
+        self.assertTrue(result["changed"])
+        self.assertTrue((self.dest / "nested").is_dir())
+
+    def test_repeated_file_path_fails(self):
+        result = self._run(
+            {"fresh.png": b"fresh"},
+            digests={"fresh.png": _sha1(b"fresh")},
+            tree=["fresh.png"],
+        )
+        self.assertNotIn("failed", result)
+
+        repeated = _call(
+            _MOD.main,
+            {
+                "src": str(self.archive),
+                "dest": f"{self.dest}/",
+                "files": [
+                    {"path": "fresh.png", "checksum": _sha1(b"fresh")},
+                    {"path": "fresh.png", "checksum": _sha1(b"fresh")},
+                ],
+                "settled": [],
+                "directories": [],
+                "tree": ["fresh.png"],
+                "mode": "0644",
+                "directory_mode": None,
+                "_ansible_tmpdir": str(self.work),
+            },
+        )
+
+        self.assertTrue(repeated["failed"])
+        self.assertIn("repeat", repeated["msg"])
+
+
+class TestOrderAndAliases(_Tree):
+    def test_files_are_written_in_the_given_order(self):
+        order = ["b/z.png", "a.png", "c.png"]
+
+        with patch.object(
+            AnsibleModule,
+            "atomic_move",
+            autospec=True,
+            side_effect=AnsibleModule.atomic_move,
+        ) as move:
+            self._run({name: name.encode() for name in order})
+
+        self.assertEqual(
+            [call.args[2] for call in move.call_args_list],
+            [str(self.dest / name) for name in order],
+        )
+
+    def test_settled_files_are_handled_in_tree_order(self):
+        settled = self._write("s.png", b"s", 0o600)
+
+        with patch.object(
+            AnsibleModule,
+            "atomic_move",
+            autospec=True,
+            side_effect=Exception("Could not replace"),
+        ):
+            result = self._run(
+                {"b.png": b"b"}, settled=["s.png"], tree=["s.png", "b.png"]
+            )
+
+        self.assertTrue(result["failed"])
+        self.assertEqual(self._mode(settled), 0o644)
+
+    def test_settled_file_after_a_failed_write_stays_untouched(self):
+        settled = self._write("s.png", b"s", 0o600)
+
+        with patch.object(
+            AnsibleModule,
+            "atomic_move",
+            autospec=True,
+            side_effect=Exception("Could not replace"),
+        ):
+            result = self._run(
+                {"b.png": b"b"}, settled=["s.png"], tree=["b.png", "s.png"]
+            )
+
+        self.assertTrue(result["failed"])
+        self.assertEqual(self._mode(settled), 0o600)
+
+    def test_path_missing_from_tree_fails(self):
+        result = self._run({"b.png": b"b"}, tree=[])
+
+        self.assertTrue(result["failed"])
+        self.assertIn("missing from tree", result["msg"])
+
+    def test_two_paths_through_a_symlinked_directory_are_refused(self):
+        (self.dest / "y").mkdir()
+        self._write("y/f.js", b"y", 0o644)
+        (self.dest / "x").symlink_to(self.dest / "y")
+
+        result = self._run(
+            {"x/f.js": b"x", "fresh.png": b"fresh"},
+            tree=["x/f.js", "y/f.js", "fresh.png"],
+        )
+
+        self.assertFalse(result["changed"])
+        self.assertIn("reach the same entry", " ".join(result["refused"]))
+        self.assertEqual(self._bytes(self.dest / "y/f.js"), b"y")
+        self.assertFalse((self.dest / "fresh.png").exists())
+
+    def test_two_missing_paths_through_a_symlinked_directory_are_refused(self):
+        (self.dest / "y").mkdir()
+        (self.dest / "x").symlink_to(self.dest / "y")
+
+        result = self._run({"x/f.js": b"x", "y/f.js": b"y"}, tree=["x/f.js", "y/f.js"])
+
+        self.assertIn("reach the same entry", " ".join(result["refused"]))
+        self.assertFalse((self.dest / "y/f.js").exists())
+
+    def test_hard_linked_targets_are_refused(self):
+        first = self._write("a.png", b"same", 0o644)
+        (self.dest / "b.png").hardlink_to(first)
+
+        result = self._run({"a.png": b"new"}, tree=["a.png", "b.png"])
+
+        self.assertIn("reach the same entry", " ".join(result["refused"]))
+        self.assertEqual(self._bytes(self.dest / "b.png"), b"same")
+
+    def test_unsafe_writes_reaches_atomic_move_from_the_environment(self):
+        with (
+            patch.dict(os.environ, {"ANSIBLE_UNSAFE_WRITES": "1"}),
+            patch.object(
+                AnsibleModule,
+                "atomic_move",
+                autospec=True,
+                side_effect=AnsibleModule.atomic_move,
+            ) as move,
+        ):
+            self._run({"icon.png": b"new"})
+
+        self.assertTrue(move.call_args.kwargs["unsafe_writes"])
+
+    def test_leaf_created_concurrently_is_accepted(self):
+        original = Path.mkdir
+
+        def racing(path, *args, **kwargs):
+            original(path, *args, **kwargs)
+            raise FileExistsError(path)
+
+        with patch.object(Path, "mkdir", autospec=True, side_effect=racing):
+            result = self._run({}, directories=["leaf"], directory_mode="0750")
+
+        self.assertTrue(result["changed"])
+        self.assertEqual(self._mode(self.dest / "leaf"), 0o750)
+
+
+class TestSameEndStateAsTheCopyModule(unittest.TestCase):
+    MODE = "0644"
+    DIRECTORY_MODE = "0750"
+    CONTENT: ClassVar[dict[str, bytes]] = {
+        "replaced.png": b"new",
+        "fresh/deep/new.svg": b"<svg/>",
+        "kept-dir/new.png": b"png",
+    }
+    SETTLED: ClassVar[tuple[str, ...]] = ("mode-drift.png",)
+    LEAVES: ClassVar[tuple[str, ...]] = ("existing-leaf", "absent/leaf")
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.root = Path(self._tmp.name)
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def _seed(self, dest):
+        dest.mkdir()
+        (dest / "replaced.png").write_bytes(b"old")
+        (dest / "replaced.png").chmod(0o600)
+        (dest / "mode-drift.png").write_bytes(b"same")
+        (dest / "mode-drift.png").chmod(0o600)
+        (dest / "kept-dir").mkdir()
+        (dest / "kept-dir").chmod(0o701)
+        (dest / "existing-leaf").mkdir()
+        (dest / "existing-leaf").chmod(0o700)
+        (dest / "untouched.png").write_bytes(b"untouched")
+
+    def _regular_copy(self, dest):
+        staging = self.root / "staging"
+        staging.mkdir()
+        for rel, data in self.CONTENT.items():
+            source = staging / ".source"
+            source.write_bytes(data)
+            result = _call(
+                copy_module.main,
+                {
+                    "src": str(source),
+                    "dest": f"{dest}/",
+                    "mode": self.MODE,
+                    "directory_mode": self.DIRECTORY_MODE,
+                    "_original_basename": rel,
+                    "follow": False,
+                    "checksum": _sha1(data),
+                },
+            )
+            self.assertNotIn("failed", result, result)
+        for rel in self.SETTLED:
+            result = _call(
+                file_module.main,
+                {
+                    "mode": self.MODE,
+                    "dest": f"{dest}/",
+                    "_original_basename": rel,
+                    "recurse": False,
+                    "state": "file",
+                },
+            )
+            self.assertNotIn("failed", result, result)
+        for rel in self.LEAVES:
+            result = _call(
+                file_module.main,
+                {
+                    "path": str(dest / rel),
+                    "state": "directory",
+                    "mode": self.DIRECTORY_MODE,
+                    "recurse": False,
+                },
+            )
+            self.assertNotIn("failed", result, result)
+
+    def _batch(self, dest):
+        archive = self.root / "archive.tar"
+        work = self.root / "work"
+        work.mkdir()
+        with tarfile.open(archive, "w") as bundle:
+            for rel, data in self.CONTENT.items():
+                member = tarfile.TarInfo(rel)
+                member.size = len(data)
+                bundle.addfile(member, io.BytesIO(data))
+        result = _call(
+            _MOD.main,
+            {
+                "src": str(archive),
+                "dest": f"{dest}/",
+                "files": [
+                    {"path": rel, "checksum": _sha1(data)}
+                    for rel, data in self.CONTENT.items()
+                ],
+                "settled": list(self.SETTLED),
+                "directories": list(self.LEAVES),
+                "tree": [*self.CONTENT, *self.SETTLED, *self.LEAVES],
+                "mode": self.MODE,
+                "directory_mode": self.DIRECTORY_MODE,
+                "_ansible_tmpdir": str(work),
+            },
+        )
+        self.assertTrue(result["changed"], result)
+
+    @staticmethod
+    def _snapshot(dest):
+        entries = {}
+        for path in sorted(dest.rglob("*")):  # nocheck: project-walk
+            info = path.lstat()
+            content = None
+            if path.is_file():
+                content = path.read_bytes()  # nocheck: cache-read
+            entries[str(path.relative_to(dest))] = (
+                stat.S_IFMT(info.st_mode),
+                stat.S_IMODE(info.st_mode),
+                info.st_uid,
+                info.st_gid,
+                content,
+            )
+        return entries
+
+    def test_batch_leaves_the_tree_the_regular_copy_leaves(self):
+        regular, batch = self.root / "regular", self.root / "batch"
+        self._seed(regular)
+        self._seed(batch)
+
+        self._regular_copy(regular)
+        self._batch(batch)
+
+        self.assertEqual(self._snapshot(batch), self._snapshot(regular))
+        self.assertEqual(self._snapshot(batch)["kept-dir"][1], 0o701)
+        self.assertEqual(self._snapshot(batch)["existing-leaf"][1], 0o750)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/python/plugins/action/test_stack_host_copy.py

@@ -1,5 +1,6 @@
 import hashlib
 import os
+import tarfile
 import tempfile
 import unittest
 from pathlib import Path
@@ -153,6 +154,7 @@ class _FakeShell:
 class _FakeConnection:
     def __init__(self):
         self._shell = _FakeShell()
+        self.become = None
 
 
 class _FakeLoader:
@@ -179,6 +181,11 @@ class _Dir:
         self.mode = mode
 
 
+class _Link:
+    def __init__(self, data):
+        self.data = data
+
+
 def _mode_bits(mode):
     return mode if isinstance(mode, int) else int(mode, 8)
 
@@ -188,6 +195,24 @@ class _FakeRemote:
         self.entries = dict(entries)
         self.calls = []
         self.args = []
+        self.transfers = []
+        self.fixups = 0
+
+    def transfer(self, local, remote_path):
+        if not remote_path.endswith(".tar"):
+            self.transfers.append(Path(local).read_bytes())  # nocheck: cache-read
+            return remote_path
+        with tarfile.open(local) as bundle:
+            self.transfers.append(
+                {
+                    member.name: bundle.extractfile(member).read()
+                    for member in bundle.getmembers()
+                }
+            )
+        return remote_path
+
+    def fixup(self, *args, **kwargs):
+        self.fixups += 1
 
     def __call__(self, module_name=None, module_args=None, task_vars=None, **kwargs):
         self.calls.append(module_name)
@@ -197,14 +222,31 @@ class _FakeRemote:
                 "changed": False,
                 "states": {p: self._state(p) for p in module_args["paths"]},
             }
+        if module_name == "ansible.legacy.copy_batch":
+            return self._copy_batch(module_args)
         if module_name == "ansible.legacy.stat":
             entry = self.entries.get(module_args["path"])
             if entry is None:
                 return {"stat": {"exists": False}}
             if isinstance(entry, _Dir):
                 return {"stat": {"exists": True, "isdir": True}}
+            if isinstance(entry, _Link):
+                digest = _sha1(entry.data) if module_args["follow"] else ""
+                return {
+                    "stat": {
+                        "exists": True,
+                        "isdir": False,
+                        "islnk": True,
+                        "checksum": digest,
+                    }
+                }
             return {
-                "stat": {"exists": True, "isdir": False, "checksum": _sha1(entry[0])}
+                "stat": {
+                    "exists": True,
+                    "isdir": False,
+                    "islnk": False,
+                    "checksum": _sha1(entry[0]),
+                }
             }
         if module_name == "ansible.legacy.file" and module_args["state"] == "directory":
             return self._ensure_directory(module_args["path"], module_args["mode"])
@@ -228,9 +270,39 @@ class _FakeRemote:
             "path": path,
         }
 
+    def _copy_batch(self, args):
+        dest = args["dest"]
+        files = [entry["path"] for entry in args["files"]]
+        refused = [
+            dest + rel
+            for rel in [*files, *args["settled"]]
+            if isinstance(self.entries.get(dest + rel), (_Dir, _Link))
+        ] + [
+            dest + rel
+            for rel in args["directories"]
+            if dest + rel in self.entries
+            and not isinstance(self.entries[dest + rel], _Dir)
+        ]
+        if refused:
+            return {"changed": False, "refused": refused}
+        archive = self.transfers[-1]
+        if list(archive) != files or any(
+            _sha1(archive[entry["path"]]) != entry["checksum"]
+            for entry in args["files"]
+        ):
+            return {"failed": True, "msg": "archive does not match files"}
+        mode = _mode_bits(args["mode"])
+        for rel in files:
+            self.entries[dest + rel] = (archive[rel], mode)
+        for rel in args["settled"]:
+            self.entries[dest + rel] = (self.entries[dest + rel][0], mode)
+        for rel in args["directories"]:
+            self.entries[dest + rel] = _Dir(_mode_bits(args["directory_mode"] or 0o755))
+        return {"changed": True}
+
     def _state(self, path):
         entry = self.entries.get(path)
-        if entry is None:
+        if entry is None or isinstance(entry, _Link):
             return {"isreg": False, "isdir": False, "mode": None, "checksum": None}
         if isinstance(entry, _Dir):
             return {
@@ -247,12 +319,8 @@ class _FakeRemote:
         }
 
 
-class TestStackHostCopyTree(unittest.TestCase):
-    FILES: ClassVar[dict[str, bytes]] = {
-        "favicon.ico": b"ico",
-        "icon.png": b"png",
-        "nested/icon.svg": b"<svg/>",
-    }
+class _TreeHarness(unittest.TestCase):
+    FILES: ClassVar[dict[str, bytes]] = {}
     DEST = "/var/www/public_cdn/roles/app/latest/img/design/"
 
     def setUp(self):
@@ -268,15 +336,19 @@ class TestStackHostCopyTree(unittest.TestCase):
     def _synced(self, mode=0o644):
         return {self.DEST + rel: (data, mode) for rel, data in self.FILES.items()}
 
-    def _run(self, remote, **extra):
+    def _run(self, remote, *, check_mode=False, diff=False, loader=None, **extra):
         task = _FakeTask(args={"src": f"{self.src}/", "dest": self.DEST, **extra})
+        task.check_mode = check_mode
+        task.diff = diff
         action = _make_action(task, _FakeTemplar(is_stack_host=True))
         action._connection = _FakeConnection()
-        action._loader = _FakeLoader()
+        action._loader = _FakeLoader() if loader is None else loader
         action._play_context = _FakePlayContext()
         action._execute_module = remote
-        action._transfer_file = lambda local, remote_path: remote_path
-        action._fixup_perms2 = lambda *a, **k: None
+        action._transfer_file = getattr(
+            remote, "transfer", lambda local, remote_path: remote_path
+        )
+        action._fixup_perms2 = getattr(remote, "fixup", lambda *a, **k: None)
         action._remove_tmp_path = lambda *a, **k: None
         action._remote_expand_user = lambda path, **k: path
 
@@ -287,6 +359,36 @@ class TestStackHostCopyTree(unittest.TestCase):
         action._make_tmp_path = make_tmp_path
         return action.run(task_vars={})
 
+    BATCH: ClassVar[list[str]] = [
+        "ansible.legacy.file_checksums",
+        "ansible.legacy.copy_batch",
+    ]
+
+    @staticmethod
+    def _batch_args(remote):
+        return remote.args[remote.calls.index("ansible.legacy.copy_batch")]
+
+    @staticmethod
+    def _copies(remote):
+        return [
+            args
+            for name, args in zip(remote.calls, remote.args, strict=True)
+            if name == "ansible.legacy.copy"
+        ]
+
+
+class TestStackHostCopyTree(_TreeHarness):
+    FILES: ClassVar[dict[str, bytes]] = {
+        "favicon.ico": b"ico",
+        "icon.png": b"png",
+        "nested/icon.svg": b"<svg/>",
+    }
+
+    def _changed(self):
+        entries = self._synced()
+        entries[self.DEST + "icon.png"] = (b"old", 0o644)
+        return entries
+
     def test_unchanged_tree_costs_one_remote_call(self):
         remote = _FakeRemote(self._synced())
 
@@ -297,7 +399,7 @@ class TestStackHostCopyTree(unittest.TestCase):
         self.assertEqual(result["dest"], self.DEST)
         self.assertEqual(result["src"], f"{self.src}/")
 
-    def test_changed_file_runs_the_regular_copy(self):
+    def test_changed_file_travels_alone_in_one_batch(self):
         entries = self._synced()
         entries[self.DEST + "icon.png"] = (b"old", 0o644)
         remote = _FakeRemote(entries)
@@ -305,9 +407,16 @@ class TestStackHostCopyTree(unittest.TestCase):
         result = self._run(remote)
 
         self.assertTrue(result["changed"])
-        self.assertIn("ansible.legacy.copy", remote.calls)
+        self.assertEqual(remote.calls, self.BATCH)
+        self.assertEqual(remote.transfers, [{"icon.png": b"png"}])
+        self.assertEqual(
+            self._batch_args(remote)["files"],
+            [{"path": "icon.png", "checksum": _sha1(b"png")}],
+        )
+        self.assertEqual(sorted(self._batch_args(remote)["tree"]), sorted(self.FILES))
+        self.assertEqual(remote.entries, self._synced())
 
-    def test_missing_file_runs_the_regular_copy(self):
+    def test_missing_file_travels_in_one_batch(self):
         entries = self._synced()
         del entries[self.DEST + "nested/icon.svg"]
         remote = _FakeRemote(entries)
@@ -315,9 +424,10 @@ class TestStackHostCopyTree(unittest.TestCase):
         result = self._run(remote)
 
         self.assertTrue(result["changed"])
-        self.assertIn("ansible.legacy.copy", remote.calls)
+        self.assertEqual(remote.calls, self.BATCH)
+        self.assertEqual(remote.transfers, [{"nested/icon.svg": b"<svg/>"}])
 
-    def test_mode_drift_runs_the_regular_copy(self):
+    def test_mode_drift_is_settled_without_sending_the_file(self):
         entries = self._synced()
         entries[self.DEST + "favicon.ico"] = (b"ico", 0o600)
         remote = _FakeRemote(entries)
@@ -325,16 +435,165 @@ class TestStackHostCopyTree(unittest.TestCase):
         result = self._run(remote)
 
         self.assertTrue(result["changed"])
-        self.assertIn("ansible.legacy.file", remote.calls)
+        self.assertEqual(remote.calls, self.BATCH)
+        self.assertEqual(remote.transfers, [{}])
+        self.assertEqual(self._batch_args(remote)["settled"], ["favicon.ico"])
+        self.assertEqual(self._batch_args(remote)["files"], [])
 
-    def test_directory_in_place_of_a_file_runs_the_regular_copy(self):
+    def test_directory_in_place_of_a_file_is_refused_then_copied_regularly(self):
         entries = self._synced()
         entries[self.DEST + "icon.png"] = _Dir(0o755)
         remote = _FakeRemote(entries)
 
         self._run(remote)
 
+        self.assertEqual(remote.calls[:2], self.BATCH)
         self.assertIn("ansible.legacy.stat", remote.calls)
+        self.assertIn("ansible.legacy.copy", remote.calls)
+
+    def test_symlink_in_place_of_a_file_is_refused_then_copied_regularly(self):
+        entries = self._synced()
+        entries[self.DEST + "icon.png"] = _Link(b"png")
+        remote = _FakeRemote(entries)
+
+        self._run(remote)
+
+        self.assertEqual(remote.calls[:2], self.BATCH)
+        copies = self._copies(remote)
+        self.assertEqual([args["_original_basename"] for args in copies], ["icon.png"])
+        self.assertFalse(copies[0]["follow"])
+
+    def test_check_mode_keeps_the_regular_copy(self):
+        remote = _FakeRemote(self._changed())
+
+        result = self._run(remote, check_mode=True)
+
+        self.assertTrue(result["changed"])
+        self.assertNotIn("ansible.legacy.copy_batch", remote.calls)
+        self.assertNotIn("ansible.legacy.copy", remote.calls)
+        self.assertIn("ansible.legacy.stat", remote.calls)
+        self.assertEqual(remote.transfers, [])
+
+    def test_diff_mode_keeps_the_regular_copy(self):
+        remote = _FakeRemote(self._changed())
+
+        with patch(
+            "plugins.action.stack_host_copy.CopyActionModule.run",
+            return_value={"changed": True, "diff": [{"after": "png"}]},
+        ) as super_run:
+            result = self._run(remote, diff=True)
+
+        super_run.assert_called_once()
+        self.assertEqual(result["diff"], [{"after": "png"}])
+        self.assertEqual(remote.calls, ["ansible.legacy.file_checksums"])
+        self.assertEqual(remote.transfers, [])
+
+    def test_failed_batch_stays_failed_without_a_second_copy(self):
+        failure = {
+            "failed": True,
+            "changed": True,
+            "msg": f"Failed to write {self.DEST}icon.png: boom",
+        }
+
+        class _FailingBatch(_FakeRemote):
+            def _copy_batch(self, args):
+                return dict(failure)
+
+        remote = _FailingBatch(self._changed())
+
+        with patch("plugins.action.stack_host_copy.CopyActionModule.run") as super_run:
+            result = self._run(remote)
+
+        super_run.assert_not_called()
+        self.assertTrue(result["failed"])
+        self.assertIn(f"{self.DEST}icon.png", result["msg"])
+        self.assertEqual(remote.calls, self.BATCH)
+
+    def test_batch_result_has_the_keys_of_the_regular_copy(self):
+        batch = self._run(_FakeRemote(self._changed()))
+        with patch.object(ActionModule, "_write_tree", return_value=None):
+            regular = self._run(_FakeRemote(self._changed()))
+
+        self.assertEqual(set(batch), set(regular))
+        self.assertEqual(
+            {key: batch[key] for key in ("changed", "dest", "src")},
+            {key: regular[key] for key in ("changed", "dest", "src")},
+        )
+
+    def test_unprivileged_become_keeps_the_regular_copy(self):
+        remote = _FakeRemote(self._changed())
+
+        with patch.object(ActionModule, "_is_become_unprivileged", return_value=True):
+            result = self._run(remote)
+
+        self.assertTrue(result["changed"])
+        self.assertNotIn("ansible.legacy.copy_batch", remote.calls)
+        self.assertIn("ansible.legacy.copy", remote.calls)
+
+    def test_dest_with_a_variable_skips_the_tree_check(self):
+        remote = _FakeRemote(self._changed())
+        task = _FakeTask(args={"src": f"{self.src}/", "dest": "/srv/$APP/img/"})
+        action = _make_action(task, _FakeTemplar(is_stack_host=True))
+        action._execute_module = remote
+
+        with patch(
+            "plugins.action.stack_host_copy.CopyActionModule.run",
+            return_value={"changed": True},
+        ) as super_run:
+            action.run(task_vars={})
+
+        super_run.assert_called_once()
+        self.assertEqual(remote.calls, [])
+
+    def test_archive_over_the_limit_keeps_the_regular_copy(self):
+        entries = self._synced()
+        entries[self.DEST + "icon.png"] = (b"old", 0o644)
+        entries[self.DEST + "favicon.ico"] = (b"old", 0o644)
+        remote = _FakeRemote(entries)
+
+        with patch("plugins.action.stack_host_copy._ARCHIVE_LIMIT", 5):
+            result = self._run(remote)
+
+        self.assertTrue(result["changed"])
+        self.assertNotIn("ansible.legacy.copy_batch", remote.calls)
+        self.assertEqual(sorted(remote.transfers), [b"ico", b"png"])
+
+    def test_files_travel_in_the_order_the_copy_action_walks(self):
+        remote = _FakeRemote({})
+
+        self._run(remote)
+
+        walked = [
+            rel
+            for _, rel in _walk_dirs_without_bytecode_caches(
+                f"{self.src}/", local_follow=True
+            )["files"]
+        ]
+        sent = [entry["path"] for entry in self._batch_args(remote)["files"]]
+        self.assertEqual(sent, walked)
+        self.assertEqual(list(remote.transfers[0]), walked)
+
+    def test_checksum_is_taken_from_the_packed_bytes(self):
+        remote = _FakeRemote(self._changed())
+        icon = self.src / "icon.png"
+        loader = _FakeLoader()
+        reads = []
+
+        def get_real_file(path, decrypt=True):
+            reads.append(path)
+            if path == str(icon) and reads.count(path) == 2:
+                icon.write_bytes(b"edited")
+            return path
+
+        loader.get_real_file = get_real_file
+
+        self._run(remote, loader=loader)
+
+        self.assertEqual(remote.transfers, [{"icon.png": b"edited"}])
+        self.assertEqual(
+            self._batch_args(remote)["files"],
+            [{"path": "icon.png", "checksum": _sha1(b"edited")}],
+        )
 
     def test_requested_mode_is_what_the_tree_check_compares(self):
         remote = _FakeRemote(self._synced(mode=0o600))
@@ -437,7 +696,7 @@ class TestStackHostCopyTree(unittest.TestCase):
         regular = _FakeRemote(self._with_leaves(0o700))
 
         settled = self._run(remote)
-        with patch.object(ActionModule, "_unchanged_tree", return_value=None):
+        with patch.object(ActionModule, "_settled_tree", return_value=None):
             copied = self._run(regular)
 
         self.assertFalse(settled["changed"])
@@ -454,7 +713,7 @@ class TestStackHostCopyTree(unittest.TestCase):
         self.assertFalse(result["changed"])
         self.assertEqual(remote.calls, ["ansible.legacy.file_checksums"])
 
-    def test_missing_remote_directory_runs_the_regular_copy(self):
+    def test_missing_remote_directory_is_ensured_in_one_batch(self):
         self._add_empty_directories()
         entries = self._with_leaves(0o755)
         del entries[self.DEST + "nested/.claude/.cc-writes"]
@@ -463,18 +722,24 @@ class TestStackHostCopyTree(unittest.TestCase):
         result = self._run(remote)
 
         self.assertTrue(result["changed"])
-        self.assertEqual(self._directory_calls(remote), self._leaf_paths())
+        self.assertEqual(remote.calls, self.BATCH)
+        self.assertEqual(
+            self._batch_args(remote)["directories"], ["nested/.claude/.cc-writes"]
+        )
+        self.assertIsNone(self._batch_args(remote)["directory_mode"])
 
-    def test_directory_mode_drift_runs_the_regular_copy(self):
+    def test_directory_mode_drift_is_settled_in_one_batch(self):
         self._add_empty_directories()
         remote = _FakeRemote(self._with_leaves(0o755))
 
         result = self._run(remote, directory_mode="0750")
 
         self.assertTrue(result["changed"])
-        self.assertEqual(self._directory_calls(remote), self._leaf_paths())
+        self.assertEqual(remote.calls, self.BATCH)
+        self.assertEqual(self._batch_args(remote)["directories"], sorted(self.LEAVES))
+        self.assertEqual(self._batch_args(remote)["directory_mode"], "0750")
 
-    def test_file_in_place_of_a_directory_runs_the_regular_copy(self):
+    def test_file_in_place_of_a_directory_is_refused_then_fails_regularly(self):
         self._add_empty_directories()
         entries = self._with_leaves(0o755)
         entries[self.DEST + ".claude/.cc-writes"] = (b"", 0o644)
@@ -482,6 +747,7 @@ class TestStackHostCopyTree(unittest.TestCase):
 
         result = self._run(remote)
 
+        self.assertEqual(remote.calls[:2], self.BATCH)
         self.assertTrue(result["failed"])
         self.assertIn(self.DEST + ".claude/.cc-writes", self._directory_calls(remote))
 
@@ -541,9 +807,10 @@ class TestStackHostCopyTree(unittest.TestCase):
         result = self._run(remote)
 
         self.assertTrue(result["changed"])
-        self.assertIn("ansible.legacy.copy", remote.calls)
+        self.assertEqual(remote.calls, self.BATCH)
         touched = [str(value) for args in remote.args for value in args.values()]
         self.assertEqual([value for value in touched if "__pycache__" in value], [])
+        self.assertEqual(remote.transfers, [{"icon.png": b"png"}])
 
     def test_bytecode_outside_a_cache_directory_stays_in_the_copy(self):
         (self.src / "legacy.pyc").write_bytes(b"bytecode")
@@ -553,7 +820,7 @@ class TestStackHostCopyTree(unittest.TestCase):
 
         self.assertTrue(result["changed"])
         self.assertIn(self.DEST + "legacy.pyc", remote.args[0]["paths"])
-        self.assertIn("ansible.legacy.copy", remote.calls)
+        self.assertEqual(remote.transfers, [{"legacy.pyc": b"bytecode"}])
 
     def test_symlinked_root_passes_the_cache_filter_unchanged(self):
         with tempfile.TemporaryDirectory() as links:
@@ -565,6 +832,79 @@ class TestStackHostCopyTree(unittest.TestCase):
         self.assertEqual(filtered["symlinks"], (str(self.src), "root"))
 
 
+class TestStackHostCopyFlatTree(_TreeHarness):
+    FILES: ClassVar[dict[str, bytes]] = {
+        f"slot-{index:02d}.png": f"palette-2-{index}".encode() for index in range(15)
+    }
+
+    def _stale(self, names):
+        entries = self._synced()
+        entries.update({self.DEST + name: (b"palette-1", 0o644) for name in names})
+        return entries
+
+    def _round_trips(self, entries, *, regular):
+        remote = _FakeRemote(entries)
+        if regular:
+            with patch.object(ActionModule, "_write_tree", return_value=None):
+                result = self._run(remote)
+        else:
+            result = self._run(remote)
+        self.assertTrue(result["changed"])
+        return {
+            "modules": len(remote.calls),
+            "transfers": len(remote.transfers),
+            "fixups": remote.fixups,
+        }
+
+    def test_all_files_changed_cost_one_transfer_and_two_module_calls(self):
+        stale = self._stale(self.FILES)
+
+        before = self._round_trips(stale, regular=True)
+        after = self._round_trips(stale, regular=False)
+
+        self.assertEqual(before, {"modules": 31, "transfers": 15, "fixups": 15})
+        self.assertEqual(after, {"modules": 2, "transfers": 1, "fixups": 1})
+
+    def test_one_changed_file_costs_one_transfer_and_two_module_calls(self):
+        stale = self._stale(["slot-07.png"])
+
+        before = self._round_trips(stale, regular=True)
+        after = self._round_trips(stale, regular=False)
+
+        self.assertEqual(before, {"modules": 31, "transfers": 1, "fixups": 1})
+        self.assertEqual(after, {"modules": 2, "transfers": 1, "fixups": 1})
+
+    def test_only_the_changed_file_is_sent(self):
+        remote = _FakeRemote(self._stale(["slot-07.png"]))
+
+        self._run(remote)
+
+        self.assertEqual(remote.transfers, [{"slot-07.png": b"palette-2-7"}])
+        self.assertEqual(remote.entries, self._synced())
+
+    def test_follow_without_symlinks_takes_the_batch(self):
+        remote = _FakeRemote(self._stale(["slot-00.png", "slot-01.png"]))
+
+        result = self._run(remote, follow=True)
+
+        self.assertTrue(result["changed"])
+        self.assertEqual(remote.calls, self.BATCH)
+
+    def test_follow_onto_a_symlinked_file_keeps_the_copy_semantics(self):
+        entries = self._synced()
+        entries[self.DEST + "slot-00.png"] = _Link(b"elsewhere")
+        remote = _FakeRemote(entries)
+
+        self._run(remote, follow=True)
+
+        self.assertEqual(remote.calls[:2], self.BATCH)
+        copies = self._copies(remote)
+        self.assertEqual(
+            [args["_original_basename"] for args in copies], ["slot-00.png"]
+        )
+        self.assertTrue(copies[0]["follow"])
+
+
 class TestStackHostCopyWalkSwap(unittest.TestCase):
     def _action(self):
         task = _FakeTask(args={"src": "x", "dest": "/opt/compose/app/x"})