diff --git a/rust/bt-ascs/src/debug.rs b/rust/bt-ascs/src/debug.rs
index 0ce995f..ae0312a 100644
--- a/rust/bt-ascs/src/debug.rs
+++ b/rust/bt-ascs/src/debug.rs
@@ -149,6 +149,7 @@
 {
     type Set = AscsCmd;
     type Error = crate::types::Error;
+    type Event = ();
 
     fn run_command(
         &self,
diff --git a/rust/bt-broadcast-assistant/src/debug.rs b/rust/bt-broadcast-assistant/src/debug.rs
index a737f08..6afa9eb 100644
--- a/rust/bt-broadcast-assistant/src/debug.rs
+++ b/rust/bt-broadcast-assistant/src/debug.rs
@@ -209,6 +209,7 @@
 {
     type Set = AssistantCmd;
     type Error = crate::assistant::Error;
+    type Event = ();
 
     fn run_command(
         &self,
diff --git a/rust/bt-common/src/debug_command.rs b/rust/bt-common/src/debug_command.rs
index dd4ea63..5d55a9f 100644
--- a/rust/bt-common/src/debug_command.rs
+++ b/rust/bt-common/src/debug_command.rs
@@ -4,9 +4,15 @@
 
 //! Debug command traits and helpers for defining commands for integration
 //! into a debug tool.
+use futures::channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender};
+use futures::task::AtomicWaker;
+use futures::{FutureExt, Stream, StreamExt};
 use log::LevelFilter;
+use std::collections::HashMap;
+use std::pin::Pin;
 use std::str::FromStr;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
+use std::task::{Context, Poll};
 
 #[must_use = "if unused the previous log level will immediately be restored"]
 pub struct ScopedVerbosityGuard {
@@ -265,10 +271,75 @@
     }
 }
 
+/// An event emitted by a [`CommandRunner`] through its
+/// [`CommandRunner::event_stream`].
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum RunnerEvent<E = ()> {
+    /// An arbitrary human-readable message from the runner or its background
+    /// tasks to be displayed by the harness.
+    Message(String),
+    /// A structured domain event specific to this runner that the harness or a
+    /// parent composite tool can react to programmatically.
+    Event(E),
+}
+
+impl<E> RunnerEvent<E> {
+    pub fn message(&self) -> Option<&str> {
+        match self {
+            Self::Message(s) => Some(s),
+            Self::Event(_) => None,
+        }
+    }
+
+    pub fn event(&self) -> Option<&E> {
+        match self {
+            Self::Message(_) => None,
+            Self::Event(e) => Some(e),
+        }
+    }
+
+    pub fn into_event(self) -> Option<E> {
+        match self {
+            Self::Message(_) => None,
+            Self::Event(e) => Some(e),
+        }
+    }
+
+    pub fn into_message(self) -> Option<String> {
+        match self {
+            Self::Message(s) => Some(s),
+            Self::Event(_) => None,
+        }
+    }
+
+    pub fn map_event<F, T>(self, f: F) -> RunnerEvent<T>
+    where
+        F: FnOnce(E) -> T,
+    {
+        match self {
+            Self::Message(m) => RunnerEvent::Message(m),
+            Self::Event(e) => RunnerEvent::Event(f(e)),
+        }
+    }
+}
+
+impl<E> From<String> for RunnerEvent<E> {
+    fn from(s: String) -> Self {
+        Self::Message(s)
+    }
+}
+
+impl<E> From<&str> for RunnerEvent<E> {
+    fn from(s: &str) -> Self {
+        Self::Message(s.to_string())
+    }
+}
+
 /// CommandRunner is used to perform a specific task based on the command set.
 pub trait CommandRunner {
     type Set: CommandSet;
     type Error: ::std::error::Error;
+    type Event;
 
     fn run_command(
         &self,
@@ -276,6 +347,15 @@
         args: Vec<String>,
     ) -> impl futures::Future<Output = Result<(), Self::Error>>;
 
+    /// Returns a stream that drives background processing for this runner and
+    /// yields display messages or structured events for the harness.
+    ///
+    /// Crates that do not need background tasks or event delivery can set
+    /// `type Event = ();` and rely on this default implementation.
+    fn event_stream(&self) -> impl Stream<Item = RunnerEvent<Self::Event>> + '_ {
+        futures::stream::pending::<RunnerEvent<Self::Event>>()
+    }
+
     fn run(
         &self,
         cmd: impl Into<CliCommand<Self::Set>>,
@@ -301,6 +381,7 @@
 impl<R: CommandRunner> CommandRunner for &R {
     type Set = R::Set;
     type Error = R::Error;
+    type Event = R::Event;
 
     fn run_command(
         &self,
@@ -309,11 +390,16 @@
     ) -> impl futures::Future<Output = Result<(), Self::Error>> {
         (**self).run_command(cmd, args)
     }
+
+    fn event_stream(&self) -> impl Stream<Item = RunnerEvent<Self::Event>> + '_ {
+        (**self).event_stream()
+    }
 }
 
 impl<R: CommandRunner> CommandRunner for Arc<R> {
     type Set = R::Set;
     type Error = R::Error;
+    type Event = R::Event;
 
     fn run_command(
         &self,
@@ -322,6 +408,309 @@
     ) -> impl futures::Future<Output = Result<(), Self::Error>> {
         (**self).run_command(cmd, args)
     }
+
+    fn event_stream(&self) -> impl Stream<Item = RunnerEvent<Self::Event>> + '_ {
+        (**self).event_stream()
+    }
+}
+
+type BoxedStream<E> = Pin<Box<dyn Stream<Item = RunnerEvent<E>> + Send + 'static>>;
+
+struct StreamsState<E> {
+    keyed: HashMap<String, BoxedStream<E>>,
+    anonymous: Vec<BoxedStream<E>>,
+}
+
+impl<E> Default for StreamsState<E> {
+    fn default() -> Self {
+        Self { keyed: HashMap::new(), anonymous: Vec::new() }
+    }
+}
+
+impl<E: Send + 'static> Stream for StreamsState<E> {
+    type Item = RunnerEvent<E>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let mut ready = None;
+        self.keyed.retain(|_key, stream| {
+            if ready.is_some() {
+                return true;
+            }
+            match stream.as_mut().poll_next(cx) {
+                Poll::Ready(Some(item)) => {
+                    ready = Some(item);
+                    true
+                }
+                Poll::Ready(None) => false,
+                Poll::Pending => true,
+            }
+        });
+        if let Some(item) = ready {
+            return Poll::Ready(Some(item));
+        }
+
+        self.anonymous.retain_mut(|stream| {
+            if ready.is_some() {
+                return true;
+            }
+            match stream.as_mut().poll_next(cx) {
+                Poll::Ready(Some(item)) => {
+                    ready = Some(item);
+                    true
+                }
+                Poll::Ready(None) => false,
+                Poll::Pending => true,
+            }
+        });
+        if let Some(item) = ready {
+            return Poll::Ready(Some(item));
+        }
+
+        Poll::Pending
+    }
+}
+
+struct RunnerBackgroundInner<E: Send + 'static> {
+    msg_sender: UnboundedSender<RunnerEvent<E>>,
+    msg_receiver: Mutex<UnboundedReceiver<RunnerEvent<E>>>,
+    streams: Mutex<StreamsState<E>>,
+    waker: AtomicWaker,
+}
+
+/// Helper for managing background streams, display messages, and structured
+/// events for a [`CommandRunner`].
+#[derive(Clone)]
+pub struct RunnerBackground<E: Send + 'static = ()> {
+    inner: Arc<RunnerBackgroundInner<E>>,
+}
+
+impl<E: Send + 'static> RunnerBackground<E> {
+    pub fn new() -> Self {
+        let (msg_sender, msg_receiver) = unbounded();
+        Self {
+            inner: Arc::new(RunnerBackgroundInner {
+                msg_sender,
+                msg_receiver: Mutex::new(msg_receiver),
+                streams: Mutex::new(StreamsState::default()),
+                waker: AtomicWaker::new(),
+            }),
+        }
+    }
+
+    /// Queues a displayable text message to be emitted by
+    /// [`Self::event_stream`].
+    pub fn println(&self, msg: impl Into<String>) {
+        let _ = self.inner.msg_sender.unbounded_send(RunnerEvent::Message(msg.into()));
+        self.inner.waker.wake();
+    }
+
+    /// Queues a structured domain event to be emitted by
+    /// [`Self::event_stream`].
+    pub fn send(&self, event: E) {
+        let _ = self.inner.msg_sender.unbounded_send(RunnerEvent::Event(event));
+        self.inner.waker.wake();
+    }
+
+    /// Returns an [`UnboundedSender`] that converts sent items into
+    /// [`RunnerEvent::Event`].
+    pub fn event_sender(&self) -> UnboundedSender<E> {
+        let (tx, rx) = unbounded::<E>();
+        self.spawn_stream(rx.map(RunnerEvent::Event));
+        tx
+    }
+
+    /// Dynamically registers or replaces a background stream under a unique
+    /// key. If a stream was already registered under `key`, the previous
+    /// stream is cancelled.
+    pub fn set_stream<K: Into<String>>(
+        &self,
+        key: K,
+        stream: impl Stream<Item = RunnerEvent<E>> + Send + 'static,
+    ) {
+        let mut streams = self.inner.streams.lock().unwrap();
+        streams.keyed.insert(key.into(), Box::pin(stream));
+        drop(streams);
+        self.inner.waker.wake();
+    }
+
+    /// Cancels the stream registered under `key`, if present. Returns `true` if
+    /// a stream was cancelled.
+    pub fn cancel_stream(&self, key: &str) -> bool {
+        let mut streams = self.inner.streams.lock().unwrap();
+        let removed = streams.keyed.remove(key).is_some();
+        drop(streams);
+        if removed {
+            self.inner.waker.wake();
+        }
+        removed
+    }
+
+    /// Spawns an anonymous background stream that runs until completion.
+    pub fn spawn_stream(&self, stream: impl Stream<Item = RunnerEvent<E>> + Send + 'static) {
+        let mut streams = self.inner.streams.lock().unwrap();
+        streams.anonymous.push(Box::pin(stream));
+        drop(streams);
+        self.inner.waker.wake();
+    }
+
+    /// Returns `true` if a background stream is currently registered under
+    /// `key`.
+    pub fn has_stream(&self, key: &str) -> bool {
+        self.inner.streams.lock().unwrap().keyed.contains_key(key)
+    }
+
+    /// Returns the total number of active background streams (keyed and
+    /// anonymous).
+    pub fn active_stream_count(&self) -> usize {
+        let streams = self.inner.streams.lock().unwrap();
+        streams.keyed.len() + streams.anonymous.len()
+    }
+
+    /// Returns a stream yielding messages and events produced by this runner.
+    pub fn event_stream(&self) -> RunnerBackgroundStream<E> {
+        RunnerBackgroundStream { inner: self.inner.clone() }
+    }
+}
+
+impl<E: Send + 'static> RunnerBackgroundInner<E> {
+    fn poll_next_event(&self, cx: &mut Context<'_>) -> Poll<Option<RunnerEvent<E>>> {
+        self.waker.register(cx.waker());
+
+        // 1. Drain directly queued messages/events from msg_receiver first.
+        if let Poll::Ready(Some(item)) = self.msg_receiver.lock().unwrap().poll_next_unpin(cx) {
+            return Poll::Ready(Some(item));
+        }
+
+        // 2. Poll registered background streams.
+        if let Poll::Ready(Some(item)) = self.streams.lock().unwrap().poll_next_unpin(cx) {
+            return Poll::Ready(Some(item));
+        }
+
+        Poll::Pending
+    }
+}
+
+impl<E: Send + 'static> Default for RunnerBackground<E> {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+/// Stream returned by [`RunnerBackground::event_stream`].
+pub struct RunnerBackgroundStream<E: Send + 'static = ()> {
+    inner: Arc<RunnerBackgroundInner<E>>,
+}
+
+impl<E: Send + 'static> Stream for RunnerBackgroundStream<E> {
+    type Item = RunnerEvent<E>;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        self.inner.poll_next_event(cx)
+    }
+}
+
+/// A harness for driving a [`CommandRunner`] alongside its background event
+/// stream.
+pub struct DebugHarness<R: CommandRunner> {
+    runner: R,
+}
+
+impl<R: CommandRunner> DebugHarness<R> {
+    pub fn new(runner: R) -> Self {
+        Self { runner }
+    }
+
+    pub fn runner(&self) -> &R {
+        &self.runner
+    }
+
+    pub fn runner_mut(&mut self) -> &mut R {
+        &mut self.runner
+    }
+
+    pub fn into_runner(self) -> R {
+        self.runner
+    }
+
+    pub fn event_stream(&self) -> impl Stream<Item = RunnerEvent<R::Event>> + '_ {
+        self.runner.event_stream()
+    }
+
+    pub async fn run(
+        &self,
+        cmd: impl Into<CliCommand<R::Set>>,
+        args: Vec<String>,
+    ) -> Result<(), R::Error> {
+        self.runner.run(cmd, args).await
+    }
+
+    async fn drive_events<Fut, F>(&self, future: Fut, mut on_event: F) -> Fut::Output
+    where
+        Fut: futures::Future,
+        F: FnMut(RunnerEvent<R::Event>),
+    {
+        let stream = self.runner.event_stream().fuse();
+        futures::pin_mut!(stream);
+        let future = future.fuse();
+        futures::pin_mut!(future);
+
+        loop {
+            futures::select! {
+                event = stream.select_next_some() => on_event(event),
+                output = future => {
+                    while let Some(Some(event)) = stream.next().now_or_never() {
+                        on_event(event);
+                    }
+                    return output;
+                }
+            }
+        }
+    }
+
+    /// Concurrently drives the runner's background `event_stream` while
+    /// executing `future`. Messages are passed to `on_message` and events
+    /// are passed to `on_event`.
+    pub async fn drive_with<Fut, FM, FE>(
+        &self,
+        future: Fut,
+        mut on_message: FM,
+        mut on_event: FE,
+    ) -> Fut::Output
+    where
+        Fut: futures::Future,
+        FM: FnMut(String),
+        FE: FnMut(R::Event),
+    {
+        self.drive_events(future, |event| match event {
+            RunnerEvent::Message(msg) => on_message(msg),
+            RunnerEvent::Event(evt) => on_event(evt),
+        })
+        .await
+    }
+
+    /// Concurrently executes `future` while collecting all `RunnerEvent`s
+    /// emitted during its execution.
+    pub async fn run_collecting_events<Fut>(
+        &self,
+        future: Fut,
+    ) -> (Fut::Output, Vec<RunnerEvent<R::Event>>)
+    where
+        Fut: futures::Future,
+    {
+        let mut events = Vec::new();
+        let output = self.drive_events(future, |event| events.push(event)).await;
+        (output, events)
+    }
+
+    /// Concurrently executes a command while collecting all `RunnerEvent`s
+    /// emitted during its execution.
+    pub async fn run_command_collecting(
+        &self,
+        cmd: impl Into<CliCommand<R::Set>>,
+        args: Vec<String>,
+    ) -> (Result<(), R::Error>, Vec<RunnerEvent<R::Event>>) {
+        self.run_collecting_events(self.run(cmd, args)).await
+    }
 }
 
 #[cfg(test)]
