feat(storage key rot): add blob header engine

examples/bench_s3_put.rs

@@ -353,7 +353,7 @@ async fn main() {
     // Full production composition: Cache(Encrypted(Retry(S3))).
     let cache_dir_b = tempfile::tempdir().expect("tempdir");
     let full_stack: Arc<dyn BlobStorageBackend> = Arc::new(CachedBlobBackend::new(
-        Arc::new(EncryptedBlobBackend::new(
+        Arc::new(EncryptedBlobBackend::new_single_aes(
             Arc::new(RetryBlobBackend::new(
                 backend.clone() as Arc<dyn BlobStorageBackend>,
                 RetryPolicy::default(),

src/common/config.rs

@@ -483,6 +483,30 @@ pub struct KeyPair {
 }
 
 impl KeyPair {
+    /// Construct a real-cipher pair with pre-decoded key material.
+    /// Convenience for tests + the pre-multi-entry legacy synthesis
+    /// path where the base64-decoded key is already in hand. New
+    /// production code paths get their pairs from
+    /// [`parse_encryption_pair_list`] which produces the same
+    /// shape.
+    pub fn new_aes_gcm(key: [u8; 32]) -> Self {
+        Self {
+            cipher: CipherKind::AesGcm256,
+            key_material: Some(key),
+        }
+    }
+
+    /// Construct a `none:` sentinel pair. Used by `entry_backend.rs`
+    /// under the always-wrap rule to synthesise a 1-pair list for
+    /// entries with no `_ENCRYPTION_KEY` declared, so those writes
+    /// still get v1 headers (plaintext-v1 flavor).
+    pub fn new_none() -> Self {
+        Self {
+            cipher: CipherKind::None,
+            key_material: None,
+        }
+    }
+
     /// Truncated SHA-256 fingerprint of the key material — 12 hex
     /// chars (6 bytes of SHA output). Used at boot for the audit-
     /// line dump so operators can eyeball which key is at each
@@ -501,6 +525,41 @@ impl KeyPair {
         let full = Sha256::digest(mat);
         Some(hex::encode(&full[..6]))
     }
+
+    /// 8-byte SHA-256 truncation used as the v1 header's `<key_fp>`
+    /// field on every encrypted blob. Read dispatch looks up the
+    /// matching [`KeyPair`] in the pair list by this value in O(1),
+    /// so a blob written under any pair in the list can be decrypted
+    /// without falling through candidate keys.
+    ///
+    /// Semantic contract with the on-disk format:
+    ///
+    /// * `CipherKind::AesGcm256` pair → 8 bytes of `sha256(key)[..8]`.
+    ///   Effectively unique per configured pair (2⁻⁶⁴ collision on
+    ///   random keys — never in practice on the ~1-3 pairs a real
+    ///   deployment has).
+    /// * `CipherKind::None` pair → all-zero. This is what marks a
+    ///   plaintext-v1 blob so the read path can dispatch "return
+    ///   post-header raw bytes" without consulting a key. Real
+    ///   ciphers cannot collide with all-zero: an
+    ///   `sha256(key)[..8] == 0` real key is 2⁻⁶⁴ improbable AND
+    ///   the parser wouldn't accept two pairs with the same fp
+    ///   (uniqueness check on raw key material — same key = same
+    ///   fp).
+    ///
+    /// Distinct truncation from [`Self::fingerprint_short`] on
+    /// purpose: this is a raw byte string embedded in every
+    /// encrypted blob (compactness matters), while the boot log
+    /// wants a legible short-hex string.
+    pub fn key_fp(&self) -> [u8; 8] {
+        use sha2::{Digest, Sha256};
+        let mut fp = [0u8; 8];
+        if let Some(mat) = self.key_material.as_ref() {
+            let full = Sha256::digest(mat);
+            fp.copy_from_slice(&full[..8]);
+        }
+        fp
+    }
 }
 
 /// Parse the `OXICLOUD_STORAGE_<name>_ENCRYPTION_KEY` env var value
@@ -3860,5 +3919,55 @@ mod tests {
             .unwrap();
             assert_ne!(pairs[0].fingerprint_short(), pairs[1].fingerprint_short());
         }
+
+        // ── key_fp (8-byte header field) tests ────────────────────
+
+        #[test]
+        fn key_fp_is_eight_bytes() {
+            let pairs = parse_encryption_pair_list("t", K1_B64).unwrap();
+            let fp = pairs[0].key_fp();
+            assert_eq!(fp.len(), 8);
+            // At least one byte non-zero (K1 = 0x00..0x1F sha256 has
+            // ample entropy; if this ever asserts we've got a truly
+            // improbable collision).
+            assert!(fp.iter().any(|b| *b != 0), "fp = {fp:?}");
+        }
+
+        #[test]
+        fn key_fp_zero_for_none_cipher() {
+            let pairs = parse_encryption_pair_list("t", "none:").unwrap();
+            assert_eq!(pairs[0].key_fp(), [0u8; 8]);
+        }
+
+        #[test]
+        fn key_fp_stable_across_calls() {
+            let pairs_a = parse_encryption_pair_list("t", K1_B64).unwrap();
+            let pairs_b = parse_encryption_pair_list("t", K1_B64).unwrap();
+            assert_eq!(pairs_a[0].key_fp(), pairs_b[0].key_fp());
+        }
+
+        #[test]
+        fn key_fp_differs_between_different_keys() {
+            let pairs = parse_encryption_pair_list(
+                "t",
+                &format!("aes-256-gcm:{K1_B64},aes-256-gcm:{K2_B64}"),
+            )
+            .unwrap();
+            assert_ne!(pairs[0].key_fp(), pairs[1].key_fp());
+        }
+
+        #[test]
+        fn key_fp_and_fingerprint_short_share_prefix() {
+            // Both derive from the same sha256(key); the on-blob fp is
+            // 8 raw bytes, the log fp is hex of the first 6. Pinning
+            // this alignment protects against a future refactor that
+            // accidentally switches one to a different hash / offset.
+            let pairs = parse_encryption_pair_list("t", K1_B64).unwrap();
+            let hex_fp = pairs[0].fingerprint_short().unwrap();
+            let raw_fp = pairs[0].key_fp();
+            // First 12 hex chars of the log fp = hex of the first 6
+            // bytes of the raw fp.
+            assert_eq!(&hex_fp[..12], &hex::encode(&raw_fp[..6]));
+        }
     }
 }

src/common/di.rs

@@ -387,7 +387,7 @@ impl AppServiceFactory {
             let key: [u8; 32] = key_bytes.try_into().expect(
                 "OXICLOUD_STORAGE_ENCRYPTION_KEY must be exactly 32 bytes (base64 of 32 bytes)",
             );
-            blob_backend = Arc::new(EncryptedBlobBackend::new(blob_backend, &key));
+            blob_backend = Arc::new(EncryptedBlobBackend::new_single_aes(blob_backend, &key));
             tracing::info!("Blob storage encryption decorator enabled (AES-256-GCM) — legacy path");
         }
 

src/infrastructure/services/dedup_service.rs

@@ -3672,7 +3672,7 @@ mod rechunk_integration_tests {
         let inner = Arc::new(LocalBlobBackend::new(&dir.path().join("blobs")));
         inner.initialize().await.expect("init backend");
         let key = EncryptedBlobBackend::generate_key();
-        let backend = Arc::new(EncryptedBlobBackend::new(inner, &key));
+        let backend = Arc::new(EncryptedBlobBackend::new_single_aes(inner, &key));
         DedupService::new(backend, pool.clone(), pool.clone())
     }
 

src/infrastructure/services/encrypted_blob_backend.rs

@@ -1,16 +1,62 @@
-//! `EncryptedBlobBackend` — AES-256-GCM encryption decorator for blob storage.
+//! `EncryptedBlobBackend` — v1 blob-format decorator for blob storage.
 //!
-//! Transparently encrypts blobs before they reach the inner backend and
-//! decrypts them on read.  Each blob gets a random 96-bit nonce which is
-//! prepended to the ciphertext.  The GCM authentication tag (16 bytes) is
-//! appended by the cipher.
+//! Wraps an inner `BlobStorageBackend` and adds two orthogonal
+//! behaviours:
+//!
+//! 1. **v1 header framing on every new write** — every blob written
+//!    by this wrapper starts with a 15-byte header
+//!    `OXCPT | <version u16> | <key_fp u64>` so future reads are
+//!    self-describing regardless of the entry's current config.
+//! 2. **Pair-list encryption** — the wrapper owns an ordered
+//!    [`KeyPair`] list. Writes use the LAST pair (the "head"); reads
+//!    dispatch on the header's `<key_fp>` field into an O(1)
+//!    fp → cipher lookup, so a blob written under any pair still in
+//!    the list decrypts without fallback attempts.
+//!
+//! See `docs/plan/storage-key-rotation.md` for the full design.
+//!
+//! ## v1 on-disk layout
+//!
+//! Encrypted-v1 (`head_cipher` is a real AEAD):
+//!
+//! ```text
+//! "OXCPT"         5 bytes  — magic marker
+//! <version>       2 bytes  — big-endian u16; v1 = 0x0001
+//! <key_fp>        8 bytes  — sha256(key material)[..8], routes reads
+//! <nonce>        12 bytes  — random per blob (AES-GCM 96-bit)
+//! <ciphertext>    N bytes  — same length as plaintext
+//! <auth_tag>     16 bytes  — AEAD authentication tag
+//! ```
+//!
+//! Plaintext-v1 (`head_cipher` is `None`, i.e. entry uses a `none:`
+//! head pair or has no encryption declared at all):
+//!
+//! ```text
+//! "OXCPT"         5 bytes  — magic marker
+//! <version>       2 bytes  — big-endian u16; v1 = 0x0001
+//! <key_fp>        8 bytes  — all zero
+//! <payload>       N bytes  — raw plaintext
+//! ```
+//!
+//! ## Legacy fallback on reads
+//!
+//! Blobs written before this wrapper existed have no OXCPT magic.
+//! Reads check the first 5 bytes:
+//!
+//! * `"OXCPT"` → v1 path (version + key_fp lookup + AEAD or raw).
+//! * anything else → **legacy path** — try `head_cipher` (if any)
+//!   as an AES-GCM decode over the pre-v1 shape
+//!   `[nonce][ciphertext][tag]`; otherwise return raw bytes.
+//!
+//! Collision probability: 2⁻⁴⁰ per blob for random legacy bytes to
+//! start with `"OXCPT"`. If it happens, subsequent version / key_fp
+//! checks fail with a hard error (`UnsupportedBlobVersion` or
+//! `NoKeyForBlob`) — never silent misread.
 //!
 //! **IMPORTANT**: BLAKE3 hashing is performed on the *plaintext* by
 //! `DedupService` before this layer sees the blob, so content-addressable
 //! dedup still works correctly.
 //!
-//! Layout on disk/S3: `[12-byte nonce][ciphertext + 16-byte GCM tag]`
-//!
 //! ## Runtime & memory characteristics
 //!
 //! GCM is all-or-nothing per blob: a blob can only be decrypted whole, so
@@ -27,6 +73,7 @@
 //! the ciphertext buffer is reused for the plaintext instead of allocating
 //! a second copy.
 
+use std::collections::HashMap;
 use std::path::{Path, PathBuf};
 use std::pin::Pin;
 
@@ -39,14 +86,42 @@ use tokio::fs;
 use crate::application::ports::blob_storage_ports::{
     BlobStorageBackend, BlobStream, StorageHealthStatus,
 };
+use crate::common::config::KeyPair;
 use crate::domain::errors::DomainError;
 
+/// v1 magic marker — every v1 blob starts with these 5 ASCII bytes.
+/// Chosen for its 2⁻⁴⁰ collision odds against random legacy data
+/// and its clean legibility in a `hexdump` (mnemonic:
+/// "OXiCloud CiPherText").
+const OXCPT_MAGIC: &[u8; 5] = b"OXCPT";
+
+/// v1 header version bytes (big-endian u16 = 0x0001). Future formats
+/// bump this in-place — the [`OXCPT_MAGIC`] stays the discriminator
+/// against legacy / non-OxiCloud files.
+const V1_VERSION_BYTES: [u8; 2] = [0x00, 0x01];
+
+/// On-blob `<key_fp>` field size — 8 bytes = 64-bit truncation of
+/// `sha256(key)`. Wide enough that random collisions are 2⁻⁶⁴; the
+/// parser also uniqueness-checks pairs on raw key material so
+/// duplicates can't sneak in.
+const KEY_FP_SIZE: usize = 8;
+
+/// Total v1 header size = magic + version + key_fp.
+const HEADER_SIZE: usize = 5 + 2 + KEY_FP_SIZE;
+
 /// Nonce size for AES-256-GCM (96 bits = 12 bytes).
 const NONCE_SIZE: usize = 12;
 
 /// AES-256-GCM authentication tag length appended after the ciphertext.
 const TAG_SIZE: usize = 16;
 
+/// AEAD overhead per blob (nonce + tag) — 28 bytes, same regardless
+/// of header framing.
+const AEAD_OVERHEAD: usize = NONCE_SIZE + TAG_SIZE;
+
+/// Per-blob overhead for an encrypted-v1 blob: header + AEAD = 43 bytes.
+const ENCRYPTED_V1_OVERHEAD: usize = HEADER_SIZE + AEAD_OVERHEAD;
+
 /// Payloads at or above this size run crypto on the blocking pool; below
 /// it the `spawn_blocking` round-trip costs more than the AES work itself.
 const CRYPTO_OFFLOAD_THRESHOLD: usize = 64 * 1024;
@@ -56,23 +131,82 @@ const CRYPTO_OFFLOAD_THRESHOLD: usize = 64 * 1024;
 /// hashers) see the same backpressure shape either way.
 const PLAINTEXT_EMIT_SIZE: usize = 64 * 1024;
 
-/// `BlobStorageBackend` decorator that encrypts blobs at rest.
+/// `BlobStorageBackend` decorator that applies v1 header framing and
+/// pair-list-driven encryption. See the module-level docs for the
+/// on-disk layout and read-fallback semantics.
 pub struct EncryptedBlobBackend {
     inner: Arc<dyn BlobStorageBackend>,
-    /// `Arc` so the per-op `clone()` handed to `offload_crypto` closures is
-    /// an atomic bump instead of copying the ~240-byte expanded AES-256
-    /// round-key schedule on every chunk read/write.
-    cipher: Arc<Aes256Gcm>,
+    /// The pair list as declared by the operator. Guaranteed
+    /// non-empty by ctor (an empty input auto-synthesises a single
+    /// `none:` pair, so the invariant holds). Preserved verbatim for
+    /// observability and for downstream K3 `storage_rotate` which
+    /// needs to walk pair indices.
+    #[allow(dead_code)]
+    pairs: Vec<KeyPair>,
+    /// `<key_fp>` → per-pair cipher, for O(1) read dispatch on v1
+    /// blobs. Excludes any `none:` pair (nothing to build). Cloned
+    /// per read via `Arc::clone` — the ~240-byte expanded AES-256
+    /// round-key schedule is amortised across every request.
+    fp_ciphers: HashMap<[u8; KEY_FP_SIZE], Arc<Aes256Gcm>>,
+    /// Cipher used by writes (last pair in the list). `None` when
+    /// the head is a `CipherKind::None` pair — in that case writes
+    /// emit plaintext-v1 (magic + version + zero fp + raw payload).
+    head_cipher: Option<Arc<Aes256Gcm>>,
+    /// Head pair's `key_fp` — embedded in every write's v1 header.
+    /// `[0u8; 8]` when head is `CipherKind::None`, matching the
+    /// plaintext-v1 shape.
+    head_key_fp: [u8; KEY_FP_SIZE],
 }
 
 impl EncryptedBlobBackend {
-    /// Create a new encryption layer wrapping `inner`.
+    /// Primary constructor. Takes an ordered pair list — the LAST
+    /// pair is the write pair (head), every pair is a candidate for
+    /// reads via `<key_fp>` dispatch.
     ///
-    /// `key` must be exactly 32 bytes (AES-256).
-    pub fn new(inner: Arc<dyn BlobStorageBackend>, key: &[u8; 32]) -> Self {
-        let cipher =
-            Arc::new(Aes256Gcm::new_from_slice(key).expect("AES-256 key must be 32 bytes"));
-        Self { inner, cipher }
+    /// Empty `pairs` is legal and treated as a single implicit
+    /// `none:` pair — the wrapper still emits v1 headers on writes
+    /// (plaintext-v1 flavor) and still magic-byte-dispatches on
+    /// reads (with legacy fallback for header-less blobs). This is
+    /// the always-wrap contract used by `entry_backend.rs` under the
+    /// K2 "normalize data" rule.
+    ///
+    /// Panics if any real-cipher pair's key material isn't 32 bytes,
+    /// which the parser guarantees — a panic here signals a
+    /// programmer bug, not operator error.
+    pub fn new(inner: Arc<dyn BlobStorageBackend>, pairs: Vec<KeyPair>) -> Self {
+        let pairs = if pairs.is_empty() {
+            vec![KeyPair::new_none()]
+        } else {
+            pairs
+        };
+        let mut fp_ciphers = HashMap::with_capacity(pairs.len());
+        for kp in &pairs {
+            if let Some(mat) = kp.key_material.as_ref() {
+                let cipher = Aes256Gcm::new_from_slice(mat)
+                    .expect("KeyPair invariant: real-cipher pair has 32-byte key");
+                fp_ciphers.insert(kp.key_fp(), Arc::new(cipher));
+            }
+        }
+        let head = pairs
+            .last()
+            .expect("post-normalisation pair list is non-empty");
+        let head_key_fp = head.key_fp();
+        let head_cipher = fp_ciphers.get(&head_key_fp).cloned();
+        Self {
+            inner,
+            pairs,
+            fp_ciphers,
+            head_cipher,
+            head_key_fp,
+        }
+    }
+
+    /// Convenience: wrap with a single AES-256-GCM pair. Same effect
+    /// as `new(inner, vec![KeyPair::new_aes_gcm(*key)])`. Used by
+    /// tests + the pre-multi-entry legacy synthesis fallback in
+    /// `di.rs`.
+    pub fn new_single_aes(inner: Arc<dyn BlobStorageBackend>, key: &[u8; 32]) -> Self {
+        Self::new(inner, vec![KeyPair::new_aes_gcm(*key)])
     }
 
     /// Generate a random 32-byte key suitable for AES-256.
@@ -84,27 +218,132 @@ impl EncryptedBlobBackend {
     }
 }
 
-/// Encrypt `data` into the on-disk layout: `[12-byte nonce][ciphertext + tag]`.
+/// Assemble an encrypted-v1 blob:
+/// `OXCPT | v1 | key_fp | nonce | ciphertext | tag`.
 ///
-/// Single output buffer, mirroring the read side's in-place detached decrypt:
-/// the payload is copied exactly once and encrypted in place with the tag
-/// appended. The old shape let `cipher.encrypt` allocate a full ciphertext
-/// `Vec` and then copied it a second time behind the nonce — one extra
-/// allocation + a full-size memcpy on every encrypted chunk write
-/// (benches/ROUND11.md §15; output bytes identical for a given nonce).
-fn encrypt_bytes(cipher: &Aes256Gcm, data: &[u8]) -> Result<Bytes, DomainError> {
+/// Single output buffer, mirroring the read side's in-place detached
+/// decrypt: the payload is copied exactly once and encrypted in
+/// place with the tag appended. The old shape (pre-K2, no header)
+/// let `cipher.encrypt` allocate a full ciphertext `Vec` and then
+/// copied it a second time behind the nonce — one extra allocation +
+/// a full-size memcpy on every encrypted chunk write
+/// (benches/ROUND11.md §15). K2 preserves the single-buffer
+/// discipline: we `extend_from_slice` header + nonce + payload, then
+/// encrypt in place from `HEADER_SIZE + NONCE_SIZE`.
+fn encrypt_v1(
+    cipher: &Aes256Gcm,
+    head_key_fp: [u8; KEY_FP_SIZE],
+    data: &[u8],
+) -> Result<Bytes, DomainError> {
     let nonce = Aes256Gcm::generate_nonce(&mut OsRng);
-    let mut out = Vec::with_capacity(NONCE_SIZE + data.len() + TAG_SIZE);
+    let mut out = Vec::with_capacity(ENCRYPTED_V1_OVERHEAD + data.len());
+    out.extend_from_slice(OXCPT_MAGIC);
+    out.extend_from_slice(&V1_VERSION_BYTES);
+    out.extend_from_slice(&head_key_fp);
     out.extend_from_slice(nonce.as_slice());
     out.extend_from_slice(data);
     let tag = cipher
-        .encrypt_in_place_detached(&nonce, b"", &mut out[NONCE_SIZE..])
+        .encrypt_in_place_detached(&nonce, b"", &mut out[HEADER_SIZE + NONCE_SIZE..])
         .map_err(|e| DomainError::internal_error("Encryption", format!("encrypt failed: {e}")))?;
     out.extend_from_slice(&tag);
     Ok(Bytes::from(out))
 }
 
-/// Decrypt the on-disk layout `[nonce][ciphertext + tag]` **in place**.
+/// Assemble a plaintext-v1 blob:
+/// `OXCPT | v1 | <8 zero bytes> | payload`.
+///
+/// No crypto, no allocation beyond the header prefix. Produced when
+/// the wrapper's head pair is `CipherKind::None`.
+fn write_plaintext_v1(data: &[u8]) -> Bytes {
+    let mut out = Vec::with_capacity(HEADER_SIZE + data.len());
+    out.extend_from_slice(OXCPT_MAGIC);
+    out.extend_from_slice(&V1_VERSION_BYTES);
+    out.extend_from_slice(&[0u8; KEY_FP_SIZE]);
+    out.extend_from_slice(data);
+    Bytes::from(out)
+}
+
+/// Read dispatch — the K2 core. Given a fetched blob and the
+/// wrapper's pair table, returns plaintext.
+///
+/// * `OXCPT` at position 0 → **v1 path**:
+///   * version check (only `0x0001` accepted today);
+///   * `key_fp == [0u8; 8]` → plaintext-v1 → return post-header
+///     bytes as-is;
+///   * else `key_fp` lookup in `fp_ciphers` → AEAD decrypt over
+///     the post-header body.
+/// * anything else → **legacy path**:
+///   * `head_cipher = Some` → AES-GCM decrypt over pre-K2 shape
+///     (`nonce | ct | tag`) — the pre-v1 world had exactly one key
+///     per entry, and that key is the head pair by construction of
+///     any pair-list upgraded from a pre-K1 config;
+///   * `head_cipher = None` → return raw bytes (pre-K2 plaintext
+///     deployment).
+///
+/// Never falls through silently: every failure returns a distinct
+/// typed error (`UnsupportedBlobVersion`, `NoKeyForBlob`, AEAD tag
+/// failure). Random legacy bytes matching `OXCPT` (2⁻⁴⁰) fail the
+/// subsequent version/key_fp check with a hard error, not silent
+/// garbage.
+fn read_dispatch(
+    fp_ciphers: &HashMap<[u8; KEY_FP_SIZE], Arc<Aes256Gcm>>,
+    head_cipher: Option<&Aes256Gcm>,
+    encrypted: Vec<u8>,
+) -> Result<Bytes, DomainError> {
+    if encrypted.len() >= 5 && &encrypted[..5] == OXCPT_MAGIC {
+        return read_v1(fp_ciphers, encrypted);
+    }
+    match head_cipher {
+        Some(cipher) => decrypt_aead_in_place(cipher, encrypted),
+        None => Ok(Bytes::from(encrypted)),
+    }
+}
+
+/// The v1 branch of `read_dispatch`, factored out for clarity.
+fn read_v1(
+    fp_ciphers: &HashMap<[u8; KEY_FP_SIZE], Arc<Aes256Gcm>>,
+    encrypted: Vec<u8>,
+) -> Result<Bytes, DomainError> {
+    if encrypted.len() < HEADER_SIZE {
+        return Err(DomainError::internal_error(
+            "Encryption",
+            format!(
+                "v1 blob too short (need at least {HEADER_SIZE} bytes for the header, got {})",
+                encrypted.len()
+            ),
+        ));
+    }
+    let version = &encrypted[5..7];
+    if version != V1_VERSION_BYTES {
+        return Err(DomainError::internal_error(
+            "Encryption",
+            format!(
+                "unsupported v1 blob version 0x{:02x}{:02x} — this build only reads 0x0001",
+                version[0], version[1]
+            ),
+        ));
+    }
+    let mut key_fp = [0u8; KEY_FP_SIZE];
+    key_fp.copy_from_slice(&encrypted[7..HEADER_SIZE]);
+    let mut body = encrypted;
+    body.drain(..HEADER_SIZE);
+    if key_fp == [0u8; KEY_FP_SIZE] {
+        // Plaintext-v1 — post-header bytes ARE the plaintext.
+        return Ok(Bytes::from(body));
+    }
+    let cipher = fp_ciphers.get(&key_fp).ok_or_else(|| {
+        DomainError::internal_error(
+            "Encryption",
+            format!(
+                "v1 blob key_fp {} does not match any configured pair — cannot decrypt",
+                hex::encode(key_fp)
+            ),
+        )
+    })?;
+    decrypt_aead_in_place(cipher, body)
+}
+
+/// Decrypt the AEAD body `[nonce][ciphertext][tag]` **in place**.
 ///
 /// Reuses the encrypted buffer for the plaintext, so peak RAM is one buffer —
 /// not ciphertext + plaintext side by side (which for legacy whole-file blobs
@@ -112,18 +351,12 @@ fn encrypt_bytes(cipher: &Aes256Gcm, data: &[u8]) -> Result<Bytes, DomainError>
 /// are lifted to the stack, the ciphertext body is decrypted in place via the
 /// detached API (mirroring the encrypt side's `encrypt_in_place_detached`), and
 /// the plaintext is returned as a zero-copy `Bytes::slice` past the nonce.
-///
-/// The prior shape did `encrypted.split_off(NONCE_SIZE)`, which allocated a
-/// fresh `Vec` and memcpy'd the entire ciphertext (up to a whole legacy blob)
-/// on every decrypted read — one full-payload allocation + copy the doc comment
-/// above claimed did not happen (benches/ROUND25.md §M1; ROUND11 §15 fixed only
-/// the encrypt side). Output plaintext is byte-identical.
-fn decrypt_bytes(cipher: &Aes256Gcm, mut encrypted: Vec<u8>) -> Result<Bytes, DomainError> {
+fn decrypt_aead_in_place(cipher: &Aes256Gcm, mut encrypted: Vec<u8>) -> Result<Bytes, DomainError> {
     let len = encrypted.len();
-    if len < NONCE_SIZE + TAG_SIZE {
+    if len < AEAD_OVERHEAD {
         return Err(DomainError::internal_error(
             "Encryption",
-            "encrypted blob too short (missing nonce/tag)",
+            "AEAD body too short (missing nonce/tag)",
         ));
     }
     // Nonce (first 12 bytes) and GCM tag (last 16 bytes) copied to the stack so
@@ -186,21 +419,15 @@ impl BlobStorageBackend for EncryptedBlobBackend {
         let inner = self.inner.clone();
         let hash = hash.to_string();
         let source = source_path.to_path_buf();
-        let cipher = self.cipher.clone();
+        let head_cipher = self.head_cipher.clone();
+        let head_key_fp = self.head_key_fp;
         Box::pin(async move {
             // Read plaintext from source
             let plaintext = fs::read(&source).await.map_err(|e| {
                 DomainError::internal_error("Encryption", format!("read source: {e}"))
             })?;
-
-            let len = plaintext.len();
-            let encrypted = offload_crypto(len, move || encrypt_bytes(&cipher, &plaintext)).await?;
-
-            // Hand the ciphertext straight to the inner backend. The previous
-            // implementation spooled it to a `.enc.tmp` file only for the
-            // inner backend to read it back — a full extra write + read of
-            // every blob that came through this path.
-            inner.put_blob_from_bytes(&hash, encrypted).await
+            let out = frame_write(head_cipher, head_key_fp, plaintext).await?;
+            inner.put_blob_from_bytes(&hash, out).await
         })
     }
 
@@ -211,11 +438,11 @@ impl BlobStorageBackend for EncryptedBlobBackend {
     ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
         let inner = self.inner.clone();
         let hash = hash.to_string();
-        let cipher = self.cipher.clone();
+        let head_cipher = self.head_cipher.clone();
+        let head_key_fp = self.head_key_fp;
         Box::pin(async move {
-            let encrypted =
-                offload_crypto(data.len(), move || encrypt_bytes(&cipher, data.as_ref())).await?;
-            inner.put_blob_from_bytes(&hash, encrypted).await
+            let out = frame_write(head_cipher, head_key_fp, data.to_vec()).await?;
+            inner.put_blob_from_bytes(&hash, out).await
         })
     }
 
@@ -226,11 +453,11 @@ impl BlobStorageBackend for EncryptedBlobBackend {
     ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
         let inner = self.inner.clone();
         let hash = hash.to_string();
-        let cipher = self.cipher.clone();
+        let head_cipher = self.head_cipher.clone();
+        let head_key_fp = self.head_key_fp;
         Box::pin(async move {
-            let encrypted =
-                offload_crypto(data.len(), move || encrypt_bytes(&cipher, data.as_ref())).await?;
-            inner.put_blob_from_bytes_unsynced(&hash, encrypted).await
+            let out = frame_write(head_cipher, head_key_fp, data.to_vec()).await?;
+            inner.put_blob_from_bytes_unsynced(&hash, out).await
         })
     }
 
@@ -250,14 +477,18 @@ impl BlobStorageBackend for EncryptedBlobBackend {
     {
         let inner = self.inner.clone();
         let hash = hash.to_string();
-        let cipher = self.cipher.clone();
+        let fp_ciphers = self.fp_ciphers.clone();
+        let head_cipher = self.head_cipher.clone();
         Box::pin(async move {
-            // GCM must see the whole message: collect ciphertext, decrypt in
-            // place off the runtime, then stream zero-copy plaintext slices.
+            // Collect the full blob, dispatch on magic bytes off the
+            // runtime, then stream zero-copy plaintext slices.
             let enc_stream = inner.get_blob_stream(&hash).await?;
             let encrypted = collect_stream(enc_stream).await?;
             let len = encrypted.len();
-            let plaintext = offload_crypto(len, move || decrypt_bytes(&cipher, encrypted)).await?;
+            let plaintext = offload_crypto(len, move || {
+                read_dispatch(&fp_ciphers, head_cipher.as_deref(), encrypted)
+            })
+            .await?;
             Ok(plaintext_stream(plaintext))
         })
     }
@@ -271,16 +502,20 @@ impl BlobStorageBackend for EncryptedBlobBackend {
     {
         let inner = self.inner.clone();
         let hash = hash.to_string();
-        let cipher = self.cipher.clone();
+        let fp_ciphers = self.fp_ciphers.clone();
+        let head_cipher = self.head_cipher.clone();
         Box::pin(async move {
-            // Decrypt the full blob, then slice the plaintext range without
-            // copying. For CDC chunks (every blob written since chunking
-            // landed) this is ≤ 1 MiB; only legacy whole-file blobs pay a
-            // full-blob decrypt here — see the module docs.
+            // Decrypt (or unwrap) the full blob, then slice the plaintext
+            // range without copying. For CDC chunks (every blob written
+            // since chunking landed) this is ≤ 1 MiB; only legacy whole-file
+            // blobs pay a full-blob decrypt here — see the module docs.
             let enc_stream = inner.get_blob_stream(&hash).await?;
             let encrypted = collect_stream(enc_stream).await?;
             let len = encrypted.len();
-            let plaintext = offload_crypto(len, move || decrypt_bytes(&cipher, encrypted)).await?;
+            let plaintext = offload_crypto(len, move || {
+                read_dispatch(&fp_ciphers, head_cipher.as_deref(), encrypted)
+            })
+            .await?;
 
             // `end` is exclusive — same contract as `LocalBlobBackend`, whose
             // implementation reads `end - start` bytes. The previous version
@@ -313,14 +548,29 @@ impl BlobStorageBackend for EncryptedBlobBackend {
         &self,
         hash: &str,
     ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
-        // The stored size includes nonce + GCM tag overhead.
-        // Return the *plaintext* size by subtracting overhead.
+        // Plaintext size = stored size - per-format overhead. The exact
+        // overhead depends on which format the blob is in (encrypted-v1
+        // = 43, plaintext-v1 = 15, legacy-encrypted = 28, legacy-plain =
+        // 0), which we can't know without inspecting bytes. We assume
+        // the blob was written under the wrapper's current head — that's
+        // true for every new write from K2 onward.
+        //
+        // For legacy blobs still on disk the estimate is off by
+        // ±(HEADER_SIZE) or so. Since `blob_size` is used for capacity
+        // metrics and admin dashboards (not byte-exact accounting —
+        // content-hash is the source of truth for that), a small drift
+        // during the legacy-blob window is acceptable. If a hot path
+        // starts depending on byte-exact `blob_size`, revisit.
         let inner = self.inner.clone();
         let hash = hash.to_string();
+        let overhead = if self.head_cipher.is_some() {
+            ENCRYPTED_V1_OVERHEAD as u64
+        } else {
+            HEADER_SIZE as u64
+        };
         Box::pin(async move {
-            let encrypted_size = inner.blob_size(&hash).await?;
-            // overhead = 12 (nonce) + 16 (GCM tag) = 28 bytes
-            Ok(encrypted_size.saturating_sub(28))
+            let stored = inner.blob_size(&hash).await?;
+            Ok(stored.saturating_sub(overhead))
         })
     }
 
@@ -330,16 +580,28 @@ impl BlobStorageBackend for EncryptedBlobBackend {
         Box<dyn std::future::Future<Output = Result<StorageHealthStatus, DomainError>> + Send + '_>,
     > {
         let inner = self.inner.clone();
+        let (outer_name, cipher_desc) = if self.head_cipher.is_some() {
+            ("encrypted", "AES-256-GCM")
+        } else {
+            ("v1-plaintext", "none")
+        };
         Box::pin(async move {
             let mut status = inner.health_check().await?;
-            status.backend_type = format!("encrypted({})", status.backend_type);
-            status.message = format!("{} | Encryption: AES-256-GCM", status.message);
+            status.backend_type = format!("{outer_name}({})", status.backend_type);
+            status.message = format!("{} | Encryption: {cipher_desc}", status.message);
             Ok(status)
         })
     }
 
     fn backend_type(&self) -> &'static str {
-        "encrypted"
+        // Choice 2/B: dynamic — reflects head-pair semantics so admin
+        // surfaces show "v1-plaintext(local)" for a `none:`-headed
+        // entry instead of misleadingly saying "encrypted(local)".
+        if self.head_cipher.is_some() {
+            "encrypted"
+        } else {
+            "v1-plaintext"
+        }
     }
 
     /// Transparent wrapper: the inner backend serves the bytes.
@@ -376,6 +638,24 @@ impl BlobStorageBackend for EncryptedBlobBackend {
     }
 }
 
+/// Frame a plaintext payload into a v1 blob per the head-pair
+/// configuration. Encrypted case runs the AEAD on the blocking pool
+/// for large payloads; plaintext case is a small header-prepend and
+/// stays inline (no crypto = no offload).
+async fn frame_write(
+    head_cipher: Option<Arc<Aes256Gcm>>,
+    head_key_fp: [u8; KEY_FP_SIZE],
+    plaintext: Vec<u8>,
+) -> Result<Bytes, DomainError> {
+    match head_cipher {
+        Some(cipher) => {
+            let len = plaintext.len();
+            offload_crypto(len, move || encrypt_v1(&cipher, head_key_fp, &plaintext)).await
+        }
+        None => Ok(write_plaintext_v1(&plaintext)),
+    }
+}
+
 /// Collect a byte stream into a single `Vec<u8>`.
 ///
 /// Modern blobs are CDC chunks (≤ `CDC_MAX_CHUNK` + nonce/tag overhead),
@@ -419,7 +699,7 @@ mod tests {
         local.initialize().await.unwrap();
 
         let key = EncryptedBlobBackend::generate_key();
-        let encrypted = EncryptedBlobBackend::new(local, &key);
+        let encrypted = EncryptedBlobBackend::new_single_aes(local, &key);
 
         // Write a test blob
         let data = b"Hello, encrypted world!";
@@ -467,7 +747,7 @@ mod tests {
         local.initialize().await.unwrap();
 
         let key = EncryptedBlobBackend::generate_key();
-        let encrypted = EncryptedBlobBackend::new(local, &key);
+        let encrypted = EncryptedBlobBackend::new_single_aes(local, &key);
 
         // 300 KiB of a repeating pattern — crosses the offload threshold and
         // spans several PLAINTEXT_EMIT_SIZE slices.
@@ -520,7 +800,7 @@ mod tests {
         local.initialize().await.unwrap();
 
         let key = EncryptedBlobBackend::generate_key();
-        let encrypted = EncryptedBlobBackend::new(local.clone(), &key);
+        let encrypted = EncryptedBlobBackend::new_single_aes(local.clone(), &key);
 
         let hash = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
         encrypted
@@ -528,16 +808,20 @@ mod tests {
             .await
             .unwrap();
 
-        // Corrupt one ciphertext byte on disk (past the 12-byte nonce).
+        // Corrupt one ciphertext byte on disk. v1 layout: 15-byte
+        // header + 12-byte nonce + ciphertext, so the first
+        // ciphertext byte is at `HEADER_SIZE + NONCE_SIZE`. Flipping
+        // it must fail AEAD tag verification.
         let path = local.local_blob_path(hash).expect("local path");
         let mut raw = std::fs::read(&path).unwrap();
-        raw[NONCE_SIZE] ^= 0xFF;
+        raw[HEADER_SIZE + NONCE_SIZE] ^= 0xFF;
         std::fs::write(&path, raw).unwrap();
 
         assert!(encrypted.get_blob_stream(hash).await.is_err());
     }
 
-    /// Decrypting with a different key must fail authentication.
+    /// Decrypting with a different key must fail — post-K2 via the
+    /// key_fp lookup (writer's fp isn't in the reader's map).
     #[tokio::test]
     async fn test_wrong_key_fails_decrypt() {
         let tmp = TempDir::new().unwrap();
@@ -545,14 +829,230 @@ mod tests {
         local.initialize().await.unwrap();
 
         let hash = "aaaabbbbccccddddaaaabbbbccccddddaaaabbbbccccddddaaaabbbbccccdddd";
-        let writer =
-            EncryptedBlobBackend::new(local.clone(), &EncryptedBlobBackend::generate_key());
+        let writer = EncryptedBlobBackend::new_single_aes(
+            local.clone(),
+            &EncryptedBlobBackend::generate_key(),
+        );
         writer
             .put_blob_from_bytes(hash, Bytes::from_static(b"locked"))
             .await
             .unwrap();
 
-        let reader = EncryptedBlobBackend::new(local, &EncryptedBlobBackend::generate_key());
-        assert!(reader.get_blob_stream(hash).await.is_err());
+        let reader =
+            EncryptedBlobBackend::new_single_aes(local, &EncryptedBlobBackend::generate_key());
+        // Result::unwrap_err needs Ok: Debug; BlobStream isn't Debug.
+        // Match directly, and extract the message off the DomainError.
+        let err = match reader.get_blob_stream(hash).await {
+            Ok(_) => panic!("expected a decrypt error, got Ok"),
+            Err(e) => e,
+        };
+        let msg = err.to_string();
+        assert!(
+            msg.contains("key_fp") || msg.contains("does not match"),
+            "expected NoKeyForBlob-shape error, got: {msg}"
+        );
+    }
+
+    // ─────────────────────────────────────────────────────────────
+    // K2 tests — v1 header format + magic-byte dispatch + legacy
+    // fallback + pair-list read routing.
+    // ─────────────────────────────────────────────────────────────
+
+    /// Every encrypted-v1 blob starts with the magic + version +
+    /// head-pair fingerprint. Pins the on-disk byte layout so a
+    /// future refactor can't silently break the format.
+    #[tokio::test]
+    async fn v1_encrypted_blob_has_expected_header() {
+        let tmp = TempDir::new().unwrap();
+        let local = Arc::new(LocalBlobBackend::new(&tmp.path().join("blobs")));
+        local.initialize().await.unwrap();
+
+        let key = [42u8; 32];
+        let backend = EncryptedBlobBackend::new_single_aes(local.clone(), &key);
+
+        let hash = "1111111111111111111111111111111111111111111111111111111111111111";
+        backend
+            .put_blob_from_bytes(hash, Bytes::from_static(b"hello world"))
+            .await
+            .unwrap();
+
+        let path = local.local_blob_path(hash).expect("local path");
+        let raw = std::fs::read(&path).unwrap();
+        assert!(
+            raw.len() >= ENCRYPTED_V1_OVERHEAD,
+            "blob too short: {}",
+            raw.len()
+        );
+        assert_eq!(&raw[..5], OXCPT_MAGIC, "missing OXCPT magic");
+        assert_eq!(&raw[5..7], &V1_VERSION_BYTES, "wrong version bytes");
+        let expected_fp = KeyPair::new_aes_gcm(key).key_fp();
+        assert_eq!(&raw[7..HEADER_SIZE], &expected_fp, "wrong key_fp in header");
+    }
+
+    /// `none:`-headed entry emits plaintext-v1 (header + raw
+    /// payload, no crypto). Round-trip must yield identity bytes.
+    #[tokio::test]
+    async fn v1_plaintext_blob_round_trips() {
+        let tmp = TempDir::new().unwrap();
+        let local = Arc::new(LocalBlobBackend::new(&tmp.path().join("blobs")));
+        local.initialize().await.unwrap();
+
+        let backend = EncryptedBlobBackend::new(local.clone(), vec![KeyPair::new_none()]);
+
+        let hash = "2222222222222222222222222222222222222222222222222222222222222222";
+        let payload = Bytes::from_static(b"cleartext bytes");
+        backend
+            .put_blob_from_bytes(hash, payload.clone())
+            .await
+            .unwrap();
+
+        // On-disk shape: magic + version + zero fp + raw payload.
+        let path = local.local_blob_path(hash).expect("local path");
+        let raw = std::fs::read(&path).unwrap();
+        assert_eq!(&raw[..5], OXCPT_MAGIC);
+        assert_eq!(&raw[5..7], &V1_VERSION_BYTES);
+        assert_eq!(&raw[7..HEADER_SIZE], &[0u8; KEY_FP_SIZE]);
+        assert_eq!(&raw[HEADER_SIZE..], payload.as_ref());
+
+        // Read must strip the header and return payload identity.
+        let stream = backend.get_blob_stream(hash).await.unwrap();
+        let round_tripped = collect_stream(stream).await.unwrap();
+        assert_eq!(round_tripped, payload.as_ref());
+    }
+
+    /// Legacy fallback: a blob written with the pre-K2 AEAD shape
+    /// (nonce | ct | tag, no OXCPT header) still decrypts via the
+    /// head-pair AES key. This is the guarantee that upgrading to
+    /// K2 doesn't break existing encrypted deployments.
+    #[tokio::test]
+    async fn legacy_encrypted_blob_still_readable() {
+        let tmp = TempDir::new().unwrap();
+        let local = Arc::new(LocalBlobBackend::new(&tmp.path().join("blobs")));
+        local.initialize().await.unwrap();
+
+        let key = [0x33u8; 32];
+        let backend = EncryptedBlobBackend::new_single_aes(local.clone(), &key);
+
+        // Craft a legacy blob by hand: AES-GCM with random nonce,
+        // no OXCPT header. Matches exactly what pre-K2 code wrote.
+        let plaintext = b"legacy secret";
+        let cipher = Aes256Gcm::new_from_slice(&key).unwrap();
+        let nonce = Aes256Gcm::generate_nonce(&mut OsRng);
+        let mut legacy = Vec::new();
+        legacy.extend_from_slice(nonce.as_slice());
+        legacy.extend_from_slice(plaintext);
+        let tag = cipher
+            .encrypt_in_place_detached(&nonce, b"", &mut legacy[NONCE_SIZE..])
+            .unwrap();
+        legacy.extend_from_slice(&tag);
+
+        // Write the raw bytes directly onto the local backend, bypassing
+        // the wrapper (else it'd add a v1 header).
+        local
+            .put_blob_from_bytes(
+                "3333333333333333333333333333333333333333333333333333333333333333",
+                Bytes::from(legacy),
+            )
+            .await
+            .unwrap();
+
+        // The wrapper's read path must dispatch on absent magic →
+        // legacy branch → head-pair AES decode.
+        let stream = backend
+            .get_blob_stream("3333333333333333333333333333333333333333333333333333333333333333")
+            .await
+            .unwrap();
+        let got = collect_stream(stream).await.unwrap();
+        assert_eq!(got, plaintext);
+    }
+
+    /// Legacy fallback for pure plaintext: entry has no encryption
+    /// (empty pair list → `none:` synthesised), reads of a raw-byte
+    /// blob written before the wrapper existed still return raw.
+    #[tokio::test]
+    async fn legacy_plaintext_blob_still_readable() {
+        let tmp = TempDir::new().unwrap();
+        let local = Arc::new(LocalBlobBackend::new(&tmp.path().join("blobs")));
+        local.initialize().await.unwrap();
+
+        let backend = EncryptedBlobBackend::new(local.clone(), vec![]);
+
+        let hash = "4444444444444444444444444444444444444444444444444444444444444444";
+        let raw = b"just some bytes, no header";
+        // Bypass the wrapper — write raw plaintext directly.
+        local
+            .put_blob_from_bytes(hash, Bytes::from_static(raw))
+            .await
+            .unwrap();
+
+        // No magic → legacy path → head is None → return bytes as-is.
+        let stream = backend.get_blob_stream(hash).await.unwrap();
+        let got = collect_stream(stream).await.unwrap();
+        assert_eq!(got, raw);
+    }
+
+    /// Pair-list key rotation: write under pair[0]'s key, add
+    /// pair[1] as head, read must still succeed via pair[0]'s
+    /// key_fp entry in the lookup table.
+    #[tokio::test]
+    async fn read_dispatches_by_key_fp_in_pair_list() {
+        let tmp = TempDir::new().unwrap();
+        let local = Arc::new(LocalBlobBackend::new(&tmp.path().join("blobs")));
+        local.initialize().await.unwrap();
+
+        let k_old = [0x11u8; 32];
+        let k_new = [0x22u8; 32];
+
+        // Write with a single-pair wrapper using k_old.
+        let writer = EncryptedBlobBackend::new_single_aes(local.clone(), &k_old);
+        let hash = "5555555555555555555555555555555555555555555555555555555555555555";
+        writer
+            .put_blob_from_bytes(hash, Bytes::from_static(b"payload"))
+            .await
+            .unwrap();
+
+        // Reader has BOTH keys: k_old at position 0, k_new at head.
+        // The blob's key_fp field points at k_old → lookup succeeds
+        // even though writes now go under k_new.
+        let reader = EncryptedBlobBackend::new(
+            local,
+            vec![KeyPair::new_aes_gcm(k_old), KeyPair::new_aes_gcm(k_new)],
+        );
+        let stream = reader.get_blob_stream(hash).await.unwrap();
+        let got = collect_stream(stream).await.unwrap();
+        assert_eq!(got, b"payload");
+    }
+
+    /// Malformed v1 blob (correct magic, unknown version bytes) →
+    /// hard error, never silent misread. Guards against the
+    /// theoretical 2⁻⁴⁰ magic-collision case on random legacy data.
+    #[tokio::test]
+    async fn unknown_v1_version_returns_hard_error() {
+        let tmp = TempDir::new().unwrap();
+        let local = Arc::new(LocalBlobBackend::new(&tmp.path().join("blobs")));
+        local.initialize().await.unwrap();
+
+        let backend = EncryptedBlobBackend::new(local.clone(), vec![]);
+
+        let hash = "6666666666666666666666666666666666666666666666666666666666666666";
+        // Magic OK, version = 0xFFFF (future format we don't know).
+        let mut bogus = Vec::from(*OXCPT_MAGIC);
+        bogus.extend_from_slice(&[0xFF, 0xFF]);
+        bogus.extend_from_slice(&[0u8; KEY_FP_SIZE]);
+        bogus.extend_from_slice(b"payload");
+        local
+            .put_blob_from_bytes(hash, Bytes::from(bogus))
+            .await
+            .unwrap();
+
+        let err = match backend.get_blob_stream(hash).await {
+            Ok(_) => panic!("expected an UnsupportedBlobVersion error, got Ok"),
+            Err(e) => e,
+        };
+        let msg = err.to_string();
+        assert!(
+            msg.contains("unsupported v1 blob version"),
+            "expected UnsupportedBlobVersion error, got: {msg}"
+        );
     }
 }

src/infrastructure/services/entry_backend.rs

@@ -229,24 +229,25 @@ pub fn build_entry_backend(
         }
     };
 
-    // 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;
-    };
+    // v1 wrapper (Choice 1/B: always wrap). Every entry gets the
+    // header-aware read/write path — even entries with no
+    // `_ENCRYPTION_KEY` at all. This normalises the on-disk format
+    // going forward: all new writes carry the OXCPT v1 header, all
+    // reads magic-byte-dispatch (with legacy fallback for
+    // header-less pre-K2 blobs). Not backwards-compatible with
+    // pre-K2 code trying to read new writes — but Ed's called it:
+    // uniform format is worth the one-way door.
     use crate::infrastructure::services::encrypted_blob_backend::EncryptedBlobBackend;
+    let pairs = entry.encryption.clone().unwrap_or_default();
+    let mode = match entry.head_cipher() {
+        Some(crate::common::config::CipherKind::AesGcm256) => "encrypted-v1",
+        _ => "plaintext-v1",
+    };
     tracing::info!(
-        "Storage entry `{}` encrypted with AES-256-GCM (head pair from env)",
-        entry.name
+        "Storage entry `{}` — {} wrapper (pairs: {})",
+        entry.name,
+        mode,
+        pairs.len()
     );
-    Arc::new(EncryptedBlobBackend::new(base, key))
+    Arc::new(EncryptedBlobBackend::new(base, pairs))
 }