diff --git a/common/optional/zfs_manager.nix b/common/optional/zfs_manager.nix
new file mode 100644
index 0000000..7250dda
--- /dev/null
+++ b/common/optional/zfs_manager.nix
@@ -0,0 +1,186 @@
+{
+  pkgs,
+  lib,
+  config,
+  ...
+}:
+let
+  cfg = config.services.zfs_manager;
+
+  snapshotOptions = {
+    options = {
+      "15_min" = lib.mkOption {
+        type = lib.types.ints.unsigned;
+        default = 0;
+        description = "How many 15 minute snapshots to keep.";
+      };
+      hourly = lib.mkOption {
+        type = lib.types.ints.unsigned;
+        default = 0;
+        description = "How many hourly snapshots to keep.";
+      };
+      daily = lib.mkOption {
+        type = lib.types.ints.unsigned;
+        default = 0;
+        description = "How many daily snapshots to keep.";
+      };
+      monthly = lib.mkOption {
+        type = lib.types.ints.unsigned;
+        default = 0;
+        description = "How many monthly snapshots to keep.";
+      };
+    };
+  };
+
+  datasetOptions = {
+    options = {
+      manageProperties = lib.mkOption {
+        type = lib.types.bool;
+        default = true;
+        description = ''
+          Whether zfs_manager owns this dataset's properties. When false the
+          dataset only contributes its snapshot retention, which is how
+          root_pool datasets are declared.
+        '';
+      };
+      createIfMissing = lib.mkOption {
+        type = lib.types.bool;
+        default = true;
+        description = ''
+          Whether zfs_manager may create this dataset when it is absent.
+
+          Set it false for a dataset that has to be provisioned by hand, such
+          as an encryption root: encryption is fixed at creation time and
+          cannot be expressed here, so creating it automatically would silently
+          produce an unencrypted dataset where an encrypted one was intended.
+          The dataset is still property checked, and its absence is reported as
+          a failure rather than quietly fixed.
+        '';
+      };
+      properties = lib.mkOption {
+        type = lib.types.attrsOf lib.types.str;
+        default = { };
+        description = ''
+          The zfs properties this dataset should have. Values are compared
+          against the live dataset and corrected when they differ.
+        '';
+      };
+      snapshots = lib.mkOption {
+        type = lib.types.submodule snapshotOptions;
+        default = cfg.defaultSnapshots;
+        description = ''
+          Snapshot retention for this dataset. Defaults to defaultSnapshots.
+        '';
+      };
+    };
+  };
+
+  # snapshot_manager.py only ever walks datasets below a pool root, so pool
+  # roots are left out of the retention table. It also indexes the table
+  # directly, which is why every entry carries all four keys.
+  snapshotTable = lib.mapAttrs (_: dataset: dataset.snapshots) (
+    lib.filterAttrs (name: _: lib.hasInfix "/" name) cfg.datasets
+  );
+
+  snapshotConfig = (pkgs.formats.toml { }).generate "snapshot_config.toml" (
+    snapshotTable // { default = cfg.defaultSnapshots; }
+  );
+
+  # Every declared dataset is emitted, including the ones whose properties are
+  # not managed, so the tool can tell "deliberately hands off" apart from
+  # "nobody has written this down yet".
+  datasetConfig = (pkgs.formats.json { }).generate "zfs_datasets.json" {
+    datasets = lib.mapAttrs (_: dataset: {
+      inherit (dataset) manageProperties createIfMissing properties;
+    }) cfg.datasets;
+  };
+in
+{
+  options = {
+    services.zfs_manager = {
+      enable = lib.mkEnableOption "declarative ZFS dataset management";
+      datasets = lib.mkOption {
+        type = lib.types.attrsOf (lib.types.submodule datasetOptions);
+        default = { };
+        example = lib.literalExpression ''
+          {
+            "media/temp".properties = {
+              sync = "disabled";
+              redundant_metadata = "none";
+            };
+          }
+        '';
+        description = ''
+          The datasets to manage, keyed by full dataset name. Missing datasets
+          are created and drifted properties are corrected. Nothing is ever
+          destroyed, and datasets that are not declared are left alone.
+
+          A name without a "/" is a pool root filesystem. Its properties are
+          managed but it is never created, pool creation stays manual.
+        '';
+      };
+      defaultSnapshots = lib.mkOption {
+        type = lib.types.submodule snapshotOptions;
+        default = { };
+        description = ''
+          Retention for undeclared datasets and for declared datasets that do
+          not override their snapshots. Emitted as the "default" table of the
+          snapshot config.
+        '';
+      };
+      dryRun = lib.mkOption {
+        type = lib.types.bool;
+        default = false;
+        description = ''
+          Log every change that would be made without touching zfs. Use this to
+          validate a new or heavily edited declaration before applying it.
+        '';
+      };
+      PYTHONPATH = lib.mkOption {
+        type = lib.types.str;
+        description = ''
+          the PYTHONPATH to use for the zfs_manager service.
+        '';
+      };
+      EnvironmentFile = lib.mkOption {
+        type = lib.types.nullOr (lib.types.coercedTo lib.types.path toString lib.types.str);
+        default = null;
+
+        description = ''
+          Single environment file for the service (e.g. /etc/zfs-manager/env).
+          Use a leading "-" to ignore if missing (systemd feature).
+        '';
+      };
+    };
+  };
+
+  config = lib.mkIf cfg.enable {
+    services.snapshot_manager.path = snapshotConfig;
+
+    systemd = {
+      services.zfs_manager = {
+        description = "ZFS Dataset Manager";
+        requires = [ "zfs-import.target" ];
+        after = [
+          "zfs-import.target"
+          "zfs-mount.service"
+        ];
+        wantedBy = [ "multi-user.target" ];
+        path = [ pkgs.zfs ];
+        # Re-run on nixos-rebuild switch whenever the declaration changes.
+        restartTriggers = [ datasetConfig ];
+        environment = {
+          PYTHONPATH = cfg.PYTHONPATH;
+        };
+        serviceConfig = {
+          Type = "oneshot";
+          RemainAfterExit = true;
+          ExecStart = "${pkgs.my_python}/bin/python -m python.tools.zfs_manager ${lib.escapeShellArg datasetConfig}${lib.optionalString cfg.dryRun " --dry-run"}";
+        }
+        // lib.optionalAttrs (cfg.EnvironmentFile != null) {
+          EnvironmentFile = cfg.EnvironmentFile;
+        };
+      };
+    };
+  };
+}
diff --git a/python/tools/snapshot_manager.py b/python/tools/snapshot_manager.py
index 866c6a5..d59ec65 100644
--- a/python/tools/snapshot_manager.py
+++ b/python/tools/snapshot_manager.py
@@ -23,6 +23,7 @@ def main(config_file: Path) -> None:
     """Main."""
     configure_logger(level="DEBUG")
     logger.info("Starting snapshot_manager")
+    failures: list[str] = []
 
     try:
         time_stamp = get_time_stamp()
@@ -34,16 +35,23 @@ def main(config_file: Path) -> None:
                 msg = f"{dataset.name} failed to create snapshot {time_stamp}"
                 logger.error(msg)
                 signal_alert(msg)
