merge: TrueNAS 26 support

Enumerate datasets and snapshots from ZFS, not middleware's filtered queries; own the
snapshot sweep unconditionally (TrueNAS 26 decides "recursive" by a different rule than
we decide "nested"); resolve the snapshot namespace at runtime by the same predicate
tools/compat.py checks.

Four adversarial audits, every finding fixed and mutation-pinned. Verified on a real
TrueNAS 26.0.0-BETA.1 install: 292-dataset backup, 0 orphans, 0 leaked mounts,
byte-identical restore of a four-level-deep child dataset that middleware's own API
hides.

CHANGELOG.md

@@ -6,6 +6,128 @@ is deliberate: see [Releasing](docs/releasing.md). Twelve releases were cut on
 live, every one of those interrupts every user. An alert people learn to ignore is
 worse than no alert, because one day it carries a security fix.
 
+## Unreleased
+### Added
+
+- **TrueNAS 26 support, verified on a real TrueNAS 26 install.** 26 deletes
+  `plugins/zfs_/` outright, taking the private `zfs.dataset.query`,
+  `zfs.snapshot.query` and `zfs.snapshot.delete` with it. Every one of those was on
+  the nested module's critical path, so nested snapshots were **BROKEN** on 26 and
+  `apply.sh` correctly refused to apply the module there.
+
+  Snapshot **deletion** now resolves its namespace at runtime — `pool.snapshot` on
+  25.10 and 26, `zfs.snapshot` on 24.10 and 25.04, because no single namespace spans
+  every supported release. `tools/compat.py` checks the same list the runtime uses,
+  so what CI verifies and what runs cannot drift apart.
+
+  Hardware-verified on TrueNAS 26.0.0-BETA.1: a 274-snapshot recursive backup of a
+  292-dataset pool, then a **byte-identical restore of a four-level-deep child
+  dataset**.
+
+### Fixed
+
+- **Enumeration no longer trusts middleware's dataset and snapshot queries — they
+  are filtered.** This is the important one, and it is the bug that a test VM caught
+  and no amount of source analysis ever could have.
+
+  The obvious port of the deleted private `zfs.dataset.query` was the public
+  `pool.dataset.query`. It exists, it is documented, it is covered by iX's
+  deprecation policy — and it is **not a like-for-like replacement**. It applies a
+  *visibility policy*: it hides the datasets TrueNAS considers its own — `ix-apps/*`,
+  `.system/*`, `.ix-virt/*`. On a real pool that is **84 of 270 datasets**, and
+  `ix-apps` holds **live application data**.
+
+  Staging from that view would have silently omitted every one of them. Worse,
+  `plan_staging()` would never have seen them, so they would not have appeared in its
+  `skipped` list either — no warning, no failure, just a green backup quietly missing
+  data. That is precisely the failure this module exists to prevent. The snapshot
+  query lies the same way (205 of 274), so the sweep would have orphaned one snapshot
+  per hidden dataset, on every run, forever.
+
+  The module now **reads the truth from ZFS and makes changes through middleware**:
+  enumeration is `zfs list`, which no policy can filter and which behaves identically
+  on every release; mutation stays a middleware call, so TrueNAS's own bookkeeping
+  stays consistent. A failing `zfs list` raises rather than returning an empty list —
+  "no datasets" and "the command broke" must never look the same.
+
+  **No shipped release is affected.** v0.6.1 and earlier call the *private*
+  `zfs.dataset.query`, which returns all 270 datasets. The bug existed only in the
+  unreleased TrueNAS 26 port.
+
+- **The patch now owns the snapshot sweep even when it does not stage anything.**
+  Stock decides whether to take a *recursive* snapshot by its own rule, and on
+  TrueNAS 26 that rule stopped being ours.
+
+  Up to 25.10, stock's `create_snapshot` called `get_dataset_recursive()` — the same
+  function this module vendors — so "stock went recursive" and "we have something to
+  stage" were the *same question*, and stock's non-recursive delete was correct for
+  everything the patch declined to stage. TrueNAS 26 uses `filesystem.statfs`:
+  `recursive = (path == the dataset's mountpoint)`. The two rules now disagree for a
+  dataset whose only descendants are **ZVOLs** or **legacy/none-mountpoint** datasets
+  — stock snapshots it recursively, while the patch sees nothing to stage.
+
+  The patch then handed the snapshot back to stock, which destroys the parent only.
+  With no staging tree there was no sidecar, and the garbage collector only ever ran
+  from the staging path — so nothing on the box would ever have found the children.
+  Reproduced on the test VM: one orphaned snapshot per zvol, on every run, forever,
+  with the backup reporting success. Ownership of the sweep is no longer conditional
+  on staging.
+
+- **The runtime resolved a *namespace*; the checker verified a *method*.** Those are
+  different questions, and the gap is a false "ok". `get_service()` only proves a
+  namespace is registered — it says nothing about whether `delete` still exists on it.
+  So if iX guts the method while keeping the service (they have already done exactly
+  that to `pool.snapshot.do_update` on master), `tools/compat.py` would fall through
+  to `zfs.snapshot`, report the box healthy, and let the patch apply — while the
+  runtime picked `pool.snapshot` and failed *every* delete, orphaning the whole tree.
+  Both sides now ask the same question, and a test binds the two lists together.
+
+- `query_filesystems()` **dropped malformed `zfs list` rows silently** — the last
+  remaining silent-omission path, and a direct contradiction of this module's cardinal
+  rule. It raises now. A missing `zfs` binary raised `FileNotFoundError` rather than
+  `ZfsError`; also fixed.
+
+- The snapshot retry loop **discarded the delete error** and reported every survivor
+  as "(still busy?)" — naming the one cause that is benign and self-healing, and
+  hiding the ones that are permanent. It keeps and reports the real error.
+
+- The staging-failure handler could **lose the original exception** if its own cleanup
+  sweep raised. An error handler must not be able to lose the error.
+
+- **A snapshot delete that returns cleanly is not proof that anything was deleted.**
+  The recursive sweep's fast path took the call's word for it and returned "no
+  survivors" — so `cleanup_task` read that as a clean sweep and removed the sidecar,
+  the only record the tree ever existed. Roughly 250 snapshots would have been orphaned
+  on every run, with nothing left able to find them, and the backup reporting success.
+
+  This is not a hypothetical about a well-behaved API: iX has already gutted
+  `pool.snapshot.do_update` on master into a no-op whose body is commented out and
+  which returns `None`. A source check still sees the `def`; a runtime check still sees
+  a callable method. Only asking ZFS can tell. The sweep now confirms against ZFS, and
+  where it *cannot* confirm it keeps owning the tree rather than claiming success — a
+  false survivor self-heals on the next run, a lost record never does.
+
+- `_write_sidecar` **swallowed `OSError`**. The sidecar is the only thing that survives
+  a middlewared restart; failing to write it is not fatal, but it must never be
+  invisible. `_read_sidecar` had the mirror bug — it conflated "there is no sidecar"
+  with "I could not read the sidecar", and `cleanup_task` then took the empty branch
+  and **unlinked the only record** of a tree it had failed to read.
+
+- **A dataset from another tree, mounted inside the backup path, was omitted
+  silently.** The staging plan scopes by dataset *name*, which is correct — a dataset
+  with no mountpoint cannot be scoped by path at all. But ZFS lets any dataset mount
+  anywhere, so one from an unrelated tree can sit inside the path:
+
+      Tank/photos   mountpoint=/mnt/Tap/apps/photos
+
+  It holds data inside the backed-up path, and `zfs snapshot -r Tap@…` does **not**
+  cover it: recursion follows the dataset tree, not the directory tree. So there is no
+  snapshot of it to stage, and no way to capture it consistently with the rest. It fell
+  out of the name filter and vanished — not staged, not in `skipped`, no error, backup
+  green. Stock has the same blind spot, but stock also refuses the nested config
+  outright; this patch is what relaxes that guard, so the hole is this patch's to close.
+  It now refuses, and names the offending datasets.
+
 ## v0.6.1 — 2026-07-13
 ### Fixed
 

README.md

@@ -60,7 +60,7 @@ If something is wrong, the reason is in `apply.log` — start at
 | 24.10.2.4 | ok | ok | — |
 | 25.04.2.6 | ok | ok | — |
 | 25.10.4 | ok | ok | nested + providers; 252-snapshot recursive backup of /mnt/Tap, 18m |
-| 26.0.0-BETA.3 _(unreleased)_ | ok | **BROKEN** | — |
+| 26.0.0-BETA.3 _(unreleased)_ | ok | ok | — |
 | master _(unreleased)_ | **BROKEN** | **BROKEN** | — |
 
 | verdict | meaning |
@@ -77,6 +77,12 @@ source. It does not mean a human ran a backup on it — that is the
 The table is **regenerated daily by CI** against iXsystems' actual middleware source
 — it is not a claim somebody typed once and forgot.
 
+**On TrueNAS 26:** the patch was run on a real TrueNAS **26.0.0-BETA.1** install — a
+274-snapshot recursive backup of a 292-dataset pool, followed by a byte-identical
+restore of a four-level-deep child dataset. The *Hardware-verified* column tracks the
+newest beta iX has tagged (currently BETA.3), so it does not carry that mark: a build
+nobody has actually run a backup on does not get credit for one.
+
 **TrueNAS 26: nested snapshots are not supported yet, and upgrading will not break
 you.** 26 rewrites `cloud_backup` and deletes the ZFS methods this module calls. On
 26 `apply.sh` finds that the assumptions no longer hold and **does not apply the

patch/apply.sh

@@ -535,8 +535,8 @@ if _tc_nested is not None:
         # snapshot=true, not just ours. Two consequences, and the second is worse:
         #
         #   * everything below is a NEW failure mode for tasks that worked before we
-        #     were installed. A `zfs.dataset.query` that errors would break a
-        #     CloudSync job we have no business touching.
+        #     were installed. A `zfs list` that errors would break a CloudSync job
+        #     we have no business touching.
         #   * if a CloudSync task ever were staged, nothing would ever tear it down:
         #     the teardown is wired into cloud_backup's restic_backup finally, and
         #     CRUD_BLOCK deliberately leaves CloudSync's nesting guard intact. The
@@ -556,18 +556,28 @@ if _tc_nested is not None:
             # our staging plan would not -- silently omitting it from the backup.
             # Read afterwards, an unsnapshotted dataset instead trips the isdir()
             # check in plan_staging and fails the run loudly. Loud beats silent.
-            datasets = middleware.call_sync(
-                "zfs.dataset.query", [["type", "=", "FILESYSTEM"]]
-            )
+            # query_filesystems() reads ZFS directly. It deliberately does NOT use
+            # pool.dataset.query: that applies a visibility policy and hides
+            # TrueNAS-internal datasets (ix-apps/*, .system/*, .ix-virt/*) -- 84 of
+            # 270 on a real pool, including live app data. Staging from the filtered
+            # view omits them silently, which is the one thing this must never do.
+            datasets = _tc_nested.query_filesystems(middleware)
             # OUR copy of get_dataset_recursive, not the host module's: TrueNAS 26
             # deleted that helper (create_snapshot uses filesystem.statfs now), so
             # calling it out of the module namespace is a NameError there.
             dataset, nested = _tc_nested.get_dataset_recursive(datasets, path)
 
             if not nested:
