feat(storage key rot): remove dead born OXICLOUD_STORAGE_<N>_ENCRYPTION_CIPHER

+ alway ovewrite on storage migration (got issue when migrating with blob already existing and a key change)

src/infrastructure/services/entry_backend.rs

@@ -229,30 +229,24 @@ pub fn build_entry_backend(
         }
     };
 
-    // Encryption decorator — presence-implies-enabled, per plan §Encryption.
-    let Some(key_b64) = entry.encryption_key_base64.as_ref() else {
+    // Encryption decorator — presence-implies-enabled, per plan
+    // §Encryption. K1 preserves single-head-pair behaviour: writes and
+    // reads use the head pair's material. K2 replaces the decorator
+    // with a header-aware read/write path that consults the full pair
+    // list; this call site becomes a wrapper construction rather than
+    // a single-key handoff at that point.
+    let Some(key) = entry.head_key_material() else {
+        // No pair list at all, OR head pair is `CipherKind::None`
+        // (mid-decrypt-migration state). Both cases mean "writes go
+        // straight to the raw backend today"; older encrypted pairs
+        // in the list stay unreachable until K2 wires the read
+        // fallback.
         return base;
     };
     use crate::infrastructure::services::encrypted_blob_backend::EncryptedBlobBackend;
-    let key_bytes = base64::Engine::decode(&base64::engine::general_purpose::STANDARD, key_b64)
-        .unwrap_or_else(|e| {
-            panic!(
-                "entry `{}` encryption key is not valid base64: {e} — parser was supposed to \
-                 catch this at boot",
-                entry.name
-            )
-        });
-    let key: [u8; 32] = key_bytes.try_into().unwrap_or_else(|v: Vec<u8>| {
-        panic!(
-            "entry `{}` encryption key decoded to {} bytes; must be 32 — parser was supposed to \
-             catch this at boot",
-            entry.name,
-            v.len()
-        )
-    });
     tracing::info!(
-        "Storage entry `{}` encrypted with AES-256-GCM (key from env)",
+        "Storage entry `{}` encrypted with AES-256-GCM (head pair from env)",
         entry.name
     );
-    Arc::new(EncryptedBlobBackend::new(base, &key))
+    Arc::new(EncryptedBlobBackend::new(base, key))
 }

src/infrastructure/services/storage_migration_service.rs

@@ -328,7 +328,10 @@ impl RecoverableJobHandler for StorageMigrationService {
         if let Some(source) = source_entry
             && entry_identity(source) == entry_identity(target_entry)
         {
-            let key_differs = source.encryption_key_base64 != target_entry.encryption_key_base64;
+            // Compare head-pair materials — the "write key" for each
+            // entry. A non-head pair difference (mid-rotation) doesn't
+            // count as a key change for the purposes of this refusal.
+            let key_differs = source.head_key_material() != target_entry.head_key_material();
             let hint = if key_differs {
                 " (encryption key differs → this looks like an in-place key rotation; \
                  create a new entry pointing at a DIFFERENT bucket / dir, migrate to it, \
@@ -446,7 +449,14 @@ impl RecoverableJobHandler for StorageMigrationService {
         };
 
         let mut copied_count = 0u64;
-        let mut skipped_count = 0u64;
+        // K1.2: with the target-skip short-circuit gone (see the
+        // detailed comment further down), no blob is ever "skipped"
+        // during a migration walk today. The counter stays wired
+        // through the log lines + `finish_completed` so K3's
+        // format-aware smart-skip can re-populate it without
+        // touching the observability surface. Not mutated in this
+        // slice — hence no `mut`.
+        let skipped_count: u64 = 0;
         let mut failed_count = 0u64;
         let mut source_missing_count = 0u64;
 
@@ -574,26 +584,33 @@ impl RecoverableJobHandler for StorageMigrationService {
                     }
                 }
 
-                // Skip when the target already has it — supports
-                // idempotent resume and cheap re-runs against a
-                // partially-migrated target.
-                match target.blob_exists(hash).await {
-                    Ok(true) => {
-                        skipped_count += 1;
-                        continue;
-                    }
-                    Ok(false) => {}
-                    Err(e) => {
-                        tracing::warn!(
-                            target: "oxicloud::migration",
-                            event = "storage_migration.blob_exists_error",
-                            run_id = %store.run_id(),
-                            hash = %hash,
-                            error = %e,
-                            "blob_exists probe on target failed; attempting copy anyway"
-                        );
-                    }
-                }
+                // Always copy — do NOT short-circuit on
+                // `target.blob_exists(hash)`. Ed hit this on 2026-08-01
+                // during S3 → local migration testing with encryption
+                // enabled on the target: the target had pre-existing
+                // plaintext blobs from an earlier local-active session,
+                // so `blob_exists` returned true and the migration
+                // silently skipped them. Result: the "encrypted"
+                // target ended up with mixed plaintext + ciphertext
+                // blobs — undetectable until a subsequent read failed.
+                //
+                // The old skip was justified by two use cases:
+                //   (a) resume idempotency — the last cursor-checkpoint
+                //       window (~100 blobs) gets re-processed on resume;
+                //   (b) target-side dedup — same content already present.
+                //
+                // Both are now handled by unconditional overwrite: the
+                // re-copy is bounded by the checkpoint window (small),
+                // and dedup-hit content is rare in practice
+                // (content-addressability means duplicate blobs ARE
+                // the same blob unless two backends were seeded
+                // separately from the same source).
+                //
+                // K3's `storage_rotate` job will restore a smart skip
+                // via the v1 header's `<key_fp>` field — "already at
+                // head format+key" then becomes cheaply detectable
+                // without reading target bytes. Until then, correct >
+                // fast.
 
                 match copy_blob(self.source.as_ref(), target.as_ref(), hash).await {
                     Ok(()) => {