diff --git a/.github/workflows/dev-image-retention.yml b/.github/workflows/dev-image-retention.yml
new file mode 100644
index 00000000..19ccab70
--- /dev/null
+++ b/.github/workflows/dev-image-retention.yml
@@ -0,0 +1,59 @@
+name: DEV API image retention
+
+on:
+  push:
+    branches: [dev]
+    paths: [.github/workflows/dev-image-retention.yml, scripts/dev_image_retention.py]
+  workflow_dispatch:
+    inputs:
+      mode:
+        type: choice
+        options: [plan, apply]
+        default: plan
+        required: true
+      plan_sha256:
+        description: Exact digest printed by the read-only plan
+        type: string
+        required: false
+
+permissions:
+  contents: read
+
+concurrency:
+  group: api-deploy-dev
+  cancel-in-progress: false
+
+jobs:
+  retention:
+    if: github.ref == 'refs/heads/dev'
+    environment: dev
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-node@v4
+        with:
+          node-version: '24'
+      - name: Install verified SSH transport
+        run: npm install --prefix "$RUNNER_TEMP/dev-migration-ssh" --ignore-scripts --no-package-lock --no-audit --no-fund ssh2@1.16.0
+      - name: Plan or apply exact unused DEV image removals
+        env:
+          NODE_PATH: ${{ runner.temp }}/dev-migration-ssh/node_modules
+          DEV_SERVER_HOST: ${{ secrets.DEV_SERVER_HOST }}
+          DEV_SERVER_USER: ${{ secrets.DEV_SERVER_USER }}
+          DEV_SERVER_PASSWORD: ${{ secrets.DEV_SERVER_PASSWORD }}
+          DEV_SSH_HOST_KEY_SHA256: ${{ vars.DEV_SSH_HOST_KEY_SHA256 }}
+          RETENTION_MODE: ${{ github.event_name == 'push' && 'plan' || inputs.mode }}
+          PLAN_SHA256: ${{ inputs.plan_sha256 }}
+        run: |
+          node <<'NODE'
+          const fs = require('node:fs');
+          const mode = process.env.RETENTION_MODE;
+          const digest = process.env.PLAN_SHA256 || 'none';
+          if (!['plan', 'apply'].includes(mode)) throw new Error('Invalid mode');
+          if (!/^(none|[0-9a-f]{64})$/.test(digest)) throw new Error('Invalid plan digest');
+          if (mode === 'apply' && !/^[0-9a-f]{64}$/.test(digest)) throw new Error('Inspect a plan first');
+          const source = fs.readFileSync('scripts/dev_image_retention.py', 'utf8');
+          process.env.DEV_DEPLOY_SCRIPT = `set -euo pipefail\npython3 - ${mode} ${digest} <<'DEV_IMAGE_RETENTION_PY'\n${source}\nDEV_IMAGE_RETENTION_PY\n`;
+          require('./scripts/dev_ci_ssh.cjs');
+          NODE
diff --git a/scripts/dev_image_retention.py b/scripts/dev_image_retention.py
new file mode 100644
index 00000000..4c7dc6db
--- /dev/null
+++ b/scripts/dev_image_retention.py
@@ -0,0 +1,221 @@
+"""Plan or remove old, reproducible DEV API images; never prune Docker globally."""
+
+import hashlib
+import json
+import os
+import re
+import subprocess
+import sys
+from datetime import datetime, timedelta, timezone
+
+TAG = re.compile(r"procollab-dev-api:([0-9a-f]{40})(?:-[0-9a-f]{12})?$")
+REPO = "/root/api"
+
+
+def command(*args):
+    return subprocess.check_output(args, text=True).strip()
+
+
+def require(condition, message):
+    if not condition:
+        raise RuntimeError(message)
+
+
+def output(message):
+    sys.stdout.write(message + "\n")
+    sys.stdout.flush()
+
+
+def digest(value):
+    return hashlib.sha256(json.dumps(value, sort_keys=True).encode()).hexdigest()
+
+
+def inventory():
+    result = {}
+    for image_id in set(command("docker", "image", "ls", "-aq", "--no-trunc").split()):
+        value = json.loads(
+            command(
+                "docker",
+                "image",
+                "inspect",
+                "--format",
+                '{"id":{{json .Id}},"tags":{{json .RepoTags}},"created":{{json .Created}}}',
+                image_id,
+            )
+        )
+        value["tags"] = sorted(value["tags"] or [])
+        result[image_id] = value
+    return result
+
+
+def containers():
+    result = []
+    for container_id in command("docker", "ps", "-aq", "--no-trunc").split():
+        result.append(
+            json.loads(
+                command(
+                    "docker",
+                    "inspect",
+                    "--format",
+                    '{"id":{{json .Id}},"image":{{json .Image}},"state":{{json .State.Status}},'
+                    '"started":{{json .State.StartedAt}},"restarts":{{json .RestartCount}}}',
+                    container_id,
+                )
+            )
+        )
+    return sorted(result, key=lambda value: value["id"])
+
+
+def volumes():
+    return sorted(
+        command(
+            "docker", "volume", "ls", "--format", "{{.Driver}} {{.Name}}"
+        ).splitlines()
+    )
+
+
+def created_at(image):
+    value = re.sub(r"(\.\d{6})\d+", r"\1", image["created"])
+    return datetime.fromisoformat(value.replace("Z", "+00:00"))
+
+
+def plan(images, container_rows, volume_rows, now, reachable):
+    used = {row["image"] for row in container_rows}
+    reproducible = []
+    for image in images.values():
+        tags = image["tags"]
+        matches = [TAG.fullmatch(tag) for tag in tags]
+        if tags and all(matches) and all(reachable(match[1]) for match in matches):
+            reproducible.append(image)
+    newest = {
+        image["id"]
+        for image in sorted(
+            reproducible, key=lambda row: (created_at(row), row["id"]), reverse=True
+        )[:3]
+    }
+    candidates = sorted(
+        (
+            image
+            for image in reproducible
+            if image["id"] not in used | newest
+            and created_at(image) < now - timedelta(hours=48)
+        ),
+        key=lambda image: image["id"],
+    )
+    proof = {
+        "images": images,
+        "containers": container_rows,
+        "volumes": volume_rows,
+        "candidates": candidates,
+    }
+    return {
+        "sha256": digest(proof),
+        "candidates": candidates,
+        "protected_count": len(images) - len(candidates),
+    }
+
+
+def apply(plan_value, expected, images_before, containers_before, volumes_before):
+    if plan_value["sha256"] != expected:
+        raise RuntimeError("Image inventory changed; inspect a new plan before cleanup")
+    remaining = dict(images_before)
+    for image in plan_value["candidates"]:
+        if (
+            containers() != containers_before
+            or volumes() != volumes_before
+            or inventory() != remaining
+        ):
+            raise RuntimeError("Docker state changed; no further images will be removed")
+        for tag in image["tags"]:
+            # Non-force removal preserves images referenced by any container.
+            # --no-prune also prevents removal of untagged parent images.
+            subprocess.run(["docker", "image", "rm", "--no-prune", tag], check=True)
+        del remaining[image["id"]]
+    if (
+        inventory() != remaining
+        or containers() != containers_before
+        or volumes() != volumes_before
+    ):
+        raise RuntimeError("Docker state differs from the reviewed removal plan")
+    output(
+        "PROTECTED_IMAGES_UNCHANGED=true\nCONTAINERS_UNCHANGED=true\nVOLUMES_UNCHANGED=true"
+    )
+
+
+def main():
+    mode, expected = sys.argv[1:]
+    require(mode in ("plan", "apply"), "Invalid operation")
+    require(
+        mode == "plan" or re.fullmatch(r"[0-9a-f]{64}", expected), "Inspect a plan first"
+    )
+    require(
+        os.geteuid() == 0 and os.path.realpath(REPO) == REPO,
+        "Unexpected host or repository",
+    )
+    command(
+        "git",
+        "-C",
+        REPO,
+        "merge-base",
+        "--is-ancestor",
+        "beef71645567128398d0afa5a2eb80dc6bc8dac7",
+        "HEAD",
+    )
+    command(
+        "grep",
+        "-Eq",
+        r"server_name[[:space:]]+dev\.procollab\.ru;",
+        "/etc/nginx/sites-enabled/dev.procollab.ru",
+    )
+    runtime = []
+    for service in ("web", "celerys"):
+        container_id = command(
+            "docker",
+            "ps",
+            "-q",
+            "--no-trunc",
+            "--filter",
+            f"label=com.docker.compose.project.working_dir={REPO}",
+            "--filter",
+            f"label=com.docker.compose.service={service}",
+        )
+        require(
+            re.fullmatch(r"[0-9a-f]{64}", container_id),
+            "Expected one running DEV service",
+        )
+        reference = command("docker", "inspect", "-f", "{{.Config.Image}}", container_id)
+        require(TAG.fullmatch(reference), "Unexpected DEV image reference")
+        runtime.append(command("docker", "inspect", "-f", "{{.Image}}", container_id))
+    require(len(set(runtime)) == 1, "DEV web and worker image mismatch")
+    images_before, containers_before, volumes_before = (
+        inventory(),
+        containers(),
+        volumes(),
+    )
+
+    def reachable(sha):
+        return (
+            subprocess.run(
+                ["git", "-C", REPO, "merge-base", "--is-ancestor", sha, "HEAD"],
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+            ).returncode
+            == 0
+        )
+
+    value = plan(
+        images_before,
+        containers_before,
+        volumes_before,
+        datetime.now(timezone.utc),
+        reachable,
+    )
+    output("DEV_IMAGE_RETENTION_PLAN=" + json.dumps(value, sort_keys=True))
+    output(command("df", "-h", "/"))
+    if mode == "apply":
+        apply(value, expected, images_before, containers_before, volumes_before)
+        output(command("df", "-h", "/"))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/test_dev_image_retention.py b/tests/test_dev_image_retention.py
new file mode 100644
index 00000000..8047456a
--- /dev/null
+++ b/tests/test_dev_image_retention.py
@@ -0,0 +1,112 @@
+import unittest
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+from scripts import dev_image_retention as service
+
+
+class DevImageRetentionTests(unittest.TestCase):
+    now = datetime(2026, 10, 25, tzinfo=timezone.utc)
+
+    def image(self, number, age):
+        return {
+            "id": f"sha256:{number:064x}",
+            "tags": [f"procollab-dev-api:{number:040x}"],
+            "created": (self.now - timedelta(days=age)).isoformat(),
+        }
+
+    def test_plan_keeps_all_container_images_newest_three_and_unknown_images(self):
+        rows = [self.image(number, number) for number in range(1, 7)]
+        foreign = self.image(7, 9)
+        foreign["tags"] = ["other-project:old"]
+        mixed = self.image(8, 9)
+        mixed["tags"].append("api-web:latest")
+        dangling = self.image(9, 9)
+        dangling["tags"] = []
+        unreachable = self.image(10, 9)
+        images = {
+            row["id"]: row for row in rows + [foreign, mixed, dangling, unreachable]
+        }
+        containers = [
+            {"id": "stopped-container", "image": rows[4]["id"], "state": "exited"}
+        ]
+        plan = service.plan(
+            images,
+            containers,
+            ["volume"],
+            self.now,
+            lambda sha: sha != f"{10:040x}",
+        )
+        self.assertEqual(
+            {row["id"] for row in plan["candidates"]}, {rows[3]["id"], rows[5]["id"]}
+        )
+        self.assertEqual(plan["protected_count"], 8)
+        # Time alone does not change the inventory digest while candidates stay identical.
+        self.assertEqual(
+            plan["sha256"],
+            service.plan(
+                images,
+                containers,
+                ["volume"],
+                self.now + timedelta(minutes=1),
+                lambda sha: sha != f"{10:040x}",
+            )["sha256"],
+        )
+
+    def test_young_images_are_not_removed_even_when_more_than_three_exist(self):
+        images = {
+            row["id"]: row
+            for row in [self.image(number, 0.1 * number) for number in range(1, 7)]
+        }
+        self.assertEqual(
+            service.plan(images, [], [], self.now, lambda sha: True)["candidates"], []
+        )
+
+    def test_nanosecond_docker_timestamp_is_supported(self):
+        self.assertEqual(
+            service.created_at({"created": "2026-10-25T00:00:00.123456789Z"}).microsecond,
+            123456,
+        )
+
+    @patch.object(service.subprocess, "run")
+    def test_wrong_review_digest_never_mutates_docker(self, run):
+        with self.assertRaises(RuntimeError):
+            service.apply({"sha256": "reviewed", "candidates": []}, "changed", {}, [], [])
+        run.assert_not_called()
+
+    @patch.object(service.subprocess, "run")
+    @patch.object(service, "containers", return_value=[{"image": "new-container"}])
+    def test_new_container_aborts_before_removing_any_tag(self, containers, run):
+        image = self.image(1, 9)
+        with self.assertRaises(RuntimeError):
+            service.apply(
+                {"sha256": "same", "candidates": [image]},
+                "same",
+                {image["id"]: image},
+                [],
+                [],
+            )
+        run.assert_not_called()
+
+    @patch.object(service, "containers", return_value=[])
+    @patch.object(service, "volumes", return_value=["volume"])
+    @patch.object(service, "inventory")
+    @patch.object(service.subprocess, "run")
+    def test_removes_only_reviewed_tags_without_force_or_parent_pruning(
+        self, run, inventory, volumes, containers
+    ):
+        image, protected = self.image(1, 9), self.image(2, 1)
+        image["tags"].append(image["tags"][0] + "-0123456789ab")
+        before = {row["id"]: row for row in (image, protected)}
+        inventory.side_effect = [before, {protected["id"]: protected}]
+        service.apply(
+            {"sha256": "same", "candidates": [image]}, "same", before, [], ["volume"]
+        )
+        self.assertEqual(run.call_count, 2)
+        for call, tag in zip(run.call_args_list, image["tags"]):
+            self.assertEqual(call.args, (["docker", "image", "rm", "--no-prune", tag],))
+            self.assertEqual(call.kwargs, {"check": True})
+
+
+if __name__ == "__main__":
+    unittest.main()