diff --git a/src/executors/docker_prune_executor.py b/src/executors/docker_prune_executor.py
new file mode 100644
index 0000000..e6a913d
--- /dev/null
+++ b/src/executors/docker_prune_executor.py
@@ -0,0 +1,109 @@
+"""
+Docker Prune Executor
+
+Scheduled, non-interactive reclaim of Docker disk usage. The host equivalent is
+system-admin-toj's scripts/disk/prune-docker.sh, which prompts per stage; a cron
+task cannot prompt, so the destructive stages are opt-in instead.
+
+Runs the docker CLI against the socket already mounted into this container.
+
+Config schema:
+    {
+        "build_cache": true,        # safe: cache is rebuilt on demand
+        "dangling_images": true,    # safe: untagged layers nothing references
+        "unused_images": false,     # re-pull on next deploy; costs bandwidth
+        "volumes": false,           # DESTRUCTIVE - see below
+        "build_cache_until_hours": 168,
+        "dry_run": false
+    }
+
+`volumes` is off by default and should stay off unless you have checked what is
+actually unattached. `docker volume prune` removes every volume not bound to a
+*running* container, which includes the data volume of anything merely stopped.
+On this host that is a plausible way to lose a database.
+
+Defaults are the two stages that only ever discard regenerable data.
+"""
+import asyncio
+import logging
+from typing import Any, Dict, List, Tuple
+
+from src.config import Settings
+
+logger = logging.getLogger(__name__)
+
+COMMAND_TIMEOUT = 900
+
+
+async def _run(args: List[str]) -> Tuple[int, str, str]:
+    proc = await asyncio.create_subprocess_exec(
+        *args,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    try:
+        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=COMMAND_TIMEOUT)
+    except asyncio.TimeoutError:
+        proc.kill()
+        await proc.wait()
+        raise Exception(f"timed out after {COMMAND_TIMEOUT}s: {' '.join(args)}")
+    return proc.returncode, stdout.decode().strip(), stderr.decode().strip()
+
+
+def _reclaimed(output: str) -> str:
+    """Pull the 'Total reclaimed space: X' line out of docker's prune output."""
+    for line in output.splitlines():
+        if "reclaimed space" in line.lower():
+            return line.split(":", 1)[1].strip()
+    return "0B"
+
+
+async def execute(config: Dict[str, Any], settings: Settings) -> str:
+    dry_run = bool(config.get("dry_run", False))
+    until_hours = int(config.get("build_cache_until_hours", 168))
+
+    stages: List[Tuple[str, List[str]]] = []
+    if config.get("build_cache", True):
+        stages.append(
+            ("build cache", ["docker", "builder", "prune", "-f", "--filter", f"until={until_hours}h"])
+        )
+    if config.get("dangling_images", True):
+        stages.append(("dangling images", ["docker", "image", "prune", "-f"]))
+    if config.get("unused_images", False):
+        stages.append(("unused images", ["docker", "image", "prune", "-a", "-f"]))
+    if config.get("volumes", False):
+        logger.warning(
+            "volume pruning is enabled; this removes volumes belonging to stopped "
+            "containers, not just orphaned ones"
+        )
+        stages.append(("volumes", ["docker", "volume", "prune", "-f"]))
+
+    if not stages:
+        return "no prune stages enabled; nothing to do"
+
+    rc, out, err = await _run(["docker", "system", "df"])
+    if rc != 0:
+        raise Exception(f"docker unavailable: {err or out}")
+    before = out
+
+    if dry_run:
+        planned = ", ".join(name for name, _ in stages)
+        logger.info("dry run; would prune: %s", planned)
+        return f"dry run - would prune: {planned}\n{before}"
+
+    results = []
+    for name, args in stages:
+        rc, out, err = await _run(args)
+        if rc != 0:
+            # Report rather than abort: a later stage may still reclaim space, and
+            # a partial reclaim is more useful than none.
+            logger.error("prune stage %r failed: %s", name, err or out)
+            results.append(f"{name}: FAILED ({(err or out).splitlines()[0] if (err or out) else 'unknown'})")
+            continue
+        results.append(f"{name}: {_reclaimed(out)}")
+        logger.info("pruned %s -> %s", name, _reclaimed(out))
+
+    summary = "; ".join(results)
+    if any("FAILED" in r for r in results):
+        raise Exception(f"one or more prune stages failed: {summary}")
+    return f"reclaimed - {summary}"
diff --git a/tests/test_docker_prune_executor.py b/tests/test_docker_prune_executor.py
new file mode 100644
index 0000000..1adfe78
--- /dev/null
+++ b/tests/test_docker_prune_executor.py
@@ -0,0 +1,135 @@
+"""
+Tests for the docker prune executor.
+
+The important property is which stages run. `volumes` removes volumes belonging
+to merely-stopped containers, so it must never be enabled by accident, and the
+safe stages must stay on by default.
+"""
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from src.config import Settings
+from src.executors import docker_prune_executor as prune
+
+
+def _runner(reclaimed="Total reclaimed space: 1.5GB", rc=0):
+    """Fake _run returning docker-shaped output for every invocation."""
+    async def run(args):
+        if args[:3] == ["docker", "system", "df"]:
+            return 0, "TYPE  TOTAL  ACTIVE  SIZE  RECLAIMABLE", ""
+        return rc, reclaimed, "" if rc == 0 else "boom"
+    return run
+
+
+@pytest.mark.executor
+@pytest.mark.unit
+class TestStageSelection:
+
+    @pytest.mark.asyncio
+    async def test_defaults_run_only_the_safe_stages(self, test_settings: Settings):
+        calls = []
+
+        async def run(args):
+            calls.append(args)
+            if args[:3] == ["docker", "system", "df"]:
+                return 0, "df output", ""
+            return 0, "Total reclaimed space: 0B", ""
+
+        with patch.object(prune, "_run", run):
+            await prune.execute({}, test_settings)
+
+        joined = [" ".join(c) for c in calls]
+        assert any("builder prune" in c for c in joined)
+        assert any("image prune -f" in c for c in joined)
+        # The destructive ones must not appear without being asked for.
+        assert not any("volume prune" in c for c in joined)
+        assert not any("image prune -a" in c for c in joined)
+
+    @pytest.mark.asyncio
+    async def test_volumes_only_when_explicitly_enabled(self, test_settings: Settings):
+        calls = []
+
+        async def run(args):
+            calls.append(args)
+            if args[:3] == ["docker", "system", "df"]:
+                return 0, "df output", ""
+            return 0, "Total reclaimed space: 2GB", ""
+
+        with patch.object(prune, "_run", run):
+            await prune.execute({"volumes": True}, test_settings)
+
+        assert any("volume prune" in " ".join(c) for c in calls)
+
+    @pytest.mark.asyncio
+    async def test_all_stages_disabled_is_a_no_op(self, test_settings: Settings):
+        with patch.object(prune, "_run", AsyncMock()) as run:
+            result = await prune.execute(
+                {"build_cache": False, "dangling_images": False}, test_settings
+            )
+        assert "nothing to do" in result
+        run.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_dry_run_executes_no_prune(self, test_settings: Settings):
+        calls = []
+
+        async def run(args):
+            calls.append(args)
+            return 0, "df output", ""
+
+        with patch.object(prune, "_run", run):
+            result = await prune.execute({"dry_run": True}, test_settings)
+
+        assert "dry run" in result
+        assert all("prune" not in " ".join(c) for c in calls)
+
+
+@pytest.mark.executor
+@pytest.mark.unit
+class TestFailureHandling:
+
+    @pytest.mark.asyncio
+    async def test_docker_unavailable_raises(self, test_settings: Settings):
+        async def run(args):
+            return 1, "", "Cannot connect to the Docker daemon"
+
+        with patch.object(prune, "_run", run):
+            with pytest.raises(Exception, match="docker unavailable"):
+                await prune.execute({}, test_settings)
+
+    @pytest.mark.asyncio
+    async def test_failed_stage_surfaces_but_others_still_run(self, test_settings: Settings):
+        attempted = []
+
+        async def run(args):
+            if args[:3] == ["docker", "system", "df"]:
+                return 0, "df output", ""
+            attempted.append(" ".join(args))
+            if "builder" in args:
+                return 1, "", "builder exploded"
+            return 0, "Total reclaimed space: 3MB", ""
+
+        with patch.object(prune, "_run", run):
+            with pytest.raises(Exception, match="one or more prune stages failed"):
+                await prune.execute({}, test_settings)
+
+        # The image stage must still have been attempted after builder failed.
+        assert any("image prune" in a for a in attempted)
+
+
+@pytest.mark.executor
+@pytest.mark.unit
+class TestOutputParsing:
+
+    @pytest.mark.parametrize(
+        "output,expected",
+        [
+            ("Total reclaimed space: 1.5GB", "1.5GB"),
+            ("deleted: sha256:abc\nTotal reclaimed space: 0B", "0B"),
+            ("no such line", "0B"),
+            ("", "0B"),
+        ],
+    )
+    def test_reclaimed_parsing(self, output, expected):
+        assert prune._reclaimed(output) == expected