diff --git a/litebox_broker_host/src/lib.rs b/litebox_broker_host/src/lib.rs
index 20fd3cb3d0..905745bb5c 100644
--- a/litebox_broker_host/src/lib.rs
+++ b/litebox_broker_host/src/lib.rs
@@ -14,7 +14,9 @@ extern crate std;
 
 use litebox_broker_core::{BrokerCore, BrokerSession, CallerCredential};
 use litebox_broker_protocol::BROKER_PROTOCOL_VERSION;
-use litebox_broker_protocol::channel::{HostControlChannel, HostReceive, PeerCredential};
+use litebox_broker_protocol::channel::{
+    HostControlChannel, HostNotificationChannel, HostReceive, PeerCredential,
+};
 use litebox_broker_protocol::error::ErrorCode;
 use litebox_broker_protocol::event::{AddEventResponse, CreateEventResponse, WaitEventResponse};
 use litebox_broker_protocol::message::{
@@ -25,15 +27,24 @@ mod error;
 
 pub use error::{BrokerHostError, Result};
 
-/// Authenticates, negotiates, and serves one broker connection over the control channel.
-pub fn serve_connection<Channel>(
+/// Authenticates, negotiates, and serves one broker association over paired
+/// control and notification channels.
+///
+/// The deployment must bind both channels to the same authenticated peer
+/// association. Active requests and responses remain on the control channel;
+/// broker-initiated readiness wakeups are sent on the notification channel.
+/// Event mutations caused by control requests return readiness in their control
+/// response and do not also emit a duplicate notification.
+pub fn serve_connection<ControlChannel, NotificationChannel, ChannelError>(
     core: &BrokerCore,
-    channel: &mut Channel,
-) -> Result<ConnectionTermination, Channel::Error>
+    control_channel: &mut ControlChannel,
+    _notification_channel: &mut NotificationChannel,
+) -> Result<ConnectionTermination, ChannelError>
 where
-    Channel: HostControlChannel,
+    ControlChannel: HostControlChannel<Error = ChannelError>,
+    NotificationChannel: HostNotificationChannel<Error = ChannelError>,
 {
-    let peer_credential = channel
+    let peer_credential = control_channel
         .peer_credential()
         .map_err(BrokerHostError::Channel)?;
     let caller_credential = match peer_credential {
@@ -43,13 +54,13 @@ where
     let session = core.create_session(caller_credential)?;
 
     loop {
-        let request = match channel
+        let request = match control_channel
             .recv_handshake_request()
             .map_err(BrokerHostError::Channel)?
         {
             HostReceive::Message(request) => request,
             HostReceive::ProtocolViolation => {
-                channel
+                control_channel
                     .send_handshake_response(&BrokerHandshakeResponse::Error(
                         ErrorCode::ProtocolState,
                     ))
@@ -69,7 +80,7 @@ where
                 broker_protocol_version: BROKER_PROTOCOL_VERSION,
             }
         };
-        channel
+        control_channel
             .send_handshake_response(&response)
             .map_err(BrokerHostError::Channel)?;
         if negotiated {
@@ -77,21 +88,14 @@ where
         }
     }
 
-    serve_request_loop(channel, &session)
-}
-
-fn serve_request_loop<Channel>(
-    channel: &mut Channel,
-    session: &BrokerSession,
-) -> Result<ConnectionTermination, Channel::Error>
-where
-    Channel: HostControlChannel,
-{
     loop {
-        let request = match channel.recv_request().map_err(BrokerHostError::Channel)? {
+        let request = match control_channel
+            .recv_request()
+            .map_err(BrokerHostError::Channel)?
+        {
             HostReceive::Message(request) => request,
             HostReceive::ProtocolViolation => {
-                channel
+                control_channel
                     .send_response(&BrokerResponse::Error(ErrorCode::ProtocolState))
                     .map_err(BrokerHostError::Channel)?;
                 return Ok(ConnectionTermination::ProtocolViolation);
@@ -99,8 +103,8 @@ where
             HostReceive::PeerClosed => break,
         };
 
-        let response = handle_request(session, request);
-        channel
+        let response = handle_request(&session, request);
+        control_channel
             .send_response(&response)
             .map_err(BrokerHostError::Channel)?;
     }
@@ -167,8 +171,11 @@ pub enum ConnectionTermination {
 mod tests {
     use super::*;
     use litebox_broker_core::{PolicyEngine, PrincipalRights};
-    use litebox_broker_protocol::event::{CreateEventRequest, WaitEventRequest};
-    use litebox_broker_protocol::message::BrokerHandshakeRequest;
+    use litebox_broker_protocol::event::{
+        AddEventRequest, ConsumeEventRequest, CreateEventRequest, EventConsumeMode,
+        WaitEventRequest,
+    };
+    use litebox_broker_protocol::message::{BrokerHandshakeRequest, BrokerNotification};
     use litebox_broker_protocol::{ObjectHandle, ProtocolVersion};
 
     #[test]
@@ -183,6 +190,7 @@ mod tests {
         serve_connection_rejects_active_request_before_negotiation(&broker);
         serve_connection_rejects_handshake_request_after_negotiation(&broker);
         serve_connection_returns_channel_error_when_response_send_fails(&broker);
+        serve_connection_returns_event_readiness_in_control_responses(&broker);
         active_request_closes_object_reference(&broker);
     }
 
@@ -198,9 +206,10 @@ mod tests {
                 Ok(HostReceive::PeerClosed),
             ]),
         );
+        let mut notifications = FakeHostNotificationChannel::default();
 
         assert_eq!(
-            serve_connection(broker, &mut channel).unwrap(),
+            serve_connection(broker, &mut channel, &mut notifications).unwrap(),
             ConnectionTermination::PeerClosed
         );
         assert_eq!(
@@ -228,9 +237,10 @@ mod tests {
             ]),
             std::vec::Vec::from([Ok(HostReceive::PeerClosed)]),
         );
+        let mut notifications = FakeHostNotificationChannel::default();
 
         assert_eq!(
-            serve_connection(broker, &mut channel).unwrap(),
+            serve_connection(broker, &mut channel, &mut notifications).unwrap(),
             ConnectionTermination::PeerClosed
         );
         assert_eq!(
@@ -251,9 +261,10 @@ mod tests {
             std::vec::Vec::from([Ok(HostReceive::ProtocolViolation)]),
             std::vec::Vec::new(),
         );
+        let mut notifications = FakeHostNotificationChannel::default();
 
         assert_eq!(
-            serve_connection(broker, &mut channel).unwrap(),
+            serve_connection(broker, &mut channel, &mut notifications).unwrap(),
             ConnectionTermination::ProtocolViolation
         );
         assert_eq!(
@@ -270,9 +281,10 @@ mod tests {
             }))]),
             std::vec::Vec::from([Ok(HostReceive::ProtocolViolation)]),
         );
+        let mut notifications = FakeHostNotificationChannel::default();
 
         assert_eq!(
-            serve_connection(broker, &mut channel).unwrap(),
+            serve_connection(broker, &mut channel, &mut notifications).unwrap(),
             ConnectionTermination::ProtocolViolation
         );
         assert_eq!(
@@ -295,14 +307,54 @@ mod tests {
             std::vec::Vec::new(),
         );
         channel.send_error = true;
+        let mut notifications = FakeHostNotificationChannel::default();
 
-        match serve_connection(broker, &mut channel) {
+        match serve_connection(broker, &mut channel, &mut notifications) {
             Err(BrokerHostError::Channel(())) => {}
             result => panic!("unexpected serve result: {result:?}"),
         }
         assert!(channel.handshake_responses.is_empty());
     }
 
+    fn serve_connection_returns_event_readiness_in_control_responses(broker: &BrokerCore) {
+        let mut channel = FakeHostControlChannel::new(
+            std::vec::Vec::from([Ok(HostReceive::Message(BrokerHandshakeRequest {
+                protocol_version: BROKER_PROTOCOL_VERSION,
+            }))]),
+            std::vec::Vec::from([Ok(HostReceive::Message(BrokerRequest::Event(
+                EventRequest::Create(CreateEventRequest { initial_count: 0 }),
+            )))]),
+        );
+        channel.enqueue_readiness_requests_after_create = true;
+        let mut notifications = FakeHostNotificationChannel::default();
+
+        assert_eq!(
+            serve_connection(broker, &mut channel, &mut notifications).unwrap(),
+            ConnectionTermination::PeerClosed
+        );
+        assert!(notifications.notifications.is_empty());
+        assert_eq!(
+            &channel.responses[1..],
+            [
+                BrokerResponse::Event(EventResponse::Add(AddEventResponse {
+                    readiness: litebox_broker_protocol::event::ReadinessState {
+                        read_ready: true,
+                        write_ready: true,
+                    },
+                })),
+                BrokerResponse::Event(EventResponse::Consume(
+                    litebox_broker_protocol::event::ConsumeEventResponse {
+                        value: 1,
+                        readiness: litebox_broker_protocol::event::ReadinessState {
+                            read_ready: false,
+                            write_ready: true,
+                        },
+                    }
+                )),
+            ]
+        );
+    }
+
     fn active_request_closes_object_reference(broker: &BrokerCore) {
         let session = broker
             .create_session(CallerCredential::Unauthenticated)
@@ -344,6 +396,7 @@ mod tests {
         requests: std::vec::Vec<core::result::Result<HostReceive<BrokerRequest>, ()>>,
         handshake_responses: std::vec::Vec<BrokerHandshakeResponse>,
         responses: std::vec::Vec<BrokerResponse>,
+        enqueue_readiness_requests_after_create: bool,
         send_error: bool,
     }
 
@@ -359,6 +412,7 @@ mod tests {
                 requests,
                 handshake_responses: std::vec::Vec::new(),
                 responses: std::vec::Vec::new(),
+                enqueue_readiness_requests_after_create: false,
                 send_error: false,
             }
         }
@@ -409,8 +463,44 @@ mod tests {
             if self.send_error {
                 return Err(());
             }
+            if self.enqueue_readiness_requests_after_create
+                && let BrokerResponse::Event(EventResponse::Create(response)) = response
+            {
+                self.requests
+                    .push(Ok(HostReceive::Message(BrokerRequest::Event(
+                        EventRequest::Add(AddEventRequest {
+                            handle: response.handle,
+                            value: 1,
+                        }),
+                    ))));
+                self.requests
+                    .push(Ok(HostReceive::Message(BrokerRequest::Event(
+                        EventRequest::Consume(ConsumeEventRequest {
+                            handle: response.handle,
+                            mode: EventConsumeMode::One,
+                        }),
+                    ))));
+                self.requests.push(Ok(HostReceive::PeerClosed));
+            }
             self.responses.push(response.clone());
             Ok(())
         }
     }
+
+    #[derive(Default)]
+    struct FakeHostNotificationChannel {
+        notifications: std::vec::Vec<BrokerNotification>,
+    }
+
+    impl HostNotificationChannel for FakeHostNotificationChannel {
+        type Error = ();
+
+        fn send_notification(
+            &mut self,
+            notification: &BrokerNotification,
+        ) -> core::result::Result<(), Self::Error> {
+            self.notifications.push(notification.clone());
+            Ok(())
+        }
+    }
 }
