diff --git a/rust/bt-csis/Cargo.toml b/rust/bt-csis/Cargo.toml
index 4eeae2e..54b3d09 100644
--- a/rust/bt-csis/Cargo.toml
+++ b/rust/bt-csis/Cargo.toml
@@ -7,9 +7,10 @@
 [dependencies]
 bt-common.workspace = true
 bt-gatt.workspace = true
+futures.workspace = true
+parking_lot.workspace = true
 thiserror.workspace = true
 
 [dev-dependencies]
 assert_matches.workspace = true
 bt-gatt = { workspace = true, features = ["test-utils"] }
-futures.workspace = true
diff --git a/rust/bt-csis/src/client.rs b/rust/bt-csis/src/client.rs
index 38c1e86..125d6e2 100644
--- a/rust/bt-csis/src/client.rs
+++ b/rust/bt-csis/src/client.rs
@@ -2,15 +2,23 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
+use std::sync::Arc;
+
+use futures::stream::{SelectAll, StreamExt};
+use parking_lot::Mutex;
 use thiserror::Error;
 
 use bt_common::packet_encoding::Decodable;
 use bt_common::Uuid;
 use bt_gatt::client::{PeerService, ServiceCharacteristic};
-use bt_gatt::types::Handle;
+use bt_gatt::types::{CharacteristicProperty, Handle};
 
 use crate::types::*;
 
