diff --git a/steward/ansible/bundled/maintenance/files/swarm_prune.py b/steward/ansible/bundled/maintenance/files/swarm_prune.py
new file mode 100644
index 0000000..66e5341
--- /dev/null
+++ b/steward/ansible/bundled/maintenance/files/swarm_prune.py
@@ -0,0 +1,111 @@
+#!/usr/bin/env python3
+"""Steward's swarm prune (maintenance/swarm_prune.yml), run on each node.
+
+Removes the images on this node that no swarm service is pinned to and no
+container uses (by id, or by the reference it was made from), and keeps every image a service is pinned to even when its
+task runs on another node. A node then holds what it needs to take over the
+other's services without a pull: the registry may be behind the node that
+failed, and a digest it has since deleted can't be pulled at all.
+
+The node must be a swarm manager, the only kind that can list services. On
+one that isn't, or when Docker doesn't answer, it removes nothing and fails,
+so a prune never runs blind. With --dry-run it says what it would remove.
+Prints one JSON line: kept, removed, failed, freed_bytes, dry_run.
+
+Stdlib only; shipped beside the playbook in Steward's built-in source.
+"""
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+
+DEADLINE_SECONDS = 300
+
+
+def docker(*args: str) -> str:
+    done = subprocess.run(["docker", *args], capture_output=True, text=True,
+                          timeout=DEADLINE_SECONDS)
+    if done.returncode != 0:
+        raise RuntimeError(f"docker {' '.join(args[:2])}: {done.stderr.strip()[-300:]}")
+    return done.stdout
+
+
+def short(repo: str) -> str:
+    """A repository the way Docker lists it locally: no docker.io/ or library/."""
+    for prefix in ("docker.io/", "index.docker.io/"):
+        if repo.startswith(prefix):
+            repo = repo[len(prefix):]
+    return repo[len("library/"):] if repo.startswith("library/") else repo
+
+
+def wanted(service_images) -> tuple[set, set]:
+    """(digests, "repo:tag" refs) the services are pinned to. A spec with a
+    digest is matched by it; one with a tag alone by that tag."""
+    digests, refs = set(), set()
+    for image in service_images:
+        ref, _, digest = (image or "").partition("@")
+        if digest:
+            digests.add(digest)
+        elif ref:
+            name = ref.rsplit("/", 1)[-1]
+            refs.add(short(ref if ":" in name else ref + ":latest"))
+    return digests, refs
+
+
+def removable(images: list, digests: set, refs: set, used: set) -> list:
+    """The images (`docker image inspect` rows) no container uses and no
+    service is pinned to."""
+    out = []
+    for im in images:
+        if im.get("Id") in used:
+            continue
+        held = {d.split("@", 1)[1] for d in im.get("RepoDigests") or [] if "@" in d}
+        tags = {short(t) for t in im.get("RepoTags") or []}
+        if held & digests or tags & refs:
+            continue
+        out.append(im)
+    return out
+
+
+def label(im: dict) -> str:
+    names = (im.get("RepoTags") or []) + (im.get("RepoDigests") or [])
+    return names[0] if names else im.get("Id", "")[7:19]
+
+
+def main(argv: list) -> int:
+    dry = "--dry-run" in argv
+    try:
+        ids = docker("service", "ls", "-q").split()
+        services = json.loads(docker("service", "inspect", *ids)) if ids else []
+        ids = docker("ps", "-aq", "--no-trunc").split()
+        containers = json.loads(docker("container", "inspect", *ids)) if ids else []
+        image_ids = sorted(set(docker("image", "ls", "-q", "--no-trunc").split()))
+        images = json.loads(docker("image", "inspect", *image_ids)) if image_ids else []
+    except (RuntimeError, ValueError, OSError, subprocess.TimeoutExpired) as exc:
+        print(json.dumps({"error": f"nothing removed: {exc}"}))
+        return 2
+    # A container's image is kept by its id and by the reference it was made
+    # from: on Docker's containerd store the id alone hasn't held a tag.
+    used = {c.get("Image") for c in containers}
+    digests, refs = wanted([s.get("Spec", {}).get("TaskTemplate", {}).get("ContainerSpec", {}).get("Image")
+                            for s in services]
+                           + [c.get("Config", {}).get("Image") for c in containers])
+    gone = removable(images, digests, refs, used)
+    removed, failed, freed = [], [], 0
+    for im in gone:
+        if not dry:
+            try:
+                docker("image", "rm", "-f", im["Id"])
+            except (RuntimeError, subprocess.TimeoutExpired) as exc:
+                failed.append({"image": label(im), "error": str(exc)[-200:]})
+                continue
+        removed.append(label(im))
+        freed += int(im.get("Size") or 0)
+    print(json.dumps({"kept": len(images) - len(gone), "removed": removed, "failed": failed,
+                      "freed_bytes": freed, "dry_run": dry}))
+    return 1 if failed else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))
diff --git a/steward/ansible/bundled/maintenance/swarm_prune.yml b/steward/ansible/bundled/maintenance/swarm_prune.yml
new file mode 100644
index 0000000..cdd587a
--- /dev/null
+++ b/steward/ansible/bundled/maintenance/swarm_prune.yml
@@ -0,0 +1,43 @@
+---
+# description: Reclaim disk on swarm nodes but keep every image a swarm service is pinned to, so each node can take over the others' services without pulling.
+# steward:category: maintenance
+# steward:confirm: true
+# Run on the swarm's nodes; each must be a manager (only a manager can list
+# services). On each node:
+#   1. docker system prune: stopped containers, unused networks, dangling
+#      images, build cache (+ prune_volumes=true: unused anonymous volumes).
+#   2. files/swarm_prune.py: removes every other image no container uses and
+#      no swarm service is pinned to, wherever that service's task runs. A
+#      node keeps the images for the services running elsewhere, so it can
+#      start them at once when the other node fails or is drained.
+# A node that can't list the swarm's services removes no images and the run
+# fails, rather than pruning blind. Check mode lists what would go.
+- name: Swarm prune
+  hosts: all
+  gather_facts: false
+  become: true
+  vars:
+    prune_volumes: false
+  tasks:
+    - name: Prune stopped containers, unused networks, dangling images and build cache
+      ansible.builtin.command:
+        argv: >-
+          {{ ['docker', 'system', 'prune', '-f']
+             + (['--volumes'] if prune_volumes | bool else []) }}
+      register: system_prune
+      changed_when: "'Total reclaimed space: 0B' not in system_prune.stdout"
+
+    - name: Remove images no container uses and no swarm service is pinned to
+      ansible.builtin.script:
+        cmd: "swarm_prune.py{{ ' --dry-run' if ansible_check_mode else '' }}"
+        executable: python3
+      check_mode: false
+      register: image_prune
+      changed_when: >-
+        not ansible_check_mode and (image_prune.stdout | from_json).removed | default([]) | length > 0
+
+    - name: Report what went
+      ansible.builtin.debug:
+        msg: >-
+          {{ (system_prune.stdout_lines | default([]) | select | list)[-1:]
+             + [image_prune.stdout | from_json] }}
diff --git a/tests/core/test_swarm_prune.py b/tests/core/test_swarm_prune.py
new file mode 100644
index 0000000..0ff7fe3
--- /dev/null
+++ b/tests/core/test_swarm_prune.py
@@ -0,0 +1,118 @@
+"""The swarm prune keeps every image a swarm service is pinned to on every
+node, removes the rest no container uses, and removes nothing when it can't
+see the swarm."""
+import importlib.util
+import json
+from pathlib import Path
+
+import pytest
+import yaml
+
+from steward.ansible.sources import discover_playbook_meta, discover_playbooks, get_sources
+
+BUILTIN = Path(get_sources({"sources": []})[0]["path"])
+
+
+@pytest.fixture
+def prune():
+    spec = importlib.util.spec_from_file_location(
+        "swarm_prune", BUILTIN / "maintenance" / "files" / "swarm_prune.py")
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+def _image(id_, tags=(), digests=(), size=100):
+    return {"Id": id_, "RepoTags": list(tags), "RepoDigests": list(digests), "Size": size}
+
+
+def test_the_playbook_is_listed_and_asks_before_it_runs():
+    assert "maintenance/swarm_prune.yml" in discover_playbooks(str(BUILTIN))
+    content = (BUILTIN / "maintenance" / "swarm_prune.yml").read_text()
+    meta = discover_playbook_meta(content)
+    assert meta["confirm"] is True and meta["category"] == "maintenance" and meta["description"]
+    play = yaml.safe_load(content)[0]
+    script = next(t for t in play["tasks"] if "ansible.builtin.script" in t)
+    assert script["check_mode"] is False and "--dry-run" in script["ansible.builtin.script"]["cmd"]
+
+
+def test_services_are_matched_by_digest_or_else_by_tag(prune):
+    digests, refs = prune.wanted([
+        "registry.example/team/app:dev@sha256:aaa",
+        "docker.io/library/redis:7",
+        "registry.example:5000/tool",
+        None,
+    ])
+    assert digests == {"sha256:aaa"}
+    assert refs == {"redis:7", "registry.example:5000/tool:latest"}
+
+
+def test_an_image_a_service_or_a_container_needs_stays(prune):
+    images = [
+        _image("sha256:1", ["registry.example/team/app:dev"], ["registry.example/team/app@sha256:aaa"]),
+        _image("sha256:2", ["registry.example/team/app:dev-old"], ["registry.example/team/app@sha256:old"]),
+        _image("sha256:3", ["redis:7"], ["redis@sha256:rrr"]),
+        _image("sha256:4", ["tool:1"]),
+        _image("sha256:5", [], ["registry.example/team/app@sha256:bbb"]),
+    ]
+    gone = prune.removable(images, {"sha256:aaa"}, {"redis:7"}, used={"sha256:4"})
+    assert [im["Id"] for im in gone] == ["sha256:2", "sha256:5"]
+
+
+def _fake_docker(monkeypatch, prune, answers, removed):
+    def docker(*args):
+        if args[:2] == ("image", "rm"):
+            removed.append(args[-1])
+            return ""
+        key = " ".join(args[:2])
+        if key not in answers:
+            raise RuntimeError(f"docker {key}: unexpected")
+        answer = answers[key]
+        if isinstance(answer, Exception):
+            raise answer
+        return answer
+    monkeypatch.setattr(prune, "docker", docker)
+
+
+def _answers():
+    return {
+        "service ls": "s1 s2\n",
+        "service inspect": json.dumps([
+            {"Spec": {"TaskTemplate": {"ContainerSpec": {"Image": "registry.example/app:dev@sha256:aaa"}}}},
+            {"Spec": {"TaskTemplate": {"ContainerSpec": {"Image": "proxy:3@sha256:ppp"}}}}]),
+        "ps -aq": "c1\n",
+        "container inspect": json.dumps([{"Image": "sha256:9", "Config": {"Image": "local:1"}}]),
+        "image ls": "sha256:1\nsha256:2\nsha256:3\nsha256:9\n",
+        "image inspect": json.dumps([
+            _image("sha256:1", ["registry.example/app:dev"], ["registry.example/app@sha256:aaa"]),
+            _image("sha256:2", ["proxy:3"], ["proxy@sha256:ppp"]),
+            _image("sha256:3", ["registry.example/app:old"], ["registry.example/app@sha256:old"], size=500),
+            _image("sha256:9", ["local:1"]),
+        ]),
+    }
+
+
+def test_a_run_removes_only_what_nothing_needs_and_says_so(prune, monkeypatch, capsys):
+    removed: list = []
+    _fake_docker(monkeypatch, prune, _answers(), removed)
+    assert prune.main([]) == 0
+    out = json.loads(capsys.readouterr().out)
+    assert removed == ["sha256:3"]
+    assert out == {"kept": 3, "removed": ["registry.example/app:old"], "failed": [],
+                   "freed_bytes": 500, "dry_run": False}
+
+
+def test_a_dry_run_removes_nothing(prune, monkeypatch, capsys):
+    removed: list = []
+    _fake_docker(monkeypatch, prune, _answers(), removed)
+    assert prune.main(["--dry-run"]) == 0
+    assert removed == [] and json.loads(capsys.readouterr().out)["removed"] == ["registry.example/app:old"]
+
+
+def test_a_node_that_cant_list_the_swarm_removes_nothing(prune, monkeypatch, capsys):
+    removed: list = []
+    answers = _answers()
+    answers["service ls"] = RuntimeError("docker service ls: This node is not a swarm manager.")
+    _fake_docker(monkeypatch, prune, answers, removed)
+    assert prune.main([]) == 2
+    assert removed == [] and "nothing removed" in json.loads(capsys.readouterr().out)["error"]