diff --git a/litebox_broker_local/src/lib.rs b/litebox_broker_local/src/lib.rs
index f90e481af1..8c2ceecd5c 100644
--- a/litebox_broker_local/src/lib.rs
+++ b/litebox_broker_local/src/lib.rs
@@ -1,11 +1,13 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT license.
 
-//! Typed broker-local control adapter for broker requests.
+//! Typed broker-local adapters for broker requests and notifications.
 //!
 //! The local control adapter owns request/response sequencing but does not own a channel.
 //! Userland, kernel, or ring-buffer deployments can provide channels by
 //! implementing [`litebox_broker_protocol::channel::LocalControlChannel`].
+//! Notification receive adapters are intentionally separate so active control
+//! requests remain strictly paired with their responses.
 
 #![no_std]
 
@@ -15,10 +17,11 @@ extern crate std;
 mod error;
 mod event;
 
-use litebox_broker_protocol::channel::LocalControlChannel;
+use litebox_broker_protocol::channel::{LocalControlChannel, LocalNotificationChannel};
 use litebox_broker_protocol::error::ErrorCode;
 use litebox_broker_protocol::message::{
-    BrokerHandshakeRequest, BrokerHandshakeResponse, BrokerRequest, BrokerResponse,
+    BrokerHandshakeRequest, BrokerHandshakeResponse, BrokerNotification, BrokerRequest,
+    BrokerResponse,
 };
 use litebox_broker_protocol::{BROKER_PROTOCOL_VERSION, ObjectHandle};
 
