diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 0c86324b7..36ab4b9ec 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -195,13 +195,13 @@ jobs:
         run: |
           set -o pipefail
           cargo test --locked -p dsm_storage_node --release --lib -- --test-threads=1 \
-            db::write_once_properties db::pg::durable_posture_tests | tee registers.log
+            db::write_once_properties db::binding_properties db::pg::durable_posture_tests | tee registers.log
           passed=$(sed -n 's/^test result: ok\. \([0-9]*\) passed.*/\1/p' registers.log | head -1)
           echo "register/durability tests executed on Postgres: ${passed:-0}"
           # May only grow. If a test is legitimately added, raise this floor in
           # the same commit; if the count DROPS, the board stopped proving
           # something it used to prove.
-          test "${passed:-0}" -ge 13
+          test "${passed:-0}" -ge 18
 
   # --------------------------------------------------------------------------
   # Formal validation: TLA+ model checking + real-code bridge harness
diff --git a/dsm_client/deterministic_state_machine/dsm/src/common/domain_tags/dsm/misc/mod.rs b/dsm_client/deterministic_state_machine/dsm/src/common/domain_tags/dsm/misc/mod.rs
index de409daa0..35b1d6732 100644
--- a/dsm_client/deterministic_state_machine/dsm/src/common/domain_tags/dsm/misc/mod.rs
+++ b/dsm_client/deterministic_state_machine/dsm/src/common/domain_tags/dsm/misc/mod.rs
@@ -102,6 +102,9 @@ pub(super) const TAGS: &[TaggedHashDomain<'static>] = &[
     TAG_DSM_STATE_ENTROPY,
     TAG_DSM_ANCHOR_PRESENTATION_V1,
     TAG_DSM_STORAGE_SET,
+    TAG_DSM_BINDING_RECORD,
+    TAG_DSM_BINDING_RECORD_SET,
+    TAG_DSM_BINDING_RECORD_SET_KEYS,
     TAG_DSM_VAULT_STATE,
     TAG_DSM_DEVTREE_DELEGATION,
     TAG_DSM_DEVTREE_DELEGATION_GENESIS_SENTINEL_V1,
diff --git a/dsm_client/deterministic_state_machine/dsm/src/common/domain_tags/dsm/misc/protocol.rs b/dsm_client/deterministic_state_machine/dsm/src/common/domain_tags/dsm/misc/protocol.rs
index 8243ed9f6..76291a4b3 100644
--- a/dsm_client/deterministic_state_machine/dsm/src/common/domain_tags/dsm/misc/protocol.rs
+++ b/dsm_client/deterministic_state_machine/dsm/src/common/domain_tags/dsm/misc/protocol.rs
@@ -126,6 +126,23 @@ pub const TAG_DSM_VAULT_STATE_PARENT_GENESIS_V2: TaggedHashDomain<'static> =
 /// both the layout and its tag go.
 pub const TAG_DSM_STORAGE_SET: TaggedHashDomain<'static> =
     crate::tagged_domain!(b"DSM/storage-set");
+/// One generic binding record (SoFi Rev 15 Def 6.20): the digest a member
+/// reports and a Class K driver compares. Over the record's canonical
+/// protobuf bytes — the node hashes what it stores, never a decoded view.
+pub const TAG_DSM_BINDING_RECORD: TaggedHashDomain<'static> =
+    crate::tagged_domain!(b"DSM/binding-record");
+/// The exact prior record SET over a sorted key set — `expected_digest` in
+/// CompareExchangeMany (§15.5). Absent cells are part of the preimage as
+/// absences, so "nothing held" has a defined digest a first writer can
+/// exchange from.
+pub const TAG_DSM_BINDING_RECORD_SET: TaggedHashDomain<'static> =
+    crate::tagged_domain!(b"DSM/binding-record-set");
+/// The sorted key set itself — `keyset_digest` inside a record, which the
+/// node requires to equal the digest of the request's own keys (Req 15.7).
+/// Distinct from `DSM/binding-keyset`, which derives ONE resource key from a
+/// vault parent (Def 6.17); this digests a set of such keys.
+pub const TAG_DSM_BINDING_RECORD_SET_KEYS: TaggedHashDomain<'static> =
+    crate::tagged_domain!(b"DSM/binding-record-set-keys");
 /// Immutable namespace for a published `AnchorPresentationV3` — the owner's
 /// complete verification bundle for one vault state (proto bytes as payload).
 /// The object is pure transport: every claim inside is re-derived by the
diff --git a/dsm_client/deterministic_state_machine/dsm/src/storage/binding_record.rs b/dsm_client/deterministic_state_machine/dsm/src/storage/binding_record.rs
new file mode 100644
index 000000000..87dd16973
--- /dev/null
+++ b/dsm_client/deterministic_state_machine/dsm/src/storage/binding_record.rs
@@ -0,0 +1,467 @@
+// SPDX-License-Identifier: MIT OR Apache-2.0
+
+//! The generic binding record and its digests — SoFi Rev 15 Def 6.20 and
+//! §15.5, the storage-engine half of `QuorumBind`.
+//!
+//! A storage member holds at most one opaque record per resource key and
+//! replaces a whole strictly-sorted key set atomically or not at all. What a
+//! member may INSPECT is exactly what this module makes inspectable — the
+//! schema, the lexicographic round, the key-set digest, and the digest of the
+//! exact prior record set — and nothing else. The value a record points at is
+//! never decoded here; Class K assigns meaning to `status` and `value_*`, and
+//! a member that attached SoFi meaning to any field would be violating §22
+//! #12, not implementing it.
+//!
+//! Everything here is pure and deterministic so the node, the SDK driver and
+//! the in-process fleet double all hash the same bytes: the node hashes what
+//! it STORES (the canonical protobuf bytes), never a decoded view of it.
+
+use crate::crypto::blake3::dsm_domain_hasher;
+use crate::common::domain_tags::{
+    TAG_DSM_BINDING_RECORD, TAG_DSM_BINDING_RECORD_SET, TAG_DSM_BINDING_RECORD_SET_KEYS,
+};
+use crate::types::proto as generated;
+use prost::Message;
+
+/// The one live schema of the generic binding record.
+pub const BINDING_RECORD_SCHEMA_V1: u32 = 1;
+
+/// Why a byte string is not a generic binding record, or a request is not a
+/// well-formed conditional exchange. Every variant is the storage-domain
+/// refusal `INVALID_STORAGE_ENCODING` (§15.5): the member did not get far
+/// enough to compare anything.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum BindingEncodingError {
+    /// Protobuf decoding failed, or the bytes are not the canonical encoding
+    /// of what they decode to. Non-canonical bytes are refused rather than
+    /// normalised, because two encodings of one record would be two digests
+    /// of one fact.
+    Noncanonical,
+    /// `schema` is not the live schema.
+    UnknownSchema(u32),
+    /// A fixed-width field had the wrong width.
+    Width { field: &'static str, got: usize },
+    /// The key set was empty.
+    EmptyKeySet,
+    /// The key set was not strictly ascending — unsorted or with duplicates.
+    /// A duplicate key would let one cell count twice in a set digest.
+    KeysNotStrictlyAscending { at: usize },
+    /// The record's `keyset_digest` does not equal the digest of the request's
+    /// own keys (Req 15.7 key-set equality).
+    KeySetMismatch,
+}
+
+impl core::fmt::Display for BindingEncodingError {
+    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+        match self {
+            Self::Noncanonical => write!(f, "binding record bytes are not a canonical encoding"),
+            Self::UnknownSchema(s) => write!(f, "binding record schema {s} is not live"),
+            Self::Width { field, got } => {
+                write!(
+                    f,
+                    "binding record field {field} has width {got}, expected 32"
+                )
+            }
+            Self::EmptyKeySet => write!(f, "a binding key set must name at least one key"),
+            Self::KeysNotStrictlyAscending { at } => write!(
+                f,
+                "binding keys must be strictly ascending; violated at index {at}"
+            ),
+            Self::KeySetMismatch => write!(
+                f,
+                "the record's keyset_digest is not the digest of the request's own keys"
+            ),
+        }
+    }
+}
+
+impl std::error::Error for BindingEncodingError {}
+
+/// A transaction round: `(counter, proposer_id)`, ordered lexicographically.
+///
+/// `counter` is a proposer-local persisted monotonic integer — never a
+/// timestamp. The derived `Ord` compares `counter` first and `proposer_id`
+/// second, which IS the lexicographic order Def 6.20 specifies; the field
+/// order of this struct is therefore load-bearing.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+pub struct Round {
+    pub counter: u64,
+    pub proposer_id: [u8; 32],
+}
+
+/// One decoded generic binding record. Constructible only through
+/// [`BindingRecord::decode_canonical`] or by a Class K driver that owns every
+/// field; a member never fabricates one.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct BindingRecord {
+    pub schema: u32,
+    pub round: Round,
+    pub tx_id: [u8; 32],
+    pub keyset_digest: [u8; 32],
+    pub value_digest: [u8; 32],
+    pub value_addr: [u8; 32],
+    pub status: u32,
+}
+
+fn fixed32(field: &'static str, v: &[u8]) -> Result<[u8; 32], BindingEncodingError> {
+    <[u8; 32]>::try_from(v).map_err(|_| BindingEncodingError::Width {
+        field,
+        got: v.len(),
+    })
+}
+
+impl BindingRecord {
+    fn to_proto(&self) -> generated::GenericBindingRecordV1 {
+        generated::GenericBindingRecordV1 {
+            schema: self.schema,
+            round_counter: self.round.counter,
+            proposer_id: self.round.proposer_id.to_vec(),
+            tx_id: self.tx_id.to_vec(),
+            keyset_digest: self.keyset_digest.to_vec(),
+            value_digest: self.value_digest.to_vec(),
+            value_addr: self.value_addr.to_vec(),
+            status: self.status,
+        }
+    }
+
+    fn from_proto(p: &generated::GenericBindingRecordV1) -> Result<Self, BindingEncodingError> {
+        if p.schema != BINDING_RECORD_SCHEMA_V1 {
+            return Err(BindingEncodingError::UnknownSchema(p.schema));
+        }
+        Ok(Self {
+            schema: p.schema,
+            round: Round {
+                counter: p.round_counter,
+                proposer_id: fixed32("proposer_id", &p.proposer_id)?,
+            },
+            tx_id: fixed32("tx_id", &p.tx_id)?,
+            keyset_digest: fixed32("keyset_digest", &p.keyset_digest)?,
+            value_digest: fixed32("value_digest", &p.value_digest)?,
+            value_addr: fixed32("value_addr", &p.value_addr)?,
+            status: p.status,
+        })
+    }
+
+    /// The canonical bytes: the protobuf encoding of the record. This is what
+    /// a member stores and hashes.
+    pub fn encode(&self) -> Vec<u8> {
+        self.to_proto().encode_to_vec()
+    }
+
+    /// Decode bytes that must be the canonical encoding of a live-schema
+    /// record with every fixed-width field at width 32. Bytes that decode but
+    /// do not re-encode to themselves are refused.
+    pub fn decode_canonical(bytes: &[u8]) -> Result<Self, BindingEncodingError> {
+        let p = generated::GenericBindingRecordV1::decode(bytes)
+            .map_err(|_| BindingEncodingError::Noncanonical)?;
+        let record = Self::from_proto(&p)?;
+        if record.encode() != bytes {
+            return Err(BindingEncodingError::Noncanonical);
+        }
+        Ok(record)
+    }
+
+    /// `H_dom(DSM/binding-record, canonical bytes)`.
+    pub fn digest(&self) -> [u8; 32] {
+        record_digest_of_bytes(&self.encode())
+    }
+}
+
+/// The digest of ALREADY-canonical record bytes, for a member that stores
+/// bytes and must not decode more than it inspects.
+pub fn record_digest_of_bytes(canonical_bytes: &[u8]) -> [u8; 32] {
+    let mut h = dsm_domain_hasher(TAG_DSM_BINDING_RECORD);
+    h.update(canonical_bytes);
+    *h.finalize().as_bytes()
+}
+
+/// A key set is valid when it is non-empty and strictly ascending — sorted
+/// and free of duplicates. The order is what makes a set digest and a
+/// deadlock-free lock order the same thing.
+pub fn validate_key_set(keys: &[[u8; 32]]) -> Result<(), BindingEncodingError> {
+    if keys.is_empty() {
+        return Err(BindingEncodingError::EmptyKeySet);
+    }
+    for (i, w) in keys.windows(2).enumerate() {
+        if w[0] >= w[1] {
+            return Err(BindingEncodingError::KeysNotStrictlyAscending { at: i + 1 });
+        }
+    }
+    Ok(())
+}
+
+/// `H_dom(DSM/binding-record-set-keys, u32_be(count) ‖ keys…)` over a valid
+/// key set. Carried inside every record as `keyset_digest`; the member
+/// requires it to equal the digest of the request's own keys.
+pub fn keyset_digest(keys: &[[u8; 32]]) -> [u8; 32] {
+    let mut h = dsm_domain_hasher(TAG_DSM_BINDING_RECORD_SET_KEYS);
+    h.update(&(keys.len() as u32).to_be_bytes());
+    for k in keys {
+        h.update(k);
+    }
+    *h.finalize().as_bytes()
+}
+
+/// One cell of a record set: a key and, if the member holds one, the digest
+/// of its canonical record bytes.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct SetCell {
+    pub key: [u8; 32],
+    pub record_digest: Option<[u8; 32]>,
+}
+
+/// `H_dom(DSM/binding-record-set, u32_be(count) ‖ for each cell in key
+/// order: key ‖ 0x00 | 0x01 ‖ record digest)`.
+///
+/// Absent cells are IN the preimage, marked `0x00` with no digest, so the
+/// digest names the exact prior set including what was empty: a first writer
+/// exchanges from the digest of all-absent, and a member that lost a row
+/// produces a different set digest rather than a coincidentally matching one.
+/// Cells must be in strictly ascending key order.
+pub fn record_set_digest(cells: &[SetCell]) -> [u8; 32] {
+    let mut h = dsm_domain_hasher(TAG_DSM_BINDING_RECORD_SET);
+    h.update(&(cells.len() as u32).to_be_bytes());
+    for c in cells {
+        h.update(&c.key);
+        match &c.record_digest {
+            None => {
+                h.update(&[0x00]);
+            }
+            Some(d) => {
+                h.update(&[0x01]);
+                h.update(d);
+            }
+        }
+    }
+    *h.finalize().as_bytes()
+}
+
+/// The digest of a key set holding nothing — what a first writer presents as
+/// `expected_digest`.
+pub fn empty_set_digest(keys: &[[u8; 32]]) -> [u8; 32] {
+    let cells: Vec<SetCell> = keys
+        .iter()
+        .map(|k| SetCell {
+            key: *k,
+            record_digest: None,
+        })
+        .collect();
+    record_set_digest(&cells)
+}
+
+/// A decoded, validated `CompareExchangeMany` request.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CompareExchange {
+    pub keys: Vec<[u8; 32]>,
+    pub expected_digest: [u8; 32],
+    pub replacement: BindingRecord,
+    /// The canonical bytes of `replacement`, exactly as received — what a
+    /// member stores. Kept beside the decoded form so the member never
+    /// re-encodes what it stores.
+    pub replacement_bytes: Vec<u8>,
+}
+
+/// Decode a `CompareExchangeManyRequestV1`, refusing non-canonical bytes,
+/// an invalid key set, and a replacement whose `keyset_digest` is not the
+/// digest of these keys.
+pub fn decode_compare_exchange(bytes: &[u8]) -> Result<CompareExchange, BindingEncodingError> {
+    let req = generated::CompareExchangeManyRequestV1::decode(bytes)
+        .map_err(|_| BindingEncodingError::Noncanonical)?;
+    if req.encode_to_vec() != bytes {
+        return Err(BindingEncodingError::Noncanonical);
+    }
+    let keys = decode_keys(&req.keys)?;
+    let expected_digest = fixed32("expected_digest", &req.expected_digest)?;
+    let rp = req
+        .replacement
+        .as_ref()
+        .ok_or(BindingEncodingError::Noncanonical)?;
+    let replacement = BindingRecord::from_proto(rp)?;
+    if replacement.keyset_digest != keyset_digest(&keys) {
+        return Err(BindingEncodingError::KeySetMismatch);
+    }
+    let replacement_bytes = rp.encode_to_vec();
+    Ok(CompareExchange {
+        keys,
+        expected_digest,
+        replacement,
+        replacement_bytes,
+    })
+}
+
+/// Decode a `ReadBindingRequestV1`, refusing non-canonical bytes and an
+/// invalid key set.
+pub fn decode_read_binding(bytes: &[u8]) -> Result<Vec<[u8; 32]>, BindingEncodingError> {
+    let req = generated::ReadBindingRequestV1::decode(bytes)
+        .map_err(|_| BindingEncodingError::Noncanonical)?;
+    if req.encode_to_vec() != bytes {
+        return Err(BindingEncodingError::Noncanonical);
+    }
+    decode_keys(&req.keys)
+}
+
+fn decode_keys(raw: &[Vec<u8>]) -> Result<Vec<[u8; 32]>, BindingEncodingError> {
+    let keys = raw
+        .iter()
+        .map(|k| fixed32("key", k))
+        .collect::<Result<Vec<_>, _>>()?;
+    validate_key_set(&keys)?;
+    Ok(keys)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn rec(counter: u64, proposer: u8, keys: &[[u8; 32]]) -> BindingRecord {
+        BindingRecord {
+            schema: BINDING_RECORD_SCHEMA_V1,
+            round: Round {
+                counter,
+                proposer_id: [proposer; 32],
+            },
+            tx_id: [0xAA; 32],
+            keyset_digest: keyset_digest(keys),
+            value_digest: [0xBB; 32],
+            value_addr: [0xCC; 32],
+            status: 1,
+        }
+    }
+
+    #[test]
+    fn a_round_orders_by_counter_then_proposer_lexicographically() {
+        let lo_lo = Round {
+            counter: 1,
+            proposer_id: [0x01; 32],
+        };
+        let lo_hi = Round {
+            counter: 1,
+            proposer_id: [0xFF; 32],
+        };
+        let hi_lo = Round {
+            counter: 2,
+            proposer_id: [0x00; 32],
+        };
+        assert!(lo_lo < lo_hi, "same counter: proposer id breaks the tie");
+        assert!(lo_hi < hi_lo, "a higher counter beats any proposer id");
+        assert_eq!(lo_lo, lo_lo);
+    }
+
+    #[test]
+    fn a_record_round_trips_canonically_and_noncanonical_bytes_are_refused() {
+        let keys = [[1u8; 32], [2u8; 32]];
+        let r = rec(3, 0x07, &keys);
+        let bytes = r.encode();
+        assert_eq!(BindingRecord::decode_canonical(&bytes).unwrap(), r);
+        // A trailing byte still decodes under protobuf's lenient reader but
+        // is not canonical.
+        let mut padded = bytes.clone();
+        padded.push(0);
+        assert_eq!(
+            BindingRecord::decode_canonical(&padded),
+            Err(BindingEncodingError::Noncanonical)
+        );
+        // A wrong-width fixed field is refused by name.
+        let mut p = generated::GenericBindingRecordV1::decode(&bytes[..]).unwrap();
+        p.tx_id = vec![1, 2, 3];
+        assert_eq!(
+            BindingRecord::decode_canonical(&p.encode_to_vec()),
+            Err(BindingEncodingError::Width {
+                field: "tx_id",
+                got: 3
+            })
+        );
+        // A dead schema is refused.
+        let mut p = generated::GenericBindingRecordV1::decode(&bytes[..]).unwrap();
+        p.schema = 2;
+        assert_eq!(
+            BindingRecord::decode_canonical(&p.encode_to_vec()),
+            Err(BindingEncodingError::UnknownSchema(2))
+        );
+    }
+
+    #[test]
+    fn a_key_set_must_be_non_empty_and_strictly_ascending() {
+        assert_eq!(
+            validate_key_set(&[]),
+            Err(BindingEncodingError::EmptyKeySet)
+        );
+        assert_eq!(
+            validate_key_set(&[[2; 32], [1; 32]]),
+            Err(BindingEncodingError::KeysNotStrictlyAscending { at: 1 })
+        );
+        assert_eq!(
+            validate_key_set(&[[1; 32], [1; 32]]),
+            Err(BindingEncodingError::KeysNotStrictlyAscending { at: 1 }),
+            "a duplicate would count one cell twice"
+        );
+        assert!(validate_key_set(&[[1; 32], [2; 32], [3; 32]]).is_ok());
+    }
+
+    #[test]
+    fn set_digests_name_absences_and_change_with_any_cell() {
+        let keys = [[1u8; 32], [2u8; 32]];
+        let empty = empty_set_digest(&keys);
+        let r = rec(1, 1, &keys);
+        let one = record_set_digest(&[
+            SetCell {
+                key: keys[0],
+                record_digest: Some(r.digest()),
+            },
+            SetCell {
+                key: keys[1],
+                record_digest: None,
+            },
+        ]);
+        let both = record_set_digest(&[
+            SetCell {
+                key: keys[0],
+                record_digest: Some(r.digest()),
+            },
+            SetCell {
+                key: keys[1],
+                record_digest: Some(r.digest()),
+            },
+        ]);
+        assert_ne!(empty, one);
+        assert_ne!(one, both);
+        assert_ne!(
+            empty,
+            empty_set_digest(&[[1u8; 32]]),
+            "the key set is part of the preimage"
+        );
+        assert_eq!(empty, empty_set_digest(&keys), "deterministic");
+    }
+
+    #[test]
+    fn a_compare_exchange_request_binds_its_own_key_set() {
+        let keys = vec![[1u8; 32], [2u8; 32]];
+        let r = rec(1, 1, &keys);
+        let req = generated::CompareExchangeManyRequestV1 {
+            keys: keys.iter().map(|k| k.to_vec()).collect(),
+            expected_digest: empty_set_digest(&keys).to_vec(),
+            replacement: Some(generated::GenericBindingRecordV1::decode(&r.encode()[..]).unwrap()),
+        };
+        let bytes = req.encode_to_vec();
+        let cx = decode_compare_exchange(&bytes).expect("well-formed");
+        assert_eq!(cx.keys, keys);
+        assert_eq!(cx.replacement, r);
+        assert_eq!(cx.replacement_bytes, r.encode());
+
+        // The same record presented over a DIFFERENT key set is refused: its
+        // keyset_digest names the keys it was built for.
+        let mut other = req.clone();
+        other.keys = vec![[1u8; 32].to_vec(), [3u8; 32].to_vec()];
+        assert_eq!(
+            decode_compare_exchange(&other.encode_to_vec()),
+            Err(BindingEncodingError::KeySetMismatch)
+        );
+        // Unsorted keys are refused before anything is compared.
+        let mut unsorted = req.clone();
+        unsorted.keys.reverse();
+        assert_eq!(
+            decode_compare_exchange(&unsorted.encode_to_vec()),
+            Err(BindingEncodingError::KeysNotStrictlyAscending { at: 1 })
+        );
+    }
+}
diff --git a/dsm_client/deterministic_state_machine/dsm/src/storage/mod.rs b/dsm_client/deterministic_state_machine/dsm/src/storage/mod.rs
index 8c71b31ab..e9be81625 100644
--- a/dsm_client/deterministic_state_machine/dsm/src/storage/mod.rs
+++ b/dsm_client/deterministic_state_machine/dsm/src/storage/mod.rs
@@ -1,3 +1,6 @@
 // SPDX-License-Identifier: MIT OR Apache-2.0
 
 //! Core defines traits only; concrete HTTP/FS/cloud impls live in dsm_sdk.
