diff --git a/rust/bt-bap/src/types.rs b/rust/bt-bap/src/types.rs
index f3946d6..5e62c73 100644
--- a/rust/bt-bap/src/types.rs
+++ b/rust/bt-bap/src/types.rs
@@ -85,6 +85,19 @@
     }
 }
 
+/// A 16-octet (128-bit) Broadcast Code used for decrypting encrypted
+/// Broadcast Audio Streams. See BAP v1.0.1 Section 3.7.2.2 for more details.
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+pub struct BroadcastCode(pub [u8; 16]);
+
+impl BroadcastCode {
+    pub const BYTE_SIZE: usize = 16;
+
+    pub fn new(raw_value: [u8; 16]) -> Self {
+        Self(raw_value)
+    }
+}
+
 /// To associate a PA, used to expose broadcast Audio Stream parameters, with a
 /// broadcast Audio Stream, the Broadcast Source shall transmit EA PDUs that
 /// include the following data. This struct represents the AD data value
diff --git a/rust/bt-bass/src/client.rs b/rust/bt-bass/src/client.rs
index d6d239f..a1188c2 100644
--- a/rust/bt-bass/src/client.rs
+++ b/rust/bt-bass/src/client.rs
@@ -13,7 +13,7 @@
 use log::warn;
 use parking_lot::Mutex;
 
-use bt_bap::types::BroadcastId;
+use bt_bap::types::{BroadcastCode, BroadcastId};
 use bt_common::core::{AddressType, AdvertisingSetId, PeriodicAdvertisingInterval};
 use bt_common::generic_audio::metadata_ltv::Metadata;
 use bt_common::packet_encoding::Decodable;