@@ -494,6 +883,7 @@
         impl CommandRunner for TestRunner {
             type Set = RunnerCmd;
             type Error = std::io::Error;
+            type Event = ();
 
             fn run_command(
                 &self,
@@ -530,4 +920,350 @@
         // Reset persistent verbosity back to Info
         log::set_max_level(LevelFilter::Info);
     }
+
+    #[test]
+    fn test_command_runner_default_event_stream() {
+        gen_commandset! {
+            DefaultCmd {
+                Action = ("action", [], [], "Default action"),
+            }
+        }
+
+        struct SimpleRunner;
+        impl CommandRunner for SimpleRunner {
+            type Set = DefaultCmd;
+            type Error = std::io::Error;
+            type Event = ();
+
+            fn run_command(
+                &self,
+                _cmd: Self::Set,
+                _args: Vec<String>,
+            ) -> impl futures::Future<Output = Result<(), Self::Error>> {
+                futures::future::ready(Ok(()))
+            }
+        }
+
+        let runner = SimpleRunner;
+        let stream = runner.event_stream();
+        futures::pin_mut!(stream);
+        let mut cx = std::task::Context::from_waker(futures::task::noop_waker_ref());
+        assert!(stream.as_mut().poll_next(&mut cx).is_pending());
+    }
+
+    #[test]
+    fn test_runner_event() {
+        let msg: RunnerEvent<u32> = RunnerEvent::Message("hello".to_string());
+        assert_eq!(msg.message(), Some("hello"));
+        assert_eq!(msg.event(), None);
+        assert_eq!(msg.clone().into_event(), None);
+        assert_eq!(msg.clone().into_message(), Some("hello".to_string()));
+
+        let evt: RunnerEvent<u32> = RunnerEvent::Event(42);
+        assert_eq!(evt.message(), None);
+        assert_eq!(evt.event(), Some(&42));
+        assert_eq!(evt.clone().into_event(), Some(42));
+        assert_eq!(evt.into_message(), None);
+
+        let str_msg: RunnerEvent<()> = "test msg".into();
+        assert_eq!(str_msg, RunnerEvent::Message("test msg".to_string()));
+
+        let string_msg: RunnerEvent<()> = String::from("string msg").into();
+        assert_eq!(string_msg, RunnerEvent::Message("string msg".to_string()));
+
+        let mapped = RunnerEvent::Event(10).map_event(|x| x * 2);
+        assert_eq!(mapped, RunnerEvent::Event(20));
+
+        let mapped_msg: RunnerEvent<i32> =
+            RunnerEvent::Message("hi".to_string()).map_event(|x: u32| x as i32);
+        assert_eq!(mapped_msg, RunnerEvent::Message("hi".to_string()));
+    }
+
+    #[test]
+    fn test_runner_background_println_and_send() {
+        futures::executor::block_on(async {
+            let bg: RunnerBackground<u32> = RunnerBackground::new();
+            bg.println("log message 1");
+            bg.send(100);
+            bg.println("log message 2");
+
+            let stream = bg.event_stream();
+            futures::pin_mut!(stream);
+
+            assert_eq!(
+                stream.next().await,
+                Some(RunnerEvent::Message("log message 1".to_string()))
+            );
+            assert_eq!(stream.next().await, Some(RunnerEvent::Event(100)));
+            assert_eq!(
+                stream.next().await,
+                Some(RunnerEvent::Message("log message 2".to_string()))
+            );
+
+            // When no more events, polling should return Pending.
+            let mut cx = std::task::Context::from_waker(futures::task::noop_waker_ref());
+            assert!(stream.as_mut().poll_next(&mut cx).is_pending());
+        });
+    }
+
+    #[test]
+    fn test_runner_background_event_sender() {
+        futures::executor::block_on(async {
+            let bg: RunnerBackground<String> = RunnerBackground::new();
+            let sender = bg.event_sender();
+
+            sender.unbounded_send("hello".to_string()).unwrap();
+            sender.unbounded_send("world".to_string()).unwrap();
+
+            let stream = bg.event_stream();
+            futures::pin_mut!(stream);
+
+            assert_eq!(stream.next().await, Some(RunnerEvent::Event("hello".to_string())));
+            assert_eq!(stream.next().await, Some(RunnerEvent::Event("world".to_string())));
+
+            drop(sender);
+            // Polling after dropping sender should clean up the stream and
+            // return Pending.
+            let mut cx = std::task::Context::from_waker(futures::task::noop_waker_ref());
+            assert!(stream.as_mut().poll_next(&mut cx).is_pending());
+            assert_eq!(bg.active_stream_count(), 0);
+        });
+    }
+
+    #[test]
+    fn test_runner_background_keyed_streams_and_cancellation() {
+        futures::executor::block_on(async {
+            let bg: RunnerBackground<u32> = RunnerBackground::new();
+            assert_eq!(bg.has_stream("discovery"), false);
+
+            let (tx1, rx1) = futures::channel::mpsc::unbounded::<RunnerEvent<u32>>();
+            bg.set_stream("discovery", rx1);
+            assert_eq!(bg.has_stream("discovery"), true);
+            assert_eq!(bg.active_stream_count(), 1);
+
+            tx1.unbounded_send(RunnerEvent::Event(1)).unwrap();
+            let stream = bg.event_stream();
+            futures::pin_mut!(stream);
+            assert_eq!(stream.next().await, Some(RunnerEvent::Event(1)));
+
+            // Replacing the stream cancels the old one.
+            let (tx2, rx2) = futures::channel::mpsc::unbounded::<RunnerEvent<u32>>();
+            bg.set_stream("discovery", rx2);
+            assert_eq!(bg.has_stream("discovery"), true);
+            // Old tx1 should have receiver dropped / closed.
+            assert!(tx1.is_closed());
+
+            tx2.unbounded_send(RunnerEvent::Event(2)).unwrap();
+            assert_eq!(stream.next().await, Some(RunnerEvent::Event(2)));
+
+            // Cancelling the stream.
+            assert!(bg.cancel_stream("discovery"));
+            assert_eq!(bg.has_stream("discovery"), false);
+            assert!(tx2.is_closed());
+            assert!(!bg.cancel_stream("discovery"));
+            assert_eq!(bg.active_stream_count(), 0);
+        });
+    }
+
+    #[test]
+    fn test_runner_background_spawn_stream_and_cleanup() {
+        futures::executor::block_on(async {
+            let bg: RunnerBackground<u32> = RunnerBackground::new();
+            let stream_data = futures::stream::iter(vec![
+                RunnerEvent::Message("from stream".to_string()),
+                RunnerEvent::Event(99),
+            ]);
+            bg.spawn_stream(stream_data);
+            assert_eq!(bg.active_stream_count(), 1);
+
+            let stream = bg.event_stream();
+            futures::pin_mut!(stream);
+
+            assert_eq!(stream.next().await, Some(RunnerEvent::Message("from stream".to_string())));
+            assert_eq!(stream.next().await, Some(RunnerEvent::Event(99)));
+
+            // Polling again will observe the stream returned None and clean it
+            // up.
+            let mut cx = std::task::Context::from_waker(futures::task::noop_waker_ref());
+            assert!(stream.as_mut().poll_next(&mut cx).is_pending());
+            assert_eq!(bg.active_stream_count(), 0);
+        });
+    }
+
+    #[test]
+    fn test_runner_background_waker_waking() {
+        let bg: RunnerBackground<u32> = RunnerBackground::new();
+        let bg_clone = bg.clone();
+        let (polled_tx, polled_rx) = futures::channel::oneshot::channel();
+
+        let handle = std::thread::spawn(move || {
+            futures::executor::block_on(polled_rx).unwrap();
+            bg_clone.println("woken up");
+            bg_clone.send(42);
+        });
+
+        futures::executor::block_on(async {
+            let stream = bg.event_stream();
+            futures::pin_mut!(stream);
+            let mut polled_tx = Some(polled_tx);
+            let first = futures::future::poll_fn(|cx| {
+                let res = stream.as_mut().poll_next(cx);
+                if let Some(tx) = polled_tx.take() {
+                    assert!(res.is_pending());
+                    tx.send(()).unwrap();
+                }
+                res
+            })
+            .await;
+            assert_eq!(first, Some(RunnerEvent::Message("woken up".to_string())));
+            assert_eq!(stream.next().await, Some(RunnerEvent::Event(42)));
+        });
+
+        handle.join().unwrap();
+    }
+
+    #[test]
+    fn test_debug_harness() {
+        gen_commandset! {
+            HarnessCmd {
+                Greet = ("greet", [], ["name"], "Greet a user"),
+                Trigger = ("trigger", [], [], "Trigger background event"),
+            }
+        }
+
+        struct HarnessRunner {
+            bg: RunnerBackground<String>,
+        }
+
+        impl CommandRunner for HarnessRunner {
+            type Set = HarnessCmd;
+            type Error = std::io::Error;
+            type Event = String;
+
+            fn run_command(
+                &self,
+                cmd: Self::Set,
+                args: Vec<String>,
+            ) -> impl futures::Future<Output = Result<(), Self::Error>> {
+                match cmd {
+                    HarnessCmd::Greet => {
+                        let name = args.first().cloned().unwrap_or_else(|| "world".to_string());
+                        self.bg.println(format!("Hello, {}!", name));
+                    }
+                    HarnessCmd::Trigger => {
+                        self.bg.send("event_fired".to_string());
+                    }
+                }
+                futures::future::ready(Ok(()))
+            }
+
+            fn event_stream(&self) -> impl Stream<Item = RunnerEvent<Self::Event>> + '_ {
+                self.bg.event_stream()
+            }
+        }
+
+        let runner = HarnessRunner { bg: RunnerBackground::new() };
+        let harness = DebugHarness::new(runner);
+
+        futures::executor::block_on(async {
+            // Test run_command_collecting
+            let (res, events) =
+                harness.run_command_collecting(HarnessCmd::Greet, vec!["Alice".to_string()]).await;
+            assert!(res.is_ok());
+            assert_eq!(events, vec![RunnerEvent::Message("Hello, Alice!".to_string())]);
+
+            // Test drive_with with Trigger
+            let mut messages = Vec::new();
+            let mut domain_events = Vec::new();
+            harness
+                .drive_with(
+                    harness.run(HarnessCmd::Trigger, vec![]),
+                    |msg| messages.push(msg),
+                    |evt| domain_events.push(evt),
+                )
+                .await
+                .unwrap();
+
+            assert!(messages.is_empty());
+            assert_eq!(domain_events, vec!["event_fired".to_string()]);
+        });
+    }
+
+    #[test]
+    fn test_debug_harness_terminating_stream() {
+        gen_commandset! {
+            FiniteCmd {
+                Action = ("action", [], [], "Finite action"),
+            }
+        }
+
+        struct FiniteStreamRunner {
+            stream_done_tx: std::sync::Mutex<Option<futures::channel::oneshot::Sender<()>>>,
+        }
+        impl CommandRunner for FiniteStreamRunner {
+            type Set = FiniteCmd;
+            type Error = std::io::Error;
+            type Event = u32;
+
+            fn run_command(
+                &self,
+                _cmd: Self::Set,
+                _args: Vec<String>,
+            ) -> impl futures::Future<Output = Result<(), Self::Error>> {
+                futures::future::ready(Ok(()))
+            }
+
+            fn event_stream(&self) -> impl Stream<Item = RunnerEvent<Self::Event>> + '_ {
+                let mut done_tx = self.stream_done_tx.lock().unwrap().take();
+                futures::stream::iter(vec![
+                    RunnerEvent::Message("finite msg".to_string()),
+                    RunnerEvent::Event(42),
+                ])
+                .chain(futures::stream::poll_fn(move |_| {
+                    if let Some(tx) = done_tx.take() {
+                        let _ = tx.send(());
+                    }
+                    Poll::Ready(None)
+                }))
+            }
+        }
+
+        let (stream_done_tx, stream_done_rx) = futures::channel::oneshot::channel::<()>();
+        let harness = DebugHarness::new(FiniteStreamRunner {
+            stream_done_tx: std::sync::Mutex::new(Some(stream_done_tx)),
+        });
+        futures::executor::block_on(async {
+            let (tx, rx) = futures::channel::oneshot::channel::<()>();
+            let handle = std::thread::spawn(move || {
+                futures::executor::block_on(stream_done_rx).unwrap();
+                let _ = tx.send(());
+            });
+
+            // The stream terminates after 2 items while rx is still pending.
+            // run_collecting_events should not busy-loop and should cleanly
+            // complete when rx resolves.
+            let (output, events) = harness.run_collecting_events(rx).await;
+            assert!(output.is_ok());
+            assert_eq!(
+                events,
+                vec![RunnerEvent::Message("finite msg".to_string()), RunnerEvent::Event(42),]
+            );
+            handle.join().unwrap();
+        });
+    }
+
+    #[test]
+    fn test_runner_background_stream_static() {
+        let bg: RunnerBackground<u32> = RunnerBackground::new();
+        bg.send(77);
+        let stream = bg.event_stream();
+
+        fn assert_static<T: 'static>(_: &T) {}
+        assert_static(&stream);
+
+        futures::executor::block_on(async {
+            futures::pin_mut!(stream);
+            assert_eq!(stream.next().await, Some(RunnerEvent::Event(77)));
+        });
+    }
 }
diff --git a/rust/bt-pacs/src/debug.rs b/rust/bt-pacs/src/debug.rs
index 342d29b..9b815d7 100644
--- a/rust/bt-pacs/src/debug.rs
+++ b/rust/bt-pacs/src/debug.rs
@@ -32,6 +32,7 @@
 impl<T: bt_gatt::GattTypes> CommandRunner for PacsDebug<T> {
     type Set = PacsCmd;
     type Error = bt_gatt::types::Error;
+    type Event = ();
 
     fn run_command(
         &self,
diff --git a/rust/bt-vcs/src/debug.rs b/rust/bt-vcs/src/debug.rs
index fd3cdc9..780f045 100644
--- a/rust/bt-vcs/src/debug.rs
+++ b/rust/bt-vcs/src/debug.rs
@@ -59,6 +59,7 @@
 impl<T: bt_gatt::GattTypes> CommandRunner for VcsDebug<T> {
     type Set = VcsCmd;
     type Error = Error;
+    type Event = ();
 
     fn run_command(
         &self,