+
+/// The generic binding record and its digests — SoFi Rev 15 Def 6.20 / §15.5.
+pub mod binding_record;
diff --git a/dsm_storage_node/src/api/mod.rs b/dsm_storage_node/src/api/mod.rs
index c91962d4f..046c791a7 100644
--- a/dsm_storage_node/src/api/mod.rs
+++ b/dsm_storage_node/src/api/mod.rs
@@ -20,5 +20,6 @@ pub mod identity;
 pub mod infra;
 pub mod objects;
 pub mod registry;
+pub mod storage;
 pub mod transport;
 pub mod vault;
diff --git a/dsm_storage_node/src/api/storage/binding.rs b/dsm_storage_node/src/api/storage/binding.rs
new file mode 100644
index 000000000..2c42d99dc
--- /dev/null
+++ b/dsm_storage_node/src/api/storage/binding.rs
@@ -0,0 +1,406 @@
+// SPDX-License-Identifier: MIT OR Apache-2.0
+
+//! Generic conditional-binding interface — SoFi Rev 15 §15.5, the node half
+//! of `QuorumBind`.
+//!
+//! This node is APPLICATION-BLIND here, by construction and by rule (§22
+//! #12). It holds opaque generic binding records under opaque resource keys
+//! and offers exactly two operations: `CompareExchangeMany`, which replaces
+//! a whole strictly-sorted key set atomically or not at all, and
+//! `ReadBinding`, which reports what it holds. It inspects only the generic
+//! storage fields the decision needs — schema, round ordering, the exact
+//! expected digest of the prior record set, key-set equality — and never
+//! decodes the value a record points at, never checks a claimant, never
+//! checks a storage set, never knows a vault, a trade, or `q`. Class K owns
+//! all of that (§15.6); a member that "helped" would be asserting authority
+//! the specification withholds from it.
+//!
+//! Every answer names the member that gave it twice: the node id (the
+//! identity echo layer) and the REGISTER INCARNATION this node is serving —
+//! the same pair the read side already requires, now on write
+//! acknowledgements too. A write ack counts toward a quorum only when both
+//! match what the caller committed (Req 15.8); node identity alone cannot
+//! distinguish a rebuilt register from the one a vault named.
+
+use std::sync::Arc;
+
+use axum::{
+    body::Bytes,
+    extract::Extension,
+    http::{header, HeaderValue, StatusCode},
+    response::{IntoResponse, Response},
+    routing::post,
+    Router,
+};
+use dsm::storage::binding_record::{
+    decode_compare_exchange, decode_read_binding, record_set_digest, SetCell,
+};
+use dsm::types::proto as generated;
+use dsm_sdk::util::text_id;
+use prost::Message;
+
+use crate::{db, AppState};
+
+/// The register incarnation this node is serving, Base32-Crockford. Stamped
+/// on EVERY answer both operations give.
+pub const INCARNATION_HEADER: &str = "x-dsm-register-incarnation";
+
+/// Write half: `CompareExchangeMany`. Mounted behind device auth by the
+/// binary — a write must come from a registered device — but the node does
+/// NOT relate the caller to the record: there is no claimant to check.
+pub fn create_write_router() -> Router<()> {
+    Router::new().route("/api/v2/storage/binding/cas", post(post_compare_exchange))
+}
+
+/// Read half: `ReadBinding`. Public.
+pub fn create_read_router(state: Arc<AppState>) -> Router<()> {
+    Router::new()
+        .route("/api/v2/storage/binding/read", post(post_read_binding))
+        .layer(Extension(state))
+}
+
+fn proto_response(status: StatusCode, bytes: Vec<u8>, incarnation: &[u8; 32]) -> Response {
+    let mut resp = (status, bytes).into_response();
+    resp.headers_mut().insert(
+        header::CONTENT_TYPE,
+        HeaderValue::from_static("application/octet-stream"),
+    );
+    if let Ok(v) = HeaderValue::from_str(&text_id::encode_base32_crockford(incarnation)) {
+        resp.headers_mut().insert(INCARNATION_HEADER, v);
+    }
+    resp
+}
+
+/// A node with no established register incarnation cannot testify: it has
+/// no register history to speak for. This is unreachable after startup —
+/// the binary establishes the incarnation before serving — and refused
+/// rather than defaulted so a test harness that forgot it fails loudly.
+fn established_incarnation(state: &AppState) -> Option<[u8; 32]> {
+    state.own_register_incarnation
+}
+
+async fn post_compare_exchange(
+    Extension(state): Extension<Arc<AppState>>,
+    body: Bytes,
+) -> Response {
+    use generated::compare_exchange_many_response_v1::Outcome;
+    let Some(incarnation) = established_incarnation(&state) else {
+        return StatusCode::SERVICE_UNAVAILABLE.into_response();
+    };
+    let member_id = state.configured_member_id.as_bytes().to_vec();
+    let answer = |status: StatusCode, outcome: Outcome, digest: [u8; 32]| {
+        let msg = generated::CompareExchangeManyResponseV1 {
+            outcome: outcome as i32,
+            resulting_digest: digest.to_vec(),
+            member_id: member_id.clone(),
+            register_incarnation: incarnation.to_vec(),
+        };
+        proto_response(status, msg.encode_to_vec(), &incarnation)
+    };
+
+    // Storage-domain checks first: canonical bytes, a valid strictly-sorted
+    // key set, and a replacement whose keyset_digest names these keys. None
+    // of this reads the value payload.
+    let cx = match decode_compare_exchange(&body) {
+        Ok(cx) => cx,
+        Err(e) => {
+            log::warn!("binding cas: invalid storage encoding: {e}");
+            return answer(StatusCode::OK, Outcome::InvalidStorageEncoding, [0u8; 32]);
+        }
+    };
+    let replacement_digest = cx.replacement.digest();
+    match db::compare_exchange_many(
+        &state.db_pool,
+        &cx.keys,
+        &cx.expected_digest,
+        &cx.replacement_bytes,
+        &replacement_digest,
+        cx.replacement.round,
+    )
+    .await
+    {
+        Ok(db::CasOutcome::Applied { resulting_digest }) => {
+            answer(StatusCode::OK, Outcome::Applied, resulting_digest)
+        }
+        Ok(db::CasOutcome::ExpectationMismatch { current_digest }) => {
+            answer(StatusCode::OK, Outcome::ExpectationMismatch, current_digest)
+        }
+        Err(e) => {
+            log::warn!("binding cas: {e}");
+            answer(
+                StatusCode::SERVICE_UNAVAILABLE,
+                Outcome::Unavailable,
+                [0u8; 32],
+            )
+        }
+    }
+}
+
+async fn post_read_binding(Extension(state): Extension<Arc<AppState>>, body: Bytes) -> Response {
+    let Some(incarnation) = established_incarnation(&state) else {
+        return StatusCode::SERVICE_UNAVAILABLE.into_response();
+    };
+    let keys = match decode_read_binding(&body) {
+        Ok(k) => k,
+        Err(e) => {
+            log::warn!("binding read: invalid storage encoding: {e}");
+            return proto_response(StatusCode::BAD_REQUEST, Vec::new(), &incarnation);
+        }
+    };
+    let held = match db::read_bindings(&state.db_pool, &keys).await {
+        Ok(h) => h,
+        Err(e) => {
+            log::warn!("binding read: {e}");
+            return proto_response(StatusCode::SERVICE_UNAVAILABLE, Vec::new(), &incarnation);
+        }
+    };
+    let cells: Vec<SetCell> = keys
+        .iter()
+        .zip(held.iter())
+        .map(|(k, h)| SetCell {
+            key: *k,
+            record_digest: h.as_ref().map(|b| b.record_digest),
+        })
+        .collect();
+    let set_digest = record_set_digest(&cells);
+    let msg = generated::ReadBindingResponseV1 {
+        cells: keys
+            .iter()
+            .zip(held)
+            .map(|(k, h)| generated::BindingCellV1 {
+                key: k.to_vec(),
+                record: h.and_then(|b| {
+                    generated::GenericBindingRecordV1::decode(b.record_bytes.as_slice()).ok()
+                }),
+            })
+            .collect(),
+        set_digest: set_digest.to_vec(),
+        member_id: state.configured_member_id.as_bytes().to_vec(),
+        register_incarnation: incarnation.to_vec(),
+    };
+    proto_response(StatusCode::OK, msg.encode_to_vec(), &incarnation)
+}
+
+#[cfg(test)]
+mod tests {
+    //! Endpoint semantics ON WHICHEVER BACKEND IS COMPILED: the four
+    //! storage-domain outcomes, both halves of the identity stamp on every
+    //! answer, asserted absence on reads, and the refusal to testify without
+    //! an established incarnation. The record's MEANING is never inspected
+    //! here because the node never inspects it.
+    #![allow(clippy::disallowed_methods)] // unwrap/expect acceptable in deterministic tests
+    use super::*;
+    use crate::db::write_once_properties::{test_pool, unique_key};
+    use crate::replication::{ReplicationConfig, ReplicationManager};
+    use dsm::storage::binding_record::{
+        empty_set_digest, keyset_digest, BindingRecord, Round, BINDING_RECORD_SCHEMA_V1,
+    };
+    use generated::compare_exchange_many_response_v1::Outcome;
+
+    const INC: [u8; 32] = [0x1C; 32];
+
+    async fn node(with_incarnation: bool) -> Arc<AppState> {
+        let pool = Arc::new(test_pool());
+        db::init_db(&pool).await.expect("init");
+        let rm = Arc::new(
+            ReplicationManager::new_for_tests(
+                ReplicationConfig {
+                    replication_factor: 3,
+                    gossip_interval_ticks: 100,
+                    failure_timeout_ticks: 300,
+                    gossip_fanout: 3,
+                    max_concurrent_jobs: 10,
+                },
+                "n1".to_string(),
+                "http://localhost:8080".to_string(),
+            )
+            .expect("replication manager for tests"),
+        );
+        let state = AppState::new("n1".into(), "127.0.0.1:1", None, pool, rm);
+        Arc::new(if with_incarnation {
+            state.with_register_incarnation(INC)
+        } else {
+            state
+        })
+    }
+
+    fn record(keys: &[[u8; 32]], counter: u64, proposer: u8, value: u8) -> BindingRecord {
+        BindingRecord {
+            schema: BINDING_RECORD_SCHEMA_V1,
+            round: Round {
+                counter,
+                proposer_id: [proposer; 32],
+            },
+            tx_id: [0xAA; 32],
+            keyset_digest: keyset_digest(keys),
+            value_digest: [value; 32],
+            value_addr: [value; 32],
+            status: 1,
+        }
+    }
+
+    fn cas_body(keys: &[[u8; 32]], expected: [u8; 32], r: &BindingRecord) -> Bytes {
+        let req = generated::CompareExchangeManyRequestV1 {
+            keys: keys.iter().map(|k| k.to_vec()).collect(),
+            expected_digest: expected.to_vec(),
+            replacement: Some(generated::GenericBindingRecordV1::decode(&r.encode()[..]).unwrap()),
+        };
+        Bytes::from(req.encode_to_vec())
+    }
+
+    async fn cas(
+        state: &Arc<AppState>,
+        body: Bytes,
+    ) -> (
+        StatusCode,
+        generated::CompareExchangeManyResponseV1,
+        Option<String>,
+    ) {
+        let resp = post_compare_exchange(Extension(state.clone()), body).await;
+        let status = resp.status();
+        let inc = resp
+            .headers()
+            .get(INCARNATION_HEADER)
+            .map(|v| v.to_str().unwrap().to_string());
+        let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
+            .await
+            .unwrap();
+        let msg = generated::CompareExchangeManyResponseV1::decode(&bytes[..]).unwrap();
+        (status, msg, inc)
+    }
+
+    async fn read(
+        state: &Arc<AppState>,
+        keys: &[[u8; 32]],
+    ) -> (StatusCode, Option<generated::ReadBindingResponseV1>) {
+        let req = generated::ReadBindingRequestV1 {
+            keys: keys.iter().map(|k| k.to_vec()).collect(),
+        };
+        let resp =
+            post_read_binding(Extension(state.clone()), Bytes::from(req.encode_to_vec())).await;
+        let status = resp.status();
+        let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
+            .await
+            .unwrap();
+        (
+            status,
+            generated::ReadBindingResponseV1::decode(&bytes[..]).ok(),
+        )
+    }
+
+    #[tokio::test]
+    async fn cas_applies_from_empty_re_acks_identical_and_refuses_stale_or_lower_round() {
+        let state = node(true).await;
+        let keys = [unique_key(0x51), unique_key(0x52)];
+        let keys = {
+            let mut k = keys;
+            k.sort();
+            k
+        };
+        let a = record(&keys, 1, 1, 0xA1);
+
+        // First writer, from the digest of nothing: APPLIED, and the answer
+        // names this member and its incarnation in BOTH the body and header.
+        let (st, msg, hdr) = cas(&state, cas_body(&keys, empty_set_digest(&keys), &a)).await;
+        assert_eq!(st, StatusCode::OK);
+        assert_eq!(msg.outcome, Outcome::Applied as i32);
+        assert_eq!(msg.member_id, b"n1".to_vec());
+        assert_eq!(msg.register_incarnation, INC.to_vec());
+        assert_eq!(
+            hdr.as_deref(),
+            Some(text_id::encode_base32_crockford(&INC).as_str())
+        );
+        let after_a: [u8; 32] = msg.resulting_digest.clone().try_into().unwrap();
+
+        // Identical replay re-acks even with a stale expectation.
+        let (_, msg, _) = cas(&state, cas_body(&keys, empty_set_digest(&keys), &a)).await;
+        assert_eq!(
+            msg.outcome,
+            Outcome::Applied as i32,
+            "byte-identical replay re-acks"
+        );
+        assert_eq!(msg.resulting_digest, after_a.to_vec());
+
+        // A different value with the stale (empty) expectation is refused and
+        // told the digest it must read against.
+        let b = record(&keys, 2, 2, 0xB2);
+        let (_, msg, _) = cas(&state, cas_body(&keys, empty_set_digest(&keys), &b)).await;
+        assert_eq!(msg.outcome, Outcome::ExpectationMismatch as i32);
+        assert_eq!(
+            msg.resulting_digest,
+            after_a.to_vec(),
+            "the current digest is reported"
+        );
+
+        // Right expectation, but a round that does not supersede: refused.
+        let low = record(&keys, 0, 9, 0xB2);
+        let (_, msg, _) = cas(&state, cas_body(&keys, after_a, &low)).await;
+        assert_eq!(msg.outcome, Outcome::ExpectationMismatch as i32);
+
+        // Right expectation AND a higher round: B supersedes A on every key.
+        let (_, msg, _) = cas(&state, cas_body(&keys, after_a, &b)).await;
+        assert_eq!(msg.outcome, Outcome::Applied as i32);
+        let (_, r) = read(&state, &keys).await;
+        let r = r.unwrap();
+        assert_eq!(r.cells.len(), 2);
+        for c in &r.cells {
+            let rec = c.record.as_ref().expect("held on every key");
+            assert_eq!(rec.value_digest, [0xB2; 32].to_vec());
+            assert_eq!(rec.round_counter, 2);
+        }
+        assert_eq!(r.set_digest, msg.resulting_digest);
+        assert_eq!(r.register_incarnation, INC.to_vec());
+    }
+
+    #[tokio::test]
+    async fn storage_domain_refusals_are_invalid_storage_encoding_and_write_nothing() {
+        let state = node(true).await;
+        let keys = {
+            let mut k = [unique_key(0x53), unique_key(0x54)];
+            k.sort();
+            k
+        };
+        let a = record(&keys, 1, 1, 0xA1);
+
+        // Non-canonical bytes.
+        let mut padded = cas_body(&keys, empty_set_digest(&keys), &a).to_vec();
+        padded.push(0);
+        let (st, msg, _) = cas(&state, Bytes::from(padded)).await;
+        assert_eq!(st, StatusCode::OK);
+        assert_eq!(msg.outcome, Outcome::InvalidStorageEncoding as i32);
+
+        // Unsorted keys.
+        let mut rev = keys;
+        rev.reverse();
+        let (_, msg, _) = cas(&state, cas_body(&rev, empty_set_digest(&keys), &a)).await;
+        assert_eq!(msg.outcome, Outcome::InvalidStorageEncoding as i32);
+
+        // A record whose keyset_digest names OTHER keys.
+        let other = [unique_key(0x55), unique_key(0x56)];
+        let foreign = record(&other, 1, 1, 0xA1);
+        let (_, msg, _) = cas(&state, cas_body(&keys, empty_set_digest(&keys), &foreign)).await;
+        assert_eq!(msg.outcome, Outcome::InvalidStorageEncoding as i32);
+
+        // None of the refusals wrote anything: absence is asserted per key.
+        let (st, r) = read(&state, &keys).await;
+        assert_eq!(st, StatusCode::OK);
+        let r = r.unwrap();
+        assert!(r.cells.iter().all(|c| c.record.is_none()), "nothing held");
+        assert_eq!(r.set_digest, empty_set_digest(&keys).to_vec());
+    }
+
+    #[tokio::test]
+    async fn a_node_without_an_established_incarnation_refuses_to_testify() {
+        let state = node(false).await;
+        let keys = [unique_key(0x57)];
+        let a = record(&keys, 1, 1, 0xA1);
+        let resp = post_compare_exchange(
+            Extension(state.clone()),
+            cas_body(&keys, empty_set_digest(&keys), &a),
+        )
+        .await;
+        assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
+        let (st, _) = read(&state, &keys).await;
+        assert_eq!(st, StatusCode::SERVICE_UNAVAILABLE);
+    }
+}
diff --git a/dsm_storage_node/src/api/storage/mod.rs b/dsm_storage_node/src/api/storage/mod.rs
new file mode 100644
index 000000000..2361b498b
--- /dev/null
+++ b/dsm_storage_node/src/api/storage/mod.rs
@@ -0,0 +1,4 @@
+// SPDX-License-Identifier: MIT OR Apache-2.0
+//! Application-blind storage primitives (SoFi Rev 15 §15): the node half of
+//! the client-driven quorum transaction.
+pub mod binding;
diff --git a/dsm_storage_node/src/api/vault/settlement_slot.rs b/dsm_storage_node/src/api/vault/settlement_slot.rs
index d1d65aaf4..54a9d7c0a 100644
--- a/dsm_storage_node/src/api/vault/settlement_slot.rs
+++ b/dsm_storage_node/src/api/vault/settlement_slot.rs
@@ -72,7 +72,7 @@ pub const OUTCOME_HEADER: &str = "x-dsm-slot-outcome";
 /// a value and never as an absence — because a rebuilt register can honestly
 /// report "nothing here" for a cell the incarnation the vault committed once
 /// held, and node identity alone cannot tell the two apart.