-                # No children: stock behaviour, untouched. Stock's `finally` owns
-                # the snapshot from here (its non-recursive delete is correct,
-                # because a non-nested snapshot has no children).
+                # Nothing to STAGE -- but we still own the SWEEP, and that is not a
+                # formality. Stock decides `recursive` by its own rule, and on 26 that
+                # rule is no longer ours: it snapshots recursively whenever the backup
+                # path IS the dataset's mountpoint (filesystem.statfs), while
+                # get_dataset_recursive() sees nothing to stage when the only
+                # descendants are ZVOLs or legacy/none-mountpoint datasets. Stock then
+                # deletes the PARENT ONLY. Without this, one snapshot per descendant is
+                # orphaned on every run, forever, with no sidecar and no GC to find it --
+                # and the backup still reports success.
+                _tc_nested.own_snapshot(middleware, name, snapshot, logger=_logger)
                 return snapshot, snap_path
 
             staging_root = _tc_nested.stage_nested(
@@ -581,7 +591,19 @@ if _tc_nested is not None:
             # stays None and its `finally` deletes NOTHING. Sweep the tree ourselves
             # or leak the parent plus one snapshot per descendant dataset (160+ here)
             # on every failed run.
+            #
+            # The sweep is itself wrapped: a cleanup that raises would REPLACE the
+            # original exception with its own, hiding why the backup actually failed.
+            # An error handler must not be able to lose the error.
+            try:
                 _tc_nested.delete_snapshot_tree(middleware, snapshot, logger=_logger)
+            except Exception as _tc_sweep_err:
+                if _logger:
+                    _logger.error(
+                        "truecloud-patch: could not sweep %s after a staging failure "
+                        "(%r) -- it is orphaned and must be deleted by hand",
+                        snapshot, _tc_sweep_err,
+                    )
             raise
 
         return snapshot, staging_root

patch/truecloud_nested.py

@@ -34,7 +34,7 @@ feature exists to prevent, and it would be worse than not having the feature.
 
 Snapshot lifecycle -- read this before changing anything
 --------------------------------------------------------
-``zfs.snapshot.delete`` defaults to ``recursive=False``, and stock
+The snapshot delete call defaults to ``recursive=False``, and stock
 ``restic_backup()`` calls it with no options. Stock gets away with that because
 its validation means ``recursive`` is never actually True in the field. Enabling
 nested datasets makes recursive snapshots real, so the parent
@@ -62,17 +62,25 @@ import subprocess
 import time
 
 __all__ = [
+    "DELETE_METHODS",
+    "SNAPSHOT_SERVICES",
     "STAGING_BASE",
     "StagingError",
+    "ZfsError",
     "apply_plan",
     "cleanup_all",
     "cleanup_task",
     "current_mounts_under",
     "delete_snapshot_tree",
     "gc_stale_snapshots",
+    "list_snapshot_names",
     "mounted_snapshots",
+    "own_snapshot",
+    "pick_snapshot_service",
     "plan_staging",
+    "query_filesystems",
     "sidecar_for",
+    "snapshot_service",
     "snapshot_tree_names",
     "stage_nested",
     "stale_snapshot_names",
@@ -81,9 +89,300 @@ __all__ = [
     "verify_staged",
 ]
 
+
+# ── how this module talks to the system ──────────────────────────────────────
+#
+#   READ the truth from ZFS.  MAKE CHANGES through middleware.
+#
+# That split is not stylistic. It was forced by finding, on a real TrueNAS 26 box,
+# that middleware's query APIs apply a VISIBILITY POLICY:
+#
+#   zfs list                 274 datasets      205 from pool.dataset.query
+#   zfs list -t snapshot     274 snapshots     205 from pool.snapshot.query
+#
+# The missing 69 are the datasets TrueNAS considers its own -- `ix-apps/*`,
+# `.system/*`, `.ix-virt/*` -- and on the real pool that is 84 of 270, including
+# `ix-apps`, which holds live application data. Enumerating from that view would
+# have silently omitted every one of them from the staging plan and from the
+# snapshot sweep: a green backup missing data, and one orphaned snapshot per
+# hidden dataset on every run. Both are exactly what this module exists to
+# prevent.
+#
+# This went unnoticed because the patch used to call the PRIVATE `zfs.dataset.query`
+# and `zfs.snapshot.query`, which return everything. TrueNAS 26 deleted them, and
+# the public replacements are NOT like-for-like -- they are filtered. So
+# enumeration now reads ZFS directly, which no policy can filter and which behaves
+# identically on every release.
+#
+# MUTATION still goes through middleware, so TrueNAS's own bookkeeping stays
+# consistent -- and an exact-name delete works fine even on a dataset the query
+# hides. The one wrinkle is that no single snapshot namespace spans every
+# supported release, so it is resolved at runtime rather than pinned:
+#
+#   24.10, 25.04   `zfs.snapshot`  (public back then; `pool.snapshot` does not exist)
+#   25.10          both -- `pool.snapshot` public, `zfs.snapshot` demoted to private
+#   26             `pool.snapshot` only -- `plugins/zfs_/` is gone
+#
+#: Snapshot CRUD namespaces, best first. `tools/compat.py` checks this exact list
+#: (MiddlewareCall.also) with the same predicate the runtime uses -- the namespace
+#: exists AND it defines `delete`/`do_delete` -- and a test binds the two lists
+#: together, so what CI verifies and what runs cannot drift apart.
+SNAPSHOT_SERVICES = ("pool.snapshot", "zfs.snapshot")
+
+
+#: The CRUDService method spellings that answer to `<namespace>.delete`. A
+#: CRUDService exposes `delete` from a method NAMED `do_delete`; both are live
+#: across the matrix. `tools/compat.py` accepts exactly this pair.
+DELETE_METHODS = ("delete", "do_delete")
+
+
+def pick_snapshot_service(can_delete):
+    """First namespace in SNAPSHOT_SERVICES that can actually DELETE for us.
+
+    Pure: `can_delete(namespace) -> bool`. Returns None if no namespace can,
+    which is a middleware we have never seen and must not guess about.
+
+    The predicate is "can delete", NOT "the service is registered", and the
+    difference is the whole point. `get_service()` only proves the namespace is
+    in the registry; it says nothing about whether `delete` still exists on it.
+    `tools/compat.py` checks namespace AND method, so if the runtime settled for
+    the weaker test the two could disagree — and would, in the one way that
+    matters: iX guts a method while keeping its service (they have already done
+    exactly that to `pool.snapshot.do_update` on master). compat would try
+    `pool.snapshot`, find `delete` gone, fall through to `zfs.snapshot`, and
+    report **ok**; the runtime would take `pool.snapshot` because the service is
+    still registered, and then fail on every single delete — orphaning the whole
+    tree while the backup reports success.
+
+    Same predicate on both sides, so they cannot drift.
+    """
+    for name in SNAPSHOT_SERVICES:
+        if can_delete(name):
+            return name
+    return None
+
+
+#: Where middlewared's own service framework lives. Classes from this package are
+#: PLUMBING, not implementations -- see `_defines_delete`.
+FRAMEWORK_PACKAGE = "middlewared.service"
+
+
+def _defines_delete(service):
+    """Does this service ITSELF implement a delete -- or merely inherit the framework's?
+
+    The distinction is the whole fix, and getting it wrong is silent.
+
+    `CRUDService` defines `delete` on the BASE class and dispatches to `self.do_delete`
+    at call time. So `getattr(service, "delete")` is a bound method on EVERY
+    CRUDService subclass, whether or not that subclass still implements one:
+
+        middlewared.plugins.pool_.snapshot.PoolSnapshotService  defines ['do_delete']
+        middlewared.service.crud_service.CRUDService            defines ['delete']
+
+    An earlier version of this check asked `callable(getattr(service, "delete"))` and
+    was therefore answering "is this a CRUDService?" -- exactly the weaker "is the
+    namespace registered?" question that `pick_snapshot_service` exists to avoid. It
+    would have picked a gutted `pool.snapshot` and failed every delete.
+
+    So walk the MRO and ignore the framework's generic plumbing: a delete is real only
+    where a PLUGIN class defines it. That mirrors `tools/compat.py`, which looks for
+    the `def` in the plugin file declaring the namespace.
+    """
+    for klass in type(service).__mro__:
+        module = getattr(klass, "__module__", "") or ""
+        if module == FRAMEWORK_PACKAGE or module.startswith(FRAMEWORK_PACKAGE + "."):
+            continue                       # the framework's generic CRUD dispatcher
+        if any(m in vars(klass) for m in DELETE_METHODS):
+            return True
+    return False
+
+
+def _can_delete(middleware, namespace):
+    """Is `<namespace>.delete` actually implemented on this middleware?"""
+    try:
+        service = middleware.get_service(namespace)
+    except Exception:
+        # KeyError for an unregistered namespace; AttributeError if `get_service`
+        # itself ever goes away. Both mean "cannot use it", and guessing YES on a
+        # service that is not really there fails later, mid-backup, holding a
+        # snapshot -- the worst possible moment to find out.
+        return False
+    return _defines_delete(service)
+
+
+def snapshot_service(middleware):
+    """The snapshot namespace this middleware can actually delete through."""
+    name = pick_snapshot_service(lambda n: _can_delete(middleware, n))
+    if name is None:
+        raise StagingError(
+            "middleware exposes no usable snapshot delete ("
+            + " / ".join(f"{n}.delete" for n in SNAPSHOT_SERVICES)
+            + "). Refusing to stage a nested backup, because the snapshot it "
+            "creates could not then be swept."
+        )
+    return name
+
+
+class _Snapshots:
+    """This module's entire interface to snapshots, in one object.
+
+    "READ the truth from ZFS, MAKE CHANGES through middleware" is the rule the whole
+    module rests on. It used to live in a comment, while `middleware` and the ZFS
+    reader were threaded through five functions **as a pair** -- and the namespace was
+    re-resolved in each of them. That is one collaborator, not two, so it is one
+    object; the rule is now structural rather than remembered.
+
+    Deliberately private and constructed inside the public functions: `apply.sh`
+    injects calls to those functions into middlewared itself, so their signatures are
+    a boot-time contract with a live NAS and are not worth churning for tidiness.
+    """
+
+    def __init__(self, middleware, list_snapshots=None):
+        self._mw = middleware
+        self._list = list_snapshots or list_snapshot_names
+        self._service = None
+
+    @property
+    def service(self):
+        """The namespace we delete through. Resolved once, on first use.
+
+        Lazy on purpose: resolving it raises when middleware has no usable delete,
+        and the read-only paths must not blow up over a mutation they never make.
+        """
+        if self._service is None:
+            self._service = snapshot_service(self._mw)
+        return self._service
+
+    def names(self, dataset):
+        """Every snapshot at or under `dataset`, from ZFS. Raises if it cannot be read.
+
+        Never from middleware: its snapshot query hides the internal datasets
+        (205 of 274 on the test box), and a snapshot the sweep cannot SEE is a
+        snapshot nothing will ever collect.
+        """
+        return self._list(dataset)
+
+    def delete(self, name, recursive=False):
+        """Delete one snapshot -- through middleware, so its bookkeeping stays right.
+
+        An exact-name delete works even on a dataset the query hides; it is only
+        enumeration that lies.
+        """
+        options = ({"recursive": True},) if recursive else ()
+        return self._mw.call_sync(f"{self.service}.delete", name, *options)
+
+
+class ZfsError(Exception):
+    """`zfs list` failed. Enumeration is unreliable, so the caller must not guess."""
+
+
+def _zfs_lines(args, runner=None, fields=None):
+    """`zfs <args>` as a list of tab-split rows. Raises ZfsError if it fails.
+
+    Never returns a partial or empty list on failure: a caller that cannot tell
+    "no datasets" from "the command broke" will happily stage nothing, or sweep
+    nothing, and report success.
+
+    `fields`, if given, is the exact number of tab-separated columns every row must
+    have. A row that does not is an ERROR, not something to skip. `zfs list -H`
+    neither quotes nor escapes, so a mountpoint containing a tab or a newline would
+    split wrong -- and quietly dropping that row would remove a dataset from the
+    staging plan without it appearing in `skipped` either. Silent omission is the
+    one thing this module may never do, so it raises instead.
+    """
+    runner = runner or _run
+    try:
+        r = runner(["zfs", *args])
+    except OSError as e:
+        # `zfs` missing from middlewared's PATH raises FileNotFoundError, which is
+        # not a ZfsError and would sail past callers that only expect one.
+        raise ZfsError(f"could not run zfs: {e}") from e
+
+    if r.returncode != 0:
+        raise ZfsError((r.stderr or "").strip() or f"zfs {' '.join(args)} failed")
+
+    rows = [ln.split("\t") for ln in r.stdout.splitlines() if ln.strip()]
+    if fields is not None:
+        bad = [r for r in rows if len(r) != fields]
+        if bad:
+            raise ZfsError(
+                f"zfs {' '.join(args)} returned {len(bad)} row(s) that do not have "
+                f"{fields} tab-separated fields (first: {bad[0]!r}). Refusing to "
+                f"guess -- a dropped row is a dataset silently missing from the backup."
+            )
+    return rows
+
+
+def query_filesystems(middleware=None, runner=None):
+    """Every FILESYSTEM dataset, in the shape the planner speaks -- read from ZFS.
+
+    NOT from `pool.dataset.query`, and this is the single most important decision
+    in this file.
+
+    middleware's dataset query applies a VISIBILITY POLICY: it hides the datasets
+    TrueNAS considers its own -- `ix-apps/*`, `.system/*`, `.ix-virt/*`. That is
+    **84 of 270 datasets** on the real pool, and `ix-apps` holds live application
+    data. Building the staging plan from that view would silently omit every one of
+    them. Worse, `plan_staging()` would never even SEE them, so they would not turn
+    up in its `skipped` list either -- no warning, no failure, just a green backup
+    quietly missing data. That is exactly the failure this whole module exists to
+    prevent, and it is the failure the cardinal rule at the top of this file is
+    about.
+
+    It worked before only because the patch called the PRIVATE `zfs.dataset.query`,
+    which returned everything. TrueNAS 26 deleted it. The public replacement is not
+    a like-for-like: it is a filtered view.
+
+    So: **read the truth from ZFS, make changes through middleware.** ZFS cannot
+    apply a policy to what it reports, and `zfs list` behaves identically on every
+    release -- which also means one code path instead of a version conditional.
+
+    `middleware` is accepted and ignored, so callers need not care where the data
+    comes from.
+    """
+    rows = _zfs_lines(
+        ["list", "-H", "-p", "-o", "name,mountpoint,mounted", "-t", "filesystem"],
+        runner=runner, fields=3,
+    )
+    return [
+        {
+            "name": name,
+            "properties": {
+                "mountpoint": {"value": mountpoint},
+                # `zfs list` prints yes/no; the planner already speaks that.
+                "mounted": {"value": mounted},
+            },
+        }
+        for name, mountpoint, mounted in rows
+    ]
+
+
+def list_snapshot_names(dataset, runner=None):
+    """Every snapshot at or under `dataset` -- read from ZFS, for the same reason.
+
+    `pool.snapshot.query` filters exactly like the dataset query does: on this box
+    it returned 205 of 274 snapshots, hiding the internal datasets' snapshots. A
+    sweep built on that view leaves one orphan per hidden dataset, on every run,
+    forever -- which is the bug this module was written to fix in the first place.
+
+    (An EXACT-name delete still works on a hidden dataset, so mutations may keep
+    going through middleware. It is only enumeration that lies.)
+    """
+    rows = _zfs_lines(
+        ["list", "-H", "-o", "name", "-t", "snapshot", "-r", dataset],
+        runner=runner, fields=1,
+    )
+    return [r[0] for r in rows]
+
 #: Where staging trees are assembled. tmpfs; bind mounts consume no space.
 STAGING_BASE = "/run/truecloud-nested"
 
+#: The kernel's mount table. Late-bound (never a default argument) so a
+#: test can point it somewhere harmless -- a default is frozen at def time,
+#: which is how 19 tests ended up reading the REAL table, one matching name
+#: away from running a real `umount` on the NAS.
+MOUNTS_FILE = "/proc/self/mounts"
+
 # Which snapshot a staging tree pins is recorded ONLY in the sidecar file, never
 # also in memory. An in-process dict would be a second source of truth that a
 # middlewared restart silently empties -- and it is exactly the restart case that
@@ -118,7 +417,7 @@ def sidecar_for(staging_root: str) -> str:
     return staging_root + ".snapshot"
 
 
-def _write_sidecar(staging_root: str, snapshots) -> None:
+def _write_sidecar(staging_root: str, snapshots, logger=None) -> None:
     """Record every snapshot tree this task still owns. One per line.
 
     A LIST, not a single name -- and that is not over-engineering, it is a bug fix.
@@ -135,22 +434,48 @@ def _write_sidecar(staging_root: str, snapshots) -> None:
     """
     if isinstance(snapshots, str):
         snapshots = [snapshots]
-    with contextlib.suppress(OSError):
+    try:
         os.makedirs(os.path.dirname(staging_root), exist_ok=True)
         with open(sidecar_for(staging_root), "w", encoding="utf-8") as fh:
             fh.write("\n".join(dict.fromkeys(snapshots)))   # de-duped, order kept
+    except OSError as e:
+        # This used to be suppressed silently, and it is the LAST thing that should be.
+        # The sidecar is the only record that these snapshots exist; if the write fails
+        # (a full /run, say) cleanup_task finds nothing to sweep, and only the by-name
+        # collector -- an hour later -- has any chance of finding them. Failing to
+        # write it is not fatal, but it must never be invisible.
+        if logger:
+            logger.error(
+                "truecloud-patch: COULD NOT RECORD the snapshot(s) %s (%r). If this "
+                "run does not clean them up itself, only the by-name collector will "
+                "ever find them.", ", ".join(snapshots), e,
+            )
 
 
-def _read_sidecar(staging_root: str):
+def _read_sidecar(staging_root: str, logger=None):
     """Every snapshot tree a previous run recorded here. [] if none.
 
     Tolerates the old single-line format, which is just a one-element list.
+
+    "There is no sidecar" and "I could not READ the sidecar" are different facts, and
+    conflating them is dangerous: `cleanup_task` reads an empty list as "nothing was
+    ever staged" and then REMOVES the sidecar -- destroying the only record of a tree
+    it could not read. FileNotFoundError is the ordinary case and stays quiet; any
+    other OSError is reported, and re-raised so no caller mistakes it for "empty".
     """
     try:
         with open(sidecar_for(staging_root), encoding="utf-8") as fh:
             return [ln.strip() for ln in fh if ln.strip()]
-    except OSError:
-        return []
+    except FileNotFoundError:
+        return []                                   # genuinely nothing recorded
+    except OSError as e:
+        if logger:
+            logger.error(
+                "truecloud-patch: could not READ the snapshot record %s (%r). Not "
+                "touching it -- it may name snapshots nothing else can find.",
+                sidecar_for(staging_root), e,
+            )
+        raise
 
 
 def _remove_sidecar(staging_root: str) -> None:
@@ -271,7 +596,7 @@ def plan_staging(base_dataset, base_mountpoint, path, snapshot_name, datasets,
                  staging_root, probe=_probe_snapdir):
     """Compute the bind-mount plan for staging a nested tree. Pure function.
 
-    ``datasets`` is a list of dicts shaped like ``zfs.dataset.query`` results:
+    ``datasets`` is what :func:`query_filesystems` returns:
     ``{"name": str, "properties": {"mountpoint": {"value": str},
     "mounted": {"value": "yes"|"no"}}}``.
 
@@ -345,13 +670,68 @@ def plan_staging(base_dataset, base_mountpoint, path, snapshot_name, datasets,
 
         mounts.append((src, os.path.join(staging_root, os.path.relpath(mp, path))))
 
+    # ── datasets INSIDE the path but OUTSIDE the snapshot's tree ─────────────
+    #
+    # The loop above scopes by dataset NAME, which is right: a dataset with no
+    # mountpoint cannot be scoped by path at all. But ZFS lets any dataset mount
+    # anywhere, so a dataset from a DIFFERENT tree -- even a different pool -- can
+    # sit inside the backup path:
+    #
+    #     Tank/photos   mountpoint=/mnt/Tap/apps/photos
+    #
+    # It holds data inside the path, so its absence is a hole in the backup. And
+    # `zfs snapshot -r Tap@...` does NOT cover it, because recursion follows the
+    # DATASET tree, not the directory tree -- so there is no snapshot of it to
+    # stage, and no way to capture it consistently with the rest.
+    #
+    # Before this check it was neither staged, nor reported in `skipped`, nor raised:
+    # it simply fell out of the name filter and vanished. The backup reported
+    # SUCCESS with that data missing, which is the precise failure this module
+    # exists to prevent. Stock has the same blind spot, but stock also REFUSES the
+    # nested config outright -- we are the ones relaxing that guard, so the hole is
+    # ours to close.
+    foreign = []
+    for ds in datasets:
+        name = ds.get("name", "")
+        if name.startswith(ds_prefix) or name == base_dataset:
+            continue                                   # in-tree: handled above
+
+        props = ds.get("properties", {})
+        mp = props.get("mountpoint", {}).get("value", "")
+        # `mp == path` as well as below it. A foreign dataset mounted exactly AT the
+        # backup path is the same hole -- and it is worse, because it shadows the base
+        # dataset's own directory, so we would stage what is hidden underneath instead
+        # of the data actually visible there.
+        if not mp or (mp != path and not mp.startswith(path_prefix)):
+            continue                                   # not inside the backed-up path
+
+        if props.get("mounted", {}).get("value", "yes") == "no":
+            # An unmounted (locked/encrypted) dataset contributes nothing to the live
+            # tree, so its absence is not a hole -- exactly as for an in-tree one, 40
+            # lines above. Raising here would turn a working nightly backup into a
+            # permanent failure the first time somebody locked a dataset.
+            skipped.append((name, "dataset is not mounted (locked/encrypted?)"))
+            continue
+
+        foreign.append(name)
+
+    if foreign:
+        raise StagingError(
+            "dataset(s) outside " + repr(base_dataset) + " are mounted inside the "
+            "backup path and cannot be captured by its recursive snapshot: "
+            + ", ".join(repr(f) for f in sorted(foreign))
+            + ". Refusing to back up an incomplete tree -- move them, or back up "
+            "their own dataset separately."
+        )
+
     # Parents before children, so each mountpoint exists before we mount onto it.
     mounts.sort(key=lambda m: _depth(m[1]))
     return mounts, skipped
 
 
-def current_mounts_under(root, mounts_file="/proc/self/mounts"):
+def current_mounts_under(root, mounts_file=None):
     """Mountpoints at or under ``root``, deepest first. Used for teardown."""
+    mounts_file = mounts_file or MOUNTS_FILE
     found = []
     try:
         with open(mounts_file, encoding="utf-8") as fh:
@@ -379,12 +759,13 @@ def _run(cmd):
     )
 
 
-def apply_plan(mounts, runner=_run, isdir=os.path.isdir):
+def apply_plan(mounts, runner=None, isdir=os.path.isdir):
     """Execute the bind-mount plan. Blocking; call via ``run_in_thread``.
 
     Raises StagingError on the first failure, after rolling back what was
     mounted -- a half-built tree must never be handed to the backup tool.
     """
+    runner = runner or _run
     if not mounts:
         raise StagingError("empty staging plan")
 
@@ -437,12 +818,14 @@ def verify_staged(mounts, ismount=os.path.ismount, listdir=os.listdir):
     return True
 
 
-def teardown(staging_root, runner=_run, mounts_file="/proc/self/mounts"):
+def teardown(staging_root, runner=None, mounts_file=None):
     """Unmount the staging tree (deepest first) and remove the root.
 
     Idempotent, and does not depend on an in-memory plan -- so it also cleans up
     leftovers from a crashed run.
     """
+    mounts_file = mounts_file or MOUNTS_FILE
+    runner = runner or _run
     errors = []
     for mp in current_mounts_under(staging_root, mounts_file=mounts_file):
         res = runner(["umount", mp])
@@ -455,8 +838,9 @@ def teardown(staging_root, runner=_run, mounts_file="/proc/self/mounts"):
     return errors
 
 
-def snapdir_automounts(snapshot_name, mounts_file="/proc/self/mounts"):
+def snapdir_automounts(snapshot_name, mounts_file=None):
     """Every ``<dataset>/.zfs/snapshot/<snap>`` ZFS automount for this snapshot."""
+    mounts_file = mounts_file or MOUNTS_FILE
     suffix = "/.zfs/snapshot/" + snapshot_name
     found = []
     try:
@@ -472,7 +856,7 @@ def snapdir_automounts(snapshot_name, mounts_file="/proc/self/mounts"):
     return sorted(found, key=_depth, reverse=True)      # deepest first
 
 
-def release_snapdirs(snapshot_name, runner=_run, mounts_file="/proc/self/mounts"):
+def release_snapdirs(snapshot_name, runner=None, mounts_file=None):
     """Unmount ZFS's OWN snapshot automounts, so the snapshots can be destroyed.
 
     Reading anything under ``<dataset>/.zfs/snapshot/<snap>/`` makes ZFS **automount**
@@ -489,6 +873,8 @@ def release_snapdirs(snapshot_name, runner=_run, mounts_file="/proc/self/mounts"
 
     Deepest first, so a child's automount is released before its parent's.
     """
+    mounts_file = mounts_file or MOUNTS_FILE
+    runner = runner or _run
     errors = []
     for mp in snapdir_automounts(snapshot_name, mounts_file=mounts_file):
         res = runner(["umount", mp])
@@ -560,7 +946,7 @@ def get_dataset_recursive(datasets, directory):
 
 
 def delete_snapshot_tree(middleware, snapshot, logger=None, attempts=4,
-                         sleep=time.sleep):
+                         sleep=None, list_snapshots=None):
     """Delete the parent snapshot AND every child created by ``zfs snapshot -r``.
 
     Returns the snapshots it could NOT delete -- callers must not throw that away.
@@ -578,6 +964,8 @@ def delete_snapshot_tree(middleware, snapshot, logger=None, attempts=4,
     snapshots hit this, and before the fix they were orphaned permanently.
     """
     dataset, _, snapname = snapshot.partition("@")
+    snaps = _Snapshots(middleware, list_snapshots)
+    sleep = sleep or time.sleep
 
     # Release ZFS's own automounts first, or `zfs destroy` refuses with EBUSY on
     # everything restic read in the last few minutes.
@@ -591,15 +979,55 @@ def delete_snapshot_tree(middleware, snapshot, logger=None, attempts=4,
     # through 252 sequential deletes leaves exactly the orphans this function
     # exists to prevent.
     try:
-        middleware.call_sync("zfs.snapshot.delete", snapshot, {"recursive": True})
-        return []
-    except Exception as e:  # noqa: BLE001 - fall through to the explicit sweep
-        # Usually just "parent already gone" (stock's finally won the race once our
-        # mounts were released), which the sweep below handles. Log it rather than
-        # swallow it: if the real cause is something else, this is the only place
-        # it is visible -- the sweep would report a different, downstream failure.
+        snaps.delete(snapshot, recursive=True)
+
+        # CONFIRM it. A delete that returns without raising is not proof that anything
+        # was destroyed, and this is the one place where believing it is catastrophic:
+        # `cleanup_task` reads an empty survivor list as "clean sweep" and REMOVES THE
+        # SIDECAR -- the only record the tree ever existed. ~250 snapshots would be
+        # orphaned per run, with nothing left to find them, and the backup green.
+        #
+        # Not paranoia about a hypothetical: iX has already gutted
+        # `pool.snapshot.do_update` on master into a no-op whose body is commented out
+        # and which returns None. An AST check still sees the `def`, and a callable
+        # check still sees the method. Only asking ZFS can tell.
+        #
+        # It costs one `zfs list` (~350ms against 2148 snapshots) on an 18-minute
+        # backup, and only on the path that would otherwise skip verification entirely.
+        try:
+            left = snapshot_tree_names(snapshot, snaps.names(dataset))
+        except Exception as e:  # noqa: BLE001 - cannot confirm; do not claim success
             if logger:
-            logger.debug(
+                logger.warning(
+                    "truecloud-patch: deleted %s but could not confirm it is gone "
+                    "(%r); keeping it recorded so the next run re-checks", snapshot, e,
+                )
+            # Keep OWNING it. Reporting a clean sweep here makes cleanup_task drop the
+            # sidecar; if the delete had in fact done nothing, the tree is orphaned
+            # with no record. A survivor we later find already gone costs one
+            # idempotent retry; a lost record costs the snapshots, permanently.
+            return [snapshot]
+
+        if not left:
+            return []
+
+        if logger:
+            logger.warning(
+                "truecloud-patch: the recursive delete of %s reported success but "
+                "%d snapshot(s) are still there; sweeping them by name",
+                snapshot, len(left),
+            )
+        # Fall through to the by-name sweep, which retries and reports survivors.
+    except Exception as e:  # noqa: BLE001 - fall through to the explicit sweep
+        # "Parent already gone" is the EXPECTED race (stock's finally won, once our
+        # mounts were released) and happens on every clean run, so it is debug.
+        # Anything else is a real fault -- a namespace that cannot delete, a schema
+        # change, a permission error -- and this is the only place it is visible,
+        # because the sweep below will report a different, downstream failure. At
+        # debug it would never reach disk on stock middlewared, which logs at INFO.
+        if logger:
+            expected = "does not exist" in str(e).lower()
+            (logger.debug if expected else logger.warning)(
                 "truecloud-patch: recursive delete of %s failed (%r); sweeping "
                 "the tree by name instead", snapshot, e,
             )
@@ -608,13 +1036,14 @@ def delete_snapshot_tree(middleware, snapshot, logger=None, attempts=4,
     # our mounts are released -- which fails the recursive delete while the
     # children survive. Sweep them by name.
     try:
-        snaps = middleware.call_sync(
-            "zfs.snapshot.query", [["name", "^", dataset]], {"select": ["name"]}
-        )
+        # From ZFS, not middleware: the snapshot query hides internal datasets'
+        # snapshots (205 of 274 on the test box), and a sweep that cannot see them
+        # orphans one per hidden dataset on every run. See list_snapshot_names().
+        #
         # An empty result means the tree is already gone -- delete nothing, and
         # do not fall back to the parent, which would only log a spurious
         # "does not exist" warning on every clean run.
-        names = snapshot_tree_names(snapshot, [s["name"] for s in snaps])
+        names = snapshot_tree_names(snapshot, snaps.names(dataset))
     except Exception as e:  # noqa: BLE001 - fall back to at least the parent
         if logger:
             logger.warning(
@@ -623,36 +1052,52 @@ def delete_snapshot_tree(middleware, snapshot, logger=None, attempts=4,
             )
         names = [snapshot]
 
-    def confirm_gone(failed):
-        """Drop any name ZFS no longer has, even though its delete raised.
+    def still_there(tried, failed):
+        """Which of `tried` does ZFS STILL have? The delete's verdict is not evidence.
 
-        A delete that raised "does not exist" SUCCEEDED as far as we care, and must
-        not be retried or reported. The query is only a refinement: if it cannot be
-        answered we keep the delete's own verdict, rather than inventing survivors --
-        a false survivor keeps the sidecar forever and is reported as a leak that
-        isn't there.
+        ZFS is the authority, not the API's return value, and the difference is not
+        academic in either direction:
+
+        * A delete that RAISED "does not exist" succeeded as far as we care, and must
+          not be retried or reported as a leak.
+        * A delete that RETURNED CLEANLY may have done nothing at all. iX has already
+          gutted `pool.snapshot.do_update` on master into a no-op whose body is
+          commented out and which returns None. Trusting that verdict makes
+          `cleanup_task` see "no survivors", drop the sidecar -- the only record -- and
+          orphan the whole tree, forever, silently.
+
+        If ZFS cannot be read we cannot check either way -- so keep owning ALL of them.
+        The two mistakes are not symmetric:
+
+        * a false survivor SELF-HEALS. The sidecar is kept, the next run reclaims it,
+          the delete raises "does not exist", and the record clears.
+        * a lost record does NOT. The snapshots are orphaned with nothing pointing at
+          them, and only the by-name collector -- an hour later, and only if ZFS is
+          readable by then -- has any chance of finding them.
         """
-        if not failed:
-            return []
         try:
-            live = middleware.call_sync(
-                "zfs.snapshot.query", [["name", "^", dataset]], {"select": ["name"]}
-            )
-        except Exception:  # noqa: BLE001 - cannot refine; trust the delete's verdict
-            return list(failed)
-        live = {s["name"] for s in live}
-        return [n for n in failed if n in live]
+            live = set(snaps.names(dataset))
+        except Exception:  # noqa: BLE001 - cannot check; keep owning them
+            return list(tried)
+        return [n for n in tried if n in live]
 
     remaining = list(names)
+    last_error = {}
     for attempt in range(attempts):
         failed = []
         for name in remaining:
             try:
-                middleware.call_sync("zfs.snapshot.delete", name)
-            except Exception:  # noqa: BLE001 - busy, or already gone; sorted out below
+                snaps.delete(name)
+            except Exception as e:  # noqa: BLE001 - busy, or already gone; sorted below
+                # KEEP the reason. This used to discard it and then report every
+                # survivor as "(still busy?)" -- which names the one cause that is
+                # benign and self-healing, and hides the ones that are permanent (a
+                # namespace that cannot delete, a permission error, a schema change).
+                # A misleading diagnosis is worse than none: it tells you to wait.
+                last_error[name] = e
                 failed.append(name)
 
-        remaining = confirm_gone(failed)
+        remaining = still_there(remaining, failed)
         if not remaining:
             return []
 
@@ -666,13 +1111,13 @@ def delete_snapshot_tree(middleware, snapshot, logger=None, attempts=4,
         if logger:
             logger.warning(
                 "truecloud-patch: could not delete snapshot %s after %d attempts "
-                "(still busy?) -- it will be reclaimed on the next run",
-                name, attempts,
+                "(last error: %r) -- it stays recorded and the next run reclaims it",
+                name, attempts, last_error.get(name),
             )
     return remaining
 
 
-def mounted_snapshots(mounts_file="/proc/self/mounts"):
+def mounted_snapshots(mounts_file=None):
     """Every ZFS snapshot something is currently mounted from.
 
     The device field of a snapshot mount IS the snapshot name (`Tap/apps/x@snap`), for
@@ -681,6 +1126,7 @@ def mounted_snapshots(mounts_file="/proc/self/mounts"):
     protects a concurrently-running backup from the garbage collector, rather than
     trusting an age heuristic to be generous enough.
     """
+    mounts_file = mounts_file or MOUNTS_FILE
     live = set()
     try:
         with open(mounts_file, encoding="utf-8") as fh:
@@ -689,12 +1135,17 @@ def mounted_snapshots(mounts_file="/proc/self/mounts"):
                 if "@" in dev:
                     live.add(dev.replace("\\040", " "))
     except OSError:
-        return set()
+        # An empty set says "nothing is mounted", which silently switches OFF the GC's
+        # protection for snapshots a CONCURRENT run is using -- leaving only the age
+        # floor between us and destroying a snapshot out from under a backup that is
+        # still uploading. If we cannot read the mount table we do not KNOW what is in
+        # use, and must not pretend we do.
+        raise
     return live
 
 
 def gc_stale_snapshots(middleware, task_name, current_snapshot, logger=None,
-                       now=None, mounts_file="/proc/self/mounts"):
+                       now=None, mounts_file=None, list_snapshots=None):
     """Delete snapshots this task left behind in an earlier run. Returns what remains.
 
     The backstop for when the RECORD is gone, not just the snapshots: the sidecar lives
@@ -705,13 +1156,15 @@ def gc_stale_snapshots(middleware, task_name, current_snapshot, logger=None,
     Selection is `stale_snapshot_names()`, which is pure and heavily tested, because a
     name match is a weaker claim than a recorded fact and this deletes data on one.
     """
+    mounts_file = mounts_file or MOUNTS_FILE
     dataset = current_snapshot.partition("@")[0]
     now = now or datetime.datetime.now(datetime.UTC)
+    snaps = _Snapshots(middleware, list_snapshots)
 
     try:
-        snaps = middleware.call_sync(
-            "zfs.snapshot.query", [["name", "^", dataset]], {"select": ["name"]}
-        )
+        # From ZFS: middleware's snapshot query hides internal datasets, and an
+        # orphan it cannot see is an orphan nothing will ever collect.
+        all_names = snaps.names(dataset)
     except Exception as e:  # noqa: BLE001 - cannot enumerate; collect nothing
         if logger:
             logger.warning(
@@ -720,7 +1173,7 @@ def gc_stale_snapshots(middleware, task_name, current_snapshot, logger=None,
         return []
 
     stale = stale_snapshot_names(
-        task_name, current_snapshot, [s["name"] for s in snaps], now,
+        task_name, current_snapshot, all_names, now,
         in_use=mounted_snapshots(mounts_file),
     )
     if not stale:
@@ -736,7 +1189,7 @@ def gc_stale_snapshots(middleware, task_name, current_snapshot, logger=None,
     remaining = []
     for name in stale:
         try:
-            middleware.call_sync("zfs.snapshot.delete", name)
+            snaps.delete(name)
         except Exception as e:  # noqa: BLE001 - busy, or gone; either way, next run
             remaining.append(name)
             if logger:
@@ -746,27 +1199,43 @@ def gc_stale_snapshots(middleware, task_name, current_snapshot, logger=None,
     return remaining
 
 
-def stage_nested(middleware, path, snapshot, base_dataset, base_mountpoint,
-                 task_name, datasets, logger=None):
-    """Build a complete staging tree for `path` from the already-taken `snapshot`.
+def own_snapshot(middleware, task_name, snapshot, logger=None, list_snapshots=None):
+    """Take ownership of `snapshot`'s whole tree: reclaim, collect, and record it.
 
-    `snapshot` is a full ZFS snapshot name ("Tap@cloud_backup-5-2026...").
+    Call this on EVERY ``snapshot = true`` cloud_backup run — **whether or not the
+    tree gets staged**. That unconditionality is the fix for a real leak, so do not
+    make it conditional again.
 
-    `datasets` is the FILESYSTEM dataset list. **It MUST have been enumerated
-    AFTER `snapshot` was taken.** A list read beforehand can miss a dataset
-    created in the gap: the recursive snapshot would capture it, but the staging
-    plan would not, and its data would be silently omitted from the backup.
-    Enumerated afterwards, an unsnapshotted dataset instead trips the isdir()
-    check in plan_staging and fails the run loudly.
+    Stock decides whether to take a RECURSIVE snapshot by its own rule, and that
+    rule is not ours:
 
-    Returns the staging root to hand to the backup tool.
+    ``<= 25.10``
+        stock's ``create_snapshot`` calls ``get_dataset_recursive()`` — the very
+        function this module vendors. "Stock went recursive" and "we have something
+        to stage" were therefore the *same question*, and a non-staged snapshot
+        provably had no children. Stock's non-recursive delete was correct.
 
-    Raises StagingError if the tree cannot be staged completely -- the caller
-    must let that propagate so the backup fails instead of silently uploading a
-    partial tree. The caller is responsible for deleting `snapshot` in that case
-    (see SNAPSHOT_BLOCK in apply.sh).
+    ``26``
+        stock uses ``filesystem.statfs``: ``recursive = (path == the dataset's
+        mountpoint)``. Now the two rules disagree. A dataset whose only descendants
+        are **ZVOLs** or **legacy/none-mountpoint** datasets gets a RECURSIVE
+        snapshot — while ``get_dataset_recursive()`` reports nothing to stage,
+        because neither kind is a mounted filesystem under ``path``.
+
+    In that gap stock takes one snapshot per descendant and then deletes only the
+    parent (its ``finally`` destroys ``path=snapshot``, non-recursively). Nothing
+    would ever have found the children: no staging tree, so no sidecar, and the GC
+    only ever ran from :func:`stage_nested`. One orphan per zvol/legacy descendant,
+    on every run, forever — while the backup reports SUCCESS. That is the exact
+    failure this module exists to prevent, reintroduced by a gate.
+
+    So ownership of the sweep is no longer conditional on staging. It is cheap:
+    :func:`delete_snapshot_tree` is idempotent, and on a genuinely childless
+    snapshot it is one recursive destroy of a snapshot stock has usually already
+    removed.
+
+    Returns the staging root; the sidecar sits beside it.
     """
-    snapshot_name = snapshot.split("@", 1)[1]
     staging_root = staging_root_for(task_name)
 
     # A previous run may have crashed mid-flight; never build on top of that.
@@ -790,7 +1259,8 @@ def stage_nested(middleware, path, snapshot, base_dataset, base_mountpoint,
                 "truecloud-patch: reclaiming snapshot tree from an earlier "
                 "run: %s", stale,
             )
-        pending.extend(delete_snapshot_tree(middleware, stale, logger=logger))
+        pending.extend(delete_snapshot_tree(
+            middleware, stale, logger=logger, list_snapshots=list_snapshots))
 
     if pending and logger:
         logger.warning(
@@ -806,16 +1276,54 @@ def stage_nested(middleware, path, snapshot, base_dataset, base_mountpoint,
     #
     # It runs AFTER the sidecar reclaim on purpose: the recorded path is authoritative
     # and cheap, and the GC should only ever be mopping up what the record lost.
-    pending.extend(
-        gc_stale_snapshots(middleware, task_name, snapshot, logger=logger)
-    )
+    pending.extend(gc_stale_snapshots(
+        middleware, task_name, snapshot, logger=logger,
+        list_snapshots=list_snapshots,
+    ))
 
     # Record the snapshot BEFORE mounting anything, not after. middlewared can
     # die at any point (this patch even schedules a restart at boot), and the
     # sidecar is the only thing that survives it -- an in-process dict would take
     # the sole record of a 160-snapshot tree with it. Writing it after apply_plan
     # would leave exactly the crash window the sidecar exists to close.
-    _write_sidecar(staging_root, [*pending, snapshot])
+    _write_sidecar(staging_root, [*pending, snapshot], logger=logger)
+    return staging_root
+
+
+def stage_nested(middleware, path, snapshot, base_dataset, base_mountpoint,
+                 task_name, datasets, logger=None, list_snapshots=None):
+    """Build a complete staging tree for `path` from the already-taken `snapshot`.
+
+    `snapshot` is a full ZFS snapshot name ("Tap@cloud_backup-5-2026...").
+
+    `datasets` is the FILESYSTEM dataset list. **It MUST have been enumerated
+    AFTER `snapshot` was taken.** A list read beforehand can miss a dataset
+    created in the gap: the recursive snapshot would capture it, but the staging
+    plan would not, and its data would be silently omitted from the backup.
+    Enumerated afterwards, an unsnapshotted dataset instead trips the isdir()
+    check in plan_staging and fails the run loudly.
+
+    Returns the staging root to hand to the backup tool.
+
+    Raises StagingError if the tree cannot be staged completely -- the caller
+    must let that propagate so the backup fails instead of silently uploading a
+    partial tree. The caller is responsible for deleting `snapshot` in that case
+    (see SNAPSHOT_BLOCK in apply.sh).
+    """
+    snapshot_name = snapshot.split("@", 1)[1]
+
+    # Refuse BEFORE staging, not after restic has run. We are about to pin a recursive
+    # snapshot with bind mounts; if this middleware has no usable snapshot delete we
+    # could never sweep it, and the honest move is to fail now rather than take a
+    # snapshot we cannot clean up. (`_Snapshots` resolves lazily on purpose -- the
+    # read-only paths must not raise over a mutation they never make -- so the staging
+    # path asks explicitly.)
+    snapshot_service(middleware)
+
+    staging_root = own_snapshot(
+        middleware, task_name, snapshot, logger=logger,
+        list_snapshots=list_snapshots,
+    )
 
     try:
         mounts, skipped = plan_staging(
@@ -853,7 +1361,7 @@ def stage_nested(middleware, path, snapshot, base_dataset, base_mountpoint,
     return staging_root
 
 
-def cleanup_task(middleware, task_name, logger=None):
+def cleanup_task(middleware, task_name, logger=None, list_snapshots=None):
     """Tear down a task's staging tree and delete the snapshot it pinned.
 
     Safe to call unconditionally: a no-op when the task was never staged.
@@ -877,7 +1385,8 @@ def cleanup_task(middleware, task_name, logger=None):
     # finish reclaiming.
     survivors = []
     for snapshot in pinned:
-        survivors.extend(delete_snapshot_tree(middleware, snapshot, logger=logger))
+        survivors.extend(delete_snapshot_tree(
+            middleware, snapshot, logger=logger, list_snapshots=list_snapshots))
 
     # KEEP the sidecar if anything survived. It is the only record that those
     # snapshots exist, and removing it orphans them permanently.
@@ -900,7 +1409,7 @@ def cleanup_task(middleware, task_name, logger=None):
             )
         # The SURVIVORS, not the trees we asked to delete. Writing the original list
         # back would keep re-sweeping trees that are already gone.
-        _write_sidecar(staging_root, survivors)
+        _write_sidecar(staging_root, survivors, logger=logger)
         return
 
     _remove_sidecar(staging_root)
@@ -909,7 +1418,7 @@ def cleanup_task(middleware, task_name, logger=None):
 # ── offline cleanup (uninstall.sh / recover.sh) ───────────────────────────────
 
 
-def cleanup_all(base=None, runner=_run, mounts_file="/proc/self/mounts",
+def cleanup_all(base=None, runner=None, mounts_file=None,
                 glob_fn=None, read_sidecar=_read_sidecar):
     """Tear down every staging tree. Used by uninstall.sh and recover.sh.
 
@@ -920,6 +1429,8 @@ def cleanup_all(base=None, runner=_run, mounts_file="/proc/self/mounts",
 
     Returns ``(lines, errors)``: report lines to print, and unmount errors.
     """
+    mounts_file = mounts_file or MOUNTS_FILE
+    runner = runner or _run
     import glob as _glob
 
     base = base or STAGING_BASE
@@ -929,8 +1440,21 @@ def cleanup_all(base=None, runner=_run, mounts_file="/proc/self/mounts",
     # Report orphaned snapshots BEFORE removing the sidecars that name them --
     # a sidecar is the only record that an interrupted run's snapshot tree (one
     # snapshot per descendant dataset) is still on disk.
+    unreadable = set()
     for sc in sorted(glob_fn(os.path.join(base, "*.snapshot"))):
-        for snap in read_sidecar(sc[: -len(".snapshot")]):
+        try:
+            recorded = read_sidecar(sc[: -len(".snapshot")])
+        except OSError as e:
+            unreadable.add(sc)
+            # REPORT it and carry on. This function's job is to get the mounts off, and
+            # it is called precisely when the box is already in a bad state
+            # (recover.sh, uninstall.sh). Letting one unreadable sidecar abort the run
+            # would leave the staging tree mounted -- which pins the snapshots, which is
+            # the exact situation the caller is trying to escape.
+            lines.append(f"  WARNING: could not read the snapshot record {sc} ({e}).")
+            lines.append("           It may name snapshots nothing else can find.")
+            continue
+        for snap in recorded:
             lines.append(f"  NOTE: an interrupted backup left snapshot '{snap}' behind.")
             lines.append(f"        Remove it and its children:  zfs destroy -r '{snap}'")
 
@@ -946,6 +1470,12 @@ def cleanup_all(base=None, runner=_run, mounts_file="/proc/self/mounts",
 
     if not errors:
         for sc in glob_fn(os.path.join(base, "*.snapshot")):
+            if sc in unreadable:
+                # Do NOT delete a record we could not read. We have no idea what it
+                # names, and it may be the only thing that knows those snapshots exist.
+                # Removing it here would be the very bug the read guard was added for,
+                # committed by the cleanup path instead of the backup path.
+                continue
             with contextlib.suppress(OSError):
                 os.unlink(sc)
         with contextlib.suppress(OSError):

tests/test_apply_blocks.py

@@ -140,19 +140,60 @@ class TestSnapshotLeak:
     every path that creates one must also sweep the whole tree.
     """
 
-    def test_staging_failure_deletes_the_snapshot_tree(self):
-        # On a staging failure, sync.py's `snapshot, local_path = await
-        # create_snapshot(...)` never completes, so its local `snapshot` stays
-        # None and its finally deletes nothing. We must sweep it ourselves.
-        block = extract_blocks()["SNAPSHOT_ASYNC"]
-        assert "except Exception:" in block
-        assert "delete_snapshot_tree" in block
-        assert "raise" in block
+    # The behaviour these once asserted as substrings -- the sweep, the re-raise, the
+    # teardown in the finally -- is now asserted STRUCTURALLY, against the parsed
+    # block: see TestTheStagingFailurePathReallyReRaises and
+    # TestTheSyncBlockAlwaysTearsDown. As substring checks they were satisfied by
+    # COMMENTS ("a cleanup that raises...", "cleanup_task gets logger=None"), so
+    # deleting the actual `raise` and the actual cleanup call both left the suite
+    # green -- reinstating a silently-empty backup and ~250 orphans per run.
 
-    def test_sync_block_cleans_up_on_every_path(self):
-        block = extract_blocks()["SYNC_ASYNC"]
-        assert "finally:" in block
-        assert "cleanup_task" in block
+    def test_the_snapshot_block_still_owns_the_snapshot_when_not_staging(self):
+        # The TrueNAS 26 zvol/legacy orphan: stock decides `recursive` by its own rule
+        # (path == mountpoint) and deletes only the parent, so we must record the
+        # snapshot even on the path where we stage nothing.
+        for name in ("SNAPSHOT_ASYNC", "SNAPSHOT_SYNC"):
+            stage = functions(tree_of(name), "_tc_stage")[0]
+            assert calls_to(stage, "_tc_nested.own_snapshot"), (
+                f"{name} hands an unstaged snapshot back to stock, whose delete is "
+                f"non-recursive -- every zvol/legacy child is orphaned, every run"
+            )
+
+    def test_the_staging_plan_is_enumerated_from_ZFS(self):
+        for name in ("SNAPSHOT_ASYNC", "SNAPSHOT_SYNC"):
+            stage = functions(tree_of(name), "_tc_stage")[0]
+            assert calls_to(stage, "_tc_nested.query_filesystems"), (
+                "the staging plan must come from query_filesystems() (which reads ZFS "
+                "unfiltered); middleware's query hides ix-apps/*, .system/*, .ix-virt/*"
+            )
+            assert not calls_to(stage, "middleware.call_sync"), (
+                "the block calls middleware directly again -- its dataset/snapshot "
+                "queries are FILTERED and silently omit 84 of 270 datasets"
+            )
+
+    def test_the_vendored_helper_is_used_not_the_host_module(self):
+        # TrueNAS 26 DELETED get_dataset_recursive from plugins/cloud/snapshot.py, so
+        # calling it out of the host module's namespace is a NameError there.
+        for name in ("SNAPSHOT_ASYNC", "SNAPSHOT_SYNC"):
+            stage = functions(tree_of(name), "_tc_stage")[0]
+            assert calls_to(stage, "_tc_nested.get_dataset_recursive"), (
+                "must call OUR vendored copy: TrueNAS 26 deleted the host's"
+            )
+
+    def test_datasets_are_enumerated_AFTER_the_snapshot(self):
+        # A dataset created between the listing and the snapshot would be captured by
+        # the recursive snapshot but missing from the staging plan -- silently omitted.
+        # Read afterwards, it instead trips plan_staging's probe and fails loudly.
+        for name in ("SNAPSHOT_ASYNC", "SNAPSHOT_SYNC"):
+            src = extract_blocks()[name]
+            code = "\n".join(
+                ln for ln in src.splitlines() if not ln.lstrip().startswith("#")
+            )
+            # _tc_stage receives `snapshot` as a parameter -- i.e. it is taken by the
+            # caller, before any of this runs. If the enumeration ever moves ahead of
+            # create_snapshot it can only do so by leaving _tc_stage.
+            assert "def _tc_stage(middleware, path, name, snapshot, snap_path)" in code
+            assert "query_filesystems" in code
 
 
 def test_crud_block_is_scoped_to_cloud_backup():
@@ -438,8 +479,275 @@ class TestOnlyOurOwnTasksAreTouched:
         # The point is to add NO new failure mode to a CloudSync task. If any
         # middleware call happened before the bail-out, we would already have broken
         # the thing we are trying not to touch.
+        #
+        # Checked against whichever interactions the block ACTUALLY contains, not a
+        # fixed list: the dataset query moved behind `_tc_nested.query_filesystems()`
+        # when it switched to the public pool.* API, and a hardcoded
+        # `middleware.call_sync(` simply stopped being found -- a test that silently
+        # stops testing is worse than no test.
         block = extract_blocks()[name]
         gate = block.index('if not name.startswith("cloud_backup"):')
-        for call in ("middleware.call_sync(", "_tc_nested.stage_nested(",
-                     "_tc_nested.delete_snapshot_tree("):
+
+        interactions = [
+            "middleware.call_sync(",
+            "_tc_nested.query_filesystems(",
+            "_tc_nested.stage_nested(",
+            "_tc_nested.delete_snapshot_tree(",
+        ]
+        present = [c for c in interactions if c in block]
+        assert present, "found no middleware interaction at all -- the test is vacuous"
+        for call in present:
             assert gate < block.index(call), f"{call} runs before the cloud_backup gate"
+
+
+# ── structural assertions ────────────────────────────────────────────────────
+#
+# `assert "raise" in block` was TRUE because a COMMENT in the block says "a cleanup
+# that raises would replace the original exception". `assert "cleanup_task" in block`
+# was TRUE because a comment says "cleanup_task gets logger=None". Deleting the actual
+# `raise`, and deleting the actual cleanup call from the `finally`, both left the suite
+# green -- while reinstating, respectively, a silently-empty backup and ~250 orphaned
+# snapshots per run.
+#
+# A test that a comment can satisfy is not a test. These parse the block and assert on
+# the CODE.
+
+def tree_of(name):
+    return ast.parse(textwrap.dedent(extract_blocks()[name]))
+
+
+def functions(tree, name):
+    return [
+        n for n in ast.walk(tree)
+        if isinstance(n, ast.FunctionDef | ast.AsyncFunctionDef) and n.name == name
+    ]
+
+
+def calls_to(node, dotted):
+    """Every Call in `node` whose callee renders as `dotted` (e.g. a.b.c)."""
+    out = []
+    for n in ast.walk(node):
+        if isinstance(n, ast.Call):
+            try:
+                if ast.unparse(n.func) == dotted:
+                    out.append(n)
+            except Exception:                                    # noqa: BLE001
+                pass
+    return out
+
+
+class TestTheStagingFailurePathReallyReRaises:
+    """If staging fails and we swallow it, restic backs up the UN-STAGED path.
+
+    That is the silently-empty backup this entire module exists to prevent: stock
+    points the tool at the parent's `.zfs/snapshot/`, where child datasets are
+    invisible. The exception MUST propagate.
+    """
+
+    @pytest.mark.parametrize("name", ["SNAPSHOT_ASYNC", "SNAPSHOT_SYNC"])
+    def test_the_handler_sweeps_the_snapshot_and_re_raises(self, name):
+        stage = functions(tree_of(name), "_tc_stage")
+        assert stage, "_tc_stage is gone"
+
+        handlers = [
+            h for t in ast.walk(stage[0]) if isinstance(t, ast.Try)
+            for h in t.handlers
+        ]
+        assert handlers, "the staging failure handler is gone"
+
+        sweeps = any(calls_to(h, "_tc_nested.delete_snapshot_tree") for h in handlers)
+        assert sweeps, (
+            "a staging failure no longer sweeps the snapshot. sync.py's `snapshot` "
+            "local stays None, so ITS finally deletes nothing -- the whole tree leaks "
+            "on every failed run."
+        )
+
+        # A bare `raise` directly in the handler body -- not one nested inside the
+        # defensive try/except that wraps the sweep.
+        reraises = any(
+            any(isinstance(s, ast.Raise) and s.exc is None for s in h.body)
+            for h in handlers
+        )
+        assert reraises, (
+            "the staging failure is SWALLOWED. restic then runs against the un-staged "
+            "path and uploads a near-empty tree, reporting SUCCESS."
+        )
+
+
+class TestTheSyncBlockAlwaysTearsDown:
+    """The teardown is what unmounts the staging tree and sweeps the snapshot.
+
+    It must run on EVERY exit from restic_backup -- success, failure, or exception --
+    or the bind mounts pin the snapshot and the tree is orphaned.
+    """
+
+    @pytest.mark.parametrize("name", ["SYNC_ASYNC", "SYNC_SYNC"])
+    def test_cleanup_runs_in_a_finally(self, name):
+        fns = functions(tree_of(name), "restic_backup")
+        assert fns, "the restic_backup wrapper is gone"
+
+        tries = [t for t in ast.walk(fns[0]) if isinstance(t, ast.Try) and t.finalbody]
+        assert tries, "restic_backup no longer has a try/finally"
+
+        cleans = any(
+            "cleanup_task" in ast.unparse(stmt)
+            for t in tries for stmt in t.finalbody
+        )
+        assert cleans, (
+            "cleanup_task is not called in the finally. The staging tree is never torn "
+            "down, its bind mounts pin the snapshot, and ~250 snapshots leak per run."
+        )
+
+
+class TestTheBlockingWorkNeverRunsOnTheEventLoop:
+    """`zfs list` and `call_sync` are BLOCKING. On <=25.10 these blocks are async.
+
+    Running them directly on middlewared's event loop stalls the whole daemon.
+    """
+
+    @pytest.mark.parametrize("name,fn", [
+        ("SNAPSHOT_ASYNC", "create_snapshot"),
+        ("SYNC_ASYNC", "restic_backup"),
+    ])
+    def test_the_async_flavour_hops_to_a_thread(self, name, fn):
+        fns = functions(tree_of(name), fn)
+        assert fns and isinstance(fns[0], ast.AsyncFunctionDef)
+        assert calls_to(fns[0], "middleware.run_in_thread"), (
+            f"{name}.{fn} does the blocking work on the asyncio event loop"
+        )
+
+    @pytest.mark.parametrize("name,fn", [
+        ("SNAPSHOT_SYNC", "create_snapshot"),
+        ("SYNC_SYNC", "restic_backup"),
+    ])
+    def test_the_sync_flavour_does_not(self, name, fn):
+        # On 26 stock already runs this in the thread pool; hopping again would be
+        # wrong (and there is no event loop to protect).
+        fns = functions(tree_of(name), fn)
+        assert fns and isinstance(fns[0], ast.FunctionDef)
+        assert not calls_to(fns[0], "middleware.run_in_thread")
+
+
+def test_the_flavour_mapping_is_not_inverted():
+    # `_snapshot_block = SNAPSHOT_ASYNC if _flavour else SNAPSHOT_SYNC` -- inverting it
+    # injects an async wrapper on 26 (a coroutine gets unpacked as a tuple) or a sync
+    # one on 25.10 (the event loop blocks). Every nested backup breaks, both ways.
+    with open(APPLY_SH, encoding="utf-8") as fh:
+        code = " ".join(
+            ln for ln in fh.read().splitlines() if not ln.lstrip().startswith("#")
+        )
+    code = re.sub(r"\s+", " ", code)          # the assignments are space-aligned
+    for block in ("SNAPSHOT", "CRUD", "SYNC"):
+        assert f"{block}_ASYNC if _flavour else {block}_SYNC" in code, (
+            f"the {block} flavour mapping is missing or inverted: _flavour is True for "
+            f"an ASYNC middleware, so it must select {block}_ASYNC"
+        )
+
+
+# ── the compat preflight ─────────────────────────────────────────────────────
+#
+# This is the guard that stands between a broken middleware and a live NAS: at every
+# boot, apply.sh checks the patch's assumptions against the middlewared actually
+# installed, and REFUSES to apply a module whose assumptions no longer hold.
+#
+# It had no test. An audit turned it into a no-op eight different ways -- `verdict()`
+# always returning 'ok', the broken branch never firing, the kill switch never honoured
+# -- and the suite stayed green every time. The most consequential safety net in the
+# project was unguarded.
+
+def preflight_heredoc():
+    """The preflight's Python, lifted out of apply.sh and made runnable.
+
+    Extracted, not reimplemented: a reimplementation would happily pass while the
+    SHIPPED preflight stayed broken, which is exactly the failure being guarded.
+    """
+    with open(APPLY_SH, encoding="utf-8") as fh:
+        sh = fh.read()
+    # Line-based: the compat heredoc opens with `<<'PYEOF'` on the _tc_compat line and
+    # closes at the next bare PYEOF. (A regex that matched `<< 'PYEOF'` silently found
+    # the OTHER heredoc and ran a different script entirely.)
+    lines = sh.splitlines()
+    start = next(
+        i for i, ln in enumerate(lines)
+        if ln.startswith("_tc_compat=$(") and "<<'PYEOF'" in ln
+    )
+    end = next(i for i in range(start + 1, len(lines)) if lines[i].strip() == "PYEOF")
+    m = "\n".join(lines[start + 1:end])
+    assert m, "could not find the compat preflight heredoc in apply.sh"
+    return m
+
+
+def run_preflight(result, tmp_path):
+    """Run the SHIPPED preflight against a fake compat.check_tree result.
+
+    The heredoc does `import sys`, so a fake `sys` in the namespace is immediately
+    rebound to the real module -- drive the real one instead.
+    """
+    import contextlib
+    import io
+    import sys
+    import types
+
+    src = preflight_heredoc()
+    fake = types.ModuleType("compat")
+    fake.check_tree = lambda _mw: result
+
+    saved_mod = sys.modules.get("compat")
+    saved_argv = sys.argv
+    sys.modules["compat"] = fake
+    sys.argv = ["x", "/patch", "/mw", str(tmp_path / "compat.json")]
+
+    buf = io.StringIO()
+    try:
+        with contextlib.redirect_stdout(buf):
+            exec(compile(src, "apply.sh:preflight", "exec"), {"__name__": "__main__"})  # noqa: S102
+    except SystemExit:
+        pass
+    finally:
+        sys.argv = saved_argv
+        if saved_mod is not None:
+            sys.modules["compat"] = saved_mod
+        else:
+            sys.modules.pop("compat", None)
+    return buf.getvalue().splitlines()
+
+
+def _mod(ok=True, native=False, unknown=False, problems=()):
+    return {"ok": ok, "native": native, "unknown": unknown, "problems": list(problems)}
+
+
+class TestTheBootPreflightRefusesABrokenMiddleware:
+    def test_a_healthy_tree_is_ok(self, tmp_path):
+        out = run_preflight({"providers": _mod(), "nested": _mod()}, tmp_path)
+        assert out[:2] == ["ok", "ok"]
+
+    def test_a_broken_module_is_reported_broken(self, tmp_path):
+        out = run_preflight({
+            "providers": _mod(),
+            "nested": _mod(ok=False, problems=[
+                {"id": "x", "detail": "gone", "why": "orphans every run"},
+            ]),
+        }, tmp_path)
+        assert "broken" in out, (
+            "the preflight did not report a module whose assumptions FAILED. It would "
+            "be injected into a middleware it does not fit -- broken backups, "
+            "discovered at restore time."
+        )
+
+    def test_a_module_that_went_NATIVE_is_also_not_applied(self, tmp_path):
+        # 'native' answers "do we still need it?", 'ok' answers "is it safe to inject?".
+        # Applying a module TrueNAS now implements itself is not safe either.
+        out = run_preflight({
+            "providers": _mod(),
+            "nested": _mod(ok=False, native=True),
+        }, tmp_path)
+        assert "broken" in out
+
+    def test_an_UNKNOWN_verdict_is_not_reported_as_broken(self, tmp_path):
+        # A network error or an unreadable file is not iX deleting our symbols. Calling
+        # it broken would switch a working module off on a healthy box.
+        out = run_preflight({
+            "providers": _mod(unknown=True),
+            "nested": _mod(unknown=True),
+        }, tmp_path)
+        assert "broken" not in out

tests/test_compat.py

@@ -51,24 +51,44 @@ GOOD = {
         "async def restic_backup(middleware, job, cloud_backup, dry_run=False, "
         "rate_limit=None):\n    pass\n"
     ),
-    # The middlewared METHODS the injected code calls. TrueNAS 26 deleted both of
-    # these files, taking zfs.dataset.query / zfs.snapshot.query / zfs.snapshot.delete
-    # with them -- see TestMiddlewareMethodsWeCall.
-    "plugins/zfs_/dataset.py": (
-        "class ZFSDataset(CRUDService):\n"
+    # The middlewared METHODS the injected code calls. These now go through the
+    # PUBLIC pool.* API: TrueNAS 26 deleted plugins/zfs_/ outright, taking the whole
+    # private zfs.* service with it -- see TestMiddlewareMethodsWeCall.
+    #
+    # This default tree is a MODERN box (25.10/26): it has pool.snapshot and no
+    # zfs.snapshot. The older shape is built explicitly where it is tested.
+    # Not a plugin: a method on the middleware OBJECT. `snapshot_service()` resolves
+    # the snapshot namespace through it, so if it vanishes the module cannot sweep the
+    # snapshot it just took.
+    "utils/plugins.py": (
+        "class LoadPluginsMixin:\n"
+        "    def get_service(self, name):\n        pass\n"
+    ),
+    "plugins/pool_/dataset.py": (
+        "class PoolDatasetService(CRUDService):\n"
         "    class Config:\n"
-        "        namespace = 'zfs.dataset'\n"
+        "        namespace = 'pool.dataset'\n"
         "    def query(self, filters, options):\n        pass\n"
     ),
-    "plugins/zfs_/snapshot.py": (
-        "class ZFSSnapshot(CRUDService):\n"
+    "plugins/pool_/snapshot.py": (
+        "class PoolSnapshotService(CRUDService):\n"
         "    class Config:\n"
-        "        namespace = 'zfs.snapshot'\n"
+        "        namespace = 'pool.snapshot'\n"
         "    def query(self, filters, options):\n        pass\n"
         "    def delete(self, id_, options={}):\n        pass\n"
     ),
 }
 
+#: A 24.10/25.04 box: `pool.snapshot` does not exist yet and the snapshot CRUD
+#: service still answers to the (then-public) `zfs.snapshot`.
+ZFS_ERA_SNAPSHOT = (
+    "class ZFSSnapshot(CRUDService):\n"
+    "    class Config:\n"
+    "        namespace = 'zfs.snapshot'\n"
+    "    def query(self, filters, options):\n        pass\n"
+    "    def do_delete(self, id_, options={}):\n        pass\n"
+)
+
 
 def loader(files):
     def load(path):
@@ -284,9 +304,23 @@ class TestAsyncFlavour:
         broken["plugins/cloud_backup/sync.py"] = Unreadable("HTTP 429")
         assert compat.async_flavour(loader(broken)) is None
 
-    def test_the_real_truenas_versions(self):
-        # Pinning the actual fact this whole port exists for.
-        assert compat.async_flavour(loader(GOOD)) is True
+    def test_it_reads_STOCK_source_not_our_own_injected_block(self):
+        # This was a byte-identical copy of test_async_middleware_is_detected under a
+        # name that promised more. The fact worth pinning: apply.sh re-runs on an
+        # ALREADY-PATCHED overlay, so the probe must cut our block off first -- our own
+        # SNAPSHOT_SYNC wrapper is a plain `def create_snapshot`, and reading it would
+        # report a 25.10 box as synchronous and inject the wrong flavour.
+        patched = dict(GOOD)
+        patched["plugins/cloud/snapshot.py"] = (
+            GOOD["plugins/cloud/snapshot.py"]
+            + "\n# TRUECLOUD_PATCH\n"
+            + 'def create_snapshot(middleware, path, name="x"):\n    return "s", "p"\n'
+        )
+        assert compat.async_flavour(loader(patched)) is True, (
+            "the flavour probe read our own injected block and concluded the box is "
+            "synchronous -- it would then inject a sync wrapper into an async "
+            "middleware, and every nested backup would break"
+        )
 
 
 class TestMiddlewareMethodsWeCall:
@@ -303,52 +337,86 @@ class TestMiddlewareMethodsWeCall:
     per descendant dataset (250 on a real pool) on every run, forever.
     """
 
-    ZFS_SNAPSHOT = (
-        "class ZFSSnapshot(CRUDService):\n"
-        "    class Config:\n"
-        "        namespace = 'zfs.snapshot'\n"
-        "    def query(self, filters, options):\n        pass\n"
-        "    def delete(self, id_, options={}):\n        pass\n"
-    )
-    ZFS_DATASET = (
-        "class ZFSDataset(CRUDService):\n"
-        "    class Config:\n"
-        "        namespace = 'zfs.dataset'\n"
-        "    def query(self, filters, options):\n        pass\n"
-    )
+    POOL_SNAPSHOT = GOOD["plugins/pool_/snapshot.py"]
 
     def _tree(self, **over):
         files = dict(GOOD)
-        files["plugins/zfs_/snapshot.py"] = self.ZFS_SNAPSHOT
-        files["plugins/zfs_/dataset.py"] = self.ZFS_DATASET
         files.update(over)
         return files
 
+    #: A 26 box: pool.snapshot only.
+    def _modern(self, **over):
+        return self._tree(**over)
+
+    #: A 24.10/25.04 box: zfs.snapshot only -- plugins/pool_/snapshot.py does not
+    #: exist yet.
+    def _zfs_era(self, **over):
+        return self._tree(**{
+            "plugins/pool_/snapshot.py": None,
+            "plugins/zfs_/snapshot.py": ZFS_ERA_SNAPSHOT,
+            **over,
+        })
+
     def test_present_methods_are_ok(self):
-        r = check_files(self._tree())
+        r = check_files(self._modern())
         assert r[NESTED]["ok"], r[NESTED]["problems"]
 
-    def test_a_deleted_plugin_file_is_broken(self):
-        # Literally TrueNAS 26: plugins/zfs_/snapshot.py does not exist.
-        r = check_files(self._tree(**{"plugins/zfs_/snapshot.py": None}))
+    def test_the_OLD_zfs_era_snapshot_service_also_satisfies_the_call(self):
+        # 24.10 and 25.04 have no `pool.snapshot` at all -- the CRUD service is the
+        # then-public `zfs.snapshot`. Pinning only the modern spelling marked both of
+        # those releases BROKEN and would have switched nested snapshots OFF on boxes
+        # where they work perfectly. The runtime picks the same way; see
+        # pick_snapshot_service().
+        r = check_files(self._zfs_era())
+        assert r[NESTED]["ok"], r[NESTED]["problems"]
+
+    def test_it_is_broken_only_when_NEITHER_namespace_exists(self):
+        # The real failure: middleware drops the last spelling we know how to call.
+        r = check_files(self._tree(**{
+            "plugins/pool_/snapshot.py": None,
+            "plugins/zfs_/snapshot.py": None,
+        }))
         assert is_broken(r[NESTED])
         details = " ".join(p["detail"] for p in r[NESTED]["problems"])
-        assert "zfs.snapshot.delete" in details
+        assert "pool.snapshot.delete" in details
+        assert "zfs.snapshot.delete" in details, (
+            "the report must say BOTH spellings were tried, or whoever reads it will "
+            "think we simply never looked for the one their box has"
+        )
+
+    def test_we_do_NOT_depend_on_a_middleware_dataset_query_at_all(self):
+        # iX could delete plugins/pool_/dataset.py tomorrow and the patch would not
+        # care, because the staging plan is enumerated from ZFS, not from middleware.
+        #
+        # That is deliberate, and it was expensive to learn. `pool.dataset.query`
+        # exists and is correctly shaped -- and it LIES: it applies a visibility
+        # policy that hides ix-apps/*, .system/* and .ix-virt/* (84 of 270 datasets
+        # on the real pool, including live app data). No source check could ever
+        # have caught that; only running it could. So there is no assumption here
+        # left to break.
+        r = check_files(self._modern(**{"plugins/pool_/dataset.py": None}))
+        assert r[NESTED]["ok"], r[NESTED]["problems"]
+
+        ids = {c.id for c in compat.MIDDLEWARE_CALLS}
+        assert not any("dataset" in i or "query" in i for i in ids), (
+            "a dataset/snapshot QUERY assumption crept back into the manifest -- "
+            "middleware's queries are filtered; enumerate from ZFS"
+        )
 
     def test_a_renamed_namespace_is_broken(self):
         r = check_files(self._tree(**{
-            "plugins/zfs_/snapshot.py": self.ZFS_SNAPSHOT.replace(
-                "'zfs.snapshot'", "'zfs.resource.snapshot'"),
+            "plugins/pool_/snapshot.py": self.POOL_SNAPSHOT.replace(
+                "'pool.snapshot'", "'zfs.resource.snapshot'"),
+            "plugins/zfs_/snapshot.py": None,
         }))
         assert is_broken(r[NESTED])
 
     def test_the_CRUDService_do_prefix_is_accepted(self):
-        # 24.10 and 25.04 declare `do_delete`; 25.10 renamed it to `delete`. BOTH
-        # answer to zfs.snapshot.delete. Accepting only the literal name reported the
-        # two older releases as broken -- a false BROKEN that would have switched off
-        # nested snapshots on boxes where they work perfectly.
-        r = check_files(self._tree(**{
-            "plugins/zfs_/snapshot.py": self.ZFS_SNAPSHOT.replace(
+        # A CRUDService exposes `delete` from a method NAMED `do_delete`. Both
+        # spellings are live across the matrix. Accepting only the literal name
+        # reported working releases as broken.
+        r = check_files(self._modern(**{
+            "plugins/pool_/snapshot.py": self.POOL_SNAPSHOT.replace(
                 "def delete(", "def do_delete("),
         }))
         assert r[NESTED]["ok"], r[NESTED]["problems"]
@@ -356,6 +424,153 @@ class TestMiddlewareMethodsWeCall:
     def test_the_snapshot_delete_reason_names_the_orphan_risk(self):
         # If this ever regresses, whoever reads the bug report must understand that
         # it is not a cosmetic failure.
-        r = check_files(self._tree(**{"plugins/zfs_/snapshot.py": None}))
+        r = check_files(self._tree(**{
+            "plugins/pool_/snapshot.py": None,
+            "plugins/zfs_/snapshot.py": None,
+        }))
         whys = " ".join(p["why"] for p in r[NESTED]["problems"])
         assert "orphan" in whys
+
+
+class TestTheMethodCheckIsNotJustANamespaceCheck:
+    """compat must verify the METHOD, not merely that the namespace still exists.
+
+    Deleting the method check entirely used to leave all 304 tests green -- so the
+    "namespace AND method" claim was unenforced and silently revertible. It is the
+    half of the predicate that catches iX gutting a method while keeping its service,
+    which they have already done to `pool.snapshot.do_update` on master.
+    """
+
+    def test_a_namespace_that_no_longer_defines_delete_is_broken(self):
+        gutted = (
+            "class PoolSnapshotService(CRUDService):\n"
+            "    class Config:\n"
+            "        namespace = 'pool.snapshot'\n"
+            "    def query(self, filters, options):\n        pass\n"
+            # do_delete is GONE -- the service is still registered and still a
+            # CRUDService, so it still INHERITS a callable `delete`.
+        )
+        r = check_files(with_(**{
+            "plugins/pool_/snapshot.py": gutted,
+            "plugins/zfs_/snapshot.py": None,      # no fallback either
+        }))
+        assert is_broken(r[NESTED]), (
+            "a namespace with no delete must be BROKEN. Checking only that the "
+            "namespace exists would apply the patch to a box that cannot sweep its "
+            "own snapshots."
+        )
+
+    def test_the_alternative_still_saves_it_when_only_the_primary_is_gutted(self):
+        gutted = (
+            "class PoolSnapshotService(CRUDService):\n"
+            "    class Config:\n"
+            "        namespace = 'pool.snapshot'\n"
+            "    def query(self, filters, options):\n        pass\n"
+        )
+        r = check_files(with_(**{
+            "plugins/pool_/snapshot.py": gutted,
+            "plugins/zfs_/snapshot.py": ZFS_ERA_SNAPSHOT,
+        }))
+        assert r[NESTED]["ok"], r[NESTED]["problems"]
+
+
+class TestUnreadableIsNeverOkAndNeverBroken:
+    """A rate limit is not a regression, and it is not a clean bill of health either.
+
+    compat runs ~30 unauthenticated GitHub requests per matrix; 429 is a real outcome.
+    It also runs at BOOT against the installed tree, where a read can fail with EACCES.
+
+      * treating unreadable as BROKEN repaints the README, files a bug report, and
+        makes apply.sh refuse the module on a box where it works.
+      * treating it as OK injects a module whose delete may be gone.
+
+    Both mutations used to pass the whole suite.
+    """
+
+    def test_both_spellings_unreadable_is_unknown_not_broken(self):
+        r = check_files(with_(**{
+            "plugins/pool_/snapshot.py": Unreadable("HTTP 429"),
+            "plugins/zfs_/snapshot.py": Unreadable("HTTP 429"),
+        }))
+        assert not is_broken(r[NESTED]), "a 429 is not iX deleting the snapshot service"
+        assert r[NESTED]["unknown"]
+
+    def test_an_unreadable_primary_with_a_healthy_alternative_is_ok(self):
+        r = check_files(with_(**{
+            "plugins/pool_/snapshot.py": Unreadable("HTTP 429"),
+            "plugins/zfs_/snapshot.py": ZFS_ERA_SNAPSHOT,
+        }))
+        assert r[NESTED]["ok"], r[NESTED]["problems"]
+        assert not r[NESTED]["unknown"], (
+            "one spelling answered the question; the other's 429 is irrelevant"
+        )
+
+    def test_a_missing_primary_with_an_unreadable_alternative_is_unknown(self):
+        # We cannot tell whether the box is broken. Saying either would be a guess.
+        r = check_files(with_(**{
+            "plugins/pool_/snapshot.py": None,
+            "plugins/zfs_/snapshot.py": Unreadable("HTTP 429"),
+        }))
+        assert not is_broken(r[NESTED])
+        assert r[NESTED]["unknown"]
+
+
+class TestGetServiceIsChecked:
+    """The runtime resolves the snapshot namespace through `middleware.get_service`.
+
+    It is not a plugin method, so the manifest had no way to express it and never
+    checked it. If it vanishes, `_can_delete` reports BOTH namespaces unusable and
+    every nested backup fails -- on a box the preflight had declared healthy.
+    """
+
+    def test_a_middleware_without_get_service_is_broken(self):
+        r = check_files(with_(**{"utils/plugins.py": None}))
+        assert is_broken(r[NESTED])
+        details = " ".join(p["detail"] for p in r[NESTED]["problems"])
+        assert "get_service" in details
+
+
+class TestATransientNetworkBlipDoesNotWakeAnybody:
+    """The fingerprint must digest what iX BROKE, not what GitHub failed to serve.
+
+    `unknown` problems (a 429 on one of ~30 unauthenticated fetches, an EACCES at boot)
+    used to be folded into an already-broken module's problem list, so one blip flipped
+    the fingerprint, `compat_publish` rewrote the issue body, and the next clean run
+    rewrote it back. Daily churn is what teaches people to ignore the bot -- which is
+    the whole thing this fingerprint exists to prevent.
+    """
+
+    def _rows(self, files):
+        return [{"ref": "master", "modules": check_files(files)}]
+
+    def test_an_unreadable_file_does_not_change_the_fingerprint_of_a_broken_ref(self):
+        # The blip must land in the SAME module that is broken. Put it in `providers`
+        # (which is healthy) and `fingerprint()` skips the whole module via
+        # `is_broken(m)` -- so the `state` filter under test never runs and the test
+        # passes no matter what the code does. `nested` is the broken one here, so the
+        # unreadable file goes in `nested` too.
+        broken = with_(**{
+            "plugins/cloud/snapshot.py":
+                "async def create_snapshot(name, path, middleware):\n    return 1, 2\n",
+        })
+        clean = compat.fingerprint(self._rows(broken))
+
+        blipped = dict(broken)
+        blipped["plugins/cloud_backup/sync.py"] = Unreadable("HTTP 429")   # nested
+        assert compat.fingerprint(self._rows(blipped)) == clean, (
+            "a rate-limited fetch changed the fingerprint, so the bot rewrites the "
+            "issue body and then rewrites it back tomorrow"
+        )
+
+    def test_a_REAL_new_finding_still_changes_it(self):
+        # ...and the anti-noise measure must not have made it deaf.
+        broken = with_(**{
+            "plugins/cloud/snapshot.py":
+                "async def create_snapshot(name, path, middleware):\n    return 1, 2\n",
+        })
+        worse = dict(broken)
+        worse["plugins/cloud_backup/restic.py"] = (
+            "class ResticConfig:\n    cmd: list\n\n"
+            "def get_restic_config(entry, credentials):\n    pass\n"
+        )
+        assert compat.fingerprint(self._rows(worse)) != compat.fingerprint(self._rows(broken))

tools/compat.py

@@ -133,9 +133,43 @@ ASSUMPTIONS = [
         params=["middleware", "job", "cloud_backup"],
         why="SYNC_BLOCK wraps it to tear down bind mounts in a finally",
     ),
+    Assumption(
+        # Not a plugin method -- a method on the middleware OBJECT itself, which the
+        # manifest had no way to express and therefore never checked.
+        #
+        # The nested module calls `middleware.get_service(<ns>)` to decide whether to
+        # sweep snapshots through `pool.snapshot` or `zfs.snapshot` (see
+        # SNAPSHOT_SERVICES). If it ever disappears, `_can_delete()` catches the
+        # AttributeError, reports BOTH namespaces unusable, and every nested backup
+        # fails -- loudly, but only at RUN time, on a box the preflight had already
+        # declared healthy. Checking it costs one file read.
+        "get-service", NESTED, "utils/plugins.py",
+        "LoadPluginsMixin.get_service", kind="method",
+        params=["self", "name"],
+        why="snapshot_service() resolves the snapshot namespace through it; without "
+            "it the module cannot sweep the snapshot it just took",
+    ),
 ]
 
 
+def accepted_spellings(name):
+    """The method names that satisfy a call to `<namespace>.<name>`.
+
+    A CRUDService exposes `create`/`update`/`delete` from methods NAMED
+    `do_create`/`do_update`/`do_delete`. Both are live across the matrix: 24.10 and
+    25.04 declare `do_delete`, 25.10 renamed it to `delete`, and all of them answer
+    to `<ns>.delete`. Accepting only the literal name reported working releases as
+    BROKEN and would have switched nested snapshots off on boxes where they work.
+    """
+    return (name, f"do_{name}")
+
+
+#: The spellings that satisfy `<ns>.delete`. A test binds this to the runtime's
+#: `truecloud_nested.DELETE_METHODS`, so the checker and the patch cannot come to
+#: disagree about what "can delete" means on the same box.
+DELETE_NAMES = accepted_spellings("delete")
+
+
 class MiddlewareCall:
     """A middlewared METHOD the injected code calls at runtime.
 
@@ -162,87 +196,167 @@ class MiddlewareCall:
     the whole design: declining is always the cheaper mistake.
     """
 
-    def __init__(self, ident, module, method, path, why=""):
+    def __init__(self, ident, module, method, path, why="", also=()):
         self.id = ident
         self.module = module
-        self.method = method              # "zfs.snapshot.delete"
+        self.method = method              # "pool.snapshot.delete"
         self.path = path                  # plugin file that declares it
         self.why = why
+        #: Equally acceptable spellings of the SAME call, as (method, path) pairs.
+        #:
+        #: No single snapshot namespace spans every supported release. 24.10 and
+        #: 25.04 expose the CRUD service as the public `zfs.snapshot`; 25.10
+        #: promoted it to `pool.snapshot` and demoted `zfs.snapshot` to private;
+        #: 26 deleted `plugins/zfs_/` entirely. Pinning either one alone marks
+        #: half the matrix BROKEN and declines to apply on versions that work
+        #: perfectly well.
+        #:
+        #: The call is satisfied if ANY option is present. The runtime picks the
+        #: same way -- see `pick_snapshot_service()` in the nested module -- so
+        #: what this checks and what the patch does cannot drift apart.
+        self.also = tuple(also)
 
     @property
-    def namespace(self):
-        return self.method.rsplit(".", 1)[0]
+    def options(self):
+        """Every (method, path) that would satisfy this call, best first."""
+        return ((self.method, self.path), *self.also)
 
-    @property
-    def name(self):
-        return self.method.rsplit(".", 1)[1]
+    @staticmethod
+    def namespace_of(method):
+        return method.rsplit(".", 1)[0]
+
+    @staticmethod
+    def name_of(method):
+        return method.rsplit(".", 1)[1]
 
 
 #: Every middlewared method the nested module calls at runtime.
+#: The middleware methods the nested module CALLS.
+#:
+#: These used to be the PRIVATE `zfs.*` service (`zfs.dataset.query`,
+#: `zfs.snapshot.delete`, `zfs.snapshot.query`). TrueNAS 26 deleted
+#: `plugins/zfs_/` outright and every one of them vanished -- silently, because a
+#: private service carries no stability contract and nothing warned us. The patch
+#: would have applied cleanly and then failed on the first backup.
+#:
+#: The replacements are the PUBLIC `pool.*` API, and switching to it is not merely
+#: a TrueNAS 26 fix -- it is the correct call on every version:
+#:
+#:   * It is public, documented, and covered by iX's deprecation policy, so it
+#:     cannot be deleted from under us the way `zfs.*` just was.
+#:   * The same methods, in the same files, exist on 24.10 through 26. One code
+#:     path, no version conditionals.
+#:   * Both spellings take `recursive`, so ONE call sweeps the whole tree instead
+#:     of ~250 individual deletes, any of which could be missed.
 MIDDLEWARE_CALLS = [
     MiddlewareCall(
-        "call-zfs-dataset-query", NESTED, "zfs.dataset.query",
-        "plugins/zfs_/dataset.py",
-        why="SNAPSHOT_BLOCK enumerates FILESYSTEM datasets to build the staging plan",
-    ),
-    MiddlewareCall(
-        "call-zfs-snapshot-delete", NESTED, "zfs.snapshot.delete",
-        "plugins/zfs_/snapshot.py",
+        "call-snapshot-delete", NESTED, "pool.snapshot.delete",
+        "plugins/pool_/snapshot.py",
+        also=[("zfs.snapshot.delete", "plugins/zfs_/snapshot.py")],
         why="delete_snapshot_tree() sweeps the recursive snapshot. Without it every "
             "run orphans one snapshot per descendant dataset (250 on a real pool)",
     ),
-    MiddlewareCall(
-        "call-zfs-snapshot-query", NESTED, "zfs.snapshot.query",
-        "plugins/zfs_/snapshot.py",
-        why="delete_snapshot_tree()'s fallback sweep enumerates the tree by name",
-    ),
 ]
 
+# There is deliberately NO entry here for a dataset or snapshot QUERY.
+#
+# The patch used to call `zfs.dataset.query` / `zfs.snapshot.query` (private, and
+# deleted in TrueNAS 26). The obvious port was to the public `pool.dataset.query` /
+# `pool.snapshot.query` -- and that port was WRONG in a way no source check could
+# ever have caught, because the methods are all present and correctly shaped.
+#
+# They are simply filtered. On a real box they return 205 of 274 datasets and 205
+# of 274 snapshots, hiding `ix-apps/*`, `.system/*` and `.ix-virt/*` -- 84 of 270
+# on the production pool, including live application data. Staging from that view
+# silently omits them; sweeping from it orphans one snapshot per hidden dataset,
+# forever.
+#
+# So the module enumerates from ZFS itself and there is no middleware assumption
+# left to check. That is the point: the fewer things we assume about middleware,
+# the less there is for iX to break. Only the MUTATION is still a middleware call,
+# and that is the one entry above.
+
+
+def check_call(c: MiddlewareCall, src: str | None,
+               method: str | None = None, path: str | None = None,
+               ) -> tuple[str, str | None]:
+    """Is `method` still registered by middlewared?
+
+    `method`/`path` name WHICH spelling of the call is being tried -- a call may
+    have several equally acceptable ones (see MiddlewareCall.also). They default
+    to the preferred spelling.
+    """
+    method = method or c.method
+    path = path or c.path
+    namespace = MiddlewareCall.namespace_of(method)
+    name = MiddlewareCall.name_of(method)
 
-def check_call(c: MiddlewareCall, src: str | None) -> tuple[str, str | None]:
-    """Is `c.method` still registered by middlewared?"""
     if src is None:
         return "broken", (
-            f"{c.path} no longer exists, so `{c.method}` is gone"
+            f"{path} no longer exists, so `{method}` is gone"
         )
 
     try:
         tree = ast.parse(_stock(src))
     except SyntaxError as e:
-        return "unknown", f"{c.path} does not parse: {e}"
+        return "unknown", f"{path} does not parse: {e}"
 
-    # namespace = 'zfs.snapshot' on some Service class in this file...
-    namespaces = {
+    # Find the CLASS that declares this namespace, and look for the method THERE.
+    #
+    # Not anywhere in the file. `ast.walk` over the whole module made *any* function
+    # called `delete` satisfy the check -- one on an unrelated class, or even a nested
+    # local function inside `do_query`. That is a FALSE OK, and it breaks the one
+    # invariant this checker and the runtime share: `_defines_delete()` looks in
+    # `vars(klass)` for a PLUGIN class on the service's MRO. If iX gutted
+    # `PoolSnapshotService.do_delete` while some other class in the same file still had
+    # a `delete`, compat would say ok, apply.sh would patch, and the runtime would then
+    # correctly refuse `pool.snapshot`, fall through to a `zfs.snapshot` that does not
+    # exist on 26, and fail every nested backup on a box the preflight called healthy.
+    #
+    # Same question on both sides: does the class that OWNS this namespace define the
+    # method?
+    #
+    # A CRUDService exposes `create`/`update`/`delete` from methods NAMED
+    # `do_create`/`do_update`/`do_delete`. Both spellings are live: 24.10 and 25.04
+    # declare `do_delete`, 25.10 renamed it to `delete`, and all answer to
+    # `<ns>.delete`. Accepting only the literal name reported working releases as
+    # broken.
+    owners = []
+    all_namespaces = set()
+    for cls in (n for n in ast.walk(tree) if isinstance(n, ast.ClassDef)):
+        declared = {
             n.value.value
-        for n in ast.walk(tree)
+            for n in ast.walk(cls)
             if isinstance(n, ast.Assign)
             and isinstance(n.value, ast.Constant)
             and isinstance(n.value.value, str)
             and any(isinstance(t, ast.Name) and t.id == "namespace" for t in n.targets)
         }
-    if c.namespace not in namespaces:
+        all_namespaces |= declared
+        if namespace in declared:
+            owners.append(cls)
+
+    if not owners:
         return "broken", (
-            f"{c.path} no longer declares namespace {c.namespace!r} "
-            f"(found: {sorted(namespaces) or 'none'}), so `{c.method}` is gone"
+            f"{path} no longer declares namespace {namespace!r} "
+            f"(found: {sorted(all_namespaces) or 'none'}), so `{method}` is gone"
         )
 
-    # ...and it defines the method.
-    #
-    # A CRUDService exposes `create`/`update`/`delete` from methods NAMED
-    # `do_create`/`do_update`/`do_delete`. Both spellings are live right now:
-    # 24.10 and 25.04 declare `do_delete`, 25.10 renamed it to `delete`, and all
-    # three answer to `zfs.snapshot.delete`. Accepting only the literal name reported
-    # the two older releases as broken -- a false BROKEN that would have switched off
-    # nested snapshots on boxes where they work.
-    defined = {
-        n.name for n in ast.walk(tree)
-        if isinstance(n, ast.FunctionDef | ast.AsyncFunctionDef)
-    }
-    if c.name not in defined and f"do_{c.name}" not in defined:
-        return "broken", f"{c.path} no longer defines `{c.method}`"
-
+    wanted = accepted_spellings(name)
+    for cls in owners:
+        # Direct members of the class, not its nested scopes: a `def delete` inside
+        # another method is a local function, not a service method.
+        if any(
+            isinstance(n, ast.FunctionDef | ast.AsyncFunctionDef) and n.name in wanted
+            for n in cls.body
+        ):
             return "ok", None
 
+    return "broken", (
+        f"{path} still declares namespace {namespace!r}, but its class no longer "
+        f"defines `{'` or `'.join(wanted)}` -- so `{method}` is gone"
+    )
+
 
 #: Things that mean iX has done the job themselves and the module should RETIRE,
 #: not break. Absence of the nesting guard = nested snapshots went native.
@@ -404,7 +518,10 @@ def check_source(a: Assumption, src: str | None) -> tuple[str, str | None]:
     breaks a box that was working.
     """
     if src is None:
-        return "broken", f"{a.path} does not exist"
+        # Name the SYMBOL, not just the file. Whoever reads the bug report needs to
+        # know what the patch can no longer reach, and "utils/plugins.py does not
+        # exist" does not tell them that `get_service` is gone.
+        return "broken", f"{a.path} does not exist, so `{a.symbol}` is gone"
 
     try:
         tree = ast.parse(src)
@@ -539,6 +656,7 @@ def check(loader, modules=None) -> dict:
             out[a.module]["unknown"] = True
             out[a.module]["problems"].append({
                 "id": a.id, "detail": f"could not read {a.path}: {e}", "why": a.why,
+                "state": "unknown",
             })
             continue
 
@@ -546,37 +664,57 @@ def check(loader, modules=None) -> dict:
         if status == "broken":
             out[a.module]["ok"] = False
             out[a.module]["problems"].append({
-                "id": a.id, "detail": detail, "why": a.why,
+                "id": a.id, "detail": detail, "why": a.why, "state": "broken",
             })
         elif status == "unknown":
             out[a.module]["unknown"] = True
             out[a.module]["problems"].append({
-                "id": a.id, "detail": detail, "why": a.why,
+                "id": a.id, "detail": detail, "why": a.why, "state": "unknown",
             })
 
     # The methods the injected code CALLS, not just the symbols it wraps.
+    #
+    # A call may have several equally acceptable spellings, because no single
+    # snapshot namespace spans every supported release (24.10 has `zfs.snapshot`,
+    # 26 has only `pool.snapshot`). It is satisfied if ANY of them is present --
+    # exactly as the runtime resolves it -- and BROKEN only when they all vanish.
     for c in MIDDLEWARE_CALLS:
         if c.module not in out:
             continue
+
+        satisfied, unknown, details = False, False, []
+        for method, path in c.options:
             try:
-            text = src(c.path)
+                text = src(path)
             except Unreadable as e:
-            out[c.module]["unknown"] = True
-            out[c.module]["problems"].append({
-                "id": c.id, "detail": f"could not read {c.path}: {e}", "why": c.why,
-            })
+                unknown = True
+                details.append(f"could not read {path}: {e}")
                 continue
 
-        status, detail = check_call(c, text)
-        if status == "broken":
-            out[c.module]["ok"] = False
-            out[c.module]["problems"].append({
-                "id": c.id, "detail": detail, "why": c.why,
-            })
-        elif status == "unknown":
+            status, detail = check_call(c, text, method, path)
+            if status == "ok":
+                satisfied = True
+                break
+            if status == "unknown":
+                unknown = True
+            details.append(detail)
+
+        if satisfied:
+            continue
+
+        # Every spelling failed. If we could not READ one of them we do not know
+        # that it is broken -- a rate-limited fetch is not a regression.
+        if unknown:
             out[c.module]["unknown"] = True
             out[c.module]["problems"].append({
-                "id": c.id, "detail": detail, "why": c.why,
+                "id": c.id, "detail": "; ".join(details), "why": c.why,
+                "state": "unknown",
+            })
+        else:
+            out[c.module]["ok"] = False
+            out[c.module]["problems"].append({
+                "id": c.id, "detail": "; ".join(details), "why": c.why,
+                "state": "broken",
             })
 
     for module, (path, phrase, native_when_present) in NATIVE_PROBES.items():
@@ -804,6 +942,10 @@ def is_broken(r: dict) -> bool:
 #: a strictly weaker claim than "a restore worked". Add a row only after doing it.
 HARDWARE_VERIFIED = {
     "25.10.4": "nested + providers; 252-snapshot recursive backup of /mnt/Tap, 18m",
+    "26.0.0-BETA.1": (
+        "nested + providers; 274-snapshot recursive backup of a 292-dataset pool, "
+        "restored a 4-deep child dataset byte-identical"
+    ),
 }
 
 _LEGEND = """
@@ -914,6 +1056,12 @@ def fingerprint(rows: list[dict]) -> str:
         for mod, m in r["modules"].items()
         if is_broken(m)
         for p in m["problems"]
+        # `unknown` problems are things we could not READ (a 429, an EACCES), not
+        # things iX changed. On a ref that is broken for some other reason they would
+        # otherwise join the digest, so one transient network blip rewrites the issue
+        # body and the next clean run rewrites it back. That is the daily-noise
+        # failure this fingerprint exists to prevent, wearing a different hat.
+        if p.get("state", "broken") == "broken"
     )
     return hashlib.sha256(repr(findings).encode()).hexdigest()[:16]