+                failures.append(msg)
                 continue
             count_lookup = get_count_lookup(config_file, dataset.name)
             logger.info(f"using {count_lookup} for {dataset.name}")
-            get_snapshots_to_delete(dataset, count_lookup)
+            failures.extend(get_snapshots_to_delete(dataset, count_lookup))
     except Exception:
         logger.exception("snapshot_manager failed")
         signal_alert("snapshot_manager failed")
         sys.exit(1)
-    else:
-        logger.info("snapshot_manager completed")
+
+    if failures:
+        logger.error(f"snapshot_manager completed with {len(failures)} errors")
+        for failure in failures:
+            logger.error(f"  {failure}")
+        sys.exit(1)
+
+    logger.info("snapshot_manager completed")
 
 
 def get_count_lookup(config_file: Path, dataset_name: str) -> dict[str, int]:
@@ -92,19 +100,29 @@ def load_config_data(config_file: Path) -> dict[str, dict[str, int]]:
 def get_snapshots_to_delete(
     dataset: Dataset,
     count_lookup: dict[str, int],
-) -> None:
+) -> list[str]:
     """Get snapshots to delete.
 
     Args:
         dataset (Dataset): the dataset
         count_lookup (dict[str, int]): the count lookup
+
+    Returns:
+        list[str]: Snapshot deletion failures encountered while pruning.
     """
+    for retention_class in ("15_min", "hourly", "daily", "monthly"):
+        count = count_lookup.get(retention_class)
+        if not isinstance(count, int) or isinstance(count, bool) or count < 0:
+            error = f"{retention_class} retention must be a non-negative integer, got {count!r}"
+            raise ValueError(error)
+
+    failures: list[str] = []
     snapshots = dataset.get_snapshots()
 
     logger.info(f"calculating snapshots for {dataset.name} to be deleted")
     if not snapshots:
         logger.info(f"{dataset.name} has no snapshots")
