diff --git a/litebox/src/broker/mod.rs b/litebox/src/broker/mod.rs
index 5334ae958..42d040bad 100644
--- a/litebox/src/broker/mod.rs
+++ b/litebox/src/broker/mod.rs
@@ -53,35 +53,31 @@ pub(crate) trait BrokerControl: Send + Sync {
     ) -> core::result::Result<ConsumeEventResponse, BrokerControlError>;
 
     fn close_object(&self, handle: ObjectHandle) -> core::result::Result<(), BrokerControlError>;
+
+    fn fail_connection(&self);
 }
 
-pub(crate) struct BrokerHandleRegistry<Platform: RawSyncPrimitivesProvider> {
-    handles: Mutex<Platform, HashMap<ObjectHandle, BrokerHandleEntry<Platform>>>,
+pub(crate) struct BrokerPollableRegistry<Platform: RawSyncPrimitivesProvider> {
+    pollables: Mutex<Platform, HashMap<ObjectHandle, Weak<Pollee<Platform>>>>,
 }
 
-impl<Platform: RawSyncPrimitivesProvider> BrokerHandleRegistry<Platform> {
+impl<Platform: RawSyncPrimitivesProvider> BrokerPollableRegistry<Platform> {
     pub(crate) fn new() -> Self {
         Self {
-            handles: Mutex::new(HashMap::new()),
+            pollables: Mutex::new(HashMap::new()),
         }
     }
 
     pub(crate) fn register_pollable(&self, handle: ObjectHandle, pollee: &Arc<Pollee<Platform>>) {
-        self.handles
-            .lock()
-            .entry(handle)
-            .or_insert_with(BrokerHandleEntry::new)
-            .register_pollable(pollee);
+        let previous = self.pollables.lock().insert(handle, Arc::downgrade(pollee));
+        assert!(
+            previous.is_none(),
+            "broker handle already has a registered pollable"
+        );
     }
 
-    pub(crate) fn unregister_pollable(&self, handle: ObjectHandle, pollee: &Arc<Pollee<Platform>>) {
-        let mut handles = self.handles.lock();
-        if let Some(entry) = handles.get_mut(&handle) {
-            entry.unregister_pollable(pollee);
-            if entry.pollables.is_empty() {
-                handles.remove(&handle);
-            }
-        }
+    pub(crate) fn unregister_pollable(&self, handle: ObjectHandle) {
+        self.pollables.lock().remove(&handle);
     }
 
     pub(crate) fn notify_readiness(&self, handle: ObjectHandle, readiness: ReadinessFlags)
@@ -92,92 +88,108 @@ impl<Platform: RawSyncPrimitivesProvider> BrokerHandleRegistry<Platform> {
         if events.is_empty() {
             return;
         }
-        let pollables = {
-            let mut handles = self.handles.lock();
-            let Some(entry) = handles.get_mut(&handle) else {
-                return;
-            };
-            entry.prune_stale_pollables();
-            let pollables = entry
-                .pollables
-                .iter()
-                .filter_map(Weak::upgrade)
-                .collect::<Vec<_>>();
-            if entry.pollables.is_empty() {
-                handles.remove(&handle);
+        let pollee = {
+            let mut pollables = self.pollables.lock();
+            let pollee = pollables.get(&handle).and_then(Weak::upgrade);
+            if pollee.is_none() {
+                pollables.remove(&handle);
             }
-            pollables
+            pollee
         };
-        for pollee in pollables {
+        if let Some(pollee) = pollee {
             pollee.notify_observers(events);
         }
     }
-}
-
-struct BrokerHandleEntry<Platform: RawSyncPrimitivesProvider> {
-    pollables: Vec<Weak<Pollee<Platform>>>,
-}
 
-impl<Platform: RawSyncPrimitivesProvider> BrokerHandleEntry<Platform> {
-    fn new() -> Self {
-        Self {
-            pollables: Vec::new(),
+    fn notify_all(&self, events: Events)
+    where
+        Platform: TimeProvider,
+    {
+        let pollables = {
+            let mut pollables = Vec::new();
+            self.pollables.lock().retain(|_, registered| {
+                let Some(pollee) = registered.upgrade() else {
+                    return false;
+                };
+                pollables.push(pollee);
+                true
+            });
+            pollables
+        };
+        for pollee in pollables {
+            pollee.notify_observers(events);
         }
     }
-
-    fn register_pollable(&mut self, pollee: &Arc<Pollee<Platform>>) {
-        self.pollables.push(Arc::downgrade(pollee));
-    }
-
-    fn unregister_pollable(&mut self, pollee: &Arc<Pollee<Platform>>) {
-        self.pollables.retain(|registered| {
-            registered
-                .upgrade()
-                .is_some_and(|registered| !Arc::ptr_eq(&registered, pollee))
-        });
-    }
-
-    fn prune_stale_pollables(&mut self) {
-        self.pollables
-            .retain(|registered| registered.strong_count() > 0);
-    }
 }
+
 pub(crate) struct BrokerLocalControl<
     Platform: RawSyncPrimitivesProvider,
     Channel: LocalControlChannel + Send,
 > {
-    local: Mutex<Platform, BrokerLocal<Channel>>,
+    local: Mutex<Platform, Option<BrokerLocal<Channel>>>,
+    pollable_registry: Arc<BrokerPollableRegistry<Platform>>,
 }
 
 impl<Platform, Channel> BrokerLocalControl<Platform, Channel>
 where
-    Platform: RawSyncPrimitivesProvider,
+    Platform: RawSyncPrimitivesProvider + TimeProvider,
     Channel: LocalControlChannel + Send,
 {
-    pub(crate) const fn new(local: BrokerLocal<Channel>) -> Self {
+    pub(crate) fn new(
+        local: BrokerLocal<Channel>,
+        pollable_registry: Arc<BrokerPollableRegistry<Platform>>,
+    ) -> Self {
         Self {
-            local: Mutex::new(local),
+            local: Mutex::new(Some(local)),
+            pollable_registry,
         }
     }
+
+    fn request<T>(
+        &self,
+        request: impl FnOnce(
+            &mut BrokerLocal<Channel>,
+        ) -> litebox_broker_local::Result<T, Channel::Error>,
+    ) -> core::result::Result<T, BrokerControlError> {
+        let (result, failed_connection) = {
+            let mut local = self.local.lock();
+            let Some(connection) = local.as_mut() else {
+                return Err(BrokerControlError::Transport);
+            };
+            let result = request(connection).map_err(BrokerControlError::from);
+            let failed_connection = if matches!(result.as_ref(), Err(BrokerControlError::Transport))
+            {
+                local.take()
+            } else {
+                None
+            };
+            (result, failed_connection)
+        };
+        if let Some(connection) = failed_connection {
+            drop(connection);
+            self.pollable_registry.notify_all(Events::ERR);
+        }
+        result
+    }
 }
 
 impl<Platform, Channel> BrokerControl for BrokerLocalControl<Platform, Channel>
 where
-    Platform: RawSyncPrimitivesProvider,
+    Platform: RawSyncPrimitivesProvider + TimeProvider,
     Channel: LocalControlChannel + Send,
 {
     fn create_event_with_count(
         &self,
         initial_count: u64,
     ) -> core::result::Result<ObjectHandle, BrokerControlError> {
-        Ok(self.local.lock().create_event_with_count(initial_count)?)
+        self.request(|local| local.create_event_with_count(initial_count))
     }
 
     fn wait_event(
         &self,
         handle: ObjectHandle,
     ) -> core::result::Result<ReadinessFlags, BrokerControlError> {
-        Ok(self.local.lock().wait_event(handle)?)
+        self.request(|local| local.wait_event(handle))
     }
 
     fn add_event(
@@ -185,7 +197,7 @@ where
         handle: ObjectHandle,
         value: u64,
     ) -> core::result::Result<ReadinessFlags, BrokerControlError> {
-        Ok(self.local.lock().add_event(handle, value)?)
+        self.request(|local| local.add_event(handle, value))
     }
 
     fn consume_event(
@@ -193,11 +205,19 @@ where
         handle: ObjectHandle,
         mode: EventConsumeMode,
     ) -> core::result::Result<ConsumeEventResponse, BrokerControlError> {
-        Ok(self.local.lock().consume_event(handle, mode)?)
+        self.request(|local| local.consume_event(handle, mode))
     }
 
     fn close_object(&self, handle: ObjectHandle) -> core::result::Result<(), BrokerControlError> {
-        Ok(self.local.lock().close_object(handle)?)
+        self.request(|local| local.close_object(handle))
+    }
+
+    fn fail_connection(&self) {
+        let connection = self.local.lock().take();
+        if let Some(connection) = connection {
+            drop(connection);
+            self.pollable_registry.notify_all(Events::ERR);
+        }
     }
 }
 
diff --git a/litebox/src/event/counter.rs b/litebox/src/event/counter.rs
index b346e1d36..4ebd50e60 100644
--- a/litebox/src/event/counter.rs
+++ b/litebox/src/event/counter.rs
@@ -12,7 +12,7 @@ use thiserror::Error;
 use crate::{
     LiteBox,
     broker::{
-        BrokerControl, BrokerHandleRegistry,
+        BrokerControl, BrokerPollableRegistry,
         error::{BrokerControlError, BrokerObjectError},
         readiness_events,
     },
@@ -46,7 +46,7 @@ pub enum EventCounterError {
 pub struct EventCounter<Platform: RawSyncPrimitivesProvider + TimeProvider> {
     broker: Arc<dyn BrokerControl>,
     handle: ObjectHandle,
-    registry: Arc<BrokerHandleRegistry<Platform>>,
+    pollable_registry: Arc<BrokerPollableRegistry<Platform>>,
     pollee: Arc<Pollee<Platform>>,
 }
 
@@ -68,13 +68,13 @@ where
             .create_event_with_count(initial_count)
             .map_err(BrokerObjectError::from)
             .map_err(EventCounterError::from)?;
-        let registry = litebox.broker_handle_registry();
+        let pollable_registry = litebox.broker_pollable_registry();
         let pollee = Arc::new(Pollee::new());
-        registry.register_pollable(handle, &pollee);
+        pollable_registry.register_pollable(handle, &pollee);
         Ok(Self {
             broker,
             handle,
-            registry,
+            pollable_registry,
             pollee,
         })
     }
@@ -143,7 +143,7 @@ where
     Platform: RawSyncPrimitivesProvider + TimeProvider,
 {
     fn drop(&mut self) {
-        self.registry.unregister_pollable(self.handle, &self.pollee);
+        self.pollable_registry.unregister_pollable(self.handle);
         let _ = self.broker.close_object(self.handle);
     }
 }
@@ -200,10 +200,13 @@ mod tests {
         let handle = ObjectHandle(7);
         let consume_attempts = Arc::new(AtomicUsize::new(0));
         let read_ready = Arc::new(AtomicBool::new(false));
+        let request_count = Arc::new(AtomicUsize::new(0));
         let local = BrokerLocal::negotiate(FakeLocalControlChannel {
-            handle,
+            next_handle: handle.0,
             consume_attempts: consume_attempts.clone(),
             read_ready: read_ready.clone(),
+            request_count,
+            fail_requests: Arc::new(AtomicBool::new(false)),
             last_request: None,
         })
         .unwrap();
@@ -246,15 +249,159 @@ mod tests {
         );
     }
 
+    #[test]
+    fn broker_association_failure_wakes_blocked_read() {
+        use std::time::{Duration, Instant};
+
+        let platform = MockPlatform::new();
+        let handle = ObjectHandle(7);
+        let consume_attempts = Arc::new(AtomicUsize::new(0));
+        let request_count = Arc::new(AtomicUsize::new(0));
+        let local = BrokerLocal::negotiate(FakeLocalControlChannel {
+            next_handle: handle.0,
+            consume_attempts: Arc::clone(&consume_attempts),
+            read_ready: Arc::new(AtomicBool::new(false)),
+            request_count: Arc::clone(&request_count),
+            fail_requests: Arc::new(AtomicBool::new(false)),
+            last_request: None,
+        })
+        .unwrap();
+        let litebox = Arc::new(LiteBox::new_with_broker_local(platform, local));
+        let counter = Arc::new(EventCounter::new(&litebox, 0).unwrap());
+
+        let (result_sender, result_receiver) = std::sync::mpsc::sync_channel(1);
+        let read_counter = Arc::clone(&counter);
+        let reader = std::thread::spawn(move || {
+            result_sender
+                .send(read_counter.read(
+                    &WaitState::new(platform).context(),
+                    false,
+                    EventCounterReadMode::One,
+                ))
+                .unwrap();
+        });
+        let deadline = Instant::now() + Duration::from_secs(1);
+        while consume_attempts.load(Ordering::SeqCst) < 2 {
+            assert!(Instant::now() < deadline);
+            std::thread::yield_now();
+        }
+
+        litebox.broker_failure_dispatcher()();
+
+        assert!(matches!(
+            result_receiver
+                .recv_timeout(Duration::from_secs(1))
+                .unwrap(),
+            Err(TryOpError::Other(EventCounterError::Io))
+        ));
+        reader.join().unwrap();
+        assert_eq!(request_count.load(Ordering::SeqCst), 3);
+        assert_eq!(counter.check_io_events(), Events::ERR);
+        assert_eq!(request_count.load(Ordering::SeqCst), 3);
+    }
+
+    #[test]
+    fn control_transport_failure_notifies_all_event_counters() {
+        let platform = MockPlatform::new();
+        let handle = ObjectHandle(7);
+        let request_count = Arc::new(AtomicUsize::new(0));
+        let fail_requests = Arc::new(AtomicBool::new(false));
+        let local = BrokerLocal::negotiate(FakeLocalControlChannel {
+            next_handle: handle.0,
+            consume_attempts: Arc::new(AtomicUsize::new(0)),
+            read_ready: Arc::new(AtomicBool::new(false)),
+            request_count: Arc::clone(&request_count),
+            fail_requests: Arc::clone(&fail_requests),
+            last_request: None,
+        })
+        .unwrap();
+        let litebox = LiteBox::new_with_broker_local(platform, local);
+        let first = EventCounter::new(&litebox, 0).unwrap();
+        let second = EventCounter::new(&litebox, 0).unwrap();
+        let first_observer = Arc::new(ErrorObserver(AtomicBool::new(false)));
+        let first_observer_dyn: Arc<dyn Observer<Events>> = first_observer.clone();
+        first.register_observer(Arc::downgrade(&first_observer_dyn), Events::ERR);
+        let second_observer = Arc::new(ErrorObserver(AtomicBool::new(false)));
+        let second_observer_dyn: Arc<dyn Observer<Events>> = second_observer.clone();
+        second.register_observer(Arc::downgrade(&second_observer_dyn), Events::ERR);
+
+        fail_requests.store(true, Ordering::SeqCst);
+
+        assert_eq!(first.check_io_events(), Events::ERR);
+        assert!(first_observer.0.load(Ordering::SeqCst));
+        assert!(second_observer.0.load(Ordering::SeqCst));
+        assert_eq!(request_count.load(Ordering::SeqCst), 3);
+        assert_eq!(second.check_io_events(), Events::ERR);
+        assert_eq!(request_count.load(Ordering::SeqCst), 3);
+    }
+
+    #[test]
+    fn broker_dispatchers_follow_objects_that_outlive_litebox() {
+        let platform = MockPlatform::new();
+        let handle = ObjectHandle(7);
+        let request_count = Arc::new(AtomicUsize::new(0));
+        let local = BrokerLocal::negotiate(FakeLocalControlChannel {
+            next_handle: handle.0,
+            consume_attempts: Arc::new(AtomicUsize::new(0)),
+            read_ready: Arc::new(AtomicBool::new(false)),
+            request_count: Arc::clone(&request_count),
+            fail_requests: Arc::new(AtomicBool::new(false)),
+            last_request: None,
+        })
+        .unwrap();
+        let litebox = LiteBox::new_with_broker_local(platform, local);
+        let counter = EventCounter::new(&litebox, 0).unwrap();
+        let read_observer = Arc::new(ReadObserver(AtomicBool::new(false)));
+        let read_observer_dyn: Arc<dyn Observer<Events>> = read_observer.clone();
+        counter.register_observer(Arc::downgrade(&read_observer_dyn), Events::IN);
+        let litebox_weak = Arc::downgrade(&litebox.x);
+        let dispatch_notification = litebox.broker_notification_dispatcher();
+        let dispatch_failure = litebox.broker_failure_dispatcher();
+
+        drop(litebox);
+
+        assert!(litebox_weak.upgrade().is_none());
+        dispatch_notification(BrokerNotification::Readiness(ReadinessNotification {
+            handle,
+            readiness: ReadinessFlags::READ,
+        }));
+        assert!(read_observer.0.load(Ordering::SeqCst));
+        dispatch_failure();
+        assert_eq!(counter.check_io_events(), Events::ERR);
+        assert_eq!(request_count.load(Ordering::SeqCst), 1);
+    }
+
+    struct ErrorObserver(AtomicBool);
+
+    impl Observer<Events> for ErrorObserver {
+        fn on_events(&self, events: &Events) {
+            if events.contains(Events::ERR) {
+                self.0.store(true, Ordering::SeqCst);
+            }
+        }
+    }
+
+    struct ReadObserver(AtomicBool);
+
+    impl Observer<Events> for ReadObserver {
+        fn on_events(&self, events: &Events) {
+            if events.contains(Events::IN) {
+                self.0.store(true, Ordering::SeqCst);
+            }
+        }
+    }
+
     struct FakeLocalControlChannel {
-        handle: ObjectHandle,
+        next_handle: u64,
         consume_attempts: Arc<AtomicUsize>,
         read_ready: Arc<AtomicBool>,
+        request_count: Arc<AtomicUsize>,
+        fail_requests: Arc<AtomicBool>,
         last_request: Option<BrokerRequest>,
     }
 
     impl LocalControlChannel for FakeLocalControlChannel {
-        type Error = core::convert::Infallible;
+        type Error = ();
 
         fn send_handshake_request(
             &mut self,
@@ -276,19 +423,22 @@ mod tests {
             request: &BrokerRequest,
         ) -> core::result::Result<(), Self::Error> {
             self.last_request = Some(request.clone());
+            self.request_count.fetch_add(1, Ordering::SeqCst);
             Ok(())
         }
 
         fn recv_response(&mut self) -> core::result::Result<Option<BrokerResponse>, Self::Error> {
+            if self.fail_requests.load(Ordering::SeqCst) {
+                self.last_request.take();
+                return Err(());
+            }
             let response = match self.last_request.take().unwrap() {
                 BrokerRequest::Event(EventRequest::Create(_)) => {
-                    BrokerResponse::Event(EventResponse::Create(CreateEventResponse {
-                        handle: self.handle,
-                    }))
+                    let handle = ObjectHandle(self.next_handle);
+                    self.next_handle += 1;
+                    BrokerResponse::Event(EventResponse::Create(CreateEventResponse { handle }))
                 }
-                BrokerRequest::Event(EventRequest::Consume(request))
-                    if request.handle == self.handle =>
-                {
+                BrokerRequest::Event(EventRequest::Consume(_)) => {
                     self.consume_attempts.fetch_add(1, Ordering::SeqCst);
                     if self.read_ready.swap(false, Ordering::SeqCst) {
                         BrokerResponse::Event(EventResponse::Consume(EventConsumption {
@@ -299,10 +449,10 @@ mod tests {
                         BrokerResponse::Error(ErrorCode::WouldBlock)
                     }
                 }
-                BrokerRequest::CloseObject(handle) if handle == self.handle => {
-                    BrokerResponse::ObjectClosed
+                BrokerRequest::CloseObject(_) => BrokerResponse::ObjectClosed,
+                request @ BrokerRequest::Event(_) => {
+                    panic!("unexpected broker request: {request:?}")
                 }
-                request => panic!("unexpected broker request: {request:?}"),
             };
             Ok(Some(response))
         }
diff --git a/litebox/src/litebox.rs b/litebox/src/litebox.rs
index 637c6301b..36838a8db 100644
--- a/litebox/src/litebox.rs
+++ b/litebox/src/litebox.rs
@@ -36,7 +36,11 @@ impl<Platform: RawSyncPrimitivesProvider> LiteBox<Platform> {
     /// If the `enforce_singleton_litebox_instance` compilation feature has been enabled, and more
     /// than one instance is made, will panic.
     pub fn new(platform: &'static Platform) -> Self {
-        Self::new_inner(platform, None)
+        Self::new_inner(
+            platform,
+            None,
+            Arc::new(broker::BrokerPollableRegistry::new()),
+        )
     }
 
     /// Create a new [`LiteBox`] instance with a negotiated broker-local control adapter installed.
@@ -45,19 +49,21 @@ impl<Platform: RawSyncPrimitivesProvider> LiteBox<Platform> {
         broker_local: BrokerLocal<Channel>,
     ) -> Self
     where
+        Platform: TimeProvider,
         Channel: LocalControlChannel + Send + 'static,
     {
-        Self::new_inner(
-            platform,
-            Some(Arc::new(
-                broker::BrokerLocalControl::<Platform, Channel>::new(broker_local),
-            )),
-        )
+        let broker_pollables = Arc::new(broker::BrokerPollableRegistry::new());
+        let broker_control = Arc::new(broker::BrokerLocalControl::<Platform, Channel>::new(
+            broker_local,
+            Arc::clone(&broker_pollables),
+        ));
+        Self::new_inner(platform, Some(broker_control), broker_pollables)
     }
 
     fn new_inner(
         platform: &'static Platform,
         broker_control: Option<Arc<dyn broker::BrokerControl>>,
+        broker_pollables: Arc<broker::BrokerPollableRegistry<Platform>>,
     ) -> Self {
         // This check ensures that there is exactly one `LiteBox` instance in the process.
         //
@@ -102,7 +108,7 @@ impl<Platform: RawSyncPrimitivesProvider> LiteBox<Platform> {
                 platform,
                 descriptors,
                 broker: broker_control,
-                broker_handles: Arc::new(broker::BrokerHandleRegistry::new()),
+                broker_pollables,
             }),
         }
     }
@@ -140,8 +146,8 @@ impl<Platform: RawSyncPrimitivesProvider> LiteBox<Platform> {
         self.x.broker.clone()
     }
 
-    pub(crate) fn broker_handle_registry(&self) -> Arc<broker::BrokerHandleRegistry<Platform>> {
-        Arc::clone(&self.x.broker_handles)
+    pub(crate) fn broker_pollable_registry(&self) -> Arc<broker::BrokerPollableRegistry<Platform>> {
+        Arc::clone(&self.x.broker_pollables)
     }
 
     /// Dispatches one broker notification to the matching local-core object.
@@ -152,7 +158,7 @@ impl<Platform: RawSyncPrimitivesProvider> LiteBox<Platform> {
         match notification {
             BrokerNotification::Readiness(notification) => self
                 .x
-                .broker_handles
+                .broker_pollables
                 .notify_readiness(notification.handle, notification.readiness),
         }
     }
@@ -162,9 +168,26 @@ impl<Platform: RawSyncPrimitivesProvider> LiteBox<Platform> {
     where
         Platform: TimeProvider + 'static,
     {
-        let litebox = self.clone();
+        let broker_pollables = Arc::downgrade(&self.x.broker_pollables);
         move |notification| {
-            litebox.dispatch_broker_notification(notification);
+            if let Some(broker_pollables) = broker_pollables.upgrade() {
+                match notification {
+                    BrokerNotification::Readiness(notification) => {
+                        broker_pollables
+                            .notify_readiness(notification.handle, notification.readiness);
+                    }
+                }
+            }
+        }
+    }
+
+    /// Returns a dispatcher that fails all broker-backed objects when the association closes.
+    pub fn broker_failure_dispatcher(&self) -> impl Fn() + Send + 'static {
+        let broker = self.x.broker.as_ref().map(Arc::downgrade);
+        move || {
+            if let Some(broker) = broker.as_ref().and_then(alloc::sync::Weak::upgrade) {
+                broker.fail_connection();
+            }
         }
     }
 }
@@ -174,5 +197,5 @@ pub(crate) struct LiteBoxX<Platform: RawSyncPrimitivesProvider> {
     pub(crate) platform: &'static Platform,
     descriptors: RwLock<Platform, Descriptors<Platform>>,
     broker: Option<Arc<dyn broker::BrokerControl>>,
-    broker_handles: Arc<broker::BrokerHandleRegistry<Platform>>,
+    broker_pollables: Arc<broker::BrokerPollableRegistry<Platform>>,
 }
diff --git a/litebox_broker_transport/src/unix_socket.rs b/litebox_broker_transport/src/unix_socket.rs
index 1b5f956a3..2f3c5ee25 100644
--- a/litebox_broker_transport/src/unix_socket.rs
+++ b/litebox_broker_transport/src/unix_socket.rs
@@ -8,6 +8,7 @@
 //! no_std protocol, local, core, and host crates.
 
 use std::io::{Error, ErrorKind, Read, Result as IoResult, Write};
+use std::net::Shutdown;
 use std::os::unix::net::UnixStream;
 use std::path::Path;
 use std::time::{Duration, Instant};
@@ -34,6 +35,11 @@ pub struct UnixStreamLocalControlChannel {
     setup_deadline: Option<Instant>,
 }
 
+/// Independently owned handle for interrupting local control-channel I/O.
+pub struct UnixStreamLocalControlCancellation {
+    stream: UnixStream,
+}
+
 impl UnixStreamLocalControlChannel {
     /// Creates a local control channel from an already-connected Unix stream.
     pub const fn from_connected(stream: UnixStream) -> Self {
@@ -62,6 +68,29 @@ impl UnixStreamLocalControlChannel {
             setup_deadline: Some(deadline),
         })
     }
+
+    /// Creates a handle that can interrupt pending control-channel I/O.
+    pub fn cancellation_handle(&self) -> IoResult<UnixStreamLocalControlCancellation> {
+        self.stream
+            .try_clone()
+            .map(|stream| UnixStreamLocalControlCancellation { stream })
+    }
+}
+
+impl UnixStreamLocalControlCancellation {
+    /// Shuts down the control stream, unblocking pending reads or writes.
+    pub fn cancel(&self) -> IoResult<()> {
+        match self.stream.shutdown(Shutdown::Both) {
+            Err(error) if error.kind() == ErrorKind::NotConnected => Ok(()),
+            result => result,
+        }
+    }
+}
+
+impl Drop for UnixStreamLocalControlChannel {
+    fn drop(&mut self) {
+        let _ = self.stream.shutdown(Shutdown::Both);
+    }
 }
 
 /// Host-side Unix-domain-socket control channel for the hosted userland POC.
@@ -403,6 +432,47 @@ mod tests {
         );
     }
 
+    #[test]
+    fn local_control_cancellation_unblocks_response_read() {
+        let (local_stream, _host_stream) = UnixStream::pair().unwrap();
+        let mut channel = UnixStreamLocalControlChannel::from_connected(local_stream);
+        let cancellation = channel.cancellation_handle().unwrap();
+        let completed = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
+        let (started_sender, started_receiver) = std::sync::mpsc::sync_channel(1);
+        let (result_sender, result_receiver) = std::sync::mpsc::sync_channel(1);
+        let reader_completed = completed.clone();
+        let reader = std::thread::spawn(move || {
+            started_sender.send(()).unwrap();
+            result_sender.send(channel.recv_response()).unwrap();
+            reader_completed.store(true, std::sync::atomic::Ordering::Release);
+        });
+
+        started_receiver
+            .recv_timeout(Duration::from_secs(1))
+            .unwrap();
+        std::thread::sleep(Duration::from_millis(50));
+        assert!(!completed.load(std::sync::atomic::Ordering::Acquire));
+        cancellation.cancel().unwrap();
+
+        assert!(result_receiver.recv_timeout(Duration::from_secs(1)).is_ok());
+        reader.join().unwrap();
+    }
+
+    #[test]
+    fn dropping_local_control_closes_connection_with_cancellation_clone() {
+        let (local_stream, mut host_stream) = UnixStream::pair().unwrap();
+        let channel = UnixStreamLocalControlChannel::from_connected(local_stream);
+        let _cancellation = channel.cancellation_handle().unwrap();
+        host_stream
+            .set_read_timeout(Some(Duration::from_secs(1)))
+            .unwrap();
+
+        drop(channel);
+
+        let mut byte = [0];
+        assert_eq!(host_stream.read(&mut byte).unwrap(), 0);
+    }
+
     #[test]
     fn host_reports_wrong_phase_request_frames_as_protocol_violations() {
         let (mut peer_stream, host_stream) = UnixStream::pair().unwrap();
diff --git a/litebox_runner_linux_userland/src/broker.rs b/litebox_runner_linux_userland/src/broker.rs
index d6bbc471d..c17b75d2c 100644
--- a/litebox_runner_linux_userland/src/broker.rs
+++ b/litebox_runner_linux_userland/src/broker.rs
@@ -10,7 +10,8 @@ use anyhow::{Context as _, Result};
 use litebox_broker_local::{BrokerLocal, BrokerNotifications};
 use litebox_broker_protocol::message::BrokerNotification;
 use litebox_broker_transport::unix_socket::{
-    UnixStreamLocalControlChannel, UnixStreamLocalNotificationChannel,
+    UnixStreamLocalControlCancellation, UnixStreamLocalControlChannel,
+    UnixStreamLocalNotificationChannel,
 };
 
 const SETUP_TIMEOUT: Duration = Duration::from_secs(5);
@@ -22,6 +23,7 @@ pub(crate) fn connect(
 ) -> Result<(
     BrokerLocal<UnixStreamLocalControlChannel>,
     BrokerNotifications<UnixStreamLocalNotificationChannel>,
+    UnixStreamLocalControlCancellation,
 )> {
     let setup_deadline = Instant::now() + SETUP_TIMEOUT;
     let control_channel = connect_with_retry(
@@ -48,26 +50,40 @@ pub(crate) fn connect(
             notification_socket_path.display()
         )
     })?;
+    let control_cancellation = control_channel
+        .cancellation_handle()
+        .context("failed to create broker control cancellation handle")?;
     let local = BrokerLocal::negotiate(control_channel).context("broker negotiation failed")?;
-    Ok((local, BrokerNotifications::new(notification_channel)))
+    Ok((
+        local,
+        BrokerNotifications::new(notification_channel),
+        control_cancellation,
+    ))
 }
 
 pub(crate) fn start_notification_receiver(
     mut notifications: BrokerNotifications<UnixStreamLocalNotificationChannel>,
+    control_cancellation: UnixStreamLocalControlCancellation,
     dispatch_notification: impl Fn(BrokerNotification) + Send + 'static,
+    dispatch_failure: impl Fn() + Send + 'static,
 ) -> Result<()> {
     std::thread::Builder::new()
         .name("litebox-broker-notifications".to_owned())
         .spawn(move || {
-            loop {
+            let receive_error = loop {
                 match notifications.recv_notification() {
                     Ok(Some(notification)) => dispatch_notification(notification),
-                    Ok(None) => break,
-                    Err(error) => {
-                        eprintln!("failed to receive broker notification: {error}");
-                        break;
-                    }
+                    Ok(None) => break None,
+                    Err(error) => break Some(error),
                 }
+            };
+            let cancellation_error = control_cancellation.cancel().err();
+            dispatch_failure();
+            if let Some(error) = receive_error {
+                eprintln!("failed to receive broker notification: {error}");
+            }
+            if let Some(error) = cancellation_error {
+                eprintln!("failed to cancel broker control channel: {error}");
             }
         })
         .context("failed to start broker notification receiver")?;
diff --git a/litebox_runner_linux_userland/src/lib.rs b/litebox_runner_linux_userland/src/lib.rs
index 9b5495b81..0fc0eee0a 100644
--- a/litebox_runner_linux_userland/src/lib.rs
+++ b/litebox_runner_linux_userland/src/lib.rs
@@ -241,14 +241,16 @@ pub fn run(cli_args: CliArgs) -> Result<()> {
     };
 
     let shim_builder = if let Some(broker_connection) = broker_connection {
-        let (broker_local, broker_notifications) = broker_connection;
+        let (broker_local, broker_notifications, broker_control_cancellation) = broker_connection;
         let litebox = litebox::LiteBox::new_with_broker_local(
             litebox_platform_multiplex::platform(),
             broker_local,
         );
         broker::start_notification_receiver(
             broker_notifications,
+            broker_control_cancellation,
             litebox.broker_notification_dispatcher(),
+            litebox.broker_failure_dispatcher(),
         )?;
         litebox_shim_linux::LinuxShimBuilder::new_with_litebox(litebox)
     } else {