@@ -29,6 +32,11 @@ pub struct BrokerLocal<Channel: LocalControlChannel> {
     channel: Channel,
 }
 
+/// Broker-local receive adapter for broker-initiated asynchronous notifications.
+pub struct BrokerNotifications<Channel: LocalNotificationChannel> {
+    channel: Channel,
+}
+
 impl<Channel: LocalControlChannel> BrokerLocal<Channel> {
     /// Negotiates the broker protocol over an already-connected control channel.
     ///
@@ -124,14 +132,33 @@ impl<Channel: LocalControlChannel> BrokerLocal<Channel> {
     }
 }
 
+impl<Channel: LocalNotificationChannel> BrokerNotifications<Channel> {
+    /// Creates a notification receiver from an already-associated notification channel.
+    pub const fn new(channel: Channel) -> Self {
+        Self { channel }
+    }
+
+    /// Receives the next broker notification.
+    ///
+    /// Returns `Ok(None)` when the broker closed the notification channel cleanly.
+    pub fn recv_notification(&mut self) -> Result<Option<BrokerNotification>, Channel::Error> {
+        self.channel
+            .recv_notification()
+            .map_err(BrokerLocalError::Channel)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
     use core::convert::Infallible;
     use litebox_broker_protocol::ObjectHandle;
     use litebox_broker_protocol::ProtocolVersion;
+    use litebox_broker_protocol::channel::LocalNotificationChannel;
     use litebox_broker_protocol::event::{CreateEventRequest, CreateEventResponse};
-    use litebox_broker_protocol::message::{EventRequest, EventResponse};
+    use litebox_broker_protocol::message::{
+        EventReadinessNotification, EventRequest, EventResponse,
+    };
 
     #[test]
     fn negotiate_returns_active_local_connection() {
@@ -219,6 +246,23 @@ mod tests {
         let _ = BrokerLocal::negotiate(channel);
     }
 
+    #[test]
+    fn notification_receiver_returns_broker_notifications() {
+        let notification = BrokerNotification::EventReadiness(EventReadinessNotification {
+            handle: ObjectHandle(7),
+            readiness: litebox_broker_protocol::event::ReadinessState {
+                read_ready: true,
+                write_ready: false,
+            },
+        });
+        let mut receiver = BrokerNotifications::new(FakeNotificationChannel {
+            notification: Some(notification.clone()),
+        });
+
+        assert_eq!(receiver.recv_notification().unwrap(), Some(notification));
+        assert_eq!(receiver.recv_notification().unwrap(), None);
+    }
+
     #[test]
     fn negotiate_rejects_broker_unsupported_version_response() {
         let broker_protocol_version = ProtocolVersion(BROKER_PROTOCOL_VERSION.0 + 1);
@@ -296,4 +340,18 @@ mod tests {
             Ok(self.response.take())
         }
     }
+
+    struct FakeNotificationChannel {
+        notification: Option<BrokerNotification>,
+    }
+
+    impl LocalNotificationChannel for FakeNotificationChannel {
+        type Error = Infallible;
+
+        fn recv_notification(
+            &mut self,
+        ) -> core::result::Result<Option<BrokerNotification>, Self::Error> {
+            Ok(self.notification.take())
+        }
+    }
 }
diff --git a/litebox_broker_userland/src/main.rs b/litebox_broker_userland/src/main.rs
index b6c2ba559b..5363b308bf 100644
--- a/litebox_broker_userland/src/main.rs
+++ b/litebox_broker_userland/src/main.rs
@@ -10,7 +10,9 @@ use std::process::Command;
 use clap::Parser;
 use litebox_broker_core::{BrokerCore, PolicyEngine, PrincipalRights};
 use litebox_broker_host::serve_connection;
-use litebox_broker_transport::unix_socket::UnixStreamHostControlChannel;
+use litebox_broker_transport::unix_socket::{
+    UnixStreamHostControlChannel, UnixStreamHostNotificationChannel,
+};
 
 #[derive(Parser, Debug)]
 struct CliArgs {
@@ -27,8 +29,10 @@ fn main() -> Result<(), Box<dyn Error>> {
     let socket_dir = tempfile::Builder::new()
         .prefix("litebox-broker-userland-")
         .tempdir()?;
-    let socket_path = socket_dir.path().join("broker.sock");
-    let listener = UnixListener::bind(&socket_path)?;
+    let control_socket_path = socket_dir.path().join("broker.sock");
+    let notification_socket_path = socket_dir.path().join("broker-notification.sock");
+    let control_listener = UnixListener::bind(&control_socket_path)?;
+    let notification_listener = UnixListener::bind(&notification_socket_path)?;
     let broker = BrokerCore::new(PolicyEngine::with_unauthenticated_rights(
         PrincipalRights::all(),
     ))?;
@@ -36,8 +40,10 @@ fn main() -> Result<(), Box<dyn Error>> {
     let mut runner_command = Command::new(&args.runner);
     runner_command
         .arg("--unstable")
-        .arg("--broker-socket")
-        .arg(&socket_path)
+        .arg("--broker-control-socket")
+        .arg(&control_socket_path)
+        .arg("--broker-notification-socket")
+        .arg(&notification_socket_path)
         .args(&args.runner_arguments);
     let mut runner = runner_command.spawn()?;
     let _runner_waiter = std::thread::spawn(move || {
@@ -47,13 +53,19 @@ fn main() -> Result<(), Box<dyn Error>> {
     });
 
     loop {
-        let (stream, _) = listener.accept()?;
+        let (control_stream, _) = control_listener.accept()?;
+        let (notification_stream, _) = notification_listener.accept()?;
         let broker = broker.clone();
         if let Err(error) = std::thread::Builder::new()
             .name("litebox-broker-connection".to_owned())
             .spawn(move || {
-                let mut channel = UnixStreamHostControlChannel::from_accepted(stream);
-                if let Err(error) = serve_connection(&broker, &mut channel) {
+                let mut control_channel =
+                    UnixStreamHostControlChannel::from_accepted(control_stream);
+                let mut notification_channel =
+                    UnixStreamHostNotificationChannel::from_accepted(notification_stream);
+                if let Err(error) =
+                    serve_connection(&broker, &mut control_channel, &mut notification_channel)
+                {
                     eprintln!("failed to serve broker connection: {error}");
                 }
             })
diff --git a/litebox_broker_userland/tests/notification_runtime.rs b/litebox_broker_userland/tests/notification_runtime.rs
new file mode 100644
index 0000000000..9fd6fb3aa3
--- /dev/null
+++ b/litebox_broker_userland/tests/notification_runtime.rs
@@ -0,0 +1,45 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+use std::os::unix::net::UnixStream;
+
+use litebox_broker_core::{BrokerCore, PolicyEngine, PrincipalRights};
+use litebox_broker_host::{ConnectionTermination, serve_connection};
+use litebox_broker_local::BrokerLocal;
+use litebox_broker_protocol::event::ReadinessState;
+use litebox_broker_transport::unix_socket::{
+    UnixStreamHostControlChannel, UnixStreamHostNotificationChannel, UnixStreamLocalControlChannel,
+};
+
+#[test]
+fn host_serves_control_requests_over_paired_userland_channels() {
+    let broker = BrokerCore::new(PolicyEngine::with_unauthenticated_rights(
+        PrincipalRights::all(),
+    ))
+    .unwrap();
+    let (local_control, host_control) = UnixStream::pair().unwrap();
+    let (_local_notification, host_notification) = UnixStream::pair().unwrap();
+
+    let host_thread = std::thread::spawn(move || {
+        let mut control = UnixStreamHostControlChannel::from_accepted(host_control);
+        let mut notification = UnixStreamHostNotificationChannel::from_accepted(host_notification);
+        serve_connection(&broker, &mut control, &mut notification)
+    });
+
+    let mut local =
+        BrokerLocal::negotiate(UnixStreamLocalControlChannel::from_connected(local_control))
+            .unwrap();
+
+    let handle = local.create_event_with_count(0).unwrap();
+    let readiness = ReadinessState {
+        read_ready: true,
+        write_ready: true,
+    };
+    assert_eq!(local.add_event(handle, 1).unwrap(), readiness);
+
+    drop(local);
+    assert_eq!(
+        host_thread.join().unwrap().unwrap(),
+        ConnectionTermination::PeerClosed
+    );
+}
diff --git a/litebox_broker_userland/tests/userland_broker.rs b/litebox_broker_userland/tests/userland_broker.rs
index 6bc055aed4..ee126be51b 100644
--- a/litebox_broker_userland/tests/userland_broker.rs
+++ b/litebox_broker_userland/tests/userland_broker.rs
@@ -10,7 +10,9 @@ use std::time::{Duration, Instant};
 
 use litebox_broker_local::BrokerLocal;
 use litebox_broker_protocol::event::ReadinessState;
-use litebox_broker_transport::unix_socket::UnixStreamLocalControlChannel;
+use litebox_broker_transport::unix_socket::{
+    UnixStreamLocalControlChannel, UnixStreamLocalNotificationChannel,
+};
 
 const RUNNER_ARGUMENT: &str = "broker-userland-test-runner";
 
@@ -30,10 +32,11 @@ fn run_parent_test() {
     // This custom-harness integration test uses its own executable as the broker's
     // runner. Cargo starts this executable without broker args, so it runs the
     // parent path here. The broker then starts the same executable with the real
-    // runner argv (`--unstable --broker-socket <path>`), which runs `run_fake_runner`.
-    // After the fake runner finishes its broker requests, it terminates the broker
-    // parent process; this lets the test exercise the long-running broker without a
-    // test-only shutdown path.
+    // runner argv (`--unstable --broker-control-socket <path>
+    // --broker-notification-socket <path>`), which runs `run_fake_runner`. After
+    // the fake runner finishes its broker requests, it terminates the broker
+    // parent process; this lets the test exercise the long-running broker
+    // without a test-only shutdown path.
     let mut broker = ChildGuard {
         child: Command::new(env!("CARGO_BIN_EXE_litebox-broker-userland"))
             .arg("--runner")
@@ -61,17 +64,24 @@ fn run_fake_runner(args: &[OsString]) {
     );
     assert_eq!(
         args.get(1).map(OsString::as_os_str),
-        Some(OsStr::new("--broker-socket"))
+        Some(OsStr::new("--broker-control-socket"))
     );
     assert_eq!(
         args.get(3).map(OsString::as_os_str),
+        Some(OsStr::new("--broker-notification-socket"))
+    );
+    assert_eq!(
+        args.get(5).map(OsString::as_os_str),
         Some(OsStr::new(RUNNER_ARGUMENT))
     );
-    assert_eq!(args.len(), 4, "unexpected runner arguments: {args:?}");
+    assert_eq!(args.len(), 6, "unexpected runner arguments: {args:?}");
 
-    let socket_path = args.get(2).unwrap();
-    let channel = connect_with_retry(Path::new(socket_path)).unwrap();
-    let mut local = BrokerLocal::negotiate(channel).unwrap();
+    let control_socket_path = args.get(2).unwrap();
+    let notification_socket_path = args.get(4).unwrap();
+    let control_channel = connect_control_with_retry(Path::new(control_socket_path)).unwrap();
+    let _notification_channel =
+        connect_notification_with_retry(Path::new(notification_socket_path)).unwrap();
+    let mut local = BrokerLocal::negotiate(control_channel).unwrap();
 
     let handle = local.create_event_with_count(0).unwrap();
     assert_eq!(
@@ -82,13 +92,11 @@ fn run_fake_runner(args: &[OsString]) {
         }
     );
 
-    assert_eq!(
-        local.add_event(handle, 1).unwrap(),
-        ReadinessState {
-            read_ready: true,
-            write_ready: true,
-        }
-    );
+    let readiness = ReadinessState {
+        read_ready: true,
+        write_ready: true,
+    };
+    assert_eq!(local.add_event(handle, 1).unwrap(), readiness);
 
     assert_eq!(
         local.wait_event(handle).unwrap(),
@@ -124,7 +132,7 @@ impl Drop for ChildGuard {
     }
 }
 
-fn connect_with_retry(socket_path: &Path) -> Result<UnixStreamLocalControlChannel> {
+fn connect_control_with_retry(socket_path: &Path) -> Result<UnixStreamLocalControlChannel> {
     let deadline = Instant::now() + Duration::from_secs(5);
     loop {
         match UnixStreamLocalControlChannel::connect_with_setup_deadline(socket_path, deadline) {
@@ -141,3 +149,23 @@ fn connect_with_retry(socket_path: &Path) -> Result<UnixStreamLocalControlChanne
         }
     }
 }
+
+fn connect_notification_with_retry(
+    socket_path: &Path,
+) -> Result<UnixStreamLocalNotificationChannel> {
+    let deadline = Instant::now() + Duration::from_secs(5);
+    loop {
+        match UnixStreamLocalNotificationChannel::connect(socket_path) {
+            Ok(channel) => return Ok(channel),
+            Err(error) if Instant::now() < deadline => {
+                if error.kind() != ErrorKind::NotFound
+                    && error.kind() != ErrorKind::ConnectionRefused
+                {
+                    return Err(error);
+                }
+                std::thread::sleep(Duration::from_millis(10));
+            }
+            Err(error) => return Err(error),
+        }
+    }
+}
diff --git a/litebox_runner_linux_userland/src/broker.rs b/litebox_runner_linux_userland/src/broker.rs
index 05737cae24..7d36e96be9 100644
--- a/litebox_runner_linux_userland/src/broker.rs
+++ b/litebox_runner_linux_userland/src/broker.rs
@@ -3,12 +3,15 @@
 
 use std::{
     path::Path,
+    thread::JoinHandle,
     time::{Duration, Instant},
 };
 
 use anyhow::{Context as _, Result};
-use litebox_broker_local::BrokerLocal;
-use litebox_broker_transport::unix_socket::UnixStreamLocalControlChannel;
+use litebox_broker_local::{BrokerLocal, BrokerNotifications};
+use litebox_broker_transport::unix_socket::{
+    UnixStreamLocalControlChannel, UnixStreamLocalNotificationChannel,
+};
 
 const SETUP_TIMEOUT: Duration = Duration::from_secs(5);
 const RETRY_DELAY: Duration = Duration::from_millis(20);
@@ -16,13 +19,63 @@ type Local = BrokerLocal<UnixStreamLocalControlChannel>;
 
 pub(crate) struct BrokerConnection {
     local: Local,
+    #[expect(
+        dead_code,
+        reason = "keeps the notification receiver thread alive while the broker connection is installed"
+    )]
+    notification_receiver_thread: JoinHandle<()>,
 }
 
-pub(crate) fn connect(socket_path: Option<&Path>) -> Result<Option<BrokerConnection>> {
-    match socket_path {
-        Some(path) => connect_to_endpoint(path).map(Some),
-        None => Ok(None),
-    }
+pub(crate) fn connect(
+    control_socket_path: &Path,
+    notification_socket_path: &Path,
+) -> Result<BrokerConnection> {
+    let setup_deadline = Instant::now() + SETUP_TIMEOUT;
+    let control_channel = connect_with_retry(
+        control_socket_path,
+        setup_deadline,
+        "timed out connecting to broker",
+        |path, deadline| UnixStreamLocalControlChannel::connect_with_setup_deadline(path, deadline),
+    )
+    .with_context(|| {
+        format!(
+            "failed to connect to broker at {}",
+            control_socket_path.display()
+        )
+    })?;
+    let notification_channel = connect_with_retry(
+        notification_socket_path,
+        setup_deadline,
+        "timed out connecting to broker notifications",
+        |path, _deadline| UnixStreamLocalNotificationChannel::connect(path),
+    )
+    .with_context(|| {
+        format!(
+            "failed to connect to broker notifications at {}",
+            notification_socket_path.display()
+        )
+    })?;
+    let local = BrokerLocal::negotiate(control_channel).context("broker negotiation failed")?;
+    let mut notifications = BrokerNotifications::new(notification_channel);
+    let notification_thread = std::thread::Builder::new()
+        .name("litebox-broker-notifications".to_owned())
+        .spawn(move || {
+            loop {
+                match notifications.recv_notification() {
+                    Ok(Some(_notification)) => {}
+                    Ok(None) => break,
+                    Err(error) => {
+                        eprintln!("failed to receive broker notification: {error}");
+                        break;
+                    }
+                }
+            }
+        })
+        .context("failed to start broker notification receiver")?;
+    Ok(BrokerConnection {
+        local,
+        notification_receiver_thread: notification_thread,
+    })
 }
 
 impl BrokerConnection {
@@ -31,26 +84,18 @@ impl BrokerConnection {
     }
 }
 
-fn connect_to_endpoint(socket_path: &Path) -> Result<BrokerConnection> {
-    let setup_deadline = Instant::now() + SETUP_TIMEOUT;
-    let local = connect_with_retry(socket_path, setup_deadline)
-        .with_context(|| format!("failed to connect to broker at {}", socket_path.display()))?;
-    Ok(BrokerConnection { local })
-}
-
-fn connect_with_retry(socket_path: &Path, setup_deadline: Instant) -> Result<Local> {
+fn connect_with_retry<Channel>(
+    socket_path: &Path,
+    setup_deadline: Instant,
+    timeout_message: &'static str,
+    mut connect: impl FnMut(&Path, Instant) -> std::io::Result<Channel>,
+) -> Result<Channel> {
     loop {
-        match UnixStreamLocalControlChannel::connect_with_setup_deadline(
-            socket_path,
-            setup_deadline,
-        ) {
-            Ok(channel) => {
-                let local = BrokerLocal::negotiate(channel).context("broker negotiation failed")?;
-                return Ok(local);
-            }
+        match connect(socket_path, setup_deadline) {
+            Ok(channel) => return Ok(channel),
             Err(error) => {
                 if Instant::now() >= setup_deadline {
-                    return Err(error).context("timed out connecting to broker");
+                    return Err(error).context(timeout_message);
                 }
             }
         }
diff --git a/litebox_runner_linux_userland/src/lib.rs b/litebox_runner_linux_userland/src/lib.rs
index efe4406325..76c4e7b5e8 100644
--- a/litebox_runner_linux_userland/src/lib.rs
+++ b/litebox_runner_linux_userland/src/lib.rs
@@ -81,14 +81,24 @@ pub struct CliArgs {
     pub program_from_tar: bool,
     /// Broker-supplied Unix socket path for the local control channel.
     #[arg(
-        long = "broker-socket",
+        long = "broker-control-socket",
         value_name = "PATH",
         value_hint = clap::ValueHint::FilePath,
         hide = true,
-        requires = "unstable",
+        requires_all = ["unstable", "broker_notification_socket"],
+        help_heading = "Unstable Options"
+    )]
+    pub broker_control_socket: Option<PathBuf>,
+    /// Broker-supplied Unix socket path for the local notification channel.
+    #[arg(
+        long = "broker-notification-socket",
+        value_name = "PATH",
+        value_hint = clap::ValueHint::FilePath,
+        hide = true,
+        requires_all = ["unstable", "broker_control_socket"],
         help_heading = "Unstable Options"
     )]
-    pub broker_socket: Option<PathBuf>,
+    pub broker_notification_socket: Option<PathBuf>,
 }
 
 struct MmappedFile {
@@ -213,7 +223,22 @@ pub fn run(cli_args: CliArgs) -> Result<()> {
     }
 
     litebox_platform_multiplex::set_platform(platform);
-    let broker_connection = broker::connect(cli_args.broker_socket.as_deref())?;
+    let broker_connection = match (
+        cli_args.broker_control_socket.as_deref(),
+        cli_args.broker_notification_socket.as_deref(),
+    ) {
+        (Some(control_socket_path), Some(notification_socket_path)) => Some(broker::connect(
+            control_socket_path,
+            notification_socket_path,
+        )?),
+        (None, None) => None,
+        (Some(_), None) => {
+            anyhow::bail!("broker notification socket is required with broker control socket")
+        }
+        (None, Some(_)) => {
+            anyhow::bail!("broker control socket is required with broker notification socket")
+        }
+    };
 
     let shim_builder = if let Some(broker_connection) = broker_connection {
         litebox_shim_linux::LinuxShimBuilder::new_with_litebox(
diff --git a/litebox_runner_linux_userland/tests/run.rs b/litebox_runner_linux_userland/tests/run.rs
index a5684a90a3..310636b955 100644
--- a/litebox_runner_linux_userland/tests/run.rs
+++ b/litebox_runner_linux_userland/tests/run.rs
@@ -123,8 +123,17 @@ impl Runner {
     }
 
     #[cfg(all(target_arch = "x86_64", target_os = "linux"))]
-    fn broker_socket(&mut self, socket_path: &Path) -> &mut Self {
-        self.command.arg("--broker-socket").arg(socket_path);
+    fn broker_sockets(
+        &mut self,
+        control_socket_path: &Path,
+        notification_socket_path: &Path,
+    ) -> &mut Self {
+        self.command
+            .arg("--broker-control-socket")
+            .arg(control_socket_path);
+        self.command
+            .arg("--broker-notification-socket")
+            .arg(notification_socket_path);
         self
     }
 
@@ -264,7 +273,8 @@ struct TestBroker {
     thread: Option<std::thread::JoinHandle<()>>,
     done_rx: std::sync::mpsc::Receiver<()>,
     close_object_count_rx: std::sync::mpsc::Receiver<usize>,
-    socket_path: PathBuf,
+    control_socket_path: PathBuf,
+    notification_socket_path: PathBuf,
 }
 
 #[cfg(all(target_arch = "x86_64", target_os = "linux"))]
@@ -284,54 +294,79 @@ impl TestBroker {
             .expect("broker test host thread missing")
             .join()
             .expect("broker test host panicked");
-        let _ = std::fs::remove_file(&self.socket_path);
+        let _ = std::fs::remove_file(&self.control_socket_path);
+        let _ = std::fs::remove_file(&self.notification_socket_path);
     }
 }
 
 #[cfg(all(target_arch = "x86_64", target_os = "linux"))]
 impl Drop for TestBroker {
     fn drop(&mut self) {
-        let _ = std::fs::remove_file(&self.socket_path);
+        let _ = std::fs::remove_file(&self.control_socket_path);
+        let _ = std::fs::remove_file(&self.notification_socket_path);
     }
 }
 
 #[cfg(all(target_arch = "x86_64", target_os = "linux"))]
 fn spawn_test_broker(
-    socket_path: &Path,
+    control_socket_path: &Path,
+    notification_socket_path: &Path,
     policy: litebox_broker_core::PolicyEngine,
     connection_count: usize,
 ) -> TestBroker {
-    let _ = std::fs::remove_file(socket_path);
+    let _ = std::fs::remove_file(control_socket_path);
+    let _ = std::fs::remove_file(notification_socket_path);
 
     let (ready_tx, ready_rx) = std::sync::mpsc::channel();
     let (done_tx, done_rx) = std::sync::mpsc::channel();
     let (close_object_count_tx, close_object_count_rx) = std::sync::mpsc::channel();
-    let server_socket_path = socket_path.to_path_buf();
-    let cleanup_socket_path = socket_path.to_path_buf();
+    let server_control_socket_path = control_socket_path.to_path_buf();
+    let server_notification_socket_path = notification_socket_path.to_path_buf();
+    let cleanup_control_socket_path = control_socket_path.to_path_buf();
+    let cleanup_notification_socket_path = notification_socket_path.to_path_buf();
     let broker_thread = std::thread::spawn(move || {
         let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
-            let listener = std::os::unix::net::UnixListener::bind(&server_socket_path)
-                .expect("failed to bind broker test socket");
+            let control_listener =
+                std::os::unix::net::UnixListener::bind(&server_control_socket_path)
+                    .expect("failed to bind broker test control socket");
+            let notification_listener =
+                std::os::unix::net::UnixListener::bind(&server_notification_socket_path)
+                    .expect("failed to bind broker test notification socket");
             let broker =
                 litebox_broker_core::BrokerCore::new(policy).expect("failed to create broker core");
             ready_tx.send(()).expect("failed to report broker ready");
 
             for _ in 0..connection_count {
-                let (stream, _) = listener
+                let (control_stream, _) = control_listener
                     .accept()
                     .expect("failed to accept broker local control connection");
-                stream
+                let (notification_stream, _) = notification_listener
+                    .accept()
+                    .expect("failed to accept broker local notification connection");
+                control_stream
                     .set_read_timeout(Some(BROKER_HELPER_TIMEOUT))
                     .expect("failed to configure broker test read timeout");
-                stream
+                control_stream
                     .set_write_timeout(Some(BROKER_HELPER_TIMEOUT))
                     .expect("failed to configure broker test write timeout");
+                notification_stream
+                    .set_read_timeout(Some(BROKER_HELPER_TIMEOUT))
+                    .expect("failed to configure broker notification test read timeout");
+                notification_stream
+                    .set_write_timeout(Some(BROKER_HELPER_TIMEOUT))
+                    .expect("failed to configure broker notification test write timeout");
                 let mut channel = CountingHostControlChannel {
-                    inner: litebox_broker_transport::unix_socket::UnixStreamHostControlChannel::from_accepted(stream),
+                    inner: litebox_broker_transport::unix_socket::UnixStreamHostControlChannel::from_accepted(control_stream),
                     close_object_count: 0,
                 };
-                let termination = litebox_broker_host::serve_connection(&broker, &mut channel)
-                    .expect("broker host failed");
+                let mut notification_channel =
+                    litebox_broker_transport::unix_socket::UnixStreamHostNotificationChannel::from_accepted(notification_stream);
+                let termination = litebox_broker_host::serve_connection(
+                    &broker,
+                    &mut channel,
+                    &mut notification_channel,
+                )
+                .expect("broker host failed");
                 assert_eq!(
                     termination,
                     litebox_broker_host::ConnectionTermination::PeerClosed
@@ -341,7 +376,8 @@ fn spawn_test_broker(
                     .expect("failed to report broker close-object count");
             }
         }));
-        let _ = std::fs::remove_file(&server_socket_path);
+        let _ = std::fs::remove_file(&server_control_socket_path);
+        let _ = std::fs::remove_file(&server_notification_socket_path);
         let _ = done_tx.send(());
         if let Err(panic) = result {
             std::panic::resume_unwind(panic);
@@ -355,7 +391,8 @@ fn spawn_test_broker(
         thread: Some(broker_thread),
         done_rx,
         close_object_count_rx,
-        socket_path: cleanup_socket_path,
+        control_socket_path: cleanup_control_socket_path,
+        notification_socket_path: cleanup_notification_socket_path,
     }
 }
 
@@ -428,9 +465,11 @@ impl<Channel: litebox_broker_protocol::channel::HostControlChannel>
 fn test_runner_broker_integration_with_rewriter() {
     let true_path = run_which("true");
     let target = common::compile("./tests/eventfd.c", "broker_eventfd_rewriter", false, false);
-    let socket_path = unique_test_socket_path("runner-broker");
+    let control_socket_path = unique_test_socket_path("runner-broker-control");
+    let notification_socket_path = unique_test_socket_path("runner-broker-notification");
     let broker_thread = spawn_test_broker(
-        &socket_path,
+        &control_socket_path,
+        &notification_socket_path,
         litebox_broker_core::PolicyEngine::with_unauthenticated_rights(
             litebox_broker_core::PrincipalRights::all(),
         ),
@@ -438,12 +477,12 @@ fn test_runner_broker_integration_with_rewriter() {
     );
 
     Runner::new(&true_path, "broker_true_rewriter")
-        .broker_socket(&socket_path)
+        .broker_sockets(&control_socket_path, &notification_socket_path)
         .run();
     assert_eq!(broker_thread.next_close_object_count(), 0);
 
     Runner::new(&target, "broker_eventfd_rewriter")
-        .broker_socket(&socket_path)
+        .broker_sockets(&control_socket_path, &notification_socket_path)
         .run();
     // eventfd.c creates eight eventfd objects; each should release one broker object.
     assert_eq!(broker_thread.next_close_object_count(), 8);