-        return
+        return failures
 
     filters = (
         ("15_min", re_compile(r"auto_\d{10}(?:15|30|45)")),
@@ -129,6 +147,9 @@ def get_snapshots_to_delete(
                 error_message = f"{dataset.name}@{snapshot} failed to delete: {error}"
                 signal_alert(error_message)
                 logger.error(error_message)
+                failures.append(error_message)
+
+    return failures
 
 
 def get_time_stamp() -> str:
diff --git a/python/tools/zfs_manager.py b/python/tools/zfs_manager.py
new file mode 100644
index 0000000..0a78f02
--- /dev/null
+++ b/python/tools/zfs_manager.py
@@ -0,0 +1,340 @@
+"""zfs_manager.
+
+Reconciles the live zfs datasets against a declaration generated by
+common/optional/zfs_manager.nix. Datasets are created and properties are
+corrected, but nothing is ever destroyed or renamed.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import sys
+from pathlib import Path  # noqa: TC003 This is required for the typer CLI
+
+import typer
+
+from python.common import configure_logger
+from python.signal_alert import signal_alert
+from python.zfs import create_dataset, get_properties, list_dataset_names, set_property
+
+logger = logging.getLogger(__name__)
+
+# Properties that can only be chosen at creation time. Attempting to zfs set
+# these fails on every run, so a mismatch is reported instead of retried.
+CREATE_ONLY_PROPERTIES = frozenset(
+    {
+        "casesensitivity",
+        "encryption",
+        "keyformat",
+        "normalization",
+        "utf8only",
+        "volblocksize",
+    },
+)
+
+# Properties whose values zfs reports in bytes but which are conventionally
+# declared with a size suffix, so "16k" and "16384" mean the same thing.
+SIZE_PROPERTIES = frozenset(
+    {
+        "quota",
+        "recordsize",
+        "refquota",
+        "refreservation",
+        "reservation",
+        "special_small_blocks",
+        "volblocksize",
+        "volsize",
+    },
+)
+
+SIZE_SUFFIXES = {"b": 1, "k": 1024, "m": 1024**2, "g": 1024**3, "t": 1024**4, "p": 1024**5}
+
+# Sources that mean the value was deliberately put on this dataset rather than
+# inherited from a parent or left at the zfs default.
+LOCAL_SOURCES = ("local", "received")
+
+
+class ReconciliationError(RuntimeError):
+    """One or more datasets could not be brought in line with the declaration."""
+
+    def __init__(self, failures: list[str]) -> None:
+        """Record the individual failures behind this run's exit code."""
+        self.failures = failures
+        super().__init__(f"ZFS reconciliation failed with {len(failures)} errors")
+
+
+def main(config_file: Path, *, dry_run: bool = False) -> None:
+    """Main.
+
+    Args:
+        config_file (Path): The path to the generated dataset declaration.
+        dry_run (bool): Log the changes that would be made without making them.
+    """
+    configure_logger(level="DEBUG")
+    logger.info(f"Starting zfs_manager {dry_run=}")
+
+    try:
+        reconcile(config_file, dry_run=dry_run)
+    except ReconciliationError as error:
+        summary = error
+    except Exception:
+        logger.exception("zfs_manager failed")
+        signal_alert("zfs_manager failed")
+        sys.exit(1)
+    else:
+        logger.info("zfs_manager completed")
+        return
+
+    # Each failure was logged and alerted as it happened. Repeating them
+    # together puts the whole picture at the end of the journal, which is what
+    # systemctl status shows. No traceback: this is an expected outcome, not a
+    # crash, and a stack trace would only bury the list.
+    logger.error(str(summary))
+    for failure in summary.failures:
+        logger.error(f"  {failure}")
+    sys.exit(1)
+
+
+def reconcile(config_file: Path, *, dry_run: bool) -> None:
+    """Bring every declared dataset in line, collecting problems as it goes.
+
+    One bad dataset must not hide the state of the others, so everything is
+    checked before anything is raised.
+
+    Args:
+        config_file (Path): The path to the generated dataset declaration.
+        dry_run (bool): Log the changes without making them.
+
+    Raises:
+        ReconciliationError: If anything could not be reconciled.
+    """
+    declared = json.loads(config_file.read_text())["datasets"]
+    existing = set(list_dataset_names())
+    unusable: set[str] = set()
+    failures: list[str] = []
+
+    # Parents before children so a newly created parent exists by the time its
+    # children are reconciled.
+    for name in sorted(declared, key=lambda name: (name.count("/"), name)):
+        entry = declared[name]
+
+        # Declared purely to record retention, its properties belong to
+        # whoever set them.
+        if not entry.get("manageProperties", True):
+            logger.debug(f"{name} is declared but its properties are not managed")
+            continue
+
+        if has_unusable_parent(name, unusable):
+            failures.append(fail(f"cannot reconcile {name}, its parent is missing"))
+            continue
+
+        if name in existing:
+            failures.extend(reconcile_dataset(name, entry["properties"], dry_run=dry_run))
+            continue
+
+        created, failure = handle_missing_dataset(name, entry, dry_run=dry_run)
+        if failure is not None:
+            failures.append(failure)
+        if created:
+            existing.add(name)
+        elif not dry_run:
+            unusable.add(name)
+
+    report_undeclared_datasets(existing, declared)
+
+    if failures:
+        raise ReconciliationError(failures)
+
+
+def fail(message: str) -> str:
+    """Log and alert a problem, and hand it back for the failure tally.
+
+    Args:
+        message (str): What went wrong.
+
+    Returns:
+        str: The same message, so the caller can collect it.
+    """
+    logger.error(message)
+    signal_alert(message)
+    return message
+
+
+def has_unusable_parent(name: str, unusable: set[str]) -> bool:
+    """Check whether an ancestor of a dataset is missing.
+
+    Args:
+        name (str): The name of the dataset.
+        unusable (set[str]): The datasets that do not exist and were not created.
+
+    Returns:
+        bool: True if any ancestor is unusable.
+    """
+    parts = name.split("/")
+    return any("/".join(parts[:depth]) in unusable for depth in range(1, len(parts)))
+
+
+def handle_missing_dataset(name: str, entry: dict, *, dry_run: bool) -> tuple[bool, str | None]:
+    """Deal with a declared dataset that is not on the system.
+
+    Pool roots are never created, and neither is anything the declaration marks
+    as provisioned outside of nix, such as an encryption root whose key
+    settings cannot be reproduced from the declaration.
+
+    Args:
+        name (str): The name of the dataset.
+        entry (dict): The declaration for this dataset.
+        dry_run (bool): Log the change without making it.
+
+    Returns:
+        tuple[bool, str | None]: Whether the dataset now exists, and a failure
+            message if there was one.
+    """
+    properties = entry["properties"]
+
+    if "/" not in name:
+        return False, fail(f"pool {name} is declared but does not exist, zfs_manager does not create pools")
+
+    if not entry.get("createIfMissing", True):
+        return False, fail(
+            f"{name} is declared but does not exist, and is marked as created outside of nix. "
+            "It has to be made by hand, see systems/jeeves/scripts/zfs.sh.",
+        )
+
+    if dry_run:
+        logger.info(f"would create {name} with {properties}")
+        return False, None
+
+    logger.info(f"creating {name} with {properties}")
+    if error := create_dataset(name, properties):
+        return False, fail(error)
+
+    return True, None
+
+
+def reconcile_dataset(name: str, properties: dict[str, str], *, dry_run: bool) -> list[str]:
+    """Bring an existing dataset in line with its declared properties.
+
+    Args:
+        name (str): The name of the dataset.
+        properties (dict[str, str]): The declared properties.
+        dry_run (bool): Log the changes without making them.
+
+    Returns:
+        list[str]: Anything that could not be put right.
+    """
+    failures: list[str] = []
+    current = get_properties(name)
+
+    for key, wanted in sorted(properties.items()):
+        current_value, _ = current.get(key, ("-", "-"))
+        if values_match(key, wanted, current_value):
+            continue
+
+        if key in CREATE_ONLY_PROPERTIES:
+            # Nothing can put this right while the dataset exists, so it is a
+            # hard failure rather than a warning that repeats unnoticed.
+            failures.append(
+                fail(
+                    f"{name} {key} is {current_value} but {wanted} is declared, "
+                    f"{key} can only be set when the dataset is created",
+                ),
+            )
+            continue
+
+        if dry_run:
+            logger.info(f"would set {key}={wanted} on {name}, currently {current_value}")
+            continue
+
+        logger.info(f"setting {key}={wanted} on {name}, was {current_value}")
+        if error := set_property(name, key, wanted):
+            failures.append(fail(error))
+
+    report_undeclared_properties(name, properties, current)
+    return failures
+
+
+def report_undeclared_properties(name: str, properties: dict[str, str], current: dict[str, tuple[str, str]]) -> None:
+    """Warn about properties set on the dataset but absent from the declaration.
+
+    Inherited and default values are silent, they are not drift. A locally set
+    value that nix does not know about was changed outside of this tool and
+    will be lost the next time the dataset is recreated, so it is worth saying.
+
+    Args:
+        name (str): The name of the dataset.
+        properties (dict[str, str]): The declared properties.
+        current (dict[str, tuple[str, str]]): The live properties keyed to (value, source).
+    """
+    for key, (value, source) in sorted(current.items()):
+        # User properties such as nixos:shutdown-time are written by other
+        # tools and are not something a dataset declaration should own.
+        if key in properties or ":" in key or source not in LOCAL_SOURCES:
+            continue
+
+        logger.warning(f"{name} has {key}={value} set outside of nix")
+        signal_alert(f"{name} has {key}={value} set outside of nix")
+
+
+def report_undeclared_datasets(existing: set[str], declared: dict[str, dict]) -> None:
+    """Warn about datasets that exist but are not declared.
+
+    These are left completely alone. They still get snapshots through the
+    default retention table.
+
+    Args:
+        existing (set[str]): The names of every live dataset.
+        declared (dict[str, dict]): The declaration.
+    """
+    for name in sorted(existing - set(declared)):
+        logger.warning(f"{name} exists but is not declared in nix")
+
+
+def values_match(key: str, wanted: str, current: str) -> bool:
+    """Compare a declared property value against the live one.
+
+    Args:
+        key (str): The property name.
+        wanted (str): The declared value.
+        current (str): The live value.
+
+    Returns:
+        bool: True if the two values mean the same thing.
+    """
+    if key in SIZE_PROPERTIES:
+        wanted_size = parse_size(wanted)
+        current_size = parse_size(current)
+        if wanted_size is not None and current_size is not None:
+            return wanted_size == current_size
+
+    return wanted == current
+
+
+def parse_size(value: str) -> int | None:
+    """Convert a zfs size such as 16k or 1M into bytes.
+
+    Args:
+        value (str): The size to convert.
+
+    Returns:
+        int | None: The size in bytes, or None if it is not a size.
+    """
+    value = value.strip()
+    if value.isdigit():
+        return int(value)
+
+    number, suffix = value[:-1], value[-1:].lower()
+    if suffix in SIZE_SUFFIXES and number.isdigit():
+        return int(number) * SIZE_SUFFIXES[suffix]
+
+    return None
+
+
+def cli() -> None:
+    """CLI."""
+    typer.run(main)
+
+
+if __name__ == "__main__":
+    cli()
diff --git a/systems/jeeves/datasets.nix b/systems/jeeves/datasets.nix
new file mode 100644
index 0000000..cd7eee6
--- /dev/null
+++ b/systems/jeeves/datasets.nix
@@ -0,0 +1,362 @@
+# Dataset declarations for jeeves, kept as plain data rather than inside
+# zfs.nix so the dataset tree stays separate from the NixOS service wiring.
+# Datasets are nested the way zfs nests them: a pool holds datasets, which can
+# hold datasets of their own. The tree is flattened into "pool/parent/child"
+# names below, which is what zfs and services.zfs_manager work in.
+#
+# Consumed by ./zfs.nix, which feeds it to services.zfs_manager.
+let
+  # Every pool on jeeves was created with the same -O options.
+  poolDefaults = mountpoint: {
+    inherit mountpoint;
+    acltype = "posix"; # zfs reports posixacl back as posix
+    atime = "off";
+    compression = "zstd";
+    dnodesize = "auto";
+    xattr = "sa";
+  };
+
+  zfsKey = "file:///root/zfs.key";
+
+  # What a dataset gets when it is not called out below, kept identical to the
+  # "default" table so the datasets that used to fall through are unchanged.
+  standard = {
+    "15_min" = 8;
+    hourly = 24;
+  };
+
+  disabledSnapshots = {
+    "15_min" = 0;
+    hourly = 0;
+    daily = 0;
+    monthly = 0;
+  };
+
+  pools = {
+    # root_pool: retention only, its properties are not managed yet.
+    root_pool = {
+      manageProperties = false;
+      datasets = {
+        home = {
+          manageProperties = false;
+          snapshots = {
+            "15_min" = 8;
+            hourly = 24;
+            daily = 14;
+          };
+        };
+        root = {
+          manageProperties = false;
+          snapshots = standard;
+        };
+        nix = {
+          manageProperties = false;
+          snapshots."15_min" = 4;
+        };
+        var = {
+          manageProperties = false;
+          snapshots = {
+            "15_min" = 8;
+            hourly = 24;
+            daily = 30;
+            monthly = 6;
+          };
+        };
+      };
+    };
+
+    media = {
+      properties = poolDefaults "/zfs/media";
+      datasets = {
+        temp = {
+          properties = {
+            redundant_metadata = "none";
+            sync = "disabled";
+          };
+          snapshots."15_min" = 2;
+        };
+        secure = {
+          # An encryption root provisioned by scripts/zfs.sh. Its create-only
+          # properties are declared for verification, but zfs_manager must
+          # never create it automatically.
+          createIfMissing = true;
+          properties = {
+            encryption = "aes-256-gcm";
+            keyformat = "hex";
+            keylocation = zfsKey;
+          };
+          snapshots = disabledSnapshots;
+          datasets = {
+            docker = {
+              properties = {
+                mountpoint = "/zfs/media/docker";
+                compression = "zstd-9";
+              };
+              snapshots = {
+                "15_min" = 3;
+                hourly = 12;
+                daily = 14;
+                monthly = 2;
+              };
+            };
+            "github-runners" = {
+              properties = {
+                mountpoint = "/zfs/media/github-runners";
+                compression = "zstd-9";
+                sync = "disabled";
+              };
+              snapshots = {
+                "15_min" = 6;
+                hourly = 2;
+                daily = 1;
+              };
+            };
+            home_assistant = {
+              properties = {
+                mountpoint = "/zfs/media/home_assistant";
+                compression = "zstd-19";
+              };
+              snapshots = standard;
+            };
+            important = {
+              properties = {
+                compression = "zstd-9";
+                copies = "2";
+              };
+              snapshots = standard;
+            };
+            notes = {
+              properties = {
+                mountpoint = "/zfs/media/notes";
+                copies = "2";
+              };
+              snapshots = {
+                "15_min" = 8;
+                hourly = 24;
+                daily = 30;
+                monthly = 12;
+              };
+            };
+            postgres = {
+              properties = {
+                mountpoint = "/zfs/media/database/postgres";
+                primarycache = "metadata";
+                recordsize = "16K";
+              };
+              snapshots = {
+                "15_min" = 8;
+                hourly = 24;
+                daily = 7;
+              };
+            };
+            "postgres-wal" = {
+              properties = {
+                compression = "lz4";
+                logbias = "latency";
+                mountpoint = "/zfs/media/database/postgres-wal";
+                primarycache = "metadata";
+                recordsize = "32K";
+                secondarycache = "none";
+                special_small_blocks = "32K";
+              };
+              snapshots = {
+                "15_min" = 4;
+                hourly = 2;
+              };
+            };
+            prometheus = {
+              properties = {
+                mountpoint = "/zfs/media/database/prometheus";
+                compression = "lz4";
+              };
+              snapshots = standard;
+            };
+            services = {
+              properties = {
+                mountpoint = "/zfs/media/services";
+                compression = "zstd-9";
+              };
+              snapshots = standard;
+            };
+            share = {
+              properties = {
+                mountpoint = "/zfs/media/share";
+                exec = "off";
+              };
+              snapshots."15_min" = 4;
+            };
+          };
+        };
+      };
+    };
+
+    storage = {
+      properties = poolDefaults "/zfs/storage";
+      datasets = {
+        nomad = {
+          properties = {
+            mountpoint = "/zfs/storage/nomad";
+            compression = "zstd-9";
+          };
+          snapshots = standard;
+        };
+        ollama = {
+          properties = {
+            compression = "zstd-19";
+            recordsize = "1M";
+            sync = "disabled";
+          };
+          snapshots."15_min" = 2;
+        };
+        secure = {
+          # An encryption root provisioned by scripts/zfs.sh. Its create-only
+          # properties are declared for verification, but zfs_manager must
+          # never create it automatically.
+          createIfMissing = false;
+          properties = {
+            encryption = "aes-256-gcm";
+            keyformat = "hex";
+            keylocation = zfsKey;
+          };
+          snapshots = disabledSnapshots;
+          datasets = {
+            archive = {
+              properties = {
+                compression = "zstd-19";
+                mountpoint = "/zfs/storage/archive";
+                recordsize = "1M";
+              };
+              snapshots = standard;
+            };
+            important = {
+              properties = {
+                compression = "zstd-19";
+                copies = "2";
+                mountpoint = "/zfs/storage/important";
+              };
+              snapshots = standard;
+            };
+            library = {
+              properties = {
+                compression = "zstd-19";
+                mountpoint = "/zfs/storage/library";
+                recordsize = "1M";
+              };
+              snapshots = standard;
+            };
+            main = {
+              properties = {
+                compression = "zstd-19";
+                mountpoint = "/zfs/storage/main";
+              };
+              snapshots = standard;
+            };
+            photos = {
+              properties = {
+                compression = "zstd-19";
+                copies = "2";
+                mountpoint = "/zfs/storage/photos";
+                recordsize = "16K";
+              };
+              snapshots = standard;
+            };
+            plex = {
+              properties = {
+                compression = "zstd-19";
+                mountpoint = "/zfs/storage/plex";
+                recordsize = "1M";
+              };
+              snapshots = {
+                "15_min" = 6;
+                hourly = 2;
+                daily = 1;
+              };
+            };
+            secrets = {
+              properties = {
+                compression = "zstd-19";
+                copies = "3";
+                mountpoint = "/zfs/storage/secrets";
+              };
+              snapshots = {
+                "15_min" = 8;
+                hourly = 24;
+                daily = 30;
+                monthly = 12;
+              };
+            };
+            syncthing = {
+              properties = {
+                compression = "zstd-19";
+                mountpoint = "/zfs/storage/syncthing";
+              };
+              snapshots = standard;
+            };
+            transmission = {
+              properties = {
+                compression = "zstd-9";
+                exec = "off";
+                mountpoint = "/zfs/storage/transmission";
+                recordsize = "1M";
+                sync = "disabled";
+              };
+              snapshots."15_min" = 4;
+            };
+          };
+        };
+      };
+    };
+
+    scratch = {
+      properties = poolDefaults "/zfs/scratch" // {
+        encryption = "aes-256-gcm";
+        keyformat = "hex";
+        keylocation = zfsKey;
+      };
+      datasets = {
+        kafka = {
+          properties = {
+            mountpoint = "/zfs/scratch/kafka";
+            recordsize = "1M";
+          };
+          snapshots = standard;
+        };
+        kestra = {
+          properties = {
+            mountpoint = "/zfs/scratch/kestra";
+            sync = "disabled";
+          };
+          snapshots = standard;
+        };
+        transmission = {
+          properties = {
+            mountpoint = "/zfs/scratch/transmission";
+            recordsize = "16K";
+            sync = "disabled";
+          };
+          snapshots."15_min" = 2;
+        };
+        uv_cache = {
+          properties.mountpoint = "/zfs/scratch/uv_cache";
+          snapshots."15_min" = 2;
+        };
+      };
+    };
+  };
+
+  # Collapse the tree into the flat "pool/parent/child" names zfs uses. Each
+  # node keeps everything except its children.
+  flatten =
+    name: node:
+    builtins.foldl' (result: child: result // flatten "${name}/${child}" node.datasets.${child}) {
+      ${name} = builtins.removeAttrs node [ "datasets" ];
+    } (builtins.attrNames (node.datasets or { }));
+
+  datasets = builtins.foldl' (result: pool: result // flatten pool pools.${pool}) { } (
+    builtins.attrNames pools
+  );
+in
+{
+  inherit datasets;
+  defaultSnapshots = standard;
+}
diff --git a/systems/jeeves/default.nix b/systems/jeeves/default.nix
index 1d47f8d..f0ab6b6 100644
--- a/systems/jeeves/default.nix
+++ b/systems/jeeves/default.nix
@@ -1,7 +1,4 @@
 { inputs, ... }:
-let
-  vars = import ./vars.nix;
-in
 {
   imports = [
     "${inputs.self}/users/dov"
@@ -15,6 +12,7 @@ in
     "${inputs.self}/common/optional/syncthing_base.nix"
     "${inputs.self}/common/optional/update.nix"
     "${inputs.self}/common/optional/zerotier.nix"
+    "${inputs.self}/common/optional/zfs_manager.nix"
     ./monitoring
     ./docker
     ./services
@@ -24,6 +22,7 @@ in
     ./programs.nix
     ./runners
     ./syncthing.nix
+    ./zfs.nix
   ];
 
   services = {
@@ -31,11 +30,6 @@ in
 
     smartd.enable = true;
 
-    snapshot_manager = {
-      path = ./snapshot_config.toml;
-      EnvironmentFile = "${vars.secrets}/services/snapshot_manager";
-    };
-
     zerotierone.joinNetworks = [ "a09acf02330d37b9" ];
   };
 
diff --git a/systems/jeeves/scripts/zfs.sh b/systems/jeeves/scripts/zfs.sh
index a0a09f2..802c851 100644
--- a/systems/jeeves/scripts/zfs.sh
+++ b/systems/jeeves/scripts/zfs.sh
@@ -1,5 +1,10 @@
 #!/bin/bash
 
+# Pool and vdev creation only. This is run by hand once per pool.
+#
+# Datasets and their properties are declared in systems/jeeves/zfs.nix and
+# reconciled by the zfs_manager service. Do not add zfs create lines here.
+
 # zpools
 
 # media
@@ -12,34 +17,10 @@ sudo zpool add storage -o ashift=12 special mirror
 sudo zpool add storage -o ashift=12 logs mirror
 
 # scratch
-sudo zpool create scratch -o ashift=12 -O acltype=posixacl -O atime=off -O dnodesize=auto -O xattr=sa -O compression=zstd -O encryption=aes-256-gcm -O keyformat=hex -O keylocation=file:///key -m /zfs/scratch
+sudo zpool create scratch -o ashift=12 -O acltype=posixacl -O atime=off -O dnodesize=auto -O xattr=sa -O compression=zstd -O encryption=aes-256-gcm -O keyformat=hex -O keylocation=file:///root/zfs.key -m /zfs/scratch
 
-# media datasets
-sudo zfs create media/temp -o sync=disabled -o redundant_metadata=none
+# The two encrypted parent datasets have to exist before zfs_manager can create
+# anything under them, since encryption cannot be set after creation.
+# These will be removed if/when the media and storage pools are encrypted in the future.
 sudo zfs create media/secure -o encryption=aes-256-gcm -o keyformat=hex -o keylocation=file:///root/zfs.key
-sudo zfs create media/secure/docker -o compression=zstd-9
-sudo zfs create media/secure/github-runners -o compression=zstd-9 -o sync=disabled
-sudo zfs create media/secure/notes -o copies=2
-sudo zfs create media/secure/postgres -o mountpoint=/zfs/media/database/postgres -o recordsize=16k -o primarycache=metadata
-sudo zfs create media/secure/postgres-wal -o mountpoint=/zfs/media/database/postgres-wal -o recordsize=32k -o primarycache=metadata -o special_small_blocks=32K -o compression=lz4 -o secondarycache=none -o logbias=latency
-sudo zfs create media/secure/prometheus -o mountpoint=/zfs/media/database/prometheus -o compression=lz4
-sudo zfs create media/secure/services -o compression=zstd-9
-sudo zfs create media/secure/share -o mountpoint=/zfs/media/share -o exec=off
-
-# scratch datasets
-sudo zfs create scratch/kafka -o mountpoint=/zfs/scratch/kafka -o recordsize=1M
-sudo zfs create scratch/transmission -o mountpoint=/zfs/scratch/transmission -o recordsize=16k -o sync=disabled -o redundant_metadata=none
-sudo zfs create scratch/uv_cache -o mountpoint=/zfs/scratch/uv_cache
-
-# storage datasets
-sudo zfs create storage/ollama -o recordsize=1M -o compression=zstd-19 -o sync=disabled
 sudo zfs create storage/secure -o encryption=aes-256-gcm -o keyformat=hex -o keylocation=file:///root/zfs.key
-sudo zfs create storage/secure/archive -o recordsize=1M -o compression=zstd-19
-sudo zfs create storage/secure/library -o recordsize=1M -o compression=zstd-19
-sudo zfs create storage/secure/main -o compression=zstd-19
-sudo zfs create storage/secure/photos -o recordsize=16K -o compression=zstd-19 -o copies=2
-sudo zfs create storage/secure/plex -o recordsize=1M -o compression=zstd-19
-sudo zfs create storage/secure/secrets -o compression=zstd-19 -o copies=3
-sudo zfs create storage/secure/syncthing -o compression=zstd-19
-sudo zfs create storage/secure/transmission -o recordsize=1M -o compression=zstd-9 -o exec=off -o sync=disabled
-sudo zfs create storage/secure/important -o compression=zstd-19 -o copies=2 -o mountpoint=/zfs/storage/important
diff --git a/systems/jeeves/snapshot_config.toml b/systems/jeeves/snapshot_config.toml
deleted file mode 100644
index cb9a3bf..0000000
--- a/systems/jeeves/snapshot_config.toml
+++ /dev/null
@@ -1,129 +0,0 @@
-["default"]
-15_min = 8
-hourly = 24
-daily = 0
-monthly = 0
-
-# root_pool
-["root_pool/home"]
-15_min = 8
-hourly = 24
-daily = 14
-monthly = 0
-
-["root_pool/root"]
-15_min = 8
-hourly = 24
-daily = 0
-monthly = 0
-
-["root_pool/nix"]
-15_min = 4
-hourly = 0
-daily = 0
-monthly = 0
-
-["root_pool/var"]
-15_min = 8
-hourly = 24
-daily = 30
-monthly = 6
-# storage
-["storage/ollama"]
-15_min = 2
-hourly = 0
-daily = 0
-monthly = 0
-
-["storage/secure"]
-15_min = 0
-hourly = 0
-daily = 0
-monthly = 0
-
-["storage/secure/plex"]
-15_min = 6
-hourly = 2
-daily = 1
-monthly = 0
-
-["storage/secure/transmission"]
-15_min = 4
-hourly = 0
-daily = 0
-monthly = 0
-
-["storage/secure/secrets"]
-15_min = 8
-hourly = 24
-daily = 30
-monthly = 12
-
-# media
-["media/temp"]
-15_min = 2
-hourly = 0
-daily = 0
-monthly = 0
-
-["media/secure"]
-15_min = 0
-hourly = 0
-daily = 0
-monthly = 0
-
-["media/secure/plex"]
-15_min = 6
-hourly = 2
-daily = 1
-monthly = 0
-
-["media/secure/postgres-wal"]
-15_min = 4
-hourly = 2
-daily = 0
-monthly = 0
-
-
-["media/secure/postgres"]
-15_min = 8
-hourly = 24
-daily = 7
-monthly = 0
-
-["media/secure/share"]
-15_min = 4
-hourly = 0
-daily = 0
-monthly = 0
-
-["media/secure/github-runners"]
-15_min = 6
-hourly = 2
-daily = 1
-monthly = 0
-
-["media/secure/notes"]
-15_min = 8
-hourly = 24
-daily = 30
-monthly = 12
-
-["media/secure/docker"]
-15_min = 3
-hourly = 12
-daily = 14
-monthly = 2
-
-# scratch
-["scratch/transmission"]
-15_min = 2
-hourly = 0
-daily = 0
-monthly = 0
-
-["scratch/uv_cache"]
-15_min = 2
-hourly = 0
-daily = 0
-monthly = 0
diff --git a/systems/jeeves/zfs.nix b/systems/jeeves/zfs.nix
new file mode 100644
index 0000000..88c43a8
--- /dev/null
+++ b/systems/jeeves/zfs.nix
@@ -0,0 +1,20 @@
+{ inputs, ... }:
+let
+  vars = import ./vars.nix;
+  jeeves_zfs = import ./datasets.nix;
+in
+{
+  services = {
+    zfs_manager = {
+      enable = true;
+      PYTHONPATH = "${inputs.self}/";
+      EnvironmentFile = "${vars.secrets}/services/snapshot_manager";
+
+      inherit (jeeves_zfs) datasets defaultSnapshots;
+    };
+
+    # Its retention config is generated from ./datasets.nix by
+    # common/optional/zfs_manager.nix, so only the credentials are set here.
+    snapshot_manager.EnvironmentFile = "${vars.secrets}/services/snapshot_manager";
+  };
+}
diff --git a/tests/test_snapshot_manager.py b/tests/test_snapshot_manager.py
index a849829..4d9b836 100644
--- a/tests/test_snapshot_manager.py
+++ b/tests/test_snapshot_manager.py
@@ -42,7 +42,7 @@ def test_main(mocker: MockerFixture, fs: FakeFilesystem) -> None:
     mock_dataset.create_snapshot.return_value = "snapshot created"
     mock_get_datasets = mocker.patch(f"{SNAPSHOT_MANAGER}.get_datasets", return_value=(mock_dataset,))
 
-    mock_get_snapshots_to_delete = mocker.patch(f"{SNAPSHOT_MANAGER}.get_snapshots_to_delete")
+    mock_get_snapshots_to_delete = mocker.patch(f"{SNAPSHOT_MANAGER}.get_snapshots_to_delete", return_value=[])
     mock_signal_alert = mocker.patch(f"{SNAPSHOT_MANAGER}.signal_alert")
     mock_snapshot_config_toml = '["default"]\n15_min = 8\nhourly = 24\ndaily = 0\nmonthly = 0\n'
     fs.create_file("/mock_snapshot_config.toml", contents=mock_snapshot_config_toml)
@@ -76,13 +76,39 @@ def test_main_create_snapshot_failure(mocker: MockerFixture, fs: FakeFilesystem)
     mock_signal_alert = mocker.patch(f"{SNAPSHOT_MANAGER}.signal_alert")
     mock_snapshot_config_toml = '["default"]\n15_min = 8\nhourly = 24\ndaily = 0\nmonthly = 0\n'
     fs.create_file("/mock_snapshot_config.toml", contents=mock_snapshot_config_toml)
-    main(Path("/mock_snapshot_config.toml"))
+    with pytest.raises(SystemExit) as exit_info:
+        main(Path("/mock_snapshot_config.toml"))
 
+    assert exit_info.value.code == 1
     mock_signal_alert.assert_called_once_with("test_dataset failed to create snapshot 2023-01-01T00:00:00")
     mock_get_datasets.assert_called_once()
     mock_get_snapshots_to_delete.assert_not_called()
 
 
+def test_main_delete_snapshot_failure(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    """Deletion failures make the service fail after processing the dataset."""
+    load_config_data.cache_clear()
+
+    mocker.patch(f"{SNAPSHOT_MANAGER}.get_time_stamp", return_value="2023-01-01T00:00:00")
+
+    mock_dataset = mocker.MagicMock(spec=Dataset)
+    mock_dataset.name = "test_dataset"
+    mock_dataset.create_snapshot.return_value = "snapshot created"
+    mocker.patch(f"{SNAPSHOT_MANAGER}.get_datasets", return_value=(mock_dataset,))
+    mocker.patch(
+        f"{SNAPSHOT_MANAGER}.get_snapshots_to_delete",
+        return_value=["test_dataset@auto_202301010000 failed to delete: busy"],
+    )
+    mocker.patch(f"{SNAPSHOT_MANAGER}.signal_alert")
+    mock_snapshot_config_toml = '["default"]\n15_min = 8\nhourly = 24\ndaily = 0\nmonthly = 0\n'
+    fs.create_file("/mock_snapshot_config.toml", contents=mock_snapshot_config_toml)
+
+    with pytest.raises(SystemExit) as exit_info:
+        main(Path("/mock_snapshot_config.toml"))
+
+    assert exit_info.value.code == 1
+
+
 def test_main_exception(mocker: MockerFixture, fs: FakeFilesystem) -> None:
     """Test main."""
     load_config_data.cache_clear()
@@ -141,6 +167,18 @@ def test_get_snapshots_to_delete_no_snapshot(mocker: MockerFixture) -> None:
     mock_dataset.delete_snapshot.assert_not_called()
 
 
+@pytest.mark.parametrize("invalid_count", [-1, "1", None, True])
+def test_invalid_retention_is_rejected_before_reading_snapshots(mocker: MockerFixture, invalid_count: object) -> None:
+    """Invalid standalone TOML values must never reach deletion logic."""
+    mock_dataset = mocker.MagicMock(spec=Dataset)
+    count_lookup = {"15_min": invalid_count, "hourly": 0, "daily": 0, "monthly": 0}
+
+    with pytest.raises(ValueError, match="15_min retention must be a non-negative integer"):
+        get_snapshots_to_delete(mock_dataset, count_lookup)  # type: ignore[arg-type]
+
+    mock_dataset.get_snapshots.assert_not_called()
+
+
 def test_get_snapshots_to_delete_errored(mocker: MockerFixture) -> None:
     """test_get_snapshots_to_delete_errored."""
     mock_snapshot_0 = create_mock_snapshot(mocker, "auto_202509150415")
@@ -153,8 +191,12 @@ def test_get_snapshots_to_delete_errored(mocker: MockerFixture) -> None:
 
     mock_signal_alert = mocker.patch(f"{SNAPSHOT_MANAGER}.signal_alert")
 
-    get_snapshots_to_delete(mock_dataset, {"15_min": 1, "hourly": 0, "daily": 0, "monthly": 0})
+    failures = get_snapshots_to_delete(
+        mock_dataset,
+        {"15_min": 1, "hourly": 0, "daily": 0, "monthly": 0},
+    )
 
+    assert failures == ["test_dataset@auto_202509150415 failed to delete: snapshot has dependent clones"]
     mock_signal_alert.assert_called_once_with(
         "test_dataset@auto_202509150415 failed to delete: snapshot has dependent clones"
     )
diff --git a/tests/test_zfs_manager.py b/tests/test_zfs_manager.py
new file mode 100644
index 0000000..d61eb09
--- /dev/null
+++ b/tests/test_zfs_manager.py
@@ -0,0 +1,382 @@
+"""test_zfs_manager."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+import pytest
+
+from python.tools.zfs_manager import main, parse_size, values_match
+
+if TYPE_CHECKING:
+    from pyfakefs.fake_filesystem import FakeFilesystem
+    from pytest_mock import MockerFixture
+
+ZFS_MANAGER = "python.tools.zfs_manager"
+CONFIG_PATH = "/mock_zfs_datasets.json"
+
+
+def write_config(
+    fs: FakeFilesystem,
+    datasets: dict[str, dict[str, str]],
+    unmanaged: list[str] | None = None,
+    never_create: list[str] | None = None,
+) -> Path:
+    """Write a dataset declaration to the fake filesystem."""
+    contents = {
+        "datasets": {
+            name: {
+                "manageProperties": True,
+                "createIfMissing": name not in (never_create or []),
+                "properties": props,
+            }
+            for name, props in datasets.items()
+        }
+        | {name: {"manageProperties": False, "createIfMissing": True, "properties": {}} for name in unmanaged or []},
+    }
+    fs.create_file(CONFIG_PATH, contents=json.dumps(contents))
+    return Path(CONFIG_PATH)
+
+
+def patch_zfs(
+    mocker: MockerFixture,
+    existing: list[str],
+    properties: dict[str, dict[str, tuple[str, str]]] | None = None,
+) -> dict[str, object]:
+    """Patch every zfs call zfs_manager makes."""
+    return {
+        "list": mocker.patch(f"{ZFS_MANAGER}.list_dataset_names", return_value=existing),
+        "get": mocker.patch(f"{ZFS_MANAGER}.get_properties", side_effect=lambda name: (properties or {}).get(name, {})),
+        "create": mocker.patch(f"{ZFS_MANAGER}.create_dataset", return_value=None),
+        "set": mocker.patch(f"{ZFS_MANAGER}.set_property", return_value=None),
+        "alert": mocker.patch(f"{ZFS_MANAGER}.signal_alert"),
+    }
+
+
+def test_creates_missing_dataset(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    zfs = patch_zfs(mocker, existing=["media", "media/secure"])
+    config = write_config(fs, {"media/secure/new": {"compression": "zstd-9"}})
+
+    main(config)
+
+    zfs["create"].assert_called_once_with("media/secure/new", {"compression": "zstd-9"})
+    zfs["set"].assert_not_called()
+    zfs["alert"].assert_not_called()
+
+
+def test_sets_drifted_property(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    zfs = patch_zfs(
+        mocker,
+        existing=["media", "media/temp"],
+        properties={"media/temp": {"compression": ("zstd", "inherited from media")}},
+    )
+    config = write_config(fs, {"media/temp": {"compression": "zstd-9"}})
+
+    main(config)
+
+    zfs["set"].assert_called_once_with("media/temp", "compression", "zstd-9")
+    zfs["create"].assert_not_called()
+
+
+def test_no_op_when_in_sync(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    zfs = patch_zfs(
+        mocker,
+        existing=["media", "media/temp"],
+        properties={"media/temp": {"sync": ("disabled", "local")}},
+    )
+    config = write_config(fs, {"media/temp": {"sync": "disabled"}})
+
+    main(config)
+
+    zfs["set"].assert_not_called()
+    zfs["create"].assert_not_called()
+    zfs["alert"].assert_not_called()
+
+
+def test_size_property_does_not_churn(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    """zfs get -p reports recordsize in bytes, the declaration uses a suffix."""
+    zfs = patch_zfs(
+        mocker,
+        existing=["media", "media/db"],
+        properties={"media/db": {"recordsize": ("16384", "local"), "special_small_blocks": ("32768", "local")}},
+    )
+    config = write_config(fs, {"media/db": {"recordsize": "16k", "special_small_blocks": "32K"}})
+
+    main(config)
+
+    zfs["set"].assert_not_called()
+
+
+def test_create_only_property_alerts_instead_of_setting(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    zfs = patch_zfs(
+        mocker,
+        existing=["media", "media/secure"],
+        properties={"media/secure": {"encryption": ("aes-256-gcm", "local")}},
+    )
+    config = write_config(fs, {"media/secure": {"encryption": "off"}})
+
+    # Nothing can fix a create-only mismatch at runtime, so it fails the run.
+    with pytest.raises(SystemExit) as exit_info:
+        main(config)
+
+    assert exit_info.value.code == 1
+    zfs["set"].assert_not_called()
+    assert zfs["alert"].call_count == 1
+    assert "can only be set when the dataset is created" in zfs["alert"].call_args.args[0]
+
+
+def test_undeclared_local_property_warns(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    zfs = patch_zfs(
+        mocker,
+        existing=["media", "media/temp"],
+        properties={"media/temp": {"exec": ("off", "local")}},
+    )
+    config = write_config(fs, {"media/temp": {}})
+
+    main(config)
+
+    zfs["alert"].assert_called_once_with("media/temp has exec=off set outside of nix")
+
+
+def test_undeclared_inherited_property_is_silent(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    zfs = patch_zfs(
+        mocker,
+        existing=["media", "media/temp"],
+        properties={
+            "media/temp": {
+                "compression": ("zstd", "inherited from media"),
+                "exec": ("on", "default"),
+                "nixos:shutdown-time": ("whenever", "local"),
+            },
+        },
+    )
+    config = write_config(fs, {"media/temp": {}})
+
+    main(config)
+
+    zfs["alert"].assert_not_called()
+
+
+def test_dry_run_makes_no_changes(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    zfs = patch_zfs(
+        mocker,
+        existing=["media", "media/temp"],
+        properties={"media/temp": {"compression": ("zstd", "local")}},
+    )
+    config = write_config(fs, {"media/temp": {"compression": "zstd-9"}, "media/new": {}})
+
+    main(config, dry_run=True)
+
+    zfs["set"].assert_not_called()
+    zfs["create"].assert_not_called()
+
+
+def test_pool_root_is_never_created(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    zfs = patch_zfs(mocker, existing=[])
+    config = write_config(fs, {"media": {"atime": "off"}})
+
+    with pytest.raises(SystemExit) as exit_info:
+        main(config)
+
+    assert exit_info.value.code == 1
+    zfs["create"].assert_not_called()
+    assert "does not create pools" in zfs["alert"].call_args.args[0]
+
+
+def test_children_skipped_when_parent_creation_fails(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    zfs = patch_zfs(mocker, existing=["media"])
+    zfs["create"].return_value = "Failed to create media/secure: key not loaded"
+    config = write_config(fs, {"media/secure": {}, "media/secure/child": {}})
+
+    with pytest.raises(SystemExit) as exit_info:
+        main(config)
+
+    assert exit_info.value.code == 1
+    zfs["create"].assert_called_once_with("media/secure", {})
+
+
+def test_unmanaged_dataset_properties_are_untouched(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    """A snapshots-only dataset is neither reconciled nor reported as unknown."""
+    zfs = patch_zfs(
+        mocker,
+        existing=["root_pool", "root_pool/var"],
+        properties={"root_pool/var": {"compression": ("lz4", "local")}},
+    )
+    config = write_config(fs, {}, unmanaged=["root_pool", "root_pool/var"])
+
+    main(config)
+
+    zfs["get"].assert_not_called()
+    zfs["set"].assert_not_called()
+    zfs["alert"].assert_not_called()
+
+
+def test_undeclared_dataset_is_left_alone(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    zfs = patch_zfs(mocker, existing=["media", "media/undeclared"])
+    config = write_config(fs, {})
+
+    main(config)
+
+    zfs["create"].assert_not_called()
+    zfs["set"].assert_not_called()
+    zfs["alert"].assert_not_called()
+
+
+def test_main_exception(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    zfs = patch_zfs(mocker, existing=[])
+    zfs["list"].side_effect = Exception("test")
+    config = write_config(fs, {})
+
+    with pytest.raises(SystemExit) as pytest_wrapped_e:
+        main(config)
+
+    assert pytest_wrapped_e.value.code == 1
+    zfs["alert"].assert_called_once_with("zfs_manager failed")
+
+
+@pytest.mark.parametrize(
+    ("value", "expected"),
+    [
+        ("16384", 16384),
+        ("16k", 16384),
+        ("16K", 16384),
+        ("1M", 1048576),
+        ("none", None),
+        ("", None),
+    ],
+)
+def test_parse_size(value, expected) -> None:
+    assert parse_size(value) == expected
+
+
+def test_values_match_falls_back_to_string_for_unparsable_sizes() -> None:
+    assert not values_match("recordsize", "none", "16384")
+    assert values_match("recordsize", "none", "none")
+    assert not values_match("compression", "zstd", "zstd-9")
+
+
+# -- failure handling: every one of these must exit non-zero -------------------
+
+
+def test_dataset_listing_failure_exits_nonzero(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    """A failed zfs list must abort, never be read as an empty system.
+
+    bash_wrapper hands back stderr as though it were output, so without the
+    return code check the reconciler would treat the error text as the dataset
+    list and conclude every declared dataset was missing.
+    """
+    zfs = patch_zfs(mocker, existing=[])
+    zfs["list"].side_effect = RuntimeError("Failed to list ZFS datasets: pool is busy")
+    config = write_config(fs, {"media/temp": {}})
+
+    with pytest.raises(SystemExit) as exit_info:
+        main(config)
+
+    assert exit_info.value.code == 1
+    zfs["create"].assert_not_called()
+    zfs["alert"].assert_called_once_with("zfs_manager failed")
+
+
+def test_create_failure_exits_nonzero(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    zfs = patch_zfs(mocker, existing=["media"])
+    zfs["create"].return_value = "Failed to create media/temp: out of space"
+    config = write_config(fs, {"media/temp": {}})
+
+    with pytest.raises(SystemExit) as exit_info:
+        main(config)
+
+    assert exit_info.value.code == 1
+    assert "out of space" in zfs["alert"].call_args.args[0]
+
+
+def test_set_failure_exits_nonzero(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    zfs = patch_zfs(
+        mocker,
+        existing=["media", "media/temp"],
+        properties={"media/temp": {"compression": ("zstd", "local")}},
+    )
+    zfs["set"].return_value = "Failed to set compression=zstd-9 on media/temp: permission denied"
+    config = write_config(fs, {"media/temp": {"compression": "zstd-9"}})
+
+    with pytest.raises(SystemExit) as exit_info:
+        main(config)
+
+    assert exit_info.value.code == 1
+    assert "permission denied" in zfs["alert"].call_args.args[0]
+
+
+def test_every_dataset_is_checked_before_failing(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    """One broken dataset must not hide the state of the others."""
+    zfs = patch_zfs(
+        mocker,
+        existing=["media", "media/one", "media/two", "media/three"],
+        properties={
+            "media/one": {"compression": ("zstd", "local")},
+            "media/two": {"compression": ("zstd", "local")},
+            "media/three": {"compression": ("zstd", "local")},
+        },
+    )
+    zfs["set"].return_value = "Failed to set compression: permission denied"
+    config = write_config(
+        fs,
+        {name: {"compression": "zstd-9"} for name in ("media/one", "media/two", "media/three")},
+    )
+
+    with pytest.raises(SystemExit) as exit_info:
+        main(config)
+
+    assert exit_info.value.code == 1
+    # All three were attempted and all three were reported, not just the first.
+    assert zfs["set"].call_count == 3
+    assert zfs["alert"].call_count == 3
+
+
+def test_dataset_marked_as_externally_created_is_never_created(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    """An encryption root must be reported as missing, not silently recreated.
+
+    Recreating it from this declaration would produce an unencrypted dataset,
+    since encryption is fixed at creation and is not declared here.
+    """
+    zfs = patch_zfs(mocker, existing=["media"])
+    config = write_config(fs, {"media/secure": {}}, never_create=["media/secure"])
+
+    with pytest.raises(SystemExit) as exit_info:
+        main(config)
+
+    assert exit_info.value.code == 1
+    zfs["create"].assert_not_called()
+    assert "created outside of nix" in zfs["alert"].call_args.args[0]
+
+
+def test_externally_created_dataset_is_still_property_checked(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    """When it does exist, it is reconciled like anything else."""
+    zfs = patch_zfs(
+        mocker,
+        existing=["media", "media/secure"],
+        properties={"media/secure": {"keylocation": ("prompt", "local")}},
+    )
+    config = write_config(
+        fs,
+        {"media/secure": {"keylocation": "file:///root/zfs.key"}},
+        never_create=["media/secure"],
+    )
+
+    main(config)
+
+    zfs["set"].assert_called_once_with("media/secure", "keylocation", "file:///root/zfs.key")
+
+
+def test_success_exits_cleanly(mocker: MockerFixture, fs: FakeFilesystem) -> None:
+    """The happy path must not raise SystemExit at all."""
+    zfs = patch_zfs(
+        mocker,
+        existing=["media", "media/temp"],
+        properties={"media/temp": {"sync": ("disabled", "local")}},
+    )
+    config = write_config(fs, {"media/temp": {"sync": "disabled"}})
+
+    main(config)
+
+    zfs["alert"].assert_not_called()