+pub mod event;
+
+pub use event::{CsisNotification, NotificationStream};
+
 #[derive(Debug, Error)]
 pub enum Error {
     #[error("GATT operation error: {0}")]
@@ -27,55 +35,134 @@
 
     #[error("Invalid characteristic value: {0}")]
     InvalidCharacteristic(String),
+
+    #[error("Notification from unexpected characteristic handle: {0:?}")]
+    UnexpectedNotification(Handle),
+}
+
+/// The values of the characteristics exposed by the CSIS service.
+#[derive(Debug, Clone)]
+pub(crate) struct CsisClientState {
+    sirk: SetIdentityResolvingKey,
+    size: Option<CoordinatedSetSize>,
+    rank: Option<SetMemberRank>,
+    lock_state: Option<SetMemberLock>,
+}
+
+impl CsisClientState {
+    pub(crate) fn sirk(&self) -> SetIdentityResolvingKey {
+        self.sirk
+    }
+
+    pub(crate) fn set_sirk(&mut self, sirk: SetIdentityResolvingKey) {
+        self.sirk = sirk;
+    }
+
+    pub(crate) fn size(&self) -> Option<CoordinatedSetSize> {
+        self.size
+    }
+
+    pub(crate) fn set_size(&mut self, size: CoordinatedSetSize) {
+        self.size = Some(size);
+    }
+
+    pub(crate) fn rank(&self) -> Option<SetMemberRank> {
+        self.rank
+    }
+
+    pub(crate) fn lock_state(&self) -> Option<SetMemberLock> {
+        self.lock_state
+    }
+
+    pub(crate) fn set_lock_state(&mut self, lock_state: SetMemberLock) {
+        self.lock_state = Some(lock_state);
+    }
 }
 
 /// A GATT client for interacting with the Coordinated Set Identification
 /// Service (CSIS). See CSIS v1.1 Section 2 & Section 5 and CSIP v1.1 Section
 /// 4.2 & Section 4.3.
 pub struct CoordinatedSetIdentificationServiceClient<T: bt_gatt::GattTypes> {
-    #[allow(dead_code)]
+    // The connection to this CSIS service instance. Held to keep it, and the
+    // notification subscriptions registered through it, alive.
     gatt_client: T::PeerService,
-    #[allow(dead_code)]
     sirk_handle: Handle,
-    #[allow(dead_code)]
     size_handle: Option<Handle>,
-    #[allow(dead_code)]
     lock_handle: Option<Handle>,
-    #[allow(dead_code)]
-    rank_handle: Option<Handle>,
+    /// Set Member Rank is not notifiable, so a change is only seen by
+    /// re-reading.
+    // TODO(b/534436497): implement re-reading.
+    _rank_handle: Option<Handle>,
 
-    // Initial read values
-    sirk: SetIdentityResolvingKey,
-    size: Option<CoordinatedSetSize>,
-    rank: Option<SetMemberRank>,
+    /// Shared with the [`NotificationStream`] so that a value updated by a
+    /// GATT notification is visible through this client's accessors.
+    state: Arc<Mutex<CsisClientState>>,
+
+    /// Handles of the notifiable characteristics. Taken by
+    /// [`Self::take_notification_stream`].
+    notifiable_handles: Option<Vec<Handle>>,
 }
 
 impl<T: bt_gatt::GattTypes> CoordinatedSetIdentificationServiceClient<T> {
-    /// Discovers CSIS characteristics and reads initial characteristic values.
-    /// See CSIS v1.1 Section 5, Table 5.1 and CSIP v1.1 Section 4.3.
+    /// Discovers CSIS characteristics and reads their initial values. See CSIS
+    /// v1.1 Section 5, Table 5.1 and CSIP v1.1 Section 4.3.
+    ///
+    /// Does not register for notifications; call
+    /// [`Self::take_notification_stream`] to subscribe.
     pub async fn create(gatt_client: T::PeerService) -> Result<Self, Error> {
         let discovered = gatt_client.discover_characteristics(None).await?;
 
-        let mut sirk_chars = Vec::new();
-        let mut size_chars = Vec::new();
-        let mut rank_chars = Vec::new();
-        let mut lock_chars = Vec::new();
+        let mut sirk = None;
+        let mut size = None;
+        let mut rank = None;
+        let mut lock = None;
 
+        let mut notifiable_handles = Vec::new();
+
+        // The CSIS service exposes at most one of each characteristic.
         for ch in discovered {
-            let service_char = ServiceCharacteristic::new(&gatt_client, ch);
-            match service_char.uuid() {
-                SET_IDENTITY_RESOLVING_KEY_UUID => sirk_chars.push(service_char),
-                COORDINATED_SET_SIZE_UUID => size_chars.push(service_char),
-                SET_MEMBER_RANK_UUID => rank_chars.push(service_char),
-                SET_MEMBER_LOCK_UUID => lock_chars.push(service_char),
-                _ => {}
+            let chr = ServiceCharacteristic::new(&gatt_client, ch);
+            let uuid = chr.uuid();
+            let handle = *chr.handle();
+
+            match uuid {
+                SET_IDENTITY_RESOLVING_KEY_UUID => {
+                    if sirk.is_some() {
+                        return Err(Error::ExtraCharacteristic(uuid));
+                    }
+                    sirk = Some((handle, Self::read_sirk(&chr).await?));
+                }
+                COORDINATED_SET_SIZE_UUID => {
+                    if size.is_some() {
+                        return Err(Error::ExtraCharacteristic(uuid));
+                    }
+                    size = Some((handle, Self::read_size(&chr).await?));
+                }
+                SET_MEMBER_LOCK_UUID => {
+                    if lock.is_some() {
+                        return Err(Error::ExtraCharacteristic(uuid));
+                    }
+                    lock = Some((handle, Self::read_lock(&chr).await?));
+                }
+                SET_MEMBER_RANK_UUID => {
+                    if rank.is_some() {
+                        return Err(Error::ExtraCharacteristic(uuid));
+                    }
+                    rank = Some((handle, Self::read_rank(&chr).await?));
+                }
+                _ => continue,
+            }
+
+            if chr.characteristic().properties.contains(CharacteristicProperty::Notify) {
+                notifiable_handles.push(handle);
             }
         }
 
-        let (sirk_handle, sirk) = Self::read_sirk(&sirk_chars).await?;
-        let (size_handle, size) = Self::read_size(&size_chars).await?;
-        let (rank_handle, rank) = Self::read_rank(&rank_chars).await?;
-        let lock_handle = Self::get_lock_handle(&lock_chars)?;
+        let (sirk_handle, sirk) =
+            sirk.ok_or(Error::MissingCharacteristic(SET_IDENTITY_RESOLVING_KEY_UUID))?;
+        let (size_handle, size) = size.unzip();
+        let (rank_handle, rank) = rank.unzip();
+        let (lock_handle, lock_state) = lock.unzip();
 
         if lock_handle.is_some() && rank_handle.is_none() {
             // CSIS v1.1 Section 5, Table 5.1 Note C.1: Mandatory if Set Member Lock is
@@ -96,98 +183,100 @@
             sirk_handle,
             size_handle,
             lock_handle,
-            rank_handle,
-            sirk,
-            size,
-            rank,
+            _rank_handle: rank_handle,
+            state: Arc::new(Mutex::new(CsisClientState { sirk, size, rank, lock_state })),
+            notifiable_handles: Some(notifiable_handles),
         })
     }
 
+    // TODO(b/534436497): Implement FromCharacteristic for the CSIS value types
+    // in place of these readers.
     async fn read_sirk(
-        chars: &[ServiceCharacteristic<'_, T>],
-    ) -> Result<(Handle, SetIdentityResolvingKey), Error> {
-        if chars.is_empty() {
-            return Err(Error::MissingCharacteristic(SET_IDENTITY_RESOLVING_KEY_UUID));
-        }
-        if chars.len() > 1 {
-            return Err(Error::ExtraCharacteristic(SET_IDENTITY_RESOLVING_KEY_UUID));
-        }
-        let char_item = &chars[0];
-        let handle = *char_item.handle();
-
+        chr: &ServiceCharacteristic<'_, T>,
+    ) -> Result<SetIdentityResolvingKey, Error> {
         let mut buf = [0; SetIdentityResolvingKey::BYTE_SIZE];
-        let bytes_read = char_item.read(&mut buf).await?;
+        let bytes_read = chr.read(&mut buf).await?;
         let (sirk_res, _) = SetIdentityResolvingKey::decode(&buf[..bytes_read]);
-        let sirk = sirk_res?;
-
-        Ok((handle, sirk))
+        Ok(sirk_res?)
     }
 
-    async fn read_size(
-        chars: &[ServiceCharacteristic<'_, T>],
-    ) -> Result<(Option<Handle>, Option<CoordinatedSetSize>), Error> {
-        if chars.is_empty() {
-            return Ok((None, None));
-        }
-        if chars.len() > 1 {
-            return Err(Error::ExtraCharacteristic(COORDINATED_SET_SIZE_UUID));
-        }
-        let char_item = &chars[0];
-        let handle = *char_item.handle();
-
+    async fn read_size(chr: &ServiceCharacteristic<'_, T>) -> Result<CoordinatedSetSize, Error> {
         let mut buf = [0; CoordinatedSetSize::BYTE_SIZE];
-        let bytes_read = char_item.read(&mut buf).await?;
+        let bytes_read = chr.read(&mut buf).await?;
         let (size_res, _) = CoordinatedSetSize::decode(&buf[..bytes_read]);
-        let size = size_res?;
-
-        Ok((Some(handle), Some(size)))
+        Ok(size_res?)
     }
 
-    async fn read_rank(
-        chars: &[ServiceCharacteristic<'_, T>],
-    ) -> Result<(Option<Handle>, Option<SetMemberRank>), Error> {
-        if chars.is_empty() {
-            return Ok((None, None));
-        }
-        if chars.len() > 1 {
-            return Err(Error::ExtraCharacteristic(SET_MEMBER_RANK_UUID));
-        }
-        let char_item = &chars[0];
-        let handle = *char_item.handle();
-
+    async fn read_rank(chr: &ServiceCharacteristic<'_, T>) -> Result<SetMemberRank, Error> {
         let mut buf = [0; SetMemberRank::BYTE_SIZE];
-        let bytes_read = char_item.read(&mut buf).await?;
+        let bytes_read = chr.read(&mut buf).await?;
         let (rank_res, _) = SetMemberRank::decode(&buf[..bytes_read]);
-        let rank = rank_res?;
-
-        Ok((Some(handle), Some(rank)))
+        Ok(rank_res?)
     }
 
-    fn get_lock_handle(chars: &[ServiceCharacteristic<'_, T>]) -> Result<Option<Handle>, Error> {
-        if chars.is_empty() {
-            return Ok(None);
+    async fn read_lock(chr: &ServiceCharacteristic<'_, T>) -> Result<SetMemberLock, Error> {
+        // CSIS v1.1 Table 5.1: Notify is mandatory for Set Member Lock.
+        if !chr.characteristic().properties.contains(CharacteristicProperty::Notify) {
+            return Err(Error::InvalidCharacteristic(
+                "Set Member Lock characteristic must support notifications".to_string(),
+            ));
         }
-        if chars.len() > 1 {
-            return Err(Error::ExtraCharacteristic(SET_MEMBER_LOCK_UUID));
-        }
-        Ok(Some(*chars[0].handle()))
+
+        let mut buf = [0; SetMemberLock::BYTE_SIZE];
+        let bytes_read = chr.read(&mut buf).await?;
+        let (lock_res, _) = SetMemberLock::decode(&buf[..bytes_read]);
+        Ok(lock_res?)
     }
 
     pub fn sirk(&self) -> SetIdentityResolvingKey {
-        self.sirk
+        self.state.lock().sirk()
     }
 
+    /// The last known value of the Coordinated Set Size characteristic, or
+    /// `None` if the Set Member does not support it.
     pub fn size(&self) -> Option<CoordinatedSetSize> {
-        self.size
+        self.state.lock().size()
     }
 
+    /// The last known value of the Set Member Rank characteristic, or `None` if
+    /// the Set Member does not support it.
     pub fn rank(&self) -> Option<SetMemberRank> {
-        self.rank
+        self.state.lock().rank()
     }
 
-    // TODO(b/534436497): Add lock() and unlock() methods for Exclusive
-    // Access & Set Member Locking. TODO(b/534436497): Add notification stream
-    // support for Lock and Size characteristics.
+    /// The last known value of the Set Member Lock characteristic, or `None` if
+    /// the Set Member does not support it.
+    pub fn lock_state(&self) -> Option<SetMemberLock> {
+        self.state.lock().lock_state()
+    }
+
+    /// Subscribes to the CSIS characteristics that support being notified and
+    /// takes the stream of the value changes they report.
+    ///
+    /// The stream is only available once; subsequent calls return `None`. It
+    /// yields no events if the Set Member does not support notifications on
+    /// any CSIS characteristic.
+    pub fn take_notification_stream(&mut self) -> Option<NotificationStream>
+    where
+        <T as bt_gatt::GattTypes>::NotificationStream: Send,
+    {
+        let handles = self.notifiable_handles.take()?;
+        let mut notification_streams = SelectAll::new();
+        for handle in handles {
+            notification_streams.push(self.gatt_client.subscribe(&handle).boxed());
+        }
+        Some(NotificationStream::new(
+            notification_streams,
+            event::NotifiedHandles {
+                sirk: self.sirk_handle,
+                size: self.size_handle,
+                lock: self.lock_handle,
+            },
+            self.state.clone(),
+        ))
+    }
+
+    // TODO(b/534436497): Add lock() and unlock(). See CSIS v1.1 Section 5.3.1.
 }
 
 #[cfg(test)]
@@ -222,6 +311,26 @@
         );
     }
 
+    /// Adds a Set Member Lock characteristic with Read, Write and Notify
+    /// properties.
+    fn add_lock_char(service: &mut FakePeerService, handle: Handle, value: Vec<u8>) {
+        service.add_characteristic(
+            Characteristic {
+                handle,
+                uuid: SET_MEMBER_LOCK_UUID,
+                properties: (&[
+                    CharacteristicProperty::Read,
+                    CharacteristicProperty::Write,
+                    CharacteristicProperty::Notify,
+                ])
+                    .into(),
+                permissions: AttributePermissions::default(),
+                descriptors: vec![],
+            },
+            value,
+        );
+    }
+
     #[test]
     fn create_success_all_chars() {
         let mut service = FakePeerService::new();
@@ -233,7 +342,7 @@
         add_char(&mut service, SIRK_HANDLE, SET_IDENTITY_RESOLVING_KEY_UUID, sirk_val);
 
         add_char(&mut service, SIZE_HANDLE, COORDINATED_SET_SIZE_UUID, vec![0x02]);
-        add_char(&mut service, LOCK_HANDLE, SET_MEMBER_LOCK_UUID, vec![0x01]); // 0x01 = Unlocked per CSIS v1.1 Section 5.3, Table 5.4
+        add_lock_char(&mut service, LOCK_HANDLE, vec![0x01]); // 0x01 = Unlocked per CSIS v1.1 Section 5.3, Table 5.4
         add_char(&mut service, RANK_HANDLE, SET_MEMBER_RANK_UUID, vec![0x01]);
 
         let create_fut = CoordinatedSetIdentificationServiceClient::<FakeTypes>::create(service);
@@ -246,11 +355,12 @@
         assert_eq!(client.sirk_handle, SIRK_HANDLE);
         assert_eq!(client.size_handle, Some(SIZE_HANDLE));
         assert_eq!(client.lock_handle, Some(LOCK_HANDLE));
-        assert_eq!(client.rank_handle, Some(RANK_HANDLE));
+        assert_eq!(client._rank_handle, Some(RANK_HANDLE));
 
         assert_eq!(client.sirk().sirk_type, SirkType::Plaintext);
         assert_eq!(client.size(), Some(CoordinatedSetSize(NonZeroU8::new(2).unwrap())));
         assert_eq!(client.rank(), Some(SetMemberRank(NonZeroU8::new(1).unwrap())));
+        assert_eq!(client.lock_state(), Some(SetMemberLock::Unlocked));
     }
 
     #[test]
@@ -272,7 +382,8 @@
         assert_eq!(client.sirk_handle, SIRK_HANDLE);
         assert_eq!(client.size_handle, None);
         assert_eq!(client.lock_handle, None);
-        assert_eq!(client.rank_handle, None);
+        assert_eq!(client._rank_handle, None);
+        assert_eq!(client.lock_state(), None);
     }
 
     #[test]
@@ -295,7 +406,7 @@
         assert_eq!(client.sirk_handle, SIRK_HANDLE);
         assert_eq!(client.size_handle, None);
         assert_eq!(client.lock_handle, None);
-        assert_eq!(client.rank_handle, Some(RANK_HANDLE));
+        assert_eq!(client._rank_handle, Some(RANK_HANDLE));
         assert_eq!(client.rank().unwrap().0.get(), 1);
     }
 
@@ -340,7 +451,7 @@
         sirk_val.extend_from_slice(&[0xAB; 16]);
         add_char(&mut service, SIRK_HANDLE, SET_IDENTITY_RESOLVING_KEY_UUID, sirk_val);
 
-        add_char(&mut service, LOCK_HANDLE, SET_MEMBER_LOCK_UUID, vec![0x01]);
+        add_lock_char(&mut service, LOCK_HANDLE, vec![0x01]);
         // Missing Rank
 
         let create_fut = CoordinatedSetIdentificationServiceClient::<FakeTypes>::create(service);
@@ -441,8 +552,8 @@
         sirk_val.extend_from_slice(&[0xAB; 16]);
         add_char(&mut service, SIRK_HANDLE, SET_IDENTITY_RESOLVING_KEY_UUID, sirk_val);
 
-        add_char(&mut service, LOCK_HANDLE, SET_MEMBER_LOCK_UUID, vec![0x01]);
-        add_char(&mut service, Handle(5), SET_MEMBER_LOCK_UUID, vec![0x01]); // Duplicate
+        add_lock_char(&mut service, LOCK_HANDLE, vec![0x01]);
+        add_lock_char(&mut service, Handle(5), vec![0x01]); // Duplicate
         add_char(&mut service, RANK_HANDLE, SET_MEMBER_RANK_UUID, vec![0x01]); // Rank is mandatory if lock present per CSIS v1.1 Table 5.1 Note C.1
 
         let create_fut = CoordinatedSetIdentificationServiceClient::<FakeTypes>::create(service);
@@ -455,6 +566,27 @@
     }
 
     #[test]
+    fn create_fails_lock_without_notify() {
+        let mut service = FakePeerService::new();
+        let mut noop_cx = Context::from_waker(noop_waker_ref());
+
+        let mut sirk_val = vec![0x01];
+        sirk_val.extend_from_slice(&[0xAB; 16]);
+        add_char(&mut service, SIRK_HANDLE, SET_IDENTITY_RESOLVING_KEY_UUID, sirk_val);
+
+        // A Lock characteristic without the mandatory Notify property.
+        add_char(&mut service, LOCK_HANDLE, SET_MEMBER_LOCK_UUID, vec![0x01]);
+        add_char(&mut service, RANK_HANDLE, SET_MEMBER_RANK_UUID, vec![0x01]);
+
+        let create_fut = CoordinatedSetIdentificationServiceClient::<FakeTypes>::create(service);
+        pin_mut!(create_fut);
+        let Poll::Ready(result) = create_fut.poll_unpin(&mut noop_cx) else {
+            panic!("Expected create to be ready");
+        };
+        assert_matches!(result.err().unwrap(), Error::InvalidCharacteristic(_));
+    }
+
+    #[test]
     fn create_fails_duplicate_rank() {
         let mut service = FakePeerService::new();
         let mut noop_cx = Context::from_waker(noop_waker_ref());
diff --git a/rust/bt-csis/src/client/event.rs b/rust/bt-csis/src/client/event.rs
new file mode 100644
index 0000000..0f78dcf
--- /dev/null
+++ b/rust/bt-csis/src/client/event.rs
@@ -0,0 +1,467 @@
+// Copyright 2026 The Fuchsia Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+
+use futures::stream::{BoxStream, FusedStream, SelectAll, Stream, StreamExt};
+use parking_lot::Mutex;
+
+use bt_common::packet_encoding::Decodable;
+use bt_gatt::client::CharacteristicNotification;
+use bt_gatt::types::Handle;
+
+use crate::client::{CsisClientState, Error};
+use crate::types::*;
+
+/// A change to a CSIS characteristic value signaled by the Set Member via a
+/// GATT notification.
+#[derive(Debug, Clone, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum CsisNotification {
+    /// The Set Identity Resolving Key characteristic value changed. The value
+    /// is reported as exposed by the server, so it may be an encrypted SIRK
+    /// (CSIS v1.1 Section 4.5).
+    SirkChanged(SetIdentityResolvingKey),
+    /// The Coordinated Set Size characteristic value changed.
+    SizeChanged(CoordinatedSetSize),
+    /// The Set Member Lock characteristic value changed (CSIS v1.1 Section
+    /// 5.3.1). A Set Member does not notify the client whose own write caused
+    /// the change.
+    LockChanged(SetMemberLock),
+}
+
+/// Handles of the CSIS characteristics whose notifications are recognized.
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct NotifiedHandles {
+    pub(crate) sirk: Handle,
+    pub(crate) size: Option<Handle>,
+    pub(crate) lock: Option<Handle>,
+}
+
+/// A stream of CSIS characteristic value changes notified by a Set Member.
+///
+/// The notified value is written back to the client's cached state before the
+/// event is yielded.
+pub struct NotificationStream {
+    notification_streams:
+        SelectAll<BoxStream<'static, Result<CharacteristicNotification, bt_gatt::types::Error>>>,
+    handles: NotifiedHandles,
+    state: Arc<Mutex<CsisClientState>>,
+    terminated: bool,
+}
+
+impl NotificationStream {
+    pub(crate) fn new(
+        notification_streams: SelectAll<
+            BoxStream<'static, Result<CharacteristicNotification, bt_gatt::types::Error>>,
+        >,
+        handles: NotifiedHandles,
+        state: Arc<Mutex<CsisClientState>>,
+    ) -> Self {
+        Self { notification_streams, handles, state, terminated: false }
+    }
+
+    /// Decodes a notification and updates the cached state.
+    fn handle_notification(
+        &mut self,
+        notif: CharacteristicNotification,
+    ) -> Result<CsisNotification, Error> {
+        if notif.handle == self.handles.sirk {
+            let (res, _) = SetIdentityResolvingKey::decode(&notif.value);
+            let sirk = res?;
+            self.state.lock().set_sirk(sirk);
+            return Ok(CsisNotification::SirkChanged(sirk));
+        }
+
+        if Some(notif.handle) == self.handles.size {
+            let (res, _) = CoordinatedSetSize::decode(&notif.value);
+            let size = res?;
+            self.state.lock().set_size(size);
+            return Ok(CsisNotification::SizeChanged(size));
+        }
+
+        if Some(notif.handle) == self.handles.lock {
+            let (res, _) = SetMemberLock::decode(&notif.value);
+            let lock = res?;
+            self.state.lock().set_lock_state(lock);
+            return Ok(CsisNotification::LockChanged(lock));
+        }
+
+        Err(Error::UnexpectedNotification(notif.handle))
+    }
+}
+
+impl Stream for NotificationStream {
+    type Item = Result<CsisNotification, Error>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        if self.terminated {
+            return Poll::Ready(None);
+        }
+
+        match futures::ready!(self.notification_streams.poll_next_unpin(cx)) {
+            Some(Ok(notif)) => Poll::Ready(Some(self.handle_notification(notif))),
+            Some(Err(e)) => Poll::Ready(Some(Err(Error::Gatt(e)))),
+            None => {
+                self.terminated = true;
+                Poll::Ready(None)
+            }
+        }
+    }
+}
+
+impl FusedStream for NotificationStream {
+    fn is_terminated(&self) -> bool {
+        self.terminated
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use assert_matches::assert_matches;
+    use bt_common::Uuid;
+    use bt_gatt::test_utils::{FakePeerService, FakeTypes};
+    use bt_gatt::types::{AttributePermissions, Characteristic, CharacteristicProperty};
+    use core::num::NonZeroU8;
+    use futures::task::noop_waker_ref;
+    use futures::FutureExt;
+
+    use crate::client::CoordinatedSetIdentificationServiceClient;
+
+    const SIRK_HANDLE: Handle = Handle(1);
+    const SIZE_HANDLE: Handle = Handle(2);
+    const LOCK_HANDLE: Handle = Handle(3);
+    const RANK_HANDLE: Handle = Handle(4);
+
+    fn add_char(
+        service: &mut FakePeerService,
+        handle: Handle,
+        uuid: Uuid,
+        value: Vec<u8>,
+        notifiable: bool,
+    ) {
+        let properties = if notifiable {
+            (&[CharacteristicProperty::Read, CharacteristicProperty::Notify]).into()
+        } else {
+            CharacteristicProperty::Read.into()
+        };
+        service.add_characteristic(
+            Characteristic {
+                handle,
+                uuid,
+                properties,
+                permissions: AttributePermissions::default(),
+                descriptors: vec![],
+            },
+            value,
+        );
+    }
+
+    fn sirk_value(key: [u8; 16]) -> Vec<u8> {
+        let mut value = vec![0x01]; // Plaintext
+        value.extend_from_slice(&key);
+        value
+    }
+
+    /// Builds a Set Member exposing SIRK, Size, Lock and Rank, with SIRK, Size
+    /// and Lock notifiable.
+    fn fake_service() -> FakePeerService {
+        let mut service = FakePeerService::new();
+        add_char(
+            &mut service,
+            SIRK_HANDLE,
+            SET_IDENTITY_RESOLVING_KEY_UUID,
+            sirk_value([0xAA; 16]),
+            true,
+        );
+        add_char(&mut service, SIZE_HANDLE, COORDINATED_SET_SIZE_UUID, vec![0x02], true);
+        add_char(&mut service, LOCK_HANDLE, SET_MEMBER_LOCK_UUID, vec![0x01], true);
+        // CSIS v1.1 Table 5.1 note C.1: Rank is mandatory when Lock is
+        // present, and is never notifiable.
+        add_char(&mut service, RANK_HANDLE, SET_MEMBER_RANK_UUID, vec![0x01], false);
+        service
+    }
+
+    fn build_client(
+        service: FakePeerService,
+    ) -> CoordinatedSetIdentificationServiceClient<FakeTypes> {
+        let mut noop_cx = Context::from_waker(noop_waker_ref());
+        let create_fut = CoordinatedSetIdentificationServiceClient::<FakeTypes>::create(service);
+        let mut create_fut = std::pin::pin!(create_fut);
+        let Poll::Ready(client) = create_fut.poll_unpin(&mut noop_cx) else {
+            panic!("Expected create to be ready");
+        };
+        client.expect("Expected create to succeed")
+    }
+
+    #[test]
+    fn create_does_not_subscribe() {
+        let service = fake_service();
+        let mut noop_cx = Context::from_waker(noop_waker_ref());
+        let mut client = build_client(service.clone());
+
+        // Dropped, nothing is subscribed yet. Differs from the value below so
+        // a client that subscribed in `create` would fail this test.
+        service.notify(
+            &SIZE_HANDLE,
+            Ok(CharacteristicNotification {
+                handle: SIZE_HANDLE,
+                value: vec![0x05],
+                maybe_truncated: false,
+            }),
+        );
+        assert_eq!(client.size(), Some(CoordinatedSetSize(NonZeroU8::new(2).unwrap())));
+
+        let mut stream = client.take_notification_stream().expect("stream available");
+        service.notify(
+            &SIZE_HANDLE,
+            Ok(CharacteristicNotification {
+                handle: SIZE_HANDLE,
+                value: vec![0x03],
+                maybe_truncated: false,
+            }),
+        );
+        let Poll::Ready(Some(Ok(event))) = stream.poll_next_unpin(&mut noop_cx) else {
+            panic!("Expected a size notification");
+        };
+        assert_eq!(
+            event,
+            CsisNotification::SizeChanged(CoordinatedSetSize(NonZeroU8::new(3).unwrap()))
+        );
+    }
+
+    #[test]
+    fn notifications_update_cached_state() {
+        let service = fake_service();
+        let mut noop_cx = Context::from_waker(noop_waker_ref());
+        let mut client = build_client(service.clone());
+        let mut stream = client.take_notification_stream().expect("stream available");
+
+        // Values read during discovery are visible before any notification.
+        assert_eq!(client.sirk().value, [0xAA; 16]);
+        assert_eq!(client.size(), Some(CoordinatedSetSize(NonZeroU8::new(2).unwrap())));
+        assert_eq!(client.lock_state(), Some(SetMemberLock::Unlocked));
+
+        assert_matches!(stream.poll_next_unpin(&mut noop_cx), Poll::Pending);
+
+        // A lock timeout expiring, or another client releasing the lock
+        // (CSIS v1.1 Section 5.3.1.1).
+        service.notify(
+            &LOCK_HANDLE,
+            Ok(CharacteristicNotification {
+                handle: LOCK_HANDLE,
+                value: vec![0x02],
+                maybe_truncated: false,
+            }),
+        );
+        let Poll::Ready(Some(Ok(event))) = stream.poll_next_unpin(&mut noop_cx) else {
+            panic!("Expected a lock notification");
+        };
+        assert_eq!(event, CsisNotification::LockChanged(SetMemberLock::Locked));
+        assert_eq!(client.lock_state(), Some(SetMemberLock::Locked));
+
+        service.notify(
+            &SIZE_HANDLE,
+            Ok(CharacteristicNotification {
+                handle: SIZE_HANDLE,
+                value: vec![0x03],
+                maybe_truncated: false,
+            }),
+        );
+        let Poll::Ready(Some(Ok(event))) = stream.poll_next_unpin(&mut noop_cx) else {
+            panic!("Expected a size notification");
+        };
+        assert_eq!(
+            event,
+            CsisNotification::SizeChanged(CoordinatedSetSize(NonZeroU8::new(3).unwrap()))
+        );
+        assert_eq!(client.size(), Some(CoordinatedSetSize(NonZeroU8::new(3).unwrap())));
+
+        service.notify(
+            &SIRK_HANDLE,
+            Ok(CharacteristicNotification {
+                handle: SIRK_HANDLE,
+                value: sirk_value([0xBB; 16]),
+                maybe_truncated: false,
+            }),
+        );
+        let Poll::Ready(Some(Ok(event))) = stream.poll_next_unpin(&mut noop_cx) else {
+            panic!("Expected a SIRK notification");
+        };
+        assert_eq!(
+            event,
+            CsisNotification::SirkChanged(SetIdentityResolvingKey {
+                sirk_type: SirkType::Plaintext,
+                value: [0xBB; 16],
+            })
+        );
+        assert_eq!(client.sirk().value, [0xBB; 16]);
+    }
+
+    #[test]
+    fn unrecognized_notifications_are_reported_as_errors() {
+        let mut service = FakePeerService::new();
+        add_char(
+            &mut service,
+            SIRK_HANDLE,
+            SET_IDENTITY_RESOLVING_KEY_UUID,
+            sirk_value([0xAA; 16]),
+            true,
+        );
+        // Not notifiable, so it is never subscribed.
+        add_char(&mut service, SIZE_HANDLE, COORDINATED_SET_SIZE_UUID, vec![0x02], false);
+        // CSIS v1.1 Table 5.1 excludes Notify for Set Member Rank, so this
+        // Set Member is not conformant. It is subscribed but never recognized.
+        add_char(&mut service, RANK_HANDLE, SET_MEMBER_RANK_UUID, vec![0x01], true);
+
+        let mut noop_cx = Context::from_waker(noop_waker_ref());
+        let mut client = build_client(service.clone());
+        let mut stream = client.take_notification_stream().expect("stream available");
+
+        // Rank is subscribed but its notifications are not recognized.
+        service.notify(
+            &RANK_HANDLE,
+            Ok(CharacteristicNotification {
+                handle: RANK_HANDLE,
+                value: vec![0x02],
+                maybe_truncated: false,
+            }),
+        );
+        assert_matches!(
+            stream.poll_next_unpin(&mut noop_cx),
+            Poll::Ready(Some(Err(Error::UnexpectedNotification(RANK_HANDLE))))
+        );
+        assert_eq!(client.rank(), Some(SetMemberRank(NonZeroU8::new(1).unwrap())));
+
+        // The SIRK characteristic is still subscribed.
+        service.notify(
+            &SIRK_HANDLE,
+            Ok(CharacteristicNotification {
+                handle: SIRK_HANDLE,
+                value: sirk_value([0xBB; 16]),
+                maybe_truncated: false,
+            }),
+        );
+        let Poll::Ready(Some(Ok(event))) = stream.poll_next_unpin(&mut noop_cx) else {
+            panic!("Expected a SIRK notification");
+        };
+        assert_matches!(event, CsisNotification::SirkChanged(_));
+
+        // The Rank subscription is still open, so the stream stays pending.
+        service.clear_notifier(&SIRK_HANDLE);
+        assert_matches!(stream.poll_next_unpin(&mut noop_cx), Poll::Pending);
+    }
+
+    #[test]
+    fn stream_terminates_when_nothing_is_notifiable() {
+        let mut service = FakePeerService::new();
+        add_char(
+            &mut service,
+            SIRK_HANDLE,
+            SET_IDENTITY_RESOLVING_KEY_UUID,
+            sirk_value([0xAA; 16]),
+            false,
+        );
+        add_char(&mut service, SIZE_HANDLE, COORDINATED_SET_SIZE_UUID, vec![0x02], false);
+        let mut noop_cx = Context::from_waker(noop_waker_ref());
+        let mut client = build_client(service);
+        let mut stream = client.take_notification_stream().expect("stream available");
+
+        assert_matches!(stream.poll_next_unpin(&mut noop_cx), Poll::Ready(None));
+        assert!(stream.is_terminated());
+    }
+
+    #[test]
+    fn stream_terminates_once_every_subscription_closes() {
+        let service = fake_service();
+        let mut noop_cx = Context::from_waker(noop_waker_ref());
+        let mut client = build_client(service.clone());
+        let mut stream = client.take_notification_stream().expect("stream available");
+
+        assert_matches!(stream.poll_next_unpin(&mut noop_cx), Poll::Pending);
+
+        // Dropping one subscription is not enough: the rest can still notify.
+        service.clear_notifier(&SIRK_HANDLE);
+        assert_matches!(stream.poll_next_unpin(&mut noop_cx), Poll::Pending);
+        service.notify(
+            &LOCK_HANDLE,
+            Ok(CharacteristicNotification {
+                handle: LOCK_HANDLE,
+                value: vec![0x02],
+                maybe_truncated: false,
+            }),
+        );
+        let Poll::Ready(Some(Ok(event))) = stream.poll_next_unpin(&mut noop_cx) else {
+            panic!("Expected a lock notification");
+        };
+        assert_eq!(event, CsisNotification::LockChanged(SetMemberLock::Locked));
+
+        // Once they have all closed, the stream ends.
+        service.clear_notifier(&SIZE_HANDLE);
+        service.clear_notifier(&LOCK_HANDLE);
+        assert_matches!(stream.poll_next_unpin(&mut noop_cx), Poll::Ready(None));
+        assert!(stream.is_terminated());
+    }
+
+    #[test]
+    fn errors_are_reported_without_ending_the_stream() {
+        let service = fake_service();
+        let mut noop_cx = Context::from_waker(noop_waker_ref());
+        let mut client = build_client(service.clone());
+        let mut stream = client.take_notification_stream().expect("stream available");
+
+        // 0x00 is RFU for the Set Member Lock characteristic.
+        service.notify(
+            &LOCK_HANDLE,
+            Ok(CharacteristicNotification {
+                handle: LOCK_HANDLE,
+                value: vec![0x00],
+                maybe_truncated: false,
+            }),
+        );
+        let Poll::Ready(Some(Err(e))) = stream.poll_next_unpin(&mut noop_cx) else {
+            panic!("Expected a decoding error");
+        };
+        assert_matches!(e, Error::Packet(_));
+        assert!(!stream.is_terminated());
+
+        // The subscription is still live after a GATT error.
+        service.notify(
+            &SIZE_HANDLE,
+            Err(bt_gatt::types::Error::PeerDisconnected(bt_common::PeerId(1))),
+        );
+        let Poll::Ready(Some(Err(e))) = stream.poll_next_unpin(&mut noop_cx) else {
+            panic!("Expected a GATT error");
+        };
+        assert_matches!(e, Error::Gatt(_));
+        assert!(!stream.is_terminated());
+
+        // The cached value is left alone and later notifications still arrive.
+        assert_eq!(client.lock_state(), Some(SetMemberLock::Unlocked));
+        service.notify(
+            &LOCK_HANDLE,
+            Ok(CharacteristicNotification {
+                handle: LOCK_HANDLE,
+                value: vec![0x02],
+                maybe_truncated: false,
+            }),
+        );
+        let Poll::Ready(Some(Ok(event))) = stream.poll_next_unpin(&mut noop_cx) else {
+            panic!("Expected a lock notification");
+        };
+        assert_eq!(event, CsisNotification::LockChanged(SetMemberLock::Locked));
+    }
+
+    #[test]
+    fn notification_stream_is_only_available_once() {
+        let service = fake_service();
+        let mut client = build_client(service);
+
+        assert!(client.take_notification_stream().is_some());
+        assert!(client.take_notification_stream().is_none());
+    }
+}
diff --git a/rust/bt-csis/src/types.rs b/rust/bt-csis/src/types.rs
index a96ee7d..d91b730 100644
--- a/rust/bt-csis/src/types.rs
+++ b/rust/bt-csis/src/types.rs
@@ -17,8 +17,7 @@
 
 // TODO(b/534436497): Add Coordinated Set Name characteristic (CSIS v1.1 Section
 // 5.5).
-// TODO(b/534436497): Add SetMemberLock enum and CsisApplicationError
-// enum.
+// TODO(b/534436497): Add CsisApplicationError enum.
 
 decodable_enum! {
     /// The type of the Set Identity Resolving Key (SIRK).
@@ -169,6 +168,50 @@
     }
 }
 
+decodable_enum! {
+    /// The Set Member Lock characteristic value. All other values are RFU.
+    /// See CSIS v1.1 Section 5.3, Table 5.4.
+    pub enum SetMemberLock<u8, bt_common::packet_encoding::Error, OutOfRange> {
+        Unlocked = 0x01,
+        Locked = 0x02,
+    }
+}
+
+impl SetMemberLock {
+    pub const BYTE_SIZE: usize = 1;
+}
+
+impl Decodable for SetMemberLock {
+    type Error = PacketError;
+
+    fn decode(buf: &[u8]) -> (core::result::Result<Self, Self::Error>, usize) {
+        if buf.is_empty() {
+            return (Err(PacketError::UnexpectedDataLength), 0);
+        }
+        let lock = match Self::try_from(buf[0]) {
+            Ok(val) => val,
+            Err(e) => return (Err(e), Self::BYTE_SIZE),
+        };
+        (Ok(lock), Self::BYTE_SIZE)
+    }
+}
+
+impl Encodable for SetMemberLock {
+    type Error = PacketError;
+
+    fn encoded_len(&self) -> usize {
+        Self::BYTE_SIZE
+    }
+
+    fn encode(&self, buf: &mut [u8]) -> Result<(), Self::Error> {
+        if buf.is_empty() {
+            return Err(PacketError::BufferTooSmall);
+        }
+        buf[0] = (*self).into();
+        Ok(())
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -247,4 +290,41 @@
         let mut empty_buf = [];
         assert_matches!(size.encode(&mut empty_buf), Err(PacketError::BufferTooSmall));
     }
+
+    #[test]
+    fn set_member_lock_decoding() {
+        let (res, consumed) = SetMemberLock::decode(&[0x01]);
+        assert_eq!(consumed, 1);
+        assert_eq!(res.unwrap(), SetMemberLock::Unlocked);
+
+        let (res, consumed) = SetMemberLock::decode(&[0x02]);
+        assert_eq!(consumed, 1);
+        assert_eq!(res.unwrap(), SetMemberLock::Locked);
+
+        // 0x00 and anything above 0x02 are RFU per CSIS v1.1 Section 5.3.
+        let (res, _) = SetMemberLock::decode(&[0x00]);
+        assert_matches!(res, Err(_));
+        let (res, _) = SetMemberLock::decode(&[0x03]);
+        assert_matches!(res, Err(_));
+
+        let (res, consumed) = SetMemberLock::decode(&[]);
+        assert_eq!(consumed, 0);
+        assert_matches!(res, Err(PacketError::UnexpectedDataLength));
+    }
+
+    #[test]
+    fn set_member_lock_encoding() {
+        let lock = SetMemberLock::Locked;
+        assert_eq!(lock.encoded_len(), 1);
+
+        let mut buf = [0; 1];
+        lock.encode(&mut buf).unwrap();
+        assert_eq!(buf[0], 0x02);
+
+        SetMemberLock::Unlocked.encode(&mut buf).unwrap();
+        assert_eq!(buf[0], 0x01);
+
+        let mut empty_buf = [];
+        assert_matches!(lock.encode(&mut empty_buf), Err(PacketError::BufferTooSmall));
+    }
 }