-pub const INCARNATION_HEADER: &str = "x-dsm-register-incarnation";
+pub use crate::api::storage::binding::INCARNATION_HEADER;
 pub const HELD_DIGEST_HEADER: &str = "x-dsm-slot-held-digest";
 pub const CLAIM_DIGEST_HEADER: &str = "x-dsm-slot-digest";
 
diff --git a/dsm_storage_node/src/db/binding.rs b/dsm_storage_node/src/db/binding.rs
new file mode 100644
index 000000000..b2024cf4f
--- /dev/null
+++ b/dsm_storage_node/src/db/binding.rs
@@ -0,0 +1,279 @@
+// SPDX-License-Identifier: MIT OR Apache-2.0
+
+//! The generic conditional-binding DECISION, shared by both backends.
+//!
+//! `CompareExchangeMany` (SoFi Rev 15 §15.5, Req 15.6) is atomic within one
+//! member and inspects only generic storage fields. The decision of whether
+//! a replacement may be applied is pure and lives here, once, so Postgres and
+//! SQLite cannot drift into two rules, and so it is testable without a
+//! database. The backends supply I/O around it: lock, read what is held,
+//! decide, write all-or-none, commit durably.
+
+use dsm::storage::binding_record::{record_set_digest, Round, SetCell};
+use std::collections::BTreeMap;
+
+/// One cell as this member holds it. `record_bytes` are the canonical bytes
+/// exactly as received; this member never re-encodes what it stores.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct StoredBinding {
+    pub key: [u8; 32],
+    pub record_bytes: Vec<u8>,
+    pub record_digest: [u8; 32],
+    pub round: Round,
+}
+
+/// What one `CompareExchangeMany` did at this member.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum CasOutcome {
+    /// Every named key now holds the replacement (or already held exactly
+    /// it). `resulting_digest` is the digest of the record set over the keys
+    /// AFTER the call — the caller's next `expected_digest`.
+    Applied { resulting_digest: [u8; 32] },
+    /// Nothing changed. Either the prior record set was not the one the
+    /// caller expected, or the replacement's round does not supersede every
+    /// round already held. `current_digest` is the set digest the caller
+    /// must read against.
+    ExpectationMismatch { current_digest: [u8; 32] },
+}
+
+/// The digest of the record set this member holds over `keys`, absences
+/// included as absences.
+pub fn set_digest_over(keys: &[[u8; 32]], held: &BTreeMap<[u8; 32], StoredBinding>) -> [u8; 32] {
+    let cells: Vec<SetCell> = keys
+        .iter()
+        .map(|k| SetCell {
+            key: *k,
+            record_digest: held.get(k).map(|b| b.record_digest),
+        })
+        .collect();
+    record_set_digest(&cells)
+}
+
+/// The replacement a caller proposes, already decoded and validated by
+/// `dsm::storage::binding_record::decode_compare_exchange`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Replacement<'a> {
+    pub bytes: &'a [u8],
+    pub digest: [u8; 32],
+    pub round: Round,
+}
+
+/// Decide whether `replacement` may be applied over `keys` given what the
+/// member holds. Checks, in order, each fail-closed:
+///
+/// 1. byte-identical replay on EVERY key → `Applied` with the current
+///    digest — idempotent, like a re-ack, so a crashed proposer converges by
+///    retrying the same bytes;
+/// 2. the exact prior set digest must equal `expected_digest`;
+/// 3. the replacement's round must be STRICTLY greater than every round
+///    already held. A matching digest proves the caller saw the state; a
+///    round that does not supersede it must still not overwrite it, and an
+///    equal round with different bytes is a proposer reusing a round for a
+///    second value, which is refused.
+pub fn decide_compare_exchange(
+    keys: &[[u8; 32]],
+    held: &BTreeMap<[u8; 32], StoredBinding>,
+    expected_digest: &[u8; 32],
+    replacement: &Replacement<'_>,
+) -> CasOutcome {
+    let current_digest = set_digest_over(keys, held);
+    let identical_everywhere = keys.iter().all(|k| {
+        held.get(k)
+            .map(|b| b.record_bytes.as_slice() == replacement.bytes)
+            .unwrap_or(false)
+    });
+    if identical_everywhere {
+        return CasOutcome::Applied {
+            resulting_digest: current_digest,
+        };
+    }
+    if &current_digest != expected_digest {
+        return CasOutcome::ExpectationMismatch { current_digest };
+    }
+    if held.values().any(|b| replacement.round <= b.round) {
+        return CasOutcome::ExpectationMismatch { current_digest };
+    }
+    let after: BTreeMap<[u8; 32], StoredBinding> = keys
+        .iter()
+        .map(|k| {
+            (
+                *k,
+                StoredBinding {
+                    key: *k,
+                    record_bytes: replacement.bytes.to_vec(),
+                    record_digest: replacement.digest,
+                    round: replacement.round,
+                },
+            )
+        })
+        .collect();
+    CasOutcome::Applied {
+        resulting_digest: set_digest_over(keys, &after),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #![allow(clippy::disallowed_methods)] // unwrap/expect acceptable in deterministic tests
+    use super::*;
+    use dsm::storage::binding_record::{empty_set_digest, record_digest_of_bytes};
+
+    fn r(counter: u64, proposer: u8, bytes: &[u8]) -> ([u8; 32], Round) {
+        (
+            record_digest_of_bytes(bytes),
+            Round {
+                counter,
+                proposer_id: [proposer; 32],
+            },
+        )
+    }
+
+    #[test]
+    fn a_first_writer_exchanges_from_the_empty_digest_and_a_stale_expectation_is_refused() {
+        let keys = [[1u8; 32], [2u8; 32]];
+        let held = BTreeMap::new();
+        let (d, round) = r(1, 1, b"A");
+        let rep = Replacement {
+            bytes: b"A",
+            digest: d,
+            round,
+        };
+        let applied = decide_compare_exchange(&keys, &held, &empty_set_digest(&keys), &rep);
+        let CasOutcome::Applied { resulting_digest } = applied else {
+            panic!("first write from the empty digest applies")
+        };
+        // Someone still expecting the empty set after that is refused with
+        // the digest they must read against.
+        let mut held2 = BTreeMap::new();
+        for k in keys {
+            held2.insert(
+                k,
+                StoredBinding {
+                    key: k,
+                    record_bytes: b"A".to_vec(),
+                    record_digest: d,
+                    round,
+                },
+            );
+        }
+        let (d2, round2) = r(2, 2, b"B");
+        let rep2 = Replacement {
+            bytes: b"B",
+            digest: d2,
+            round: round2,
+        };
+        assert_eq!(
+            decide_compare_exchange(&keys, &held2, &empty_set_digest(&keys), &rep2),
+            CasOutcome::ExpectationMismatch {
+                current_digest: resulting_digest
+            }
+        );
+        // With the right expectation AND a higher round, B supersedes A.
+        assert!(matches!(
+            decide_compare_exchange(&keys, &held2, &resulting_digest, &rep2),
+            CasOutcome::Applied { .. }
+        ));
+    }
+
+    #[test]
+    fn a_round_that_does_not_supersede_is_refused_even_with_the_right_expectation() {
+        let keys = [[1u8; 32]];
+        let (d, round) = r(5, 9, b"A");
+        let mut held = BTreeMap::new();
+        held.insert(
+            keys[0],
+            StoredBinding {
+                key: keys[0],
+                record_bytes: b"A".to_vec(),
+                record_digest: d,
+                round,
+            },
+        );
+        let current = set_digest_over(&keys, &held);
+        // Lower counter: refused.
+        let (d2, r2) = r(4, 0xFF, b"B");
+        assert!(matches!(
+            decide_compare_exchange(
+                &keys,
+                &held,
+                &current,
+                &Replacement {
+                    bytes: b"B",
+                    digest: d2,
+                    round: r2
+                }
+            ),
+            CasOutcome::ExpectationMismatch { .. }
+        ));
+        // Same counter, same proposer, DIFFERENT bytes: a proposer reusing a
+        // round for a second value — refused.
+        let (d3, r3) = r(5, 9, b"C");
+        assert!(matches!(
+            decide_compare_exchange(
+                &keys,
+                &held,
+                &current,
+                &Replacement {
+                    bytes: b"C",
+                    digest: d3,
+                    round: r3
+                }
+            ),
+            CasOutcome::ExpectationMismatch { .. }
+        ));
+        // Same counter, HIGHER proposer id: lexicographically greater — applies.
+        let (d4, r4) = r(5, 10, b"D");
+        assert!(matches!(
+            decide_compare_exchange(
+                &keys,
+                &held,
+                &current,
+                &Replacement {
+                    bytes: b"D",
+                    digest: d4,
+                    round: r4
+                }
+            ),
+            CasOutcome::Applied { .. }
+        ));
+    }
+
+    #[test]
+    fn byte_identical_replay_re_acks_without_needing_the_expectation() {
+        let keys = [[1u8; 32], [2u8; 32]];
+        let (d, round) = r(1, 1, b"A");
+        let mut held = BTreeMap::new();
+        for k in keys {
+            held.insert(
+                k,
+                StoredBinding {
+                    key: k,
+                    record_bytes: b"A".to_vec(),
+                    record_digest: d,
+                    round,
+                },
+            );
+        }
+        let current = set_digest_over(&keys, &held);
+        let rep = Replacement {
+            bytes: b"A",
+            digest: d,
+            round,
+        };
+        // A stale (wrong) expectation does not matter for an identical replay:
+        // the member already holds exactly these bytes everywhere.
+        assert_eq!(
+            decide_compare_exchange(&keys, &held, &[0xEE; 32], &rep),
+            CasOutcome::Applied {
+                resulting_digest: current
+            }
+        );
+        // But identical on only SOME keys is not a replay; it is a
+        // partial state, and the expectation must match.
+        held.remove(&keys[1]);
+        assert!(matches!(
+            decide_compare_exchange(&keys, &held, &[0xEE; 32], &rep),
+            CasOutcome::ExpectationMismatch { .. }
+        ));
+    }
+}
diff --git a/dsm_storage_node/src/db/binding_properties.rs b/dsm_storage_node/src/db/binding_properties.rs
new file mode 100644
index 000000000..ba7e4c842
--- /dev/null
+++ b/dsm_storage_node/src/db/binding_properties.rs
@@ -0,0 +1,223 @@
+// SPDX-License-Identifier: MIT OR Apache-2.0
+
+//! The generic conditional-binding properties, ON WHICHEVER BACKEND IS
+//! COMPILED (SoFi Rev 15 Req 15.6, 15.9, 15.11): all-or-none over a key set,
+//! exactly one winner among concurrent exchanges from the same expectation,
+//! rounds that only move forward, overlapping key sets that serialise, and
+//! a record set that survives re-opening the store. CI executes this file
+//! against a real Postgres as well as the in-memory default; the Postgres
+//! build refuses to run without `DSM_TEST_DATABASE_URL` rather than skipping.
+
+#![allow(clippy::disallowed_methods)] // unwrap/expect acceptable in deterministic tests
+
+use crate::db;
+use crate::db::write_once_properties::{reopenable_pool, test_pool, unique_key};
+use dsm::storage::binding_record::{
+    empty_set_digest, keyset_digest, BindingRecord, Round, BINDING_RECORD_SCHEMA_V1,
+};
+
+fn keys(tag: u8, n: usize) -> Vec<[u8; 32]> {
+    let mut k: Vec<[u8; 32]> = (0..n).map(|_| unique_key(tag)).collect();
+    k.sort();
+    k
+}
+
+fn record(keys: &[[u8; 32]], counter: u64, proposer: u8, value: u8) -> (Vec<u8>, [u8; 32], Round) {
+    let r = BindingRecord {
+        schema: BINDING_RECORD_SCHEMA_V1,
+        round: Round {
+            counter,
+            proposer_id: [proposer; 32],
+        },
+        tx_id: [0xAA; 32],
+        keyset_digest: keyset_digest(keys),
+        value_digest: [value; 32],
+        value_addr: [value; 32],
+        status: 1,
+    };
+    (r.encode(), r.digest(), r.round)
+}
+
+async fn cas(
+    pool: &db::DBPool,
+    keys: &[[u8; 32]],
+    expected: [u8; 32],
+    rec: &(Vec<u8>, [u8; 32], Round),
+) -> db::CasOutcome {
+    db::compare_exchange_many(pool, keys, &expected, &rec.0, &rec.1, rec.2)
+        .await
+        .expect("cas")
+}
+
+/// All-or-none: a set whose expectation is wrong on ONE key applies to NONE,
+/// and the keys already held keep their record.
+#[tokio::test]
+async fn an_exchange_applies_to_every_key_or_to_none() {
+    let pool = test_pool();
+    db::init_db(&pool).await.expect("init");
+    let ks = keys(0x61, 3);
+    // Seed only the middle key so the set is partially held.
+    let seed = record(&ks[1..2], 1, 1, 0x11);
+    assert!(matches!(
+        cas(&pool, &ks[1..2], empty_set_digest(&ks[1..2]), &seed).await,
+        db::CasOutcome::Applied { .. }
+    ));
+    // A caller who expects the whole set empty is refused; nothing changes.
+    let a = record(&ks, 2, 2, 0xA1);
+    let out = cas(&pool, &ks, empty_set_digest(&ks), &a).await;
+    assert!(matches!(out, db::CasOutcome::ExpectationMismatch { .. }));
+    let held = db::read_bindings(&pool, &ks).await.expect("read");
+    assert!(
+        held[0].is_none() && held[2].is_none(),
+        "unheld keys stay unheld"
+    );
+    assert_eq!(
+        held[1].as_ref().unwrap().record_bytes,
+        seed.0,
+        "the held key keeps its record"
+    );
+    // With the true expectation (the read's digest) and a higher round, ALL
+    // three keys change together.
+    let db::CasOutcome::ExpectationMismatch { current_digest } = out else {
+        unreachable!()
+    };
+    assert!(matches!(
+        cas(&pool, &ks, current_digest, &a).await,
+        db::CasOutcome::Applied { .. }
+    ));
+    let held = db::read_bindings(&pool, &ks).await.expect("read");
+    assert!(held
+        .iter()
+        .all(|h| h.as_ref().map(|b| b.record_bytes == a.0).unwrap_or(false)));
+}
+
+/// N racers exchanging from the same (empty) expectation with distinct
+/// values: exactly one is APPLIED; every other is refused with the winner's
+/// resulting digest, and a read agrees.
+#[tokio::test]
+async fn concurrent_exchanges_from_one_expectation_yield_exactly_one_application() {
+    let pool = test_pool();
+    db::init_db(&pool).await.expect("init");
+    let ks = keys(0x62, 2);
+    let empty = empty_set_digest(&ks);
+    let mut handles = Vec::new();
+    for i in 0..16u8 {
+        let pool = pool.clone();
+        let ks = ks.clone();
+        handles.push(tokio::spawn(async move {
+            let rec = record(&ks, 1, i + 1, 0x70 + i);
+            cas(&pool, &ks, empty, &rec).await
+        }));
+    }
+    let mut applied = 0;
+    let mut reported = std::collections::BTreeSet::new();
+    for h in handles {
+        match h.await.expect("racer") {
+            db::CasOutcome::Applied { resulting_digest } => {
+                applied += 1;
+                reported.insert(resulting_digest);
+            }
+            db::CasOutcome::ExpectationMismatch { current_digest } => {
+                reported.insert(current_digest);
+            }
+        }
+    }
+    assert_eq!(applied, 1, "exactly one racer applies");
+    assert_eq!(reported.len(), 1, "every loser is told THE winner's digest");
+    let held = db::read_bindings(&pool, &ks).await.expect("read");
+    assert!(held.iter().all(|h| h.is_some()));
+}
+
+/// Rounds only move forward: after a higher round is held, a lower or equal
+/// round is refused even when the caller's expectation is exact.
+#[tokio::test]
+async fn a_round_never_moves_backwards() {
+    let pool = test_pool();
+    db::init_db(&pool).await.expect("init");
+    let ks = keys(0x63, 1);
+    let hi = record(&ks, 7, 3, 0xA1);
+    let db::CasOutcome::Applied { resulting_digest } =
+        cas(&pool, &ks, empty_set_digest(&ks), &hi).await
+    else {
+        panic!("first exchange applies")
+    };
+    let lower = record(&ks, 6, 0xFF, 0xB2);
+    assert!(matches!(
+        cas(&pool, &ks, resulting_digest, &lower).await,
+        db::CasOutcome::ExpectationMismatch { .. }
+    ));
+    let same_round_other_value = record(&ks, 7, 3, 0xC3);
+    assert!(matches!(
+        cas(&pool, &ks, resulting_digest, &same_round_other_value).await,
+        db::CasOutcome::ExpectationMismatch { .. }
+    ));
+    let held = db::read_bindings(&pool, &ks).await.expect("read");
+    assert_eq!(
+        held[0].as_ref().unwrap().record_bytes,
+        hi.0,
+        "the higher round is still held"
+    );
+}
+
+/// Overlapping key sets {A,B} and {B,C} exchanged from their own empty
+/// expectations: at most one can apply, because B can hold only one record.
+#[tokio::test]
+async fn overlapping_key_sets_serialise_on_the_shared_key() {
+    let pool = test_pool();
+    db::init_db(&pool).await.expect("init");
+    let mut abc = keys(0x64, 3);
+    abc.sort();
+    let (a, b, c) = (abc[0], abc[1], abc[2]);
+    let ab = vec![a, b];
+    let bc = vec![b, c];
+    let t1 = record(&ab, 1, 1, 0xA1);
+    let t2 = record(&bc, 1, 2, 0xB2);
+    let (p1, p2) = (pool.clone(), pool.clone());
+    let (ab1, bc1) = (ab.clone(), bc.clone());
+    let h1 = tokio::spawn(async move { cas(&p1, &ab1, empty_set_digest(&ab1), &t1).await });
+    let h2 = tokio::spawn(async move { cas(&p2, &bc1, empty_set_digest(&bc1), &t2).await });
+    let (o1, o2) = (h1.await.unwrap(), h2.await.unwrap());
+    let applied = [&o1, &o2]
+        .iter()
+        .filter(|o| matches!(o, db::CasOutcome::Applied { .. }))
+        .count();
+    assert_eq!(applied, 1, "B cannot belong to two transactions");
+    // Whichever lost sees the truth on B when it reads its own key set.
+    let held_b = db::read_bindings(&pool, &[b]).await.expect("read")[0].clone();
+    assert!(held_b.is_some(), "B is held by the winner");
+}
+
+/// A record set survives re-opening the store: the digest a caller was told
+/// is the digest it reads back, and a stale expectation is still refused.
+#[tokio::test]
+async fn a_record_set_survives_reopening_the_store() {
+    let path = std::env::temp_dir()
+        .join("dsm-binding-reopen")
+        .to_string_lossy()
+        .to_string();
+    let _ = std::fs::remove_file(&path);
+    let ks = keys(0x65, 2);
+    let a = record(&ks, 1, 1, 0xA1);
+    let told = {
+        let pool = reopenable_pool(&path);
+        db::init_db(&pool).await.expect("init");
+        let db::CasOutcome::Applied { resulting_digest } =
+            cas(&pool, &ks, empty_set_digest(&ks), &a).await
+        else {
+            panic!("applies")
+        };
+        resulting_digest
+    };
+    let pool = reopenable_pool(&path);
+    db::init_db(&pool).await.expect("init");
+    let held = db::read_bindings(&pool, &ks).await.expect("read");
+    assert!(held
+        .iter()
+        .all(|h| h.as_ref().map(|b| b.record_bytes == a.0).unwrap_or(false)));
+    let b = record(&ks, 2, 2, 0xB2);
+    assert!(matches!(
+        cas(&pool, &ks, empty_set_digest(&ks), &b).await,
+        db::CasOutcome::ExpectationMismatch { current_digest } if current_digest == told
+    ));
+    let _ = std::fs::remove_file(&path);
+}
diff --git a/dsm_storage_node/src/db/mod.rs b/dsm_storage_node/src/db/mod.rs
index 268963feb..201dc9b43 100644
--- a/dsm_storage_node/src/db/mod.rs
+++ b/dsm_storage_node/src/db/mod.rs
@@ -7,6 +7,10 @@
 //!
 //! Feature flag `local-dev` switches the implementation at compile time.
 
+/// The generic conditional-binding decision, one rule for both backends.
+pub mod binding;
+pub use binding::{CasOutcome, StoredBinding};
+
 #[cfg(not(feature = "local-dev"))]
 mod pg;
 
@@ -23,3 +27,8 @@ pub use sqlite::*;
 /// run against whichever backend is compiled.
 #[cfg(test)]
 pub(crate) mod write_once_properties;
+
+/// The generic conditional-binding properties, run against whichever
+/// backend is compiled.
+#[cfg(test)]
+pub(crate) mod binding_properties;
diff --git a/dsm_storage_node/src/db/pg.rs b/dsm_storage_node/src/db/pg.rs
index 56378925d..474219d6b 100644
--- a/dsm_storage_node/src/db/pg.rs
+++ b/dsm_storage_node/src/db/pg.rs
@@ -396,12 +396,159 @@ pub async fn register_incarnation(pool: &Pool) -> Result<[u8; 32]> {
         .map_err(|_| anyhow!("stored register incarnation is not 32 bytes"))
 }
 
+// ===================== Generic conditional binding (Rev 15 §15.5) =====================
+
+use super::binding::{decide_compare_exchange, CasOutcome, Replacement, StoredBinding};
+
+fn row_to_binding(row: &tokio_postgres::Row) -> Result<StoredBinding> {
+    let key: Vec<u8> = row.get(0);
+    let record_bytes: Vec<u8> = row.get(1);
+    let record_digest: Vec<u8> = row.get(2);
+    let round_counter: i64 = row.get(3);
+    let proposer_id: Vec<u8> = row.get(4);
+    Ok(StoredBinding {
+        key: key
+            .try_into()
+            .map_err(|_| anyhow!("stored binding key is not 32 bytes"))?,
+        record_bytes,
+        record_digest: record_digest
+            .try_into()
+            .map_err(|_| anyhow!("stored binding digest is not 32 bytes"))?,
+        round: dsm::storage::binding_record::Round {
+            counter: u64::try_from(round_counter)
+                .map_err(|_| anyhow!("stored round counter is negative"))?,
+            proposer_id: proposer_id
+                .try_into()
+                .map_err(|_| anyhow!("stored proposer id is not 32 bytes"))?,
+        },
+    })
+}
+
+/// `CompareExchangeMany` at ONE member (Req 15.6): every named key changes to
+/// the replacement record or none does, decided by [`decide_compare_exchange`]
+/// inside one durable transaction. This member knows nothing of `q`, contacts
+/// no peer, and decides nothing about what the record means.
+///
+/// Concurrency: a transaction-scoped advisory lock is taken per key, in the
+/// caller-supplied strictly ascending key order, so two exchanges over
+/// overlapping key sets serialise instead of deadlocking or interleaving.
+/// Absent keys cannot be row-locked, which is why the lock is advisory.
+#[allow(clippy::too_many_arguments)]
+pub async fn compare_exchange_many(
+    pool: &Pool,
+    keys: &[[u8; 32]],
+    expected_digest: &[u8; 32],
+    replacement_bytes: &[u8],
+    replacement_digest: &[u8; 32],
+    replacement_round: dsm::storage::binding_record::Round,
+) -> Result<CasOutcome> {
+    dsm::storage::binding_record::validate_key_set(keys)
+        .map_err(|e| anyhow!("compare_exchange_many: {e}"))?;
+    let mut client = pool.get().await?;
+    let tx = begin_durable_write(&mut client).await?;
+    // The lock id is the key's leading eight bytes as a signed 64-bit
+    // integer. Keys are 32-byte digests, so those bytes are uniformly
+    // distributed and a collision between distinct keys is negligible; and a
+    // collision would only serialise two exchanges that did not need it,
+    // never let two interleave. Computed here rather than in SQL so no text
+    // cast is involved.
+    for k in keys {
+        let lock_id = i64::from_be_bytes([k[0], k[1], k[2], k[3], k[4], k[5], k[6], k[7]]);
+        tx.execute("SELECT pg_advisory_xact_lock($1)", &[&lock_id])
+            .await?;
+    }
+    let key_vecs: Vec<Vec<u8>> = keys.iter().map(|k| k.to_vec()).collect();
+    let rows = tx
+        .query(
+            "SELECT key, record_bytes, record_digest, round_counter, proposer_id
+               FROM binding_records WHERE key = ANY($1)",
+            &[&key_vecs],
+        )
+        .await?;
+    let mut held = std::collections::BTreeMap::new();
+    for r in &rows {
+        let b = row_to_binding(r)?;
+        held.insert(b.key, b);
+    }
+    let outcome = decide_compare_exchange(
+        keys,
+        &held,
+        expected_digest,
+        &Replacement {
+            bytes: replacement_bytes,
+            digest: *replacement_digest,
+            round: replacement_round,
+        },
+    );
+    if let CasOutcome::Applied { .. } = &outcome {
+        let counter_i64 = i64::try_from(replacement_round.counter)
+            .map_err(|_| anyhow!("round counter does not fit in i64"))?;
+        let stmt = tx
+            .prepare_cached(
+                "INSERT INTO binding_records
+                   (key, record_bytes, record_digest, round_counter, proposer_id)
+                 VALUES ($1, $2, $3, $4, $5)
+                 ON CONFLICT (key) DO UPDATE SET
+                   record_bytes = EXCLUDED.record_bytes,
+                   record_digest = EXCLUDED.record_digest,
+                   round_counter = EXCLUDED.round_counter,
+                   proposer_id = EXCLUDED.proposer_id",
+            )
+            .await?;
+        for k in keys {
+            tx.execute(
+                &stmt,
+                &[
+                    &k.as_slice(),
+                    &replacement_bytes,
+                    &replacement_digest.as_slice(),
+                    &counter_i64,
+                    &replacement_round.proposer_id.as_slice(),
+                ],
+            )
+            .await?;
+        }
+    }
+    tx.commit().await?;
+    Ok(outcome)
+}
+
+/// `ReadBinding` at one member: what it holds for each key, in key order,
+/// `None` where it holds nothing. Absence is this member's assertion.
+pub async fn read_bindings(pool: &Pool, keys: &[[u8; 32]]) -> Result<Vec<Option<StoredBinding>>> {
+    dsm::storage::binding_record::validate_key_set(keys)
+        .map_err(|e| anyhow!("read_bindings: {e}"))?;
+    let client = pool.get().await?;
+    let key_vecs: Vec<Vec<u8>> = keys.iter().map(|k| k.to_vec()).collect();
+    let rows = client
+        .query(
+            "SELECT key, record_bytes, record_digest, round_counter, proposer_id
+               FROM binding_records WHERE key = ANY($1)",
+            &[&key_vecs],
+        )
+        .await?;
+    let mut held = std::collections::BTreeMap::new();
+    for r in &rows {
+        let b = row_to_binding(r)?;
+        held.insert(b.key, b);
+    }
+    Ok(keys.iter().map(|k| held.remove(k)).collect())
+}
+
 /// Initialize database schema for storage node.
 pub async fn init_db(pool: &Pool) -> Result<()> {
     let client = pool.get().await?;
     client
         .batch_execute(
-            r#"CREATE TABLE IF NOT EXISTS register_incarnation (
+            r#"CREATE TABLE IF NOT EXISTS binding_records (
+                    key           BYTEA PRIMARY KEY,
+                    record_bytes  BYTEA NOT NULL,
+                    record_digest BYTEA NOT NULL,
+                    round_counter BIGINT NOT NULL,
+                    proposer_id   BYTEA NOT NULL
+                );
+
+                CREATE TABLE IF NOT EXISTS register_incarnation (
                     only_row    SMALLINT PRIMARY KEY CHECK (only_row = 1),
                     incarnation BYTEA NOT NULL
                 );
diff --git a/dsm_storage_node/src/db/sqlite.rs b/dsm_storage_node/src/db/sqlite.rs
index a564eb9b9..a3dc7d972 100644
--- a/dsm_storage_node/src/db/sqlite.rs
+++ b/dsm_storage_node/src/db/sqlite.rs
@@ -193,11 +193,140 @@ pub async fn register_incarnation(pool: &DBPool) -> Result<[u8; 32]> {
     .await
 }
 
+// ===================== Generic conditional binding (Rev 15 §15.5) =====================
+
+use super::binding::{decide_compare_exchange, CasOutcome, Replacement, StoredBinding};
+
+/// One `binding_records` row as SQLite returns it, before it is decoded into a
+/// [`StoredBinding`]: `(record_bytes, record_digest, round_counter, proposer_id)`.
+type BindingRow = (Vec<u8>, Vec<u8>, i64, Vec<u8>);
+
+fn read_held(
+    conn: &rusqlite::Connection,
+    keys: &[[u8; 32]],
+) -> Result<std::collections::BTreeMap<[u8; 32], StoredBinding>> {
+    let mut held = std::collections::BTreeMap::new();
+    let mut stmt = conn.prepare_cached(
+        "SELECT record_bytes, record_digest, round_counter, proposer_id
+           FROM binding_records WHERE key = ?1",
+    )?;
+    for k in keys {
+        let row: Option<BindingRow> = stmt
+            .query_row(params![k.to_vec()], |r| {
+                Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?))
+            })
+            .optional()?;
+        if let Some((record_bytes, record_digest, counter, proposer_id)) = row {
+            held.insert(
+                *k,
+                StoredBinding {
+                    key: *k,
+                    record_bytes,
+                    record_digest: record_digest
+                        .try_into()
+                        .map_err(|_| anyhow!("stored binding digest is not 32 bytes"))?,
+                    round: dsm::storage::binding_record::Round {
+                        counter: u64::try_from(counter)
+                            .map_err(|_| anyhow!("stored round counter is negative"))?,
+                        proposer_id: proposer_id
+                            .try_into()
+                            .map_err(|_| anyhow!("stored proposer id is not 32 bytes"))?,
+                    },
+                },
+            );
+        }
+    }
+    Ok(held)
+}
+
+/// `CompareExchangeMany` at ONE member (Req 15.6): all named keys change to
+/// the replacement or none do, decided by the shared
+/// [`decide_compare_exchange`] inside one durable transaction. The single
+/// connection behind `with_conn` serialises exchanges, so overlapping key
+/// sets cannot interleave.
+pub async fn compare_exchange_many(
+    pool: &DBPool,
+    keys: &[[u8; 32]],
+    expected_digest: &[u8; 32],
+    replacement_bytes: &[u8],
+    replacement_digest: &[u8; 32],
+    replacement_round: dsm::storage::binding_record::Round,
+) -> Result<CasOutcome> {
+    dsm::storage::binding_record::validate_key_set(keys)
+        .map_err(|e| anyhow!("compare_exchange_many: {e}"))?;
+    let keys = keys.to_vec();
+    let expected = *expected_digest;
+    let bytes = replacement_bytes.to_vec();
+    let digest = *replacement_digest;
+    with_conn(pool, move |conn| {
+        conn.execute_batch("PRAGMA synchronous=FULL;")?;
+        let tx = conn.unchecked_transaction()?;
+        let held = read_held(&tx, &keys)?;
+        let outcome = decide_compare_exchange(
+            &keys,
+            &held,
+            &expected,
+            &Replacement {
+                bytes: &bytes,
+                digest,
+                round: replacement_round,
+            },
+        );
+        if let CasOutcome::Applied { .. } = &outcome {
+            let counter_i64 = i64::try_from(replacement_round.counter)
+                .map_err(|_| anyhow!("round counter does not fit in i64"))?;
+            for k in &keys {
+                tx.execute(
+                    "INSERT INTO binding_records
+                       (key, record_bytes, record_digest, round_counter, proposer_id)
+                     VALUES (?1, ?2, ?3, ?4, ?5)
+                     ON CONFLICT(key) DO UPDATE SET
+                       record_bytes = excluded.record_bytes,
+                       record_digest = excluded.record_digest,
+                       round_counter = excluded.round_counter,
+                       proposer_id = excluded.proposer_id",
+                    params![
+                        k.to_vec(),
+                        bytes,
+                        digest.to_vec(),
+                        counter_i64,
+                        replacement_round.proposer_id.to_vec()
+                    ],
+                )?;
+            }
+        }
+        tx.commit()?;
+        Ok(outcome)
+    })
+    .await
+}
+
+/// `ReadBinding` at one member: what it holds for each key, in key order,
+/// `None` where it holds nothing. Absence is this member's assertion.
+pub async fn read_bindings(pool: &DBPool, keys: &[[u8; 32]]) -> Result<Vec<Option<StoredBinding>>> {
+    dsm::storage::binding_record::validate_key_set(keys)
+        .map_err(|e| anyhow!("read_bindings: {e}"))?;
+    let keys = keys.to_vec();
+    with_conn(pool, move |conn| {
+        let mut held = read_held(conn, &keys)?;
+        Ok(keys.iter().map(|k| held.remove(k)).collect())
+    })
+    .await
+}
+
 /// Initialize database schema (SQLite version).
 pub async fn init_db(pool: &DBPool) -> Result<()> {
     with_conn(pool, |conn| {
         conn.execute_batch(
-            r#"CREATE TABLE IF NOT EXISTS register_incarnation (
+            r#"CREATE TABLE IF NOT EXISTS binding_records (
+                    key           BLOB PRIMARY KEY,
+                    record_bytes  BLOB NOT NULL,
+                    record_digest BLOB NOT NULL,
+                    round_counter INTEGER NOT NULL,
+                    proposer_id   BLOB NOT NULL
+                );
+
+                CREATE TABLE IF NOT EXISTS register_incarnation (
                     only_row    INTEGER PRIMARY KEY CHECK (only_row = 1),
                     incarnation BLOB NOT NULL
                 );
diff --git a/dsm_storage_node/src/lib.rs b/dsm_storage_node/src/lib.rs
index 6f20b31a8..ef5fa1cbd 100644
--- a/dsm_storage_node/src/lib.rs
+++ b/dsm_storage_node/src/lib.rs
@@ -20,6 +20,13 @@ use replication::StorageNodeId;
 #[derive(Clone)]
 pub struct AppState {
     pub node_id: StorageNodeId,
+    /// The protocol identity EXACTLY as configured (`[node] id`) — the string
+    /// a client's catalog names, the value the identity echo layer emits, and
+    /// therefore the `member_id` every generic-binding answer carries. Kept
+    /// beside the canonical 32-byte `node_id` because the two are different
+    /// facts: one is how peers address this node, the other is who a quorum
+    /// counts.
+    pub configured_member_id: String,
     pub hsts_max_age: Option<u64>,
     pub db_pool: Arc<db::DBPool>,
     pub replication_manager: Arc<replication::ReplicationManager>,
@@ -29,6 +36,11 @@ pub struct AppState {
     /// it). `None` = not configured: the settlement-slot register refuses every
     /// claim (fail closed) rather than accepting claims for an unknown set.
     pub storage_set: Option<Arc<NodeStorageSet>>,
+    /// The register incarnation THIS node is serving — minted once into its
+    /// own database at first boot, established before anything is served.
+    /// Stamped on every generic-binding answer so a caller can tell this
+    /// register history from a rebuilt one wearing the same node id.
+    pub own_register_incarnation: Option<[u8; 32]>,
     /// The DSM network this node serves (`node.network_id` in config). Gates
     /// the ERA faucet-ticket register: the canonical faucet identity is
     /// NETWORK-SCOPED (`era_faucet_id(network_id)`), so a node that does not
@@ -117,15 +129,18 @@ impl AppState {
         db_pool: Arc<db::DBPool>,
         replication_manager: Arc<replication::ReplicationManager>,
     ) -> Self {
+        let configured_member_id = node_id_input.clone();
         let node_id =
             StorageNodeId::from_base32_or_derive(&node_id_input, address_or_seed.as_bytes());
         Self {
             node_id,
+            configured_member_id,
             hsts_max_age,
             db_pool,
             replication_manager,
             current_tick: Arc::new(AtomicI64::new(0)),
             storage_set: None,
+            own_register_incarnation: None,
             network_id: None,
         }
     }
@@ -136,6 +151,12 @@ impl AppState {
         self
     }
 
+    /// Record the register incarnation this node established at startup.
+    pub fn with_register_incarnation(mut self, incarnation: [u8; 32]) -> Self {
+        self.own_register_incarnation = Some(incarnation);
+        self
+    }
+
     pub fn with_storage_set(mut self, set: NodeStorageSet) -> Self {
         self.storage_set = Some(Arc::new(set));
         self
@@ -161,6 +182,27 @@ pub fn economic_register_write_router(state: Arc<AppState>) -> axum::Router<()>
         .layer(Extension(state))
 }
 
+/// The write half of the generic conditional-binding interface (Rev 15
+/// §15.5): `CompareExchangeMany` behind device auth. The node relates the
+/// caller to nothing — there is no claimant in a generic record — device
+/// auth only says a registered device is writing.
+pub fn generic_binding_write_router(state: Arc<AppState>) -> axum::Router<()> {
+    let auth_state = Arc::new(auth::AuthState {
+        db_pool: state.db_pool.clone(),
+    });
+    api::storage::binding::create_write_router()
+        .layer(axum::middleware::from_fn_with_state(
+            auth_state,
+            auth::device_auth,
+        ))
+        .layer(Extension(state))
+}
+
+/// The public read half of the same interface: `ReadBinding`.
+pub fn generic_binding_read_router(state: Arc<AppState>) -> axum::Router<()> {
+    api::storage::binding::create_read_router(state)
+}
+
 /// The public READ half of the same two registers. The binary rate-limits it;
 /// the conformance suite mounts it bare.
 pub fn economic_register_read_router(state: Arc<AppState>) -> axum::Router<()> {
diff --git a/dsm_storage_node/src/main.rs b/dsm_storage_node/src/main.rs
index 985cfb809..6505946f0 100644
--- a/dsm_storage_node/src/main.rs
+++ b/dsm_storage_node/src/main.rs
@@ -321,6 +321,12 @@ fn build_router(state: Arc<AppState>, config: &ServerConfig, benchmark_mode: boo
     // every response is NORMATIVE for these registers: quorum reads count a
     // response only when the echo equals the member queried. Assembled by
     // the library so the conformance suite drives exactly what is served.
+    // Generic conditional binding (Rev 15 §15.5): application-blind
+    // CompareExchangeMany behind device auth, ReadBinding public.
+    let generic_binding_write_router =
+        dsm_storage_node::generic_binding_write_router(state.clone());
+    let generic_binding_read_router = dsm_storage_node::generic_binding_read_router(state.clone())
+        .layer(public_rate_layer.clone());
     let economic_register_write_router =
         dsm_storage_node::economic_register_write_router(state.clone());
     let economic_register_read_router =
@@ -373,6 +379,8 @@ fn build_router(state: Arc<AppState>, config: &ServerConfig, benchmark_mode: boo
         .merge(dlv_slot_router)
         .merge(slot_claim_write_router)
         .merge(economic_register_write_router)
+        .merge(generic_binding_write_router)
+        .merge(generic_binding_read_router)
         .merge(slot_claim_read_router)
         .merge(economic_register_read_router)
         .merge(recovery_capsule_router)
@@ -530,6 +538,7 @@ async fn async_main() -> Result<()> {
     let own_incarnation = db::register_incarnation(&db_pool)
         .await
         .context("failed to establish this node's register incarnation")?;
+    state = state.with_register_incarnation(own_incarnation);
     log::info!(
         "register incarnation for node {}: {}",
         server_config.node_id,
diff --git a/proto/dsm_app.proto b/proto/dsm_app.proto
index 49417daf6..26642578f 100644
--- a/proto/dsm_app.proto
+++ b/proto/dsm_app.proto
@@ -1653,6 +1653,81 @@ message SettlementSlotClaimV2 {
   bytes  signature               = 2 [(dsm_max_len)=65535];
 }
 
+// ── Generic conditional-binding interface (SoFi Rev 15 §15.5, Def 6.20) ──
+//
+// The node-side binding interface is APPLICATION-BLIND. A member holds at most
+// one opaque generic binding record per resource key and applies a
+// replacement to a whole strictly-sorted key set atomically or not at all
+// (Req 15.6). It inspects ONLY the generic storage fields — schema, round
+// ordering, the exact expected digest of the prior record set, and key-set
+// equality (Req 15.7). It never decodes the value at `value_addr` and attaches
+// no settlement, vault, claimant, or economic meaning to any field (§22 #12).
+// Class K — never the node — decides what a record means (§15.6).
+message GenericBindingRecordV1 {
+  uint32 schema        = 1;
+  // round = (counter, proposer_id), ordered lexicographically. `counter` is a
+  // proposer-local persisted monotonic integer — never a timestamp. A
+  // recovering proposer chooses a round strictly greater than every round it
+  // must supersede (Def 6.20).
+  uint64 round_counter = 2;
+  bytes  proposer_id   = 3 [(dsm_fixed_len)=32];
+  bytes  tx_id         = 4 [(dsm_fixed_len)=32];
+  // H(DSM/binding-record-set-keys || sorted keys). The node requires it to
+  // equal the digest of the request's own key set (key-set equality).
+  bytes  keyset_digest = 5 [(dsm_fixed_len)=32];
+  bytes  value_digest  = 6 [(dsm_fixed_len)=32];
+  bytes  value_addr    = 7 [(dsm_fixed_len)=32];
+  // Opaque to the node. Class K assigns meaning (prepared / accepted / …).
+  uint32 status        = 8;
+}
+
+message CompareExchangeManyRequestV1 {
+  // Strictly sorted, distinct, opaque 32-byte resource keys.
+  repeated bytes keys                = 1 [(dsm_fixed_len)=32];
+  // Digest of the EXACT prior generic record set over `keys` (absent cells
+  // included as absent). The empty set has a defined digest so a first writer
+  // can exchange from nothing.
+  bytes  expected_digest             = 2 [(dsm_fixed_len)=32];
+  GenericBindingRecordV1 replacement = 3;
+}
+
+message CompareExchangeManyResponseV1 {
+  enum Outcome {
+    APPLIED                  = 0;
+    EXPECTATION_MISMATCH     = 1;
+    UNAVAILABLE              = 2;
+    INVALID_STORAGE_ENCODING = 3;
+  }
+  Outcome outcome              = 1;
+  // Digest of the record set over `keys` AFTER this call — what a caller
+  // must present as `expected_digest` next time.
+  bytes   resulting_digest     = 2 [(dsm_fixed_len)=32];
+  bytes   member_id            = 3 [(dsm_max_len)=256];
+  // The register incarnation this member is serving. A write acknowledgement
+  // counts toward a quorum only when BOTH member_id and register_incarnation
+  // equal what the caller committed — the same rule the read side applies.
+  bytes   register_incarnation = 4 [(dsm_fixed_len)=32];
+}
+
+message ReadBindingRequestV1 {
+  repeated bytes keys = 1 [(dsm_fixed_len)=32];
+}
+
+// One cell of the answer. `record` is absent when the member holds nothing
+// for `key`; absence is an assertion by the member, never an inference.
+message BindingCellV1 {
+  bytes key                     = 1 [(dsm_fixed_len)=32];
+  GenericBindingRecordV1 record = 2;
+}
+
+message ReadBindingResponseV1 {
+  repeated BindingCellV1 cells = 1;
+  // Digest of the exact current record set over the requested keys.
+  bytes  set_digest            = 2 [(dsm_fixed_len)=32];
+  bytes  member_id             = 3 [(dsm_max_len)=256];
+  bytes  register_incarnation  = 4 [(dsm_fixed_len)=32];
+}
+
 // A trader's claim on one cell of the economic root register: the write-once
 // value at K_root = H("DSM/trader-economic-root-register-key/v1" || 0x00 || G
 // || DevID || u64_be(economic_position)).