@@ -371,7 +371,7 @@
     ) -> Result<(), Error> {
         let source_id = self.get_source_id(&broadcast_id)?;
 
-        let op = SetBroadcastCodeOperation::new(source_id, broadcast_code.clone());
+        let op = SetBroadcastCodeOperation::new(source_id, BroadcastCode::new(broadcast_code));
         self.write_to_bascp(op).await?;
 
         // Save the broadcast code we sent.
diff --git a/rust/bt-bass/src/lib.rs b/rust/bt-bass/src/lib.rs
index e32b283..3c3b2fa 100644
--- a/rust/bt-bass/src/lib.rs
+++ b/rust/bt-bass/src/lib.rs
@@ -3,10 +3,13 @@
 // found in the LICENSE file.
 
 pub mod client;
-pub mod server;
-pub mod types;
 pub use crate::client::error::Error as ClientError;
-pub use crate::server::Error as ServerError;
+
+pub mod server;
+pub use crate::server::error::Error as ServerError;
+pub use crate::server::ServerEvent;
+
+pub mod types;
 
 #[cfg(any(test, feature = "test-utils"))]
 pub mod test_utils;
diff --git a/rust/bt-bass/src/server.rs b/rust/bt-bass/src/server.rs
index c7bbde7..715a803 100644
--- a/rust/bt-bass/src/server.rs
+++ b/rust/bt-bass/src/server.rs
@@ -4,22 +4,28 @@
 
 //! Implements the Broadcast Audio Scan Service (BASS) server role.
 
-use bt_gatt::server::{LocalService, Server as _, ServiceDefinition, ServiceId};
+use bt_bap::types::BroadcastCode;
+use bt_common::packet_encoding::{Decodable, Encodable};
+use bt_common::PeerId;
+use bt_gatt::server::{
+    LocalService, ReadResponder, Server as _, ServiceDefinition, ServiceEvent, ServiceId,
+    WriteResponder,
+};
 use bt_gatt::types::{
-    AttributePermissions, CharacteristicProperty, Handle, SecurityLevels, ServiceKind,
+    AttributePermissions, CharacteristicProperty, GattError, Handle, SecurityLevels, ServiceKind,
 };
 use bt_gatt::Characteristic;
+use futures::stream::Stream;
 use futures::Future;
 use pin_project::pin_project;
-use std::collections::HashMap;
+use std::collections::{BTreeMap, HashMap};
+use std::num::NonZeroU8;
+use std::task::{Poll, Waker};
 
-use crate::types::{
-    BroadcastReceiveState, SourceId, BROADCAST_AUDIO_SCAN_CONTROL_POINT_UUID,
-    BROADCAST_AUDIO_SCAN_SERVICE_UUID, BROADCAST_RECEIVE_STATE_UUID,
-};
+use crate::types::*;
 
 pub mod error;
-pub use error::Error;
+use error::Error;
 
 /// Service identifier assigned to the published BASS GATT service instance.
 const BASS_SERVICE_ID: ServiceId = ServiceId::new(1);
@@ -35,9 +41,6 @@
 /// See BASS v1.0 Section 3.2.
 #[derive(Debug)]
 pub(crate) struct PublishedReceiveStateCharacteristic {
-    /// 1-indexed source identifier.
-    /// See BASS v1.0 Section 3.2.1.
-    source_id: SourceId,
     /// Handle assigned to this GATT characteristic.
     handle: Handle,
     /// Current Broadcast Receive State value.
@@ -48,19 +51,26 @@
 #[pin_project(project = LocalServiceProj)]
 enum LocalServiceState<T: bt_gatt::ServerTypes> {
     /// Service definition has not been registered in the GATT database.
-    NotPublished,
+    NotPublished {
+        waker: Option<Waker>,
+    },
     /// Service registration is in progress.
     Preparing {
         #[pin]
         fut: T::LocalServiceFut,
     },
     /// Service registration is complete and active in the GATT database.
-    Published { service: T::LocalService },
+    Published {
+        service: T::LocalService,
+        #[pin]
+        events: T::ServiceEventStream,
+    },
+    Terminated,
 }
 
 impl<T: bt_gatt::ServerTypes> Default for LocalServiceState<T> {
     fn default() -> Self {
-        Self::NotPublished
+        Self::NotPublished { waker: None }
     }
 }
 
@@ -68,28 +78,45 @@
     fn is_published(&self) -> bool {
         matches!(self, LocalServiceState::Published { .. })
     }
+}
 
-    fn poll_publish(
+impl<T: bt_gatt::ServerTypes> Stream for LocalServiceState<T> {
+    type Item = Result<bt_gatt::server::ServiceEvent<T>, Error>;
+
+    fn poll_next(
         mut self: std::pin::Pin<&mut Self>,
         cx: &mut std::task::Context<'_>,
-    ) -> std::task::Poll<Result<(), Error>> {
-        match self.as_mut().project() {
-            LocalServiceProj::NotPublished => std::task::Poll::Pending,
-            LocalServiceProj::Preparing { fut } => {
-                let service_result = futures::ready!(fut.poll(cx));
-                match service_result {
-                    Ok(service) => {
-                        let _ = service.publish();
-                        self.set(LocalServiceState::Published { service });
-                        std::task::Poll::Ready(Ok(()))
-                    }
-                    Err(e) => {
-                        self.set(LocalServiceState::NotPublished);
-                        std::task::Poll::Ready(Err(Error::Gatt(e)))
-                    }
+    ) -> Poll<Option<Self::Item>> {
+        loop {
+            match self.as_mut().project() {
+                LocalServiceProj::Terminated => return Poll::Ready(None),
+                LocalServiceProj::NotPublished { waker } => {
+                    *waker = Some(cx.waker().clone());
+                    return Poll::Pending;
+                }
+                LocalServiceProj::Preparing { fut } => {
+                    let service_result = futures::ready!(fut.poll(cx));
+                    let Ok(service) = service_result else {
+                        self.as_mut().set(LocalServiceState::NotPublished { waker: None });
+                        return Poll::Ready(Some(Err(Error::Gatt(service_result.err().unwrap()))));
+                    };
+                    let events = service.publish();
+                    self.as_mut().set(LocalServiceState::Published { service, events });
+                }
+                LocalServiceProj::Published { service: _, events } => {
+                    return match futures::ready!(events.poll_next(cx)) {
+                        Some(Ok(event)) => Poll::Ready(Some(Ok(event))),
+                        Some(Err(e)) => {
+                            self.as_mut().set(LocalServiceState::Terminated);
+                            Poll::Ready(Some(Err(Error::Gatt(e))))
+                        }
+                        None => {
+                            self.as_mut().set(LocalServiceState::Terminated);
+                            Poll::Ready(None)
+                        }
+                    };
                 }
             }
-            LocalServiceProj::Published { .. } => std::task::Poll::Ready(Ok(())),
         }
     }
 }
@@ -166,37 +193,318 @@
 
         let _ = service_def.add_characteristic(Self::build_control_point());
 
-        // Broadcast Receive State characteristics (Read, Notify; Encryption Required)
+        // Broadcast Receive State characteristics (Read, Notify; Encryption Required).
+        // Handle(1) is allocated to the Control Point Characteristic.
         const FIRST_RECEIVE_STATE_HANDLE: Handle = Handle(2);
         let num_receive_states = self.receive_states.len();
-        let mut receive_state_characteristics = HashMap::with_capacity(num_receive_states);
-        let mut source_id_to_handle = HashMap::with_capacity(num_receive_states);
-        for (i, mut state) in self.receive_states.into_iter().enumerate() {
-            let source_id = (i + 1) as u8;
+        let mut state = ServerState::new(num_receive_states);
+
+        for (i, mut receive_state) in self.receive_states.into_iter().enumerate() {
             let handle = Handle(FIRST_RECEIVE_STATE_HANDLE.0 + i as u64);
             let _ = service_def.add_characteristic(Self::build_receive_state(handle));
 
-            // Override Source ID as this is assigned by the server. See BASS v1.0 Section
-            // 3.2.1.
-            if let BroadcastReceiveState::NonEmpty(ref mut receive_state) = state {
-                receive_state.source_id = source_id;
+            if let BroadcastReceiveState::NonEmpty(ref mut r) = receive_state {
+                r.source_id = state
+                    .allocate_source_id(handle)
+                    .map_err(|e| Error::Gatt(bt_gatt::types::Error::Gatt(e)))?;
             }
 
-            receive_state_characteristics
-                .insert(handle, PublishedReceiveStateCharacteristic { source_id, handle, state });
-            source_id_to_handle.insert(source_id, handle);
+            state.receive_state_characteristics.insert(
+                handle,
+                PublishedReceiveStateCharacteristic { handle, state: receive_state },
+            );
         }
 
-        let next_source_id = (num_receive_states + 1) as SourceId;
+        Ok(Server { service_def, local_service: Default::default(), state })
+    }
+}
 
-        Ok(Server {
-            service_def,
-            local_service: Default::default(),
-            control_point_handle: CONTROL_POINT_HANDLE,
-            receive_state_characteristics,
-            source_id_to_handle,
-            next_source_id,
-        })
+/// Events emitted by the [`Server`] when a remote GATT client writes to the
+/// Control Point.
+#[derive(Debug, PartialEq)]
+pub enum ServerEvent {
+    /// A client requested a change in remote scanning state.
+    RemoteScanState { peer_id: PeerId, is_scanning: bool },
+    /// A client requested adding a new Broadcast Source.
+    AddSource { peer_id: PeerId, source_id: SourceId, operation: AddSourceOperation },
+    /// A client requested modifying an existing Broadcast Source.
+    ModifySource { peer_id: PeerId, source_id: SourceId, operation: ModifySourceOperation },
+    /// A client provided a Broadcast Code for an encrypted Broadcast Source.
+    SetBroadcastCode { peer_id: PeerId, source_id: SourceId, broadcast_code: BroadcastCode },
+    /// A client requested removing a Broadcast Source.
+    RemoveSource { peer_id: PeerId, source_id: SourceId },
+}
+
+impl ServerEvent {
+    /// Attempts to decode a raw control point write request into a
+    /// [`ServerEvent`].
+    fn decode(peer_id: PeerId, val_bytes: &[u8]) -> Result<Self, GattError> {
+        if val_bytes.is_empty() {
+            return Err(GattError::WriteRequestRejected);
+        }
+
+        let raw_opcode = val_bytes[0];
+        let Ok(opcode) = ControlPointOpcode::try_from(raw_opcode) else {
+            return Err(ERROR_OPCODE_NOT_SUPPORTED);
+        };
+
+        match opcode {
+            ControlPointOpcode::RemoteScanStopped => {
+                let _ = decode_control_point_op::<RemoteScanStoppedOperation>(val_bytes)?;
+                Ok(ServerEvent::RemoteScanState { peer_id, is_scanning: false })
+            }
+            ControlPointOpcode::RemoteScanStarted => {
+                let _ = decode_control_point_op::<RemoteScanStartedOperation>(val_bytes)?;
+                Ok(ServerEvent::RemoteScanState { peer_id, is_scanning: true })
+            }
+            ControlPointOpcode::AddSource => {
+                let operation = decode_control_point_op::<AddSourceOperation>(val_bytes)?;
+                // `source_id` will be assigned internally by the Server.
+                Ok(ServerEvent::AddSource { peer_id, source_id: 0, operation })
+            }
+            ControlPointOpcode::ModifySource => {
+                let operation = decode_control_point_op::<ModifySourceOperation>(val_bytes)?;
+                Ok(ServerEvent::ModifySource { peer_id, source_id: operation.source_id, operation })
+            }
+            ControlPointOpcode::SetBroadcastCode => {
+                let operation = decode_control_point_op::<SetBroadcastCodeOperation>(val_bytes)?;
+                Ok(ServerEvent::SetBroadcastCode {
+                    peer_id,
+                    source_id: operation.source_id,
+                    broadcast_code: operation.broadcast_code,
+                })
+            }
+            ControlPointOpcode::RemoveSource => {
+                let operation = decode_control_point_op::<RemoveSourceOperation>(val_bytes)?;
+                Ok(ServerEvent::RemoveSource { peer_id, source_id: operation.0 })
+            }
+        }
+    }
+}
+
+/// Attempts to decode a control point operation payload.
+fn decode_control_point_op<O: Decodable>(val_bytes: &[u8]) -> Result<O, GattError> {
+    let (Ok(op), consumed) = O::decode(val_bytes) else {
+        return Err(GattError::WriteRequestRejected);
+    };
+    if consumed != val_bytes.len() {
+        return Err(GattError::WriteRequestRejected);
+    }
+    Ok(op)
+}
+
+/// Internal state of the BASS GATT Server.
+#[derive(Debug)]
+struct ServerState {
+    /// Broadcast Receive State characteristics identified by the assigned
+    /// GATT handle.
+    receive_state_characteristics: BTreeMap<Handle, PublishedReceiveStateCharacteristic>,
+    /// Unique IDs tracking each assigned GATT Handle.
+    source_id_to_handle: HashMap<SourceId, Handle>,
+    /// Next available Source ID to be assigned to an empty Receive State
+    /// characteristic.
+    /// Must be in [1, 255]. See BASS spec v1.0 Section 3.2.1.
+    next_source_id: NonZeroU8,
+}
+
+impl ServerState {
+    fn new(capacity: usize) -> Self {
+        Self {
+            receive_state_characteristics: BTreeMap::new(),
+            source_id_to_handle: HashMap::with_capacity(capacity),
+            next_source_id: NonZeroU8::MIN,
+        }
+    }
+
+    /// Allocates the next available, unused [`SourceId`] and maps it to
+    /// `handle`.
+    ///
+    /// Returns the assigned ID on success.
+    /// Returns `GattError::InsufficientResources` if all 255 Source IDs
+    /// are currently in use.
+    fn allocate_source_id(&mut self, handle: Handle) -> Result<SourceId, GattError> {
+        // Upper bound on the maximum number of unique SourceIDs (u8 max).
+        for _ in 0..NonZeroU8::MAX.get() {
+            let id = self.next_source_id.get();
+            self.next_source_id = self.next_source_id.checked_add(1).unwrap_or(NonZeroU8::MIN);
+            if !self.source_id_to_handle.contains_key(&id) {
+                self.source_id_to_handle.insert(id, handle);
+                return Ok(id);
+            }
+        }
+        Err(GattError::InsufficientResources)
+    }
+
+    /// Handles a read request for a characteristic on the BASS server.
+    fn handle_read(&self, handle: Handle, offset: usize, responder: impl ReadResponder) {
+        // Only the Broadcast Receive State characteristics can be read.
+        // See BASS spec v1.0 Section 3.3.2.
+        if handle == CONTROL_POINT_HANDLE {
+            responder.error(GattError::ReadNotPermitted);
+            return;
+        }
+
+        let Some(chrc) = self.receive_state_characteristics.get(&handle) else {
+            responder.error(GattError::InvalidHandle);
+            return;
+        };
+
+        let len = chrc.state.encoded_len();
+        if offset > len {
+            responder.error(GattError::InvalidOffset);
+            return;
+        }
+
+        let mut buf = vec![0u8; len];
+        match chrc.state.encode(&mut buf) {
+            Ok(_) => responder.respond(&buf[offset..]),
+            Err(_) => responder.error(GattError::UnlikelyError),
+        }
+    }
+
+    /// Handles a write request for a characteristic on the BASS server.
+    fn handle_write(
+        &mut self,
+        peer_id: PeerId,
+        handle: Handle,
+        value: &[u8],
+        responder: impl WriteResponder,
+    ) -> Option<ServerEvent> {
+        // Only the Broadcast Audio Scan Control Point characteristic can be written to
+        // by a client. See BASS v1.0 Section 3.3.
+        if handle != CONTROL_POINT_HANDLE {
+            responder.error(GattError::WriteNotPermitted);
+            return None;
+        }
+
+        match self.handle_control_point_write(peer_id, value) {
+            Ok(event) => {
+                responder.acknowledge();
+                Some(event)
+            }
+            Err(err) => {
+                responder.error(err);
+                None
+            }
+        }
+    }
+
+    /// Returns true if the specified `source_id` is assigned to a non-empty
+    /// Receive State characteristic.
+    fn is_valid_source_id(&self, source_id: SourceId) -> bool {
+        self.source_id_to_handle
+            .get(&source_id)
+            .and_then(|h| self.receive_state_characteristics.get(h))
+            .map_or(false, |s| !s.state.is_empty())
+    }
+
+    fn handle_control_point_write(
+        &mut self,
+        peer_id: PeerId,
+        val_bytes: &[u8],
+    ) -> Result<ServerEvent, GattError> {
+        let mut event = ServerEvent::decode(peer_id, val_bytes)?;
+
+        match &mut event {
+            ServerEvent::RemoteScanState { .. } => {}
+            ServerEvent::AddSource { source_id, operation, .. } => {
+                *source_id = self.handle_add_source_write(operation)?;
+            }
+            ServerEvent::ModifySource { operation, .. } => {
+                self.handle_modify_source_write(operation)?;
+            }
+            ServerEvent::SetBroadcastCode { source_id, .. } => {
+                if !self.is_valid_source_id(*source_id) {
+                    return Err(ERROR_INVALID_SOURCE_ID);
+                }
+            }
+            ServerEvent::RemoveSource { source_id, .. } => {
+                self.handle_remove_source_write(*source_id)?;
+            }
+        }
+
+        Ok(event)
+    }
+
+    fn handle_add_source_write(
+        &mut self,
+        operation: &AddSourceOperation,
+    ) -> Result<SourceId, GattError> {
+        // Find the next available empty Receive State characteristic entry.
+        let handle = {
+            let Some(chrc) =
+                self.receive_state_characteristics.values().find(|c| c.state.is_empty())
+            else {
+                return Err(GattError::InsufficientResources);
+            };
+            chrc.handle
+        };
+
+        let source_id = self.allocate_source_id(handle)?;
+        let chrc =
+            self.receive_state_characteristics.get_mut(&handle).expect("just checked existence");
+        let pa_sync_state = PaSyncState::from_pa_sync(operation.pa_sync);
+        let subgroups = operation
+            .subgroups
+            .iter()
+            .map(|s| BigSubgroup::new(Some(s.bis_sync.clone())).with_metadata(s.metadata.clone()))
+            .collect();
+
+        let initial_state = ReceiveState::new(
+            source_id,
+            operation.advertiser_address_type,
+            operation.advertiser_address,
+            operation.advertising_sid,
+            operation.broadcast_id,
+            pa_sync_state,
+            EncryptionStatus::NotEncrypted,
+            subgroups,
+        );
+
+        chrc.state = BroadcastReceiveState::NonEmpty(initial_state);
+        Ok(source_id)
+    }
+
+    fn handle_modify_source_write(
+        &mut self,
+        operation: &ModifySourceOperation,
+    ) -> Result<(), GattError> {
+        let Some(handle) = self.source_id_to_handle.get(&operation.source_id) else {
+            return Err(ERROR_INVALID_SOURCE_ID);
+        };
+        let Some(chrc) = self.receive_state_characteristics.get_mut(handle) else {
+            return Err(ERROR_INVALID_SOURCE_ID);
+        };
+        let BroadcastReceiveState::NonEmpty(ref mut state) = chrc.state else {
+            return Err(ERROR_INVALID_SOURCE_ID);
+        };
+
+        let subgroups = operation
+            .subgroups
+            .iter()
+            .map(|s| BigSubgroup::new(Some(s.bis_sync.clone())).with_metadata(s.metadata.clone()))
+            .collect();
+
+        state.subgroups = subgroups;
+        state.pa_sync_state = PaSyncState::from_pa_sync(operation.pa_sync);
+
+        Ok(())
+    }
+
+    fn handle_remove_source_write(&mut self, source_id: SourceId) -> Result<(), GattError> {
+        let Some(handle) = self.source_id_to_handle.get(&source_id).copied() else {
+            return Err(ERROR_INVALID_SOURCE_ID);
+        };
+        let Some(chrc) = self.receive_state_characteristics.get_mut(&handle) else {
+            return Err(ERROR_INVALID_SOURCE_ID);
+        };
+        if chrc.state.is_empty() {
+            return Err(ERROR_INVALID_SOURCE_ID);
+        }
+
+        let _ = self.source_id_to_handle.remove(&source_id);
+        chrc.state = BroadcastReceiveState::Empty;
+        Ok(())
     }
 }
 
@@ -209,14 +517,7 @@
     service_def: ServiceDefinition,
     #[pin]
     local_service: LocalServiceState<T>,
-    control_point_handle: Handle,
-    /// Broadcast Receive State characteristics mapped by GATT handle.
-    receive_state_characteristics: HashMap<Handle, PublishedReceiveStateCharacteristic>,
-    /// Map from Source ID to characteristic handle.
-    source_id_to_handle: HashMap<SourceId, Handle>,
-    /// Next available Source ID to be assigned to an empty Receive State
-    /// characteristic.
-    next_source_id: SourceId,
+    state: ServerState,
 }
 
 impl<T: bt_gatt::ServerTypes> Server<T> {
@@ -225,66 +526,59 @@
         self.local_service.is_published()
     }
 
-    /// Helper for polling publication to transition from Preparing to Published
-    /// state.
-    // TODO(b/534436439): Remove once the Stream implementation is defined.
-    #[cfg(test)]
-    pub(crate) fn poll_publish(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> std::task::Poll<Result<(), Error>> {
-        self.project().local_service.poll_publish(cx)
-    }
-
     /// Publishes the service to the GATT database.
     pub fn publish(&mut self, server: T::Server) -> Result<(), Error> {
-        if matches!(
-            self.local_service,
-            LocalServiceState::Preparing { .. } | LocalServiceState::Published { .. }
-        ) {
+        if !matches!(self.local_service, LocalServiceState::NotPublished { .. }) {
             return Err(Error::AlreadyPublished);
         }
 
-        let LocalServiceState::NotPublished = std::mem::replace(
+        let LocalServiceState::NotPublished { waker } = std::mem::replace(
             &mut self.local_service,
             LocalServiceState::Preparing { fut: server.prepare(self.service_def.clone()) },
         ) else {
             unreachable!();
         };
-        Ok(())
-    }
 
-    /// Handles a read request for a characteristic on the BASS server.
-    // TODO(b/534436439): Remove once the Stream implementation is defined.
-    #[cfg(test)]
-    fn handle_read(
-        &self,
-        handle: Handle,
-        offset: usize,
-        responder: impl bt_gatt::server::ReadResponder,
-    ) {
-        use bt_common::packet_encoding::Encodable;
-        use bt_gatt::types::GattError;
-
-        if handle == self.control_point_handle {
-            responder.error(GattError::ReadNotPermitted);
-            return;
+        if let Some(w) = waker {
+            w.wake();
         }
 
-        let Some(slot) = self.receive_state_characteristics.get(&handle) else {
-            responder.error(GattError::InvalidHandle);
-            return;
-        };
+        Ok(())
+    }
+}
 
-        let mut buf = vec![0u8; slot.state.encoded_len()];
-        if slot.state.encode(&mut buf).is_ok() {
-            if offset > buf.len() {
-                responder.error(GattError::InvalidOffset);
-            } else {
-                responder.respond(&buf[offset..]);
+impl<T: bt_gatt::ServerTypes> Stream for Server<T> {
+    type Item = Result<ServerEvent, Error>;
+
+    fn poll_next(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> Poll<Option<Self::Item>> {
+        loop {
+            let mut this = self.as_mut().project();
+            let gatt_event = match futures::ready!(this.local_service.as_mut().poll_next(cx)) {
+                Some(Ok(event)) => event,
+                Some(Err(e)) => return Poll::Ready(Some(Err(e))),
+                None => return Poll::Ready(None),
+            };
+
+            match gatt_event {
+                ServiceEvent::Read { handle, offset, responder, .. } => {
+                    this.state.handle_read(handle, offset as usize, responder);
+                }
+                ServiceEvent::Write { handle, peer_id, value, responder, .. } => {
+                    let val_vec = value.to_owned();
+                    if let Some(event) =
+                        this.state.handle_write(peer_id, handle, &val_vec, responder)
+                    {
+                        return Poll::Ready(Some(Ok(event)));
+                    }
+                }
+                _ => {
+                    // TODO(b/534436439): Track CCC for Broadcast Receive State
+                    // characteristics and `PeerInfo` if needed.
+                }
             }
-        } else {
-            responder.error(GattError::UnlikelyError);
         }
     }
 }
@@ -293,29 +587,15 @@
 mod tests {
     use super::*;
     use bt_bap::types::BroadcastId;
-    use bt_common::core::{AddressType, AdvertisingSetId};
-    use bt_gatt::server::ReadResponder;
+    use bt_common::core::{AddressType, AdvertisingSetId, PeriodicAdvertisingInterval};
     use bt_gatt::test_utils::{FakeServer, FakeServerEvent, FakeTypes};
     use bt_gatt::types::GattError;
+    use futures::task::Context;
     use futures::FutureExt;
-    use parking_lot::Mutex;
-    use std::sync::Arc;
+    use futures::StreamExt;
+    use std::task::Poll;
 
-    use crate::types::{EncryptionStatus, PaSyncState, ReceiveState};
-
-    struct TestReadResponder {
-        result: Arc<Mutex<Option<Result<Vec<u8>, GattError>>>>,
-    }
-
-    impl ReadResponder for TestReadResponder {
-        fn respond(self, value: &[u8]) {
-            *self.result.lock() = Some(Ok(value.to_vec()));
-        }
-
-        fn error(self, error: GattError) {
-            *self.result.lock() = Some(Err(error));
-        }
-    }
+    use crate::types::{EncryptionStatus, PaSync, PaSyncState, ReceiveState};
 
     fn make_test_receive_state(source_id: SourceId) -> BroadcastReceiveState {
         BroadcastReceiveState::NonEmpty(ReceiveState::new(
@@ -345,12 +625,36 @@
         assert!(matches!(builder.build::<FakeTypes>(), Err(Error::ExceedsMaxReceiveStates)));
     }
 
+    fn setup_test_server(
+        receive_states: Vec<BroadcastReceiveState>,
+    ) -> (
+        Server<FakeTypes>,
+        FakeServer,
+        futures::channel::mpsc::UnboundedReceiver<FakeServerEvent>,
+        Context<'static>,
+    ) {
+        let mut noop_cx = Context::from_waker(futures::task::noop_waker_ref());
+        let (fake_gatt_server, mut event_receiver) = FakeServer::new();
+        let mut event_stream = event_receiver.next();
+
+        let mut builder = ServerBuilder::new();
+        for state in receive_states {
+            builder = builder.add_receive_state_characteristic(state);
+        }
+        let mut server = builder.build::<FakeTypes>().expect("building server works");
+
+        server.publish(fake_gatt_server.clone()).expect("publish ok");
+        let _ = server.poll_next_unpin(&mut noop_cx);
+        assert!(matches!(
+            event_stream.poll_unpin(&mut noop_cx),
+            Poll::Ready(Some(FakeServerEvent::Published { .. }))
+        ));
+
+        (server, fake_gatt_server, event_receiver, noop_cx)
+    }
+
     #[test]
     fn publish_server() {
-        use futures::task::Context;
-        use futures::StreamExt;
-        use std::task::Poll;
-
         let mut noop_cx = Context::from_waker(futures::task::noop_waker_ref());
 
         let mut server = ServerBuilder::new()
@@ -367,8 +671,8 @@
         assert!(server.publish(fake_gatt_server.clone()).is_ok());
         assert!(server.publish(fake_gatt_server).is_err());
 
-        // Poll publish to complete preparation and transition to Published state
-        assert!(std::pin::Pin::new(&mut server).poll_publish(&mut noop_cx).is_ready());
+        // Polling stream completes publication preparation
+        let _ = server.poll_next_unpin(&mut noop_cx);
         assert!(server.is_published());
 
         let Poll::Ready(Some(FakeServerEvent::Published { id, definition })) =
@@ -384,57 +688,419 @@
 
     #[test]
     fn read_receive_state() {
-        let server = ServerBuilder::new()
-            .add_receive_state_characteristic(BroadcastReceiveState::Empty)
-            .add_receive_state_characteristic(make_test_receive_state(2))
-            .build::<FakeTypes>()
-            .expect("building server works");
+        let (mut server, fake_gatt_server, mut event_receiver, mut noop_cx) =
+            setup_test_server(vec![BroadcastReceiveState::Empty, make_test_receive_state(2)]);
+        let mut event_stream = event_receiver.next();
+        assert!(server.is_published());
 
         // 1. Read empty Receive State slot (Handle 2)
-        let res = Arc::new(Mutex::new(None));
-        server.handle_read(Handle(2), 0, TestReadResponder { result: res.clone() });
-        assert_eq!(res.lock().take().unwrap().expect("ok"), vec![]);
+        fake_gatt_server.incoming_read(PeerId(1), BASS_SERVICE_ID, Handle(2), 0);
+        let _ = server.poll_next_unpin(&mut noop_cx);
+        let Poll::Ready(Some(FakeServerEvent::ReadResponded { handle, value, .. })) =
+            event_stream.poll_unpin(&mut noop_cx)
+        else {
+            panic!("Expected ReadResponded event");
+        };
+        assert_eq!(handle, Handle(2));
+        assert_eq!(value.unwrap(), vec![]);
 
         // 2. Read populated Receive State slot (Handle 3)
-        let res = Arc::new(Mutex::new(None));
-        server.handle_read(Handle(3), 0, TestReadResponder { result: res.clone() });
-        assert!(!res.lock().take().unwrap().expect("ok").is_empty());
+        fake_gatt_server.incoming_read(PeerId(1), BASS_SERVICE_ID, Handle(3), 0);
+        let _ = server.poll_next_unpin(&mut noop_cx);
+        let Poll::Ready(Some(FakeServerEvent::ReadResponded { handle, value, .. })) =
+            event_stream.poll_unpin(&mut noop_cx)
+        else {
+            panic!("Expected ReadResponded event");
+        };
+        assert_eq!(handle, Handle(3));
+        assert!(!value.unwrap().is_empty());
 
         // 3. Read unknown handle (Handle 99)
-        let res = Arc::new(Mutex::new(None));
-        server.handle_read(Handle(99), 0, TestReadResponder { result: res.clone() });
-        assert_eq!(res.lock().take().unwrap().expect_err("err"), GattError::InvalidHandle);
+        fake_gatt_server.incoming_read(PeerId(1), BASS_SERVICE_ID, Handle(99), 0);
+        let _ = server.poll_next_unpin(&mut noop_cx);
+        let Poll::Ready(Some(FakeServerEvent::ReadResponded { handle, value, .. })) =
+            event_stream.poll_unpin(&mut noop_cx)
+        else {
+            panic!("Expected ReadResponded event");
+        };
+        assert_eq!(handle, Handle(99));
+        assert!(matches!(
+            value.unwrap_err(),
+            bt_gatt::types::Error::Gatt(GattError::InvalidHandle)
+        ));
+    }
+
+    #[test]
+    fn control_point_remote_scan_state_changed() {
+        let (mut server, fake_gatt_server, mut event_receiver, mut noop_cx) =
+            setup_test_server(vec![BroadcastReceiveState::Empty]);
+        let mut event_stream = event_receiver.next();
+
+        // Simulate RemoteScanStarted request
+        let raw_cmd = vec![ControlPointOpcode::RemoteScanStarted as u8];
+        fake_gatt_server.incoming_write(PeerId(1), BASS_SERVICE_ID, Handle(1), 0, raw_cmd);
+
+        let Poll::Ready(Some(Ok(event))) = server.poll_next_unpin(&mut noop_cx) else {
+            panic!("Expected ServerEvent");
+        };
+        assert_eq!(event, ServerEvent::RemoteScanState { peer_id: PeerId(1), is_scanning: true });
+
+        let Poll::Ready(Some(FakeServerEvent::WriteResponded { handle, value, .. })) =
+            event_stream.poll_unpin(&mut noop_cx)
+        else {
+            panic!("Expected WriteResponded event");
+        };
+        assert_eq!(handle, Handle(1));
+        assert!(value.is_ok());
+
+        // Simulate RemoteScanStopped request
+        let raw_cmd = vec![ControlPointOpcode::RemoteScanStopped as u8];
+        fake_gatt_server.incoming_write(PeerId(1), BASS_SERVICE_ID, Handle(1), 0, raw_cmd);
+
+        let Poll::Ready(Some(Ok(event))) = server.poll_next_unpin(&mut noop_cx) else {
+            panic!("Expected ServerEvent");
+        };
+        assert_eq!(event, ServerEvent::RemoteScanState { peer_id: PeerId(1), is_scanning: false });
+
+        let Poll::Ready(Some(FakeServerEvent::WriteResponded { handle, value, .. })) =
+            event_stream.poll_unpin(&mut noop_cx)
+        else {
+            panic!("Expected WriteResponded event");
+        };
+        assert_eq!(handle, Handle(1));
+        assert!(value.is_ok());
+    }
+
+    #[test]
+    fn control_point_invalid_opcode_fails() {
+        let (mut server, fake_gatt_server, mut event_receiver, mut noop_cx) =
+            setup_test_server(vec![BroadcastReceiveState::Empty]);
+        let mut event_stream = event_receiver.next();
+
+        // Simulate invalid opcode request
+        fake_gatt_server.incoming_write(PeerId(1), BASS_SERVICE_ID, Handle(1), 0, vec![0xFF]);
+
+        let _ = server.poll_next_unpin(&mut noop_cx);
+
+        let Poll::Ready(Some(FakeServerEvent::WriteResponded { handle, value, .. })) =
+            event_stream.poll_unpin(&mut noop_cx)
+        else {
+            panic!("Expected WriteResponded event");
+        };
+        assert_eq!(handle, Handle(1));
+        assert!(matches!(
+            value.unwrap_err(),
+            bt_gatt::types::Error::Gatt(GattError::ApplicationError80)
+        ));
+    }
+
+    #[test]
+    fn control_point_unallocated_source_id_fails() {
+        let (mut server, fake_gatt_server, mut event_receiver, mut noop_cx) =
+            setup_test_server(vec![BroadcastReceiveState::Empty]);
+        let mut event_stream = event_receiver.next();
+
+        // Operation targeting SourceId = 1 (slot exists, but is empty)
+        let modify_op = ModifySourceOperation::new(
+            1,
+            PaSync::DoNotSync,
+            PeriodicAdvertisingInterval(0xFFFF),
+            vec![],
+        );
+        let mut raw_bytes = vec![0u8; modify_op.encoded_len()];
+        modify_op.encode(&mut raw_bytes).expect("encode ok");
+
+        fake_gatt_server.incoming_write(PeerId(1), BASS_SERVICE_ID, Handle(1), 0, raw_bytes);
+
+        let _ = server.poll_next_unpin(&mut noop_cx);
+
+        let Poll::Ready(Some(FakeServerEvent::WriteResponded { handle, value, .. })) =
+            event_stream.poll_unpin(&mut noop_cx)
+        else {
+            panic!("Expected WriteResponded event");
+        };
+        assert_eq!(handle, Handle(1));
+        assert!(matches!(
+            value.unwrap_err(),
+            bt_gatt::types::Error::Gatt(GattError::ApplicationError81)
+        ));
+    }
+
+    #[test]
+    fn control_point_nonexistent_source_id_fails() {
+        let (mut server, fake_gatt_server, mut event_receiver, mut noop_cx) =
+            setup_test_server(vec![BroadcastReceiveState::Empty]);
+        let mut event_stream = event_receiver.next();
+
+        // ModifySource operation targeting SourceId = 99 (does not exist)
+        let modify_op = ModifySourceOperation::new(
+            99,
+            PaSync::DoNotSync,
+            PeriodicAdvertisingInterval(0xFFFF),
+            vec![],
+        );
+        let mut raw_bytes = vec![0u8; modify_op.encoded_len()];
+        modify_op.encode(&mut raw_bytes).expect("encode ok");
+
+        fake_gatt_server.incoming_write(PeerId(1), BASS_SERVICE_ID, Handle(1), 0, raw_bytes);
+
+        let _ = server.poll_next_unpin(&mut noop_cx);
+
+        let Poll::Ready(Some(FakeServerEvent::WriteResponded { handle, value, .. })) =
+            event_stream.poll_unpin(&mut noop_cx)
+        else {
+            panic!("Expected WriteResponded event");
+        };
+        assert_eq!(handle, Handle(1));
+        assert!(matches!(
+            value.unwrap_err(),
+            bt_gatt::types::Error::Gatt(GattError::ApplicationError81)
+        ));
+    }
+
+    #[test]
+    fn control_point_extra_bytes_fails() {
+        let (mut server, fake_gatt_server, mut event_receiver, mut noop_cx) =
+            setup_test_server(vec![BroadcastReceiveState::Empty]);
+        let mut event_stream = event_receiver.next();
+
+        // RemoteScanStarted opcode (1 byte) with extra trailing byte (0xFF)
+        let raw_cmd = vec![ControlPointOpcode::RemoteScanStarted as u8, 0xFF];
+        fake_gatt_server.incoming_write(PeerId(1), BASS_SERVICE_ID, Handle(1), 0, raw_cmd);
+
+        let _ = server.poll_next_unpin(&mut noop_cx);
+
+        let Poll::Ready(Some(FakeServerEvent::WriteResponded { handle, value, .. })) =
+            event_stream.poll_unpin(&mut noop_cx)
+        else {
+            panic!("Expected WriteResponded event");
+        };
+        assert_eq!(handle, Handle(1));
+        assert!(matches!(
+            value.unwrap_err(),
+            bt_gatt::types::Error::Gatt(GattError::WriteRequestRejected)
+        ));
+    }
+
+    #[test]
+    fn control_point_add_source_success() {
+        let (mut server, fake_gatt_server, mut event_receiver, mut noop_cx) =
+            setup_test_server(vec![BroadcastReceiveState::Empty]);
+        let mut event_stream = event_receiver.next();
+
+        let add_op = AddSourceOperation::new(
+            AddressType::Public,
+            [1, 2, 3, 4, 5, 6],
+            AdvertisingSetId::try_from(1).unwrap(),
+            BroadcastId::try_from(0x123456).unwrap(),
+            PaSync::DoNotSync,
+            PeriodicAdvertisingInterval(0xFFFF),
+            vec![],
+        );
+        let mut raw_bytes = vec![0u8; add_op.encoded_len()];
+        add_op.encode(&mut raw_bytes).expect("encode ok");
+
+        fake_gatt_server.incoming_write(PeerId(1), BASS_SERVICE_ID, Handle(1), 0, raw_bytes);
+
+        let Poll::Ready(Some(Ok(event))) = server.poll_next_unpin(&mut noop_cx) else {
+            panic!("Expected ServerEvent");
+        };
+        assert_eq!(
+            event,
+            ServerEvent::AddSource { peer_id: PeerId(1), source_id: 1, operation: add_op }
+        );
+
+        let Poll::Ready(Some(FakeServerEvent::WriteResponded { handle, value, .. })) =
+            event_stream.poll_unpin(&mut noop_cx)
+        else {
+            panic!("Expected WriteResponded event");
+        };
+        assert_eq!(handle, Handle(1));
+        assert!(value.is_ok());
+    }
+
+    #[test]
+    fn control_point_modify_source_success() {
+        let (mut server, fake_gatt_server, mut event_receiver, mut noop_cx) =
+            setup_test_server(vec![make_test_receive_state(1)]);
+        let mut event_stream = event_receiver.next();
+
+        let modify_op = ModifySourceOperation::new(
+            1,
+            PaSync::SyncPastAvailable,
+            PeriodicAdvertisingInterval(0x0010),
+            vec![],
+        );
+        let mut raw_bytes = vec![0u8; modify_op.encoded_len()];
+        modify_op.encode(&mut raw_bytes).expect("encode ok");
+
+        fake_gatt_server.incoming_write(PeerId(1), BASS_SERVICE_ID, Handle(1), 0, raw_bytes);
+
+        let Poll::Ready(Some(Ok(event))) = server.poll_next_unpin(&mut noop_cx) else {
+            panic!("Expected ServerEvent");
+        };
+        assert_eq!(
+            event,
+            ServerEvent::ModifySource { peer_id: PeerId(1), source_id: 1, operation: modify_op }
+        );
+
+        let Poll::Ready(Some(FakeServerEvent::WriteResponded { handle, value, .. })) =
+            event_stream.poll_unpin(&mut noop_cx)
+        else {
+            panic!("Expected WriteResponded event");
+        };
+        assert_eq!(handle, Handle(1));
+        assert!(value.is_ok());
+    }
+
+    #[test]
+    fn control_point_set_broadcast_code_success() {
+        let (mut server, fake_gatt_server, mut event_receiver, mut noop_cx) =
+            setup_test_server(vec![make_test_receive_state(1)]);
+        let mut event_stream = event_receiver.next();
+
+        let code = BroadcastCode::new([0xAB; 16]);
+        let set_code_op = SetBroadcastCodeOperation::new(1, code);
+        let mut raw_bytes = vec![0u8; set_code_op.encoded_len()];
+        set_code_op.encode(&mut raw_bytes).expect("encode ok");
+
+        fake_gatt_server.incoming_write(PeerId(1), BASS_SERVICE_ID, Handle(1), 0, raw_bytes);
+
+        let Poll::Ready(Some(Ok(event))) = server.poll_next_unpin(&mut noop_cx) else {
+            panic!("Expected ServerEvent");
+        };
+        assert_eq!(
+            event,
+            ServerEvent::SetBroadcastCode {
+                peer_id: PeerId(1),
+                source_id: 1,
+                broadcast_code: code,
+            }
+        );
+
+        let Poll::Ready(Some(FakeServerEvent::WriteResponded { handle, value, .. })) =
+            event_stream.poll_unpin(&mut noop_cx)
+        else {
+            panic!("Expected WriteResponded event");
+        };
+        assert_eq!(handle, Handle(1));
+        assert!(value.is_ok());
+    }
+
+    #[test]
+    fn control_point_remove_source_success() {
+        let (mut server, fake_gatt_server, mut event_receiver, mut noop_cx) =
+            setup_test_server(vec![make_test_receive_state(1)]);
+        let mut event_stream = event_receiver.next();
+
+        let remove_op = RemoveSourceOperation::new(1);
+        let mut raw_bytes = vec![0u8; remove_op.encoded_len()];
+        remove_op.encode(&mut raw_bytes).expect("encode ok");
+
+        fake_gatt_server.incoming_write(PeerId(1), BASS_SERVICE_ID, Handle(1), 0, raw_bytes);
+
+        let Poll::Ready(Some(Ok(event))) = server.poll_next_unpin(&mut noop_cx) else {
+            panic!("Expected ServerEvent");
+        };
+        assert_eq!(event, ServerEvent::RemoveSource { peer_id: PeerId(1), source_id: 1 });
+
+        let Poll::Ready(Some(FakeServerEvent::WriteResponded { handle, value, .. })) =
+            event_stream.poll_unpin(&mut noop_cx)
+        else {
+            panic!("Expected WriteResponded event");
+        };
+        assert_eq!(handle, Handle(1));
+        assert!(value.is_ok());
+    }
+
+    #[test]
+    fn write_non_control_point_fails() {
+        let (mut server, fake_gatt_server, mut event_receiver, mut noop_cx) =
+            setup_test_server(vec![BroadcastReceiveState::Empty]);
+        let mut event_stream = event_receiver.next();
+
+        // Writing to BroadcastReceiveState characteristic handle (Handle 2)
+        fake_gatt_server.incoming_write(PeerId(1), BASS_SERVICE_ID, Handle(2), 0, vec![0x01]);
+
+        let _ = server.poll_next_unpin(&mut noop_cx);
+
+        let Poll::Ready(Some(FakeServerEvent::WriteResponded { handle, value, .. })) =
+            event_stream.poll_unpin(&mut noop_cx)
+        else {
+            panic!("Expected WriteResponded event");
+        };
+        assert_eq!(handle, Handle(2));
+        assert!(matches!(
+            value.unwrap_err(),
+            bt_gatt::types::Error::Gatt(GattError::WriteNotPermitted)
+        ));
+    }
+
+    #[test]
+    fn control_point_empty_payload_fails() {
+        let (mut server, fake_gatt_server, mut event_receiver, mut noop_cx) =
+            setup_test_server(vec![BroadcastReceiveState::Empty]);
+        let mut event_stream = event_receiver.next();
+
+        // Writing empty payload to Control Point
+        fake_gatt_server.incoming_write(PeerId(1), BASS_SERVICE_ID, Handle(1), 0, vec![]);
+
+        let _ = server.poll_next_unpin(&mut noop_cx);
+
+        let Poll::Ready(Some(FakeServerEvent::WriteResponded { handle, value, .. })) =
+            event_stream.poll_unpin(&mut noop_cx)
+        else {
+            panic!("Expected WriteResponded event");
+        };
+        assert_eq!(handle, Handle(1));
+        assert!(matches!(
+            value.unwrap_err(),
+            bt_gatt::types::Error::Gatt(GattError::WriteRequestRejected)
+        ));
     }
 
     #[test]
     fn read_control_point_fails() {
-        let server = ServerBuilder::new()
-            .add_receive_state_characteristic(BroadcastReceiveState::Empty)
-            .build::<FakeTypes>()
-            .expect("building server works");
+        let (mut server, fake_gatt_server, mut event_receiver, mut noop_cx) =
+            setup_test_server(vec![BroadcastReceiveState::Empty]);
+        let mut event_stream = event_receiver.next();
 
-        let res = Arc::new(Mutex::new(None));
-        server.handle_read(CONTROL_POINT_HANDLE, 0, TestReadResponder { result: res.clone() });
-        assert_eq!(res.lock().take().unwrap().expect_err("err"), GattError::ReadNotPermitted);
+        fake_gatt_server.incoming_read(PeerId(1), BASS_SERVICE_ID, CONTROL_POINT_HANDLE, 0);
+        let _ = server.poll_next_unpin(&mut noop_cx);
+        let Poll::Ready(Some(FakeServerEvent::ReadResponded { handle, value, .. })) =
+            event_stream.poll_unpin(&mut noop_cx)
+        else {
+            panic!("Expected ReadResponded event");
+        };
+        assert_eq!(handle, CONTROL_POINT_HANDLE);
+        assert!(matches!(
+            value.unwrap_err(),
+            bt_gatt::types::Error::Gatt(GattError::ReadNotPermitted)
+        ));
     }
 
     #[test]
     fn read_receive_state_invalid_offset() {
-        let server = ServerBuilder::new()
-            .add_receive_state_characteristic(BroadcastReceiveState::Empty)
-            .build::<FakeTypes>()
-            .expect("building server works");
+        let (mut server, fake_gatt_server, mut event_receiver, mut noop_cx) =
+            setup_test_server(vec![BroadcastReceiveState::Empty]);
+        let mut event_stream = event_receiver.next();
 
-        let res = Arc::new(Mutex::new(None));
-        server.handle_read(Handle(2), 10, TestReadResponder { result: res.clone() });
-        assert_eq!(res.lock().take().unwrap().expect_err("err"), GattError::InvalidOffset);
+        fake_gatt_server.incoming_read(PeerId(1), BASS_SERVICE_ID, Handle(2), 10);
+        let _ = server.poll_next_unpin(&mut noop_cx);
+        let Poll::Ready(Some(FakeServerEvent::ReadResponded { handle, value, .. })) =
+            event_stream.poll_unpin(&mut noop_cx)
+        else {
+            panic!("Expected ReadResponded event");
+        };
+        assert_eq!(handle, Handle(2));
+        assert!(matches!(
+            value.unwrap_err(),
+            bt_gatt::types::Error::Gatt(GattError::InvalidOffset)
+        ));
     }
 
     #[test]
-    fn poll_publish_failure() {
-        use futures::task::Context;
-        use std::task::Poll;
-
+    fn publish_failure() {
         let mut noop_cx = Context::from_waker(futures::task::noop_waker_ref());
 
         let mut server = ServerBuilder::new()
@@ -446,11 +1112,13 @@
         fake_gatt_server
             .set_next_prepare_result(Err(bt_gatt::types::Error::Gatt(GattError::UnlikelyError)));
 
-        assert!(server.publish(fake_gatt_server.clone()).is_ok());
+        assert!(server.publish(fake_gatt_server).is_ok());
 
         assert!(matches!(
-            std::pin::Pin::new(&mut server).poll_publish(&mut noop_cx),
-            Poll::Ready(Err(Error::Gatt(bt_gatt::types::Error::Gatt(GattError::UnlikelyError))))
+            server.poll_next_unpin(&mut noop_cx),
+            Poll::Ready(Some(Err(Error::Gatt(bt_gatt::types::Error::Gatt(
+                GattError::UnlikelyError
+            )))))
         ));
 
         // Verifying server reset state to NotPublished so it is not published
@@ -461,15 +1129,41 @@
     fn builder_normalizes_source_id() {
         // Create state with an arbitrary source_id = 99
         let state = make_test_receive_state(99);
-        let server = ServerBuilder::new()
-            .add_receive_state_characteristic(state)
-            .build::<FakeTypes>()
-            .expect("building server works");
+        let (mut server, fake_gatt_server, mut event_receiver, mut noop_cx) =
+            setup_test_server(vec![state]);
+        let mut event_stream = event_receiver.next();
 
         // Verify slot's internal state and encoded read value have source_id = 1
-        let res = Arc::new(Mutex::new(None));
-        server.handle_read(Handle(2), 0, TestReadResponder { result: res.clone() });
-        let buf = res.lock().take().unwrap().expect("ok");
+        fake_gatt_server.incoming_read(PeerId(1), BASS_SERVICE_ID, Handle(2), 0);
+        let _ = server.poll_next_unpin(&mut noop_cx);
+        let Poll::Ready(Some(FakeServerEvent::ReadResponded { handle: _, value, .. })) =
+            event_stream.poll_unpin(&mut noop_cx)
+        else {
+            panic!("Expected ReadResponded event");
+        };
+        let buf = value.expect("ok");
         assert_eq!(buf[0], 1);
     }
+
+    #[test]
+    fn allocate_source_id_skips_in_use_ids() {
+        let mut state = ServerState::new(5);
+        state.source_id_to_handle.insert(1, Handle(10));
+        // Force the next ID to be 1 to simulate wrap around.
+        state.next_source_id = NonZeroU8::new(1).unwrap();
+
+        let allocated_id = state.allocate_source_id(Handle(20)).unwrap();
+        assert_eq!(allocated_id, 2);
+        assert_eq!(state.source_id_to_handle.get(&2), Some(&Handle(20)));
+    }
+
+    #[test]
+    fn allocate_source_id_error_when_full() {
+        let mut state = ServerState::new(255);
+        for id in 1..=255 {
+            state.source_id_to_handle.insert(id, Handle(u64::from(id)));
+        }
+
+        assert_eq!(state.allocate_source_id(Handle(300)), Err(GattError::InsufficientResources));
+    }
 }
diff --git a/rust/bt-bass/src/types.rs b/rust/bt-bass/src/types.rs
index 2418d9e..64bf76c 100644
--- a/rust/bt-bass/src/types.rs
+++ b/rust/bt-bass/src/types.rs
@@ -2,12 +2,13 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
-use bt_bap::types::BroadcastId;
+use bt_bap::types::{BroadcastCode, BroadcastId};
 use bt_common::core::ltv::LtValue;
 use bt_common::core::{AddressType, AdvertisingSetId, PeriodicAdvertisingInterval};
 use bt_common::generic_audio::metadata_ltv::*;
 use bt_common::packet_encoding::{Decodable, Encodable, Error as PacketError};
 use bt_common::{decodable_enum, Uuid};
+use bt_gatt::types::GattError;
 use std::str::FromStr;
 
 pub const ADDRESS_BYTE_SIZE: usize = 6;
@@ -21,6 +22,11 @@
 pub const BROADCAST_AUDIO_SCAN_CONTROL_POINT_UUID: Uuid = Uuid::from_u16(0x2BC7);
 pub const BROADCAST_RECEIVE_STATE_UUID: Uuid = Uuid::from_u16(0x2BC8);
 
+/// ATT Application Error Codes for the Broadcast Audio Scan Service.
+/// See BASS v1.0 Table 3.2.
+pub const ERROR_OPCODE_NOT_SUPPORTED: GattError = GattError::ApplicationError80;
+pub const ERROR_INVALID_SOURCE_ID: GattError = GattError::ApplicationError81;
+
 pub type SourceId = u8;
 
 /// Index into the vector of BIG subgroups. Valid value range is [0 to len of
@@ -284,10 +290,10 @@
 /// See Broadcast Audio Scan Service spec v1.0 Section 3.1.1.5 for details.
 #[derive(Debug, PartialEq)]
 pub struct ModifySourceOperation {
-    source_id: SourceId,
-    pa_sync: PaSync,
-    pa_interval: PeriodicAdvertisingInterval,
-    subgroups: Vec<BigSubgroup>,
+    pub(crate) source_id: SourceId,
+    pub(crate) pa_sync: PaSync,
+    pub(crate) pa_interval: PeriodicAdvertisingInterval,
+    pub(crate) subgroups: Vec<BigSubgroup>,
 }
 
 impl ModifySourceOperation {
@@ -382,16 +388,16 @@
 /// See Broadcast Audio Scan Service spec v1.0 Section 3.1.1.6 for details.
 #[derive(Debug, PartialEq)]
 pub struct SetBroadcastCodeOperation {
-    source_id: SourceId,
-    broadcast_code: [u8; 16],
+    pub(crate) source_id: SourceId,
+    pub(crate) broadcast_code: BroadcastCode,
 }
 
 impl SetBroadcastCodeOperation {
-    const BROADCAST_CODE_LEN: usize = 16;
+    const BROADCAST_CODE_LEN: usize = BroadcastCode::BYTE_SIZE;
     const PACKET_SIZE: usize =
         ControlPointOpcode::BYTE_SIZE + SOURCE_ID_BYTE_SIZE + Self::BROADCAST_CODE_LEN;
 
-    pub fn new(source_id: SourceId, broadcast_code: [u8; 16]) -> Self {
+    pub fn new(source_id: SourceId, broadcast_code: BroadcastCode) -> Self {
         SetBroadcastCodeOperation { source_id, broadcast_code }
     }
 }
@@ -413,8 +419,9 @@
         let decode_fn = || {
             let _ = Self::check_opcode(buf[0])?;
             let source_id = buf[1];
-            let mut broadcast_code = [0; Self::BROADCAST_CODE_LEN];
-            broadcast_code.copy_from_slice(&buf[2..2 + Self::BROADCAST_CODE_LEN]);
+            let mut raw_code = [0; Self::BROADCAST_CODE_LEN];
+            raw_code.copy_from_slice(&buf[2..2 + Self::BROADCAST_CODE_LEN]);
+            let broadcast_code = BroadcastCode::new(raw_code);
             Ok((Self { source_id, broadcast_code }, Self::PACKET_SIZE))
         };
 
@@ -435,7 +442,7 @@
 
         buf[0] = Self::opcode() as u8;
         buf[1] = self.source_id;
-        buf[2..2 + Self::BROADCAST_CODE_LEN].copy_from_slice(&self.broadcast_code);
+        buf[2..2 + Self::BROADCAST_CODE_LEN].copy_from_slice(&self.broadcast_code.0);
         Ok(())
     }
 
@@ -446,7 +453,7 @@
 
 /// See Broadcast Audio Scan Service spec v1.0 Section 3.1.1.7 for details.
 #[derive(Debug, PartialEq)]
-pub struct RemoveSourceOperation(SourceId);
+pub struct RemoveSourceOperation(pub(crate) SourceId);
 
 impl RemoveSourceOperation {
     const PACKET_SIZE: usize = ControlPointOpcode::BYTE_SIZE + SOURCE_ID_BYTE_SIZE;
@@ -762,7 +769,6 @@
         + EncryptionStatus::MIN_PACKET_SIZE
         + NUM_SUBGROUPS_BYTE_SIZE;
 
-    #[cfg(any(test, feature = "test-utils"))]
     pub fn new(
         source_id: u8,
         source_address_type: AddressType,
@@ -920,6 +926,17 @@
     }
 }
 
+impl PaSyncState {
+    /// Returns the initial [`PaSyncState`] corresponding to a requested
+    /// [`PaSync`].
+    pub fn from_pa_sync(pa_sync: PaSync) -> Self {
+        match pa_sync {
+            PaSync::SyncPastAvailable => Self::SyncInfoRequest,
+            _ => Self::NotSynced,
+        }
+    }
+}
+
 /// Represents BIG_Encryption and Bad_Code params from BASS spec v.1.0 Table
 /// 3.9.
 #[derive(Clone, Copy, Debug, PartialEq)]
@@ -1310,10 +1327,10 @@
         // Encoding.
         let op = SetBroadcastCodeOperation::new(
             0x0A,
-            [
+            BroadcastCode::new([
                 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x10, 0x11, 0x12, 0x13, 0x14,
                 0x15, 0x16,
-            ],
+            ]),
         );
         assert_eq!(op.encoded_len(), 18);
         let mut buf = vec![0; op.encoded_len()];
diff --git a/rust/bt-gatt/src/types.rs b/rust/bt-gatt/src/types.rs
index c65d05b..07527e3 100644
--- a/rust/bt-gatt/src/types.rs
+++ b/rust/bt-gatt/src/types.rs
@@ -196,7 +196,7 @@
 /// use in correlating with ServiceEvents from peers.  Stacks must translate
 /// the actual GATT handles to the handles provided in the ServiceDefinition
 /// for these events.
-#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
 pub struct Handle(pub u64);
 
 /// Whether a service is marked as Primary or Secondary on the server.