diff --git a/CHANGELOG.md b/CHANGELOG.md
index 705d122..a4b5551 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,14 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 
 ## [Unreleased]
 
+### Added
+- `pg_dump_executor`: dumps every postgres-shared database individually (`docker exec
+  postgres-shared pg_dump -Fc`, version-matched inside that container rather than through this
+  image's PG17 client), plus a globals-only dump of roles. Each archive is verified with
+  `pg_restore --list` before being promoted — a zero-TOC archive is treated as a failed dump,
+  not a kept one. One database failing does not stop the others; the run is still reported
+  CRITICAL and every failure is named. Not yet registered as a scheduled task (T-7).
+
 ## [1.9.0] - 2026-08-11
 
 ### Changed
diff --git a/src/executors/pg_dump_executor.py b/src/executors/pg_dump_executor.py
new file mode 100644
index 0000000..d43ca34
--- /dev/null
+++ b/src/executors/pg_dump_executor.py
@@ -0,0 +1,311 @@
+"""
+Postgres Dump Executor
+
+Backs up every database on postgres-shared with `pg_dump`, one file per
+database plus a globals-only dump of roles, so any single database can be
+restored on its own rather than only as part of a full-cluster file copy.
+
+Why this exists: the only thing that currently copies postgres-shared is
+config_backup_executor, which tars `/data/docker-data` — including the live
+data directory — while the server runs, without `pg_backup_start`. That is a
+file copy of a running cluster, with no per-database restore and no
+verification that the copy is even usable. This executor is the proven
+replacement; config_backup_executor is left untouched until it is retired by
+a separate decision.
+
+Why the dump runs server-side, through `docker exec`, instead of this
+image's own pg_dump: postgres-shared is PostgreSQL 16.15; this scheduler
+image carries the client tools for PostgreSQL 17.10. A 17 custom-format
+archive cannot be read by a 16 `pg_restore`, and 17's plain SQL dumps emit
+`SET transaction_timeout`, which 16 rejects outright. `docker exec
+postgres-shared pg_dump ...` runs version 16's own pg_dump inside that
+container, so the archive is version-matched by construction. The mismatch
+runs the other way for verification: 17's `pg_restore --list` reads a 16
+archive without trouble, so it is run directly — no `docker exec` needed,
+because the scheduler container already has it.
+
+Config schema:
+    {
+        "container": "postgres-shared",      # required; the server's container name
+        "databases": "all",                  # "all", or an explicit list of names
+        "exclude": ["template0", "template1",
+                     "test_scheduler", "core_api_test"],
+        "globals": true,                      # also dump roles via pg_dumpall --globals-only
+        "output_dir": "/backups/postgres",
+        "retention_days": 14,
+        "timeout_per_db": 900                 # seconds; applies to each dump individually
+    }
+
+"all" is discovered with `select datname from pg_database where not
+datistemplate`, then `exclude` is subtracted — so the exclude list is a
+safety net even when the template databases are already filtered out
+upstream.
+
+Every archive is written to `<name>-<stamp>.dump.partial`, verified, then
+renamed and chmod 0600 — the same partial-then-rename-then-chmod shape as
+portainer_backup_executor, because a PostgreSQL dump contains table data and
+the globals dump contains role password hashes.
+
+Failure semantics: one database failing does not stop the others, and their
+files are kept. The run as a whole is still reported CRITICAL, and every
+failed target is named in the result string, by construction — that string
+is built by joining each target's own outcome, not a separate summary of it.
+"""
+import asyncio
+import logging
+import re
+import time
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import List
+
+from src.config import Settings
+from src.executors import health_report
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_OUTPUT_DIR = "/backups/postgres"
+DEFAULT_RETENTION_DAYS = 14
+DEFAULT_TIMEOUT_PER_DB = 900
+
+DISCOVER_SQL = "select datname from pg_database where not datistemplate"
+
+# Matches every filename this executor writes and nothing else: "<name>" is
+# either a database name (letters, digits, underscore, hyphen, dot — real
+# names here include "sonarr-log") or the literal "globals", followed by the
+# UTC run stamp and the right extension. `_prune` uses this to delete only
+# what it wrote, never a file placed here by something else.
+FILENAME_RE = re.compile(r"^(?:globals|[A-Za-z0-9_.\-]+)-\d{8}T\d{6}Z\.(?:dump|sql)$")
+
+
+def _toc_entry_count(list_output: str) -> int:
+    """Count TOC entry lines from `pg_restore --list` output.
+
+    Every header and comment line in that output starts with ';'; the actual
+    entries pg_restore would restore do not. Counting on that distinction is
+    more direct than parsing the "; TOC Entries: N" summary line, and does not
+    depend on that line's exact wording surviving a pg_restore version bump.
+    """
+    return sum(
+        1 for line in list_output.splitlines()
+        if line.strip() and not line.lstrip().startswith(";")
+    )
+
+
+async def _verify_archive(path: Path) -> int:
+    """Run the image's own pg_restore --list against a finished dump.
+
+    A 200-looking response that is not a usable archive is worse than no
+    backup at all, because it looks fine until the day it is needed — the
+    same reasoning portainer_backup_executor applies to its own archive. The
+    only signal pg_restore --list gives back is the TOC; zero entries is
+    treated the same as a file that would not open.
+    """
+    proc = await asyncio.create_subprocess_exec(
+        "pg_restore", "--list", str(path),
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    stdout, stderr = await proc.communicate()
+    if proc.returncode != 0:
+        raise Exception(f"pg_restore --list failed: {stderr.decode(errors='replace').strip()[:300]}")
+    entries = _toc_entry_count(stdout.decode(errors="replace"))
+    if entries <= 0:
+        raise Exception("archive has zero TOC entries")
+    return entries
+
+
+async def _discover_databases(container: str) -> List[str]:
+    proc = await asyncio.create_subprocess_exec(
+        "docker", "exec", container, "psql", "-U", "postgres", "-tAc", DISCOVER_SQL,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    stdout, stderr = await proc.communicate()
+    if proc.returncode != 0:
+        raise Exception(f"database discovery failed: {stderr.decode(errors='replace').strip()[:300]}")
+    return [line.strip() for line in stdout.decode().splitlines() if line.strip()]
+
+
+async def _stream_subprocess_to_file(args: List[str], destination: Path, timeout: int) -> None:
+    """Run args, writing its stdout to destination as it arrives, within timeout.
+
+    Same create-subprocess-then-wait_for-then-kill shape as
+    docker_prune_executor._run; the body differs because a dump's stdout is a
+    binary stream that can run into the hundreds of MB, not the few lines of
+    text `communicate()` holds comfortably — so this reads and writes it in
+    chunks instead of buffering it whole.
+    """
+    proc = await asyncio.create_subprocess_exec(
+        *args,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+
+    async def _drain() -> bytes:
+        with open(destination, "wb") as f:
+            while True:
+                chunk = await proc.stdout.read(1 << 20)
+                if not chunk:
+                    break
+                f.write(chunk)
+        return await proc.stderr.read()
+
+    try:
+        stderr = await asyncio.wait_for(_drain(), timeout=timeout)
+    except asyncio.TimeoutError:
+        proc.kill()
+        await proc.wait()
+        raise Exception(f"timed out after {timeout}s: {' '.join(args)}")
+
+    rc = await proc.wait()
+    if rc != 0:
+        raise Exception(f"exited {rc}: {stderr.decode(errors='replace').strip()[:300]}")
+
+
+async def _dump_database(container: str, db: str, destination: Path, timeout: int) -> None:
+    await _stream_subprocess_to_file(
+        ["docker", "exec", container, "pg_dump", "-U", "postgres", "-Fc", db],
+        destination, timeout,
+    )
+
+
+async def _dump_globals(container: str, destination: Path, timeout: int) -> None:
+    await _stream_subprocess_to_file(
+        ["docker", "exec", container, "pg_dumpall", "-U", "postgres", "--globals-only"],
+        destination, timeout,
+    )
+
+
+def _prune(output_dir: Path, retention_days: int) -> int:
+    """Delete this executor's own files past the retention window."""
+    cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days)
+    removed = 0
+    for path in output_dir.iterdir():
+        if not path.is_file() or not FILENAME_RE.match(path.name):
+            continue
+        if datetime.fromtimestamp(path.stat().st_mtime, timezone.utc) < cutoff:
+            path.unlink()
+            removed += 1
+            logger.info("pruned old postgres dump: %s", path.name)
+    return removed
+
+
+async def _run(config: dict, settings: Settings) -> str:
+    container = config.get("container", "")
+    if not container:
+        raise ValueError("Missing required config: 'container'")
+
+    output_dir = Path(config.get("output_dir", DEFAULT_OUTPUT_DIR))
+    exclude = set(config.get("exclude") or [])
+    want_globals = bool(config.get("globals", True))
+
+    retention_days = config.get("retention_days", DEFAULT_RETENTION_DAYS)
+    if not isinstance(retention_days, int) or isinstance(retention_days, bool) or retention_days < 1:
+        raise ValueError(f"retention_days must be a positive integer, got {retention_days!r}")
+
+    timeout_per_db = config.get("timeout_per_db", DEFAULT_TIMEOUT_PER_DB)
+    if not isinstance(timeout_per_db, int) or isinstance(timeout_per_db, bool) or timeout_per_db < 1:
+        raise ValueError(f"timeout_per_db must be a positive integer, got {timeout_per_db!r}")
+
+    databases_cfg = config.get("databases", "all")
+    if databases_cfg == "all":
+        discovered = await _discover_databases(container)
+    elif isinstance(databases_cfg, list) and all(isinstance(d, str) for d in databases_cfg):
+        discovered = databases_cfg
+    else:
+        raise ValueError(f"'databases' must be \"all\" or a list of names, got {databases_cfg!r}")
+
+    targets = [d for d in discovered if d not in exclude]
+    if not targets and not want_globals:
+        raise ValueError("no databases selected (all excluded) and globals disabled; nothing to do")
+
+    output_dir.mkdir(parents=True, exist_ok=True)
+    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
+    started = time.monotonic()
+
+    succeeded: List[str] = []
+    failed: List[str] = []
+
+    for db in targets:
+        final = output_dir / f"{db}-{stamp}.dump"
+        partial = output_dir / f"{final.name}.partial"
+        try:
+            await _dump_database(container, db, partial, timeout_per_db)
+            entries = await _verify_archive(partial)
+            partial.replace(final)
+            final.chmod(0o600)
+            size_mb = final.stat().st_size / 1_048_576
+            succeeded.append(f"{db} ({entries} entries, {size_mb:.2f} MB)")
+        except Exception as exc:  # noqa: BLE001 - one database's failure must not stop the rest
+            logger.error("pg_dump failed for %s: %s", db, exc)
+            failed.append(f"FAILED {db}: {exc}")
+        finally:
+            if partial.exists():
+                partial.unlink()
+
+    if want_globals:
+        final = output_dir / f"globals-{stamp}.sql"
+        partial = output_dir / f"{final.name}.partial"
+        try:
+            await _dump_globals(container, partial, timeout_per_db)
+            # pg_restore --list cannot read a plain-SQL globals dump; a
+            # non-empty file is the only check available for it.
+            if partial.stat().st_size == 0:
+                raise Exception("globals dump is empty")
+            partial.replace(final)
+            final.chmod(0o600)  # role password hashes live in here
+            succeeded.append(f"globals ({final.stat().st_size} bytes)")
+        except Exception as exc:  # noqa: BLE001
+            logger.error("globals dump failed: %s", exc)
+            failed.append(f"FAILED globals: {exc}")
+        finally:
+            if partial.exists():
+                partial.unlink()
+
+    removed = _prune(output_dir, retention_days)
+    kept = len([p for p in output_dir.iterdir() if p.is_file() and FILENAME_RE.match(p.name)])
+    elapsed = time.monotonic() - started
+    total = len(targets) + (1 if want_globals else 0)
+
+    summary = (
+        f"dumped {len(succeeded)}/{total} targets in {elapsed:.1f}s "
+        f"({'; '.join(succeeded + failed)}); kept {kept}, pruned {removed} older than {retention_days}d"
+    )
+    logger.info(summary)
+
+    if failed:
+        raise Exception(summary)
+    return summary
+
+
+async def execute(config: dict, settings: Settings) -> str:
+    """Run the dump and report its own outcome to check_history (D-33, T-69).
+
+    The report wraps the work rather than living inside it, so the failure
+    path cannot be forgotten: an exception is reported as critical and then
+    re-raised, leaving the task's own status untouched. Reporting only
+    success would reproduce exactly the blind spot this replaces — a monitor
+    that cannot tell a failed backup from one that has not run.
+    """
+    try:
+        output = await _run(config, settings)
+    except Exception as exc:
+        await health_report.report_async(
+            settings,
+            domain="backup",
+            status=health_report.CRITICAL,
+            source="scheduler/pg_dump_executor",
+            summary=f"backup failed: {str(exc)[:300]}",
+            metrics={"job": "scheduler/pg_dump_executor", "error": str(exc)[:400]},
+        )
+        raise
+    await health_report.report_async(
+        settings,
+        domain="backup",
+        status=health_report.OK,
+        source="scheduler/pg_dump_executor",
+        summary=output[:400],
+        metrics={"job": "scheduler/pg_dump_executor"},
+    )
+    return output
diff --git a/tests/test_pg_dump_executor.py b/tests/test_pg_dump_executor.py
new file mode 100644
index 0000000..f4c8f4a
--- /dev/null
+++ b/tests/test_pg_dump_executor.py
@@ -0,0 +1,364 @@
+"""
+Tests for the postgres dump executor.
+
+The point of this executor is a backup that is actually restorable, so most
+of these cover the failure paths: an empty TOC behind a successful dump, one
+database failing while the others are kept, retention touching only its own
+files, and a per-database timeout actually firing. Every subprocess call is
+mocked — this suite must pass with postgres-shared unreachable (D-26).
+"""
+import asyncio
+from datetime import datetime, timedelta, timezone
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from src.config import Settings
+from src.executors import pg_dump_executor as pde
+
+
+def _config(tmp_path, **over):
+    cfg = {
+        "container": "postgres-shared",
+        "databases": "all",
+        "exclude": [],
+        "globals": False,  # most tests don't exercise the globals path; see TestGlobals
+        "output_dir": str(tmp_path),
+        "retention_days": 14,
+        "timeout_per_db": 900,
+    }
+    cfg.update(over)
+    return cfg
+
+
+async def _writing_dump(container, db, destination, timeout):
+    """A fake _dump_database that behaves like the real one: it writes bytes
+    to the destination it is given."""
+    destination.write_bytes(f"dump-of-{db}".encode())
+
+
+@pytest.mark.executor
+@pytest.mark.unit
+class TestConfigValidation:
+
+    @pytest.mark.asyncio
+    async def test_missing_container_rejected(self, test_settings: Settings, tmp_path):
+        with pytest.raises(ValueError, match="container"):
+            await pde.execute(_config(tmp_path, container=""), test_settings)
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("bad", [0, -1, "14", None, True])
+    async def test_bad_retention_rejected(self, bad, test_settings: Settings, tmp_path):
+        with pytest.raises(ValueError, match="retention_days"):
+            await pde.execute(_config(tmp_path, retention_days=bad), test_settings)
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("bad", [0, -1, "900", None, True])
+    async def test_bad_timeout_per_db_rejected(self, bad, test_settings: Settings, tmp_path):
+        with pytest.raises(ValueError, match="timeout_per_db"):
+            await pde.execute(_config(tmp_path, timeout_per_db=bad), test_settings)
+
+    @pytest.mark.asyncio
+    async def test_bad_databases_value_rejected(self, test_settings: Settings, tmp_path):
+        with pytest.raises(ValueError, match="databases"):
+            await pde.execute(_config(tmp_path, databases={"not": "valid"}), test_settings)
+
+    @pytest.mark.asyncio
+    async def test_everything_excluded_and_no_globals_rejected(self, test_settings: Settings, tmp_path):
+        with patch.object(pde, "_discover_databases", AsyncMock(return_value=["a", "b"])):
+            with pytest.raises(ValueError, match="nothing to do"):
+                await pde.execute(
+                    _config(tmp_path, exclude=["a", "b"], globals=False), test_settings
+                )
+
+
+@pytest.mark.executor
+@pytest.mark.unit
+class TestDiscovery:
+
+    @pytest.mark.asyncio
+    async def test_discover_databases_parses_rows(self):
+        proc = MagicMock()
+        proc.communicate = AsyncMock(return_value=(b"alpha\nbeta\n\n", b""))
+        proc.returncode = 0
+        with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=proc)) as mock_exec:
+            result = await pde._discover_databases("postgres-shared")
+        assert result == ["alpha", "beta"]
+        args = mock_exec.call_args.args
+        assert args[:3] == ("docker", "exec", "postgres-shared")
+        assert "psql" in args
+
+    @pytest.mark.asyncio
+    async def test_discover_databases_failure_raises(self):
+        proc = MagicMock()
+        proc.communicate = AsyncMock(return_value=(b"", b"connection refused"))
+        proc.returncode = 1
+        with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=proc)):
+            with pytest.raises(Exception, match="database discovery failed"):
+                await pde._discover_databases("postgres-shared")
+
+    @pytest.mark.asyncio
+    async def test_all_minus_excludes_selects_correct_targets(self, test_settings: Settings, tmp_path):
+        attempted = []
+
+        async def fake_dump(container, db, destination, timeout):
+            attempted.append(db)
+            destination.write_bytes(b"x")
+
+        with patch.object(pde, "_discover_databases", AsyncMock(
+                return_value=["keep", "template0", "test_scheduler"])), \
+             patch.object(pde, "_dump_database", fake_dump), \
+             patch.object(pde, "_verify_archive", AsyncMock(return_value=1)):
+            await pde.execute(
+                _config(tmp_path, exclude=["template0", "test_scheduler"]), test_settings
+            )
+
+        assert attempted == ["keep"]
+
+    @pytest.mark.asyncio
+    async def test_explicit_database_list_also_respects_exclude(self, test_settings: Settings, tmp_path):
+        attempted = []
+
+        async def fake_dump(container, db, destination, timeout):
+            attempted.append(db)
+            destination.write_bytes(b"x")
+
+        with patch.object(pde, "_discover_databases", AsyncMock()) as discover, \
+             patch.object(pde, "_dump_database", fake_dump), \
+             patch.object(pde, "_verify_archive", AsyncMock(return_value=1)):
+            await pde.execute(
+                _config(tmp_path, databases=["one", "two"], exclude=["two"]), test_settings
+            )
+
+        discover.assert_not_called()  # explicit list means no discovery call at all
+        assert attempted == ["one"]
+
+
+@pytest.mark.executor
+@pytest.mark.unit
+class TestDumpBehaviour:
+
+    @pytest.mark.asyncio
+    async def test_writes_verified_archive_per_database_mode_0600(self, test_settings: Settings, tmp_path):
+        with patch.object(pde, "_discover_databases", AsyncMock(return_value=["alpha", "beta"])), \
+             patch.object(pde, "_dump_database", _writing_dump), \
+             patch.object(pde, "_verify_archive", AsyncMock(return_value=5)):
+            result = await pde.execute(_config(tmp_path), test_settings)
+
+        files = sorted(tmp_path.glob("*.dump"))
+        assert [f.name.split("-")[0] for f in files] == ["alpha", "beta"]
+        assert "5 entries" in result
+        for f in files:
+            assert oct(f.stat().st_mode)[-3:] == "600"
+        # no .partial left behind
+        assert list(tmp_path.glob("*.partial")) == []
+
+    @pytest.mark.asyncio
+    async def test_one_failing_database_fails_run_but_keeps_others(self, test_settings: Settings, tmp_path):
+        async def fake_dump(container, db, destination, timeout):
+            if db == "bad":
+                raise Exception("pg_dump exploded")
+            destination.write_bytes(b"x")
+
+        with patch.object(pde, "_discover_databases", AsyncMock(return_value=["good", "bad"])), \
+             patch.object(pde, "_dump_database", fake_dump), \
+             patch.object(pde, "_verify_archive", AsyncMock(return_value=3)):
+            with pytest.raises(Exception, match="FAILED bad: pg_dump exploded"):
+                await pde.execute(_config(tmp_path), test_settings)
+
+        files = list(tmp_path.glob("*.dump"))
+        assert len(files) == 1
+        assert files[0].name.startswith("good-")
+        assert list(tmp_path.glob("*.partial")) == []
+
+    @pytest.mark.asyncio
+    async def test_empty_toc_deletes_partial_and_counts_as_failure(self, test_settings: Settings, tmp_path):
+        async def fake_verify(path):
+            raise Exception("archive has zero TOC entries")
+
+        with patch.object(pde, "_discover_databases", AsyncMock(return_value=["alpha"])), \
+             patch.object(pde, "_dump_database", _writing_dump), \
+             patch.object(pde, "_verify_archive", fake_verify):
+            with pytest.raises(Exception, match="FAILED alpha: archive has zero TOC entries"):
+                await pde.execute(_config(tmp_path), test_settings)
+
+        assert list(tmp_path.glob("*.dump")) == []
+        assert list(tmp_path.glob("*.partial")) == []
+
+
+@pytest.mark.executor
+@pytest.mark.unit
+class TestGlobals:
+
+    @pytest.mark.asyncio
+    async def test_globals_dump_written_and_mode_0600(self, test_settings: Settings, tmp_path):
+        async def fake_globals(container, destination, timeout):
+            destination.write_bytes(b"CREATE ROLE postgres;")
+
+        with patch.object(pde, "_discover_databases", AsyncMock(return_value=["alpha"])), \
+             patch.object(pde, "_dump_database", _writing_dump), \
+             patch.object(pde, "_verify_archive", AsyncMock(return_value=2)), \
+             patch.object(pde, "_dump_globals", fake_globals):
+            result = await pde.execute(_config(tmp_path, globals=True), test_settings)
+
+        files = list(tmp_path.glob("globals-*.sql"))
+        assert len(files) == 1
+        assert oct(files[0].stat().st_mode)[-3:] == "600"
+        assert "globals" in result
+
+    @pytest.mark.asyncio
+    async def test_empty_globals_dump_is_a_failure_but_keeps_db_dumps(self, test_settings: Settings, tmp_path):
+        async def fake_globals(container, destination, timeout):
+            destination.write_bytes(b"")  # empty
+
+        with patch.object(pde, "_discover_databases", AsyncMock(return_value=["alpha"])), \
+             patch.object(pde, "_dump_database", _writing_dump), \
+             patch.object(pde, "_verify_archive", AsyncMock(return_value=2)), \
+             patch.object(pde, "_dump_globals", fake_globals):
+            with pytest.raises(Exception, match="FAILED globals: globals dump is empty"):
+                await pde.execute(_config(tmp_path, globals=True), test_settings)
+
+        assert len(list(tmp_path.glob("*.dump"))) == 1   # alpha kept
+        assert list(tmp_path.glob("globals-*.sql")) == []
+        assert list(tmp_path.glob("*.partial")) == []
+
+
+@pytest.mark.executor
+@pytest.mark.unit
+class TestVerifyArchive:
+
+    def test_toc_entry_count_counts_non_comment_lines(self):
+        output = (
+            ";\n"
+            "; Archive created at 2026-10-19 09:45:07 UTC\n"
+            ";     TOC Entries: 2\n"
+            ";\n"
+            "216; 1259 44268 TABLE public scheduled_tasks test_scheduler_user\n"
+            "215; 1259 44267 SEQUENCE public scheduled_tasks_id_seq test_scheduler_user\n"
+        )
+        assert pde._toc_entry_count(output) == 2
+
+    def test_toc_entry_count_zero_for_header_only(self):
+        output = (
+            ";\n"
+            "; Archive created at 2026-10-19 09:45:07 UTC\n"
+            ";     TOC Entries: 0\n"
+            ";\n"
+        )
+        assert pde._toc_entry_count(output) == 0
+
+    @pytest.mark.asyncio
+    async def test_verify_archive_raises_on_nonzero_exit(self, tmp_path):
+        proc = MagicMock()
+        proc.communicate = AsyncMock(return_value=(b"", b"not an archive"))
+        proc.returncode = 1
+        with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=proc)):
+            with pytest.raises(Exception, match="pg_restore --list failed"):
+                await pde._verify_archive(tmp_path / "broken.dump")
+
+    @pytest.mark.asyncio
+    async def test_verify_archive_raises_on_zero_entries(self, tmp_path):
+        proc = MagicMock()
+        proc.communicate = AsyncMock(return_value=(b";\n;     TOC Entries: 0\n;\n", b""))
+        proc.returncode = 0
+        with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=proc)):
+            with pytest.raises(Exception, match="zero TOC entries"):
+                await pde._verify_archive(tmp_path / "empty.dump")
+
+
+@pytest.mark.executor
+@pytest.mark.unit
+class TestTimeout:
+
+    @pytest.mark.asyncio
+    async def test_dump_times_out_per_database(self, tmp_path):
+        """The actual asyncio.wait_for/kill mechanism, not a stand-in for it."""
+
+        class HangingStdout:
+            async def read(self, n):
+                await asyncio.sleep(10)
+                return b""
+
+        class EmptyStderr:
+            async def read(self):
+                return b""
+
+        fake_proc = MagicMock()
+        fake_proc.stdout = HangingStdout()
+        fake_proc.stderr = EmptyStderr()
+        fake_proc.kill = MagicMock()
+        fake_proc.wait = AsyncMock(return_value=0)
+
+        with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=fake_proc)):
+            with pytest.raises(Exception, match="timed out after"):
+                await pde._dump_database(
+                    "postgres-shared", "alpha", tmp_path / "alpha.partial", timeout=0.05
+                )
+
+        fake_proc.kill.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_timeout_failure_handled_like_any_other_db_failure(self, test_settings: Settings, tmp_path):
+        async def fake_dump(container, db, destination, timeout):
+            if db == "slow":
+                raise Exception(f"timed out after {timeout}s: pg_dump")
+            destination.write_bytes(b"x")
+
+        with patch.object(pde, "_discover_databases", AsyncMock(return_value=["fast", "slow"])), \
+             patch.object(pde, "_dump_database", fake_dump), \
+             patch.object(pde, "_verify_archive", AsyncMock(return_value=1)):
+            with pytest.raises(Exception, match="FAILED slow: timed out after"):
+                await pde.execute(_config(tmp_path, timeout_per_db=1), test_settings)
+
+        files = list(tmp_path.glob("*.dump"))
+        assert len(files) == 1 and files[0].name.startswith("fast-")
+
+
+@pytest.mark.executor
+@pytest.mark.unit
+class TestRetention:
+
+    def _age(self, path, days):
+        import os
+        old = (datetime.now(timezone.utc) - timedelta(days=days)).timestamp()
+        os.utime(path, (old, old))
+
+    def test_prunes_only_past_the_window(self, tmp_path):
+        fresh = tmp_path / "alpha-20261019T094500Z.dump"
+        stale = tmp_path / "alpha-20260101T120000Z.dump"
+        for f in (fresh, stale):
+            f.write_bytes(b"x")
+        self._age(stale, 45)
+
+        assert pde._prune(tmp_path, 14) == 1
+        assert fresh.exists() and not stale.exists()
+
+    def test_leaves_unrelated_files_alone(self, tmp_path):
+        """Retention must not touch anything it did not write."""
+        other = tmp_path / "important-database-dump.tar.gz"
+        named_alike = tmp_path / "alpha-backup-manual.dump"
+        for f in (other, named_alike):
+            f.write_bytes(b"x")
+            self._age(f, 400)
+
+        assert pde._prune(tmp_path, 14) == 0
+        assert other.exists() and named_alike.exists()
+
+    def test_prune_leaves_partial_files_alone(self, tmp_path):
+        """A .partial is never this executor's finished output; retention
+        must never be the thing that deletes one out from under a dump that
+        is still writing it."""
+        partial = tmp_path / "alpha-20260101T120000Z.dump.partial"
+        partial.write_bytes(b"x")
+        self._age(partial, 400)
+
+        assert pde._prune(tmp_path, 14) == 0
+        assert partial.exists()
+
+    def test_prunes_globals_files_too(self, tmp_path):
+        stale = tmp_path / "globals-20260101T120000Z.sql"
+        stale.write_bytes(b"x")
+        self._age(stale, 400)
+
+        assert pde._prune(tmp_path, 14) == 1
+        assert not stale.exists()