fix(consistency): deep mode must verify storage, not the cache

`backend_consistency ?deep=true` re-reads every chunk and re-hashes it
to catch silent bit-rot, and records `blob_corrupted` (severity
`data_loss`) naming `backend.backend_type()`. It was reading through the
live backend — which for a remote backend includes `CachedBlobBackend`,
whose `get_blob_stream` returns the local cached file and never touches
the remote on a hit.

So the attribution was false in both directions: rot on S3 hidden by a
good cached copy, and rot in the cache reported against a healthy S3 —
the second sending an operator to the wrong layer entirely.

Surfaced by a real run: 2022 chunks, 321 ms shallow, 1.5 s deep. That is
0.74 ms per chunk for a full read plus BLAKE3, sequential, over S3 —
impossible, and explained by every chunk being cache-warm. A genuine
uncached sweep is tens of seconds.

Adds `BlobStorageBackend::uncached()`, defaulting to `None`.
`CachedBlobBackend` returns its inner; `Retry` and `Swappable` forward
so the unwrap reaches the cache through them. `Swappable` resolves via
`current()` rather than capturing a handle, because it sits OUTSIDE the
cache — a DI-time snapshot would keep pointing at pre-cutover storage
and audit the backend a migration just moved away from.

Only the cache is peeled. The cache stores plaintext and the content
hash is over plaintext, so unwrapping past the encryption decorator
would hand back ciphertext and fail every blob it checked.

**No change to normal reads.** `uncached()` is called in exactly one
place, and the unwrapped handle is used at exactly one call site
(`verify_bytes`). Enumeration, every other job, and every request path
still go through the cached stack.

`?storage=<entry>` was already correct — `build_entry_backend` has no
cache decorator — so this only fixes the live-backend path, which is the
one that was silently fast.

Also reports `verified` in the run extras, on every run including zero.
A deep run that verified nothing and one that verified everything were
otherwise indistinguishable in the outcome, which is what made a 1.5 s
"deep" sweep look plausible in the first place. Same lesson as
`orphans_covered` on the Azure fallback: a check that cannot report its
own coverage will eventually be believed when it should not be.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

src/application/ports/blob_storage_ports.rs

@@ -220,6 +220,34 @@ pub trait BlobStorageBackend: Send + Sync + 'static {
     /// return `None`; callers that need a local file must stream + spool.
     fn local_blob_path(&self, hash: &str) -> Option<PathBuf>;
 
+    /// The same storage with any read-through cache peeled off, or
+    /// `None` when this backend is not a cache.
+    ///
+    /// **For verification only** — normal reads must keep going through
+    /// the cache, which is the point of having one.
+    ///
+    /// A cache answers reads from its own copy, so re-hashing through
+    /// one checks the cache rather than storage: rot on the remote is
+    /// hidden by a good cached copy, and rot in the cache is blamed on a
+    /// healthy remote. The second is worse, because it sends an operator
+    /// to the wrong layer. `backend_consistency ?deep=true` is the only
+    /// caller.
+    ///
+    /// Peels **only** the cache. The cache sits outside the encryption
+    /// decorator and stores plaintext, while the content hash is over
+    /// plaintext, so unwrapping further would hand back ciphertext and
+    /// fail every blob it checked.
+    ///
+    /// Implement by returning the inner backend. Pass-through wrappers
+    /// (hot-swap, retry) forward to whatever they wrap, so the unwrap
+    /// still reaches the cache — and, for hot-swap, resolves through
+    /// `current()` so it survives a migration cutover rather than
+    /// pinning the pre-cutover storage. Everything else inherits the
+    /// `None` default and is used as-is.
+    fn uncached(&self) -> Option<std::sync::Arc<dyn BlobStorageBackend>> {
+        None
+    }
+
     /// How many chunk fetches the CDC reader may run concurrently when
     /// reassembling a file (`read_blob_stream`'s `buffered(N)` read-ahead).
     ///

src/infrastructure/services/backend_consistency_service.rs

@@ -352,11 +352,33 @@ impl RecoverableJobHandler for BackendConsistencyCheck {
         // actually verified.
         let deep = args.get_bool("deep");
 
+        // Verify through storage, never through a read-through cache.
+        //
+        // A cache answers from its own copy, so re-hashing through one
+        // checks the CACHE: rot on the remote is masked by a good cached
+        // copy, and rot in the cache is recorded as `blob_corrupted`
+        // against a healthy remote — sending an operator to the wrong
+        // layer. The finding names `backend.backend_type()`, so that
+        // attribution has to be true.
+        //
+        // Only the cache is peeled; the decryptor stays, because the
+        // cache holds plaintext and the content hash is over plaintext.
+        // `?storage=<entry>` already builds an uncached stack, so this
+        // only changes the live-backend path — which is the one that was
+        // silently fast.
+        let verify_backend = backend.uncached().unwrap_or_else(|| backend.clone());
+        // Counter, not just a flag: a deep run that verified nothing and
+        // a deep run that verified everything are otherwise
+        // indistinguishable in the outcome, which is exactly the
+        // ambiguity that made a 1.5s "deep" sweep over 2022 chunks look
+        // plausible.
+        let mut verified_count = 0u64;
         if deep {
             tracing::info!(
                 target: "oxicloud::consistency",
                 event = "backend_consistency.deep_mode_active",
                 run_id = %store.run_id(),
+                cached_read_bypassed = backend.uncached().is_some(),
                 "deep mode: re-reading + re-hashing every matched blob (bit-rot detection)"
             );
         }
@@ -477,10 +499,14 @@ impl RecoverableJobHandler for BackendConsistencyCheck {
                     event = "backend_consistency.completed",
                     run_id = %store.run_id(),
                     finding_count = finding_count,
+                    verified = verified_count,
                     "backend_consistency completed with {} finding(s)",
                     finding_count
                 );
-                return RunOutcome::completed();
+                return RunOutcome::completed_with(serde_json::json!({
+                    "deep":     deep,
+                    "verified": verified_count,
+                }));
             }
 
             // ── Merge-join, not a one-sided probe ────────────────
@@ -562,8 +588,10 @@ impl RecoverableJobHandler for BackendConsistencyCheck {
                     // verified then.
                     (Some(b), Some(d)) if b.hash == **d => {
                         if deep && in_range(&b.hash) {
-                            finding_count +=
+                            verified_count += 1;
-                                self.verify_bytes(store, backend.as_ref(), &b.hash).await;
+                            finding_count += self
+                                .verify_bytes(store, verify_backend.as_ref(), &b.hash)
+                                .await;
                         }
                         bi.next();
                         di.next();
@@ -668,10 +696,18 @@ impl RecoverableJobHandler for BackendConsistencyCheck {
                     event = "backend_consistency.completed",
                     run_id = %store.run_id(),
                     finding_count = finding_count,
+                    verified = verified_count,
                     "backend_consistency completed with {} finding(s)",
                     finding_count
                 );
-                return RunOutcome::completed();
+                // `verified` is reported on EVERY run, zero included:
+                // absent-vs-zero is exactly the distinction an operator
+                // needs, and omitting it on a shallow run would make
+                // "deep verified nothing" look like "this was shallow".
+                return RunOutcome::completed_with(serde_json::json!({
+                    "deep":     deep,
+                    "verified": verified_count,
+                }));
             }
         }
     }

src/infrastructure/services/cached_blob_backend.rs

@@ -413,6 +413,14 @@ impl BlobStorageBackend for CachedBlobBackend {
         if path.exists() { Some(path) } else { None }
     }
 
+    /// This decorator IS the cache, so peeling it yields the real
+    /// storage. See [`BlobStorageBackend::uncached`] for why an
+    /// integrity check must not read through here — every other caller
+    /// keeps using the cache.
+    fn uncached(&self) -> Option<Arc<dyn BlobStorageBackend>> {
+        Some(self.inner.clone())
+    }
+
     /// Enumeration MUST delegate to the primary (inner) backend, not
     /// the local cache. The cache is by definition a subset (only
     /// recently-accessed blobs); walking the cache would look like

src/infrastructure/services/retry_blob_backend.rs

@@ -356,6 +356,14 @@ impl BlobStorageBackend for RetryBlobBackend {
         self.inner.local_blob_path(hash)
     }
 
+    /// Pass-through wrapper — forward, so an unwrap started above still
+    /// reaches the cache. Inheriting the `None` default would silently
+    /// end the search at this layer and leave verification reading
+    /// through the cache after all.
+    fn uncached(&self) -> Option<Arc<dyn BlobStorageBackend>> {
+        self.inner.uncached()
+    }
+
     /// Enumeration delegates to inner. Retry semantics apply per
     /// call, not per batch — a single list call that fails after
     /// exhausting retries surfaces the error to the tenant, which

src/infrastructure/services/swappable_blob_backend.rs

@@ -205,6 +205,15 @@ impl BlobStorageBackend for SwappableBlobBackend {
         self.current().local_blob_path(hash)
     }
 
+    /// Resolved through `current()`, not captured once: this wrapper sits
+    /// OUTSIDE the cache, so a migration cutover replaces the entire
+    /// cached stack beneath it. A handle taken at DI time would keep
+    /// pointing at the pre-cutover storage and audit the backend that
+    /// was just migrated away from.
+    fn uncached(&self) -> Option<Arc<dyn BlobStorageBackend>> {
+        self.current().uncached()
+    }
+
     fn read_prefetch(&self) -> usize {
         self.current().read_prefetch()
     }