diff --git a/crates/buttplug_core/src/util/task/mod.rs b/crates/buttplug_core/src/util/task/mod.rs
index 0391a9f0e..847a917cc 100644
--- a/crates/buttplug_core/src/util/task/mod.rs
+++ b/crates/buttplug_core/src/util/task/mod.rs
@@ -17,7 +17,15 @@
 mod registry;
 mod scope;
 
-pub use registry::{TaskEvent, TaskId, TaskInfo, TaskOutcome, TaskRegistry, registry};
+pub use registry::{
+  TaskEvent,
+  TaskId,
+  TaskInfo,
+  TaskOutcome,
+  TaskRegistry,
+  TaskSpawnError,
+  registry,
+};
 pub use scope::TaskScope;
 
 use crate::util::async_manager;
diff --git a/crates/buttplug_core/src/util/task/registry.rs b/crates/buttplug_core/src/util/task/registry.rs
index 97f986e28..361cce292 100644
--- a/crates/buttplug_core/src/util/task/registry.rs
+++ b/crates/buttplug_core/src/util/task/registry.rs
@@ -7,10 +7,20 @@
 
 use dashmap::DashMap;
 use std::sync::{
+  Mutex,
   OnceLock,
   atomic::{AtomicU64, Ordering},
 };
 use tokio::sync::broadcast;
+use tokio_util::sync::CancellationToken;
+
+/// Why a task could not be spawned into its scope.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
+pub enum TaskSpawnError {
+  /// The scope was cancelled before registration could win the race.
+  #[error("task scope is closed")]
+  ScopeClosed,
+}
 
 /// Unique identifier for a registered task. Process-lifetime unique.
 ///
@@ -71,6 +81,8 @@ pub enum TaskEvent {
 #[derive(Debug)]
 pub struct TaskRegistry {
   tasks: DashMap<u64, TaskInfo>,
+  /// Short-held gate ordering registration against scope cancellation.
+  gate: Mutex<()>,
   counter: AtomicU64,
   root_counter: AtomicU64,
   events: broadcast::Sender<TaskEvent>,
@@ -86,6 +98,7 @@ impl TaskRegistry {
   pub(super) fn new() -> Self {
     Self {
       tasks: DashMap::new(),
+      gate: Mutex::new(()),
       counter: AtomicU64::new(1),
       root_counter: AtomicU64::new(1),
       events: broadcast::channel(256).0,
@@ -98,6 +111,47 @@ impl TaskRegistry {
   }
 
   pub(super) fn register(&self, path: String, detached: bool) -> TaskId {
+    let _gate = self
+      .gate
+      .lock()
+      .unwrap_or_else(|poisoned| poisoned.into_inner());
+    self.register_locked(path, detached)
+  }
+
+  pub(super) fn register_scoped(
+    &self,
+    path: String,
+    token: &CancellationToken,
+  ) -> Result<(TaskId, CancellationToken), TaskSpawnError> {
+    let _gate = self
+      .gate
+      .lock()
+      .unwrap_or_else(|poisoned| poisoned.into_inner());
+    if token.is_cancelled() {
+      return Err(TaskSpawnError::ScopeClosed);
+    }
+    let task_token = token.child_token();
+    let id = self.register_locked(path, false);
+    Ok((id, task_token))
+  }
+
+  pub(super) fn cancel(&self, token: &CancellationToken) {
+    let _gate = self
+      .gate
+      .lock()
+      .unwrap_or_else(|poisoned| poisoned.into_inner());
+    token.cancel();
+  }
+
+  #[cfg(test)]
+  pub(crate) fn test_hold_gate(&self) -> std::sync::MutexGuard<'_, ()> {
+    self
+      .gate
+      .lock()
+      .unwrap_or_else(|poisoned| poisoned.into_inner())
+  }
+
+  fn register_locked(&self, path: String, detached: bool) -> TaskId {
     let id = TaskId(self.counter.fetch_add(1, Ordering::Relaxed));
     self.tasks.insert(
       id.0,
diff --git a/crates/buttplug_core/src/util/task/scope.rs b/crates/buttplug_core/src/util/task/scope.rs
index 035a2518b..1027781b4 100644
--- a/crates/buttplug_core/src/util/task/scope.rs
+++ b/crates/buttplug_core/src/util/task/scope.rs
@@ -5,7 +5,7 @@
 // Licensed under the BSD 3-Clause license. See LICENSE file in the project root
 // for full license information.
 
-use super::registry::{TaskId, TaskOutcome, registry};
+use super::registry::{TaskId, TaskOutcome, TaskSpawnError, registry};
 use crate::util::async_manager;
 use std::future::Future;
 use tokio_util::sync::CancellationToken;
@@ -56,40 +56,44 @@ impl TaskScope {
 
   /// Request cancellation of every task in this scope's subtree.
   pub fn cancel(&self) {
-    self.token.cancel();
+    registry().cancel(&self.token);
   }
 
   /// Cancel the subtree and wait until every task under this scope has
   /// deregistered. Wrap in a timeout if the subtree may contain
   /// uncooperative tasks.
   pub async fn shutdown(&self) {
-    self.token.cancel();
+    registry().cancel(&self.token);
     registry().wait_empty_under(&self.path).await;
   }
 
   /// Spawn a task owned by this scope. The closure receives the task's own
   /// child token; long-running tasks MUST select on it.
   #[cfg(not(feature = "wasm"))]
-  pub fn spawn<F, Fut>(&self, name: &str, f: F)
+  pub fn spawn<F, Fut>(&self, name: &str, f: F) -> Result<(), TaskSpawnError>
   where
     F: FnOnce(CancellationToken) -> Fut,
     Fut: Future<Output = ()> + Send + 'static,
   {
-    let (id, task_token, span) = self.register_task(name);
+    let (id, task_token, span) = self.register_task(name)?;
+    let guard = DeregisterGuard::new(id, Some(task_token.clone()));
     let fut = f(task_token.clone());
-    async_manager::spawn(finish_task(fut, id, task_token), span);
+    async_manager::spawn(finish_task(fut, guard), span);
+    Ok(())
   }
 
   /// Spawn a task owned by this scope (WASM, no Send required).
   #[cfg(feature = "wasm")]
-  pub fn spawn<F, Fut>(&self, name: &str, f: F)
+  pub fn spawn<F, Fut>(&self, name: &str, f: F) -> Result<(), TaskSpawnError>
   where
     F: FnOnce(CancellationToken) -> Fut,
     Fut: Future<Output = ()> + 'static,
   {
-    let (id, task_token, span) = self.register_task(name);
+    let (id, task_token, span) = self.register_task(name)?;
+    let guard = DeregisterGuard::new(id, Some(task_token.clone()));
     let fut = f(task_token.clone());
-    async_manager::spawn(finish_task(fut, id, task_token), span);
+    async_manager::spawn(finish_task(fut, guard), span);
+    Ok(())
   }
 
   /// Consume the scope and spawn a task that holds it alive for its own
@@ -97,47 +101,53 @@ impl TaskScope {
   /// protocol subscription handlers): drop-cancel must not fire before the
   /// task runs, but parent cancellation still propagates.
   #[cfg(not(feature = "wasm"))]
-  pub fn spawn_and_hold<F, Fut>(self, name: &str, f: F)
+  pub fn spawn_and_hold<F, Fut>(self, name: &str, f: F) -> Result<(), TaskSpawnError>
   where
     F: FnOnce(CancellationToken) -> Fut,
     Fut: Future<Output = ()> + Send + 'static,
   {
-    let (id, task_token, span) = self.register_task(name);
+    let (id, task_token, span) = self.register_task(name)?;
+    let guard = DeregisterGuard::new(id, Some(task_token.clone()));
     let fut = f(task_token.clone());
     async_manager::spawn(
       async move {
         let _hold = self;
-        finish_task(fut, id, task_token).await;
+        finish_task(fut, guard).await;
       },
       span,
     );
+    Ok(())
   }
 
   /// Consume the scope and spawn a task that holds it alive (WASM, no Send).
   #[cfg(feature = "wasm")]
-  pub fn spawn_and_hold<F, Fut>(self, name: &str, f: F)
+  pub fn spawn_and_hold<F, Fut>(self, name: &str, f: F) -> Result<(), TaskSpawnError>
   where
     F: FnOnce(CancellationToken) -> Fut,
     Fut: Future<Output = ()> + 'static,
   {
-    let (id, task_token, span) = self.register_task(name);
+    let (id, task_token, span) = self.register_task(name)?;
+    let guard = DeregisterGuard::new(id, Some(task_token.clone()));
     let fut = f(task_token.clone());
     async_manager::spawn(
       async move {
         let _hold = self;
-        finish_task(fut, id, task_token).await;
+        finish_task(fut, guard).await;
       },
       span,
     );
+    Ok(())
   }
 
-  fn register_task(&self, name: &str) -> (TaskId, CancellationToken, tracing::Span) {
+  fn register_task(
+    &self,
+    name: &str,
+  ) -> Result<(TaskId, CancellationToken, tracing::Span), TaskSpawnError> {
     let path = format!("{}/{}", self.path, name);
-    let id = registry().register(path.clone(), false);
-    let task_token = self.token.child_token();
+    let (id, task_token) = registry().register_scoped(path.clone(), &self.token)?;
     // Span names must be const in tracing; the dynamic path goes in a field.
     let span = tracing::span!(tracing::Level::INFO, "buttplug_task", task.path = %path);
-    (id, task_token, span)
+    Ok((id, task_token, span))
   }
 }
 
@@ -174,14 +184,13 @@ impl Drop for DeregisterGuard {
   }
 }
 
-async fn finish_task(fut: impl Future<Output = ()>, id: TaskId, token: CancellationToken) {
-  let _guard = DeregisterGuard::new(id, Some(token));
+async fn finish_task(fut: impl Future<Output = ()>, _guard: DeregisterGuard) {
   fut.await;
 }
 
 impl Drop for TaskScope {
   fn drop(&mut self) {
-    self.token.cancel();
+    registry().cancel(&self.token);
   }
 }
 
@@ -189,6 +198,7 @@ impl Drop for TaskScope {
 mod test {
   use super::*;
   use crate::util::task::registry;
+  use crate::util::task::registry::TaskEvent;
   use std::time::Duration;
 
   #[test]
@@ -206,12 +216,132 @@ mod test {
     assert_eq!(child.path(), format!("{}/devices", root.path()));
   }
 
+  #[test]
+  fn test_spawn_rejected_after_cancel_does_not_invoke_closure() {
+    let root = TaskScope::root("rejecttest");
+    let path = root.path().to_owned();
+    root.cancel();
+    let invoked = std::cell::Cell::new(false);
+    let result = root.spawn("worker", |_token| {
+      invoked.set(true);
+      async {}
+    });
+    assert_eq!(result, Err(TaskSpawnError::ScopeClosed));
+    assert!(!invoked.get());
+    assert_eq!(registry().live_count_under(&path), 0);
+  }
+
+  #[test]
+  fn test_rejected_spawn_emits_no_started_event() {
+    let root = TaskScope::root("rejecteventtest");
+    let task_path = format!("{}/worker", root.path());
+    let mut events = registry().event_stream();
+    root.cancel();
+    assert_eq!(
+      root.spawn("worker", |_token| async {}),
+      Err(TaskSpawnError::ScopeClosed)
+    );
+    while let Ok(event) = events.try_recv() {
+      if let TaskEvent::Started { path, .. } = event {
+        assert_ne!(path, task_path, "rejected spawn emitted TaskStarted");
+      }
+    }
+  }
+
+  #[test]
+  fn test_retained_child_rejects_after_parent_cancel() {
+    let root = TaskScope::root("retainedchildtest");
+    let child = root.child("child");
+    root.cancel();
+    assert_eq!(
+      child.spawn("worker", |_token| async {}),
+      Err(TaskSpawnError::ScopeClosed)
+    );
+  }
+
+  #[tokio::test]
+  async fn test_retained_arc_rejects_after_shutdown() {
+    let root = std::sync::Arc::new(TaskScope::root("retainedarctest"));
+    let path = root.path().to_owned();
+    let task_root = root.clone();
+    let (ready, wait) = tokio::sync::oneshot::channel();
+    task_root
+      .spawn("worker", move |token| async move {
+        let _ = ready.send(());
+        token.cancelled().await;
+      })
+      .expect("worker registration should succeed");
+    wait.await.expect("worker did not start");
+    root.shutdown().await;
+    assert_eq!(
+      root.spawn("late", |_token| async {}),
+      Err(TaskSpawnError::ScopeClosed)
+    );
+    assert_eq!(registry().live_count_under(&path), 0);
+  }
+
+  #[tokio::test]
+  async fn test_sync_panic_deregisters_spawn() {
+    let root = TaskScope::root("syncpanictest");
+    let path = root.path().to_owned();
+    let mut events = registry().event_stream();
+    let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+      let _ = root.spawn("panicker", |_token| -> std::future::Ready<()> {
+        panic!("intentional construction panic");
+      });
+    }));
+    assert!(result.is_err());
+    assert_eq!(registry().live_count_under(&path), 0);
+    let task_path = format!("{}/panicker", root.path());
+    let mut saw_started = false;
+    let mut saw_panicked = false;
+    while let Ok(event) = events.try_recv() {
+      match event {
+        TaskEvent::Started { path, .. } if path == task_path => saw_started = true,
+        TaskEvent::Ended { path, outcome, .. } if path == task_path => {
+          saw_panicked = outcome == TaskOutcome::Panicked;
+        }
+        _ => {}
+      }
+    }
+    assert!(saw_started);
+    assert!(saw_panicked);
+  }
+
+  #[test]
+  fn test_sync_panic_deregisters_spawn_and_hold() {
+    let root = TaskScope::root("syncpanicholdtest");
+    let path = root.path().to_owned();
+    let task_path = format!("{}/panicker", root.path());
+    let mut events = registry().event_stream();
+    let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+      let _ = root.spawn_and_hold("panicker", |_token| -> std::future::Ready<()> {
+        panic!("intentional construction panic");
+      });
+    }));
+    assert!(result.is_err());
+    assert_eq!(registry().live_count_under(&path), 0);
+    let mut saw_started = false;
+    let mut saw_panicked = false;
+    while let Ok(event) = events.try_recv() {
+      match event {
+        TaskEvent::Started { path, .. } if path == task_path => saw_started = true,
+        TaskEvent::Ended { path, outcome, .. } if path == task_path => {
+          saw_panicked = outcome == TaskOutcome::Panicked;
+        }
+        _ => {}
+      }
+    }
+    assert!(saw_started);
+    assert!(saw_panicked);
+  }
+
   #[tokio::test]
   async fn test_spawn_registers_and_completes() {
     let root = TaskScope::root("spawntest");
     let path = root.path().to_owned();
     let (tx, rx) = tokio::sync::oneshot::channel();
-    root.spawn("worker", |_token| async move {
+    let _ = root.spawn("worker", |_token| async move {
       let _ = tx.send(());
     });
     rx.await.unwrap();
@@ -225,7 +355,7 @@ mod test {
     let root = TaskScope::root("canceltest");
     let path = root.path().to_owned();
     let child = root.child("inner");
-    child.spawn("worker", |token| async move {
+    let _ = child.spawn("worker", |token| async move {
       token.cancelled().await;
     });
     root.cancel();
@@ -238,7 +368,7 @@ mod test {
   async fn test_drop_cancels() {
     let root = TaskScope::root("droptest");
     let path = root.path().to_owned();
-    root.spawn("worker", |token| async move {
+    let _ = root.spawn("worker", |token| async move {
       token.cancelled().await;
     });
     drop(root);
@@ -250,7 +380,7 @@ mod test {
   #[tokio::test]
   async fn test_shutdown_awaits_subtree() {
     let root = TaskScope::root("shutdowntest");
-    root.spawn("worker", |token| async move {
+    let _ = root.spawn("worker", |token| async move {
       token.cancelled().await;
       // Simulate cleanup work after observing cancellation.
       tokio::time::sleep(Duration::from_millis(20)).await;
@@ -268,7 +398,7 @@ mod test {
     // spawned panic. Without the guard, this wait would time out.
     let root = TaskScope::root("panictest");
     let path = root.path().to_owned();
-    root.spawn("panicker", |_token| async move {
+    let _ = root.spawn("panicker", |_token| async move {
       panic!("intentional panic for deregistration test");
     });
     tokio::time::timeout(Duration::from_secs(1), registry().wait_empty_under(&path))
@@ -276,6 +406,108 @@ mod test {
       .expect("panicking task did not deregister — registry entry leaked");
   }
 
+  #[tokio::test]
+  async fn test_registration_winning_before_cancel_is_awaited_by_shutdown() {
+    let root = std::sync::Arc::new(TaskScope::root("gatewintest"));
+    let path = root.path().to_owned();
+    let gate = registry().test_hold_gate();
+    let spawn_root = root.clone();
+    let handle = tokio::runtime::Handle::current();
+    let spawn_thread = std::thread::spawn(move || {
+      let _enter = handle.enter();
+      spawn_root
+        .spawn("worker", |token| async move {
+          token.cancelled().await;
+        })
+        .expect("registration should succeed after gate release");
+    });
+    std::thread::yield_now();
+    drop(gate);
+    spawn_thread.join().expect("spawn thread panicked");
+    tokio::time::timeout(Duration::from_secs(1), root.shutdown())
+      .await
+      .expect("shutdown did not await the successful registration");
+    assert_eq!(registry().live_count_under(&path), 0);
+  }
+
+  #[tokio::test]
+  async fn test_concurrent_spawn_shutdown_gate_has_only_valid_outcomes() {
+    use registry::TaskEvent;
+
+    let root = std::sync::Arc::new(TaskScope::root("gateracetest"));
+    let path = root.path().to_owned();
+    let task_path = format!("{path}/worker");
+    let mut events = registry().event_stream();
+    let gate = registry().test_hold_gate();
+    let barrier = std::sync::Arc::new(std::sync::Barrier::new(3));
+    let spawn_root = root.clone();
+    let spawn_barrier = barrier.clone();
+    let spawn_task = tokio::task::spawn_blocking(move || {
+      spawn_barrier.wait();
+      spawn_root.spawn("worker", |token| async move {
+        token.cancelled().await;
+      })
+    });
+    let cancel_root = root.clone();
+    let cancel_barrier = barrier.clone();
+    let cancel_task = tokio::task::spawn_blocking(move || {
+      cancel_barrier.wait();
+      cancel_root.cancel();
+    });
+    barrier.wait();
+    drop(gate);
+    let spawn_result = spawn_task.await.expect("spawn task panicked");
+    cancel_task.await.expect("cancel task panicked");
+    tokio::time::timeout(Duration::from_secs(1), root.shutdown())
+      .await
+      .expect("shutdown did not drain the subtree");
+    assert_eq!(registry().live_count_under(&path), 0);
+
+    match spawn_result {
+      Ok(()) => {
+        let mut saw_started = false;
+        let mut saw_ended = false;
+        tokio::time::timeout(Duration::from_secs(1), async {
+          while !(saw_started && saw_ended) {
+            match events.recv().await {
+              Ok(TaskEvent::Started { path, .. }) if path == task_path => saw_started = true,
+              Ok(TaskEvent::Ended { path, .. }) if path == task_path => saw_ended = true,
+              Ok(_) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
+              Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
+            }
+          }
+        })
+        .await
+        .expect("successful spawn lifecycle events did not arrive");
+        assert!(
+          saw_started,
+          "successful spawn emitted no matching TaskStarted"
+        );
+        assert!(saw_ended, "successful spawn emitted no matching TaskEnded");
+      }
+      Err(TaskSpawnError::ScopeClosed) => {
+        let drain = tokio::time::timeout(Duration::from_millis(100), async {
+          loop {
+            match events.recv().await {
+              Ok(TaskEvent::Started { path, .. }) | Ok(TaskEvent::Ended { path, .. })
+                if path == task_path =>
+              {
+                panic!("rejected spawn emitted a matching lifecycle event");
+              }
+              Ok(_) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
+              Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
+            }
+          }
+        })
+        .await;
+        assert!(
+          drain.is_err(),
+          "event drain ended without its bounded timeout"
+        );
+      }
+    }
+  }
+
   #[tokio::test]
   async fn test_spawn_and_hold_keeps_scope_alive() {
     use registry::TaskEvent;
@@ -288,7 +520,7 @@ mod test {
     let sub_scope = root.child("subscription");
     let (tx, rx) = tokio::sync::oneshot::channel();
     // Consuming spawn: the scope moves INTO the task and must not cancel it.
-    sub_scope.spawn_and_hold("worker", |_token| async move {
+    let _ = sub_scope.spawn_and_hold("worker", |_token| async move {
       tokio::time::sleep(Duration::from_millis(20)).await;
       let _ = tx.send(());
     });
diff --git a/crates/buttplug_server/CHANGELOG.md b/crates/buttplug_server/CHANGELOG.md
index f4dd07672..381a0a814 100644
--- a/crates/buttplug_server/CHANGELOG.md
+++ b/crates/buttplug_server/CHANGELOG.md
@@ -1,3 +1,10 @@
+## Unreleased
+
+## Compatibility Notice
+
+- `ProtocolHandler::handle_input_subscribe_cmd` now receives a `TaskScope` argument. This is technically a public Rust API signature change, so custom external implementations must update their method signature.
+- `ProtocolHandler` is currently intended for protocol implementations shipped inside this repository; it is not treated as a stable external extension interface. In line with ADR 0008, servers are not intended to be independently reimplemented. This policy does not make the public trait change non-breaking for existing external implementations.
+
 # 10.0.4 (2026-06-01)
 
 ## Features
diff --git a/crates/buttplug_server/src/device/device_handle.rs b/crates/buttplug_server/src/device/device_handle.rs
index 759dc4824..50a9f86ce 100644
--- a/crates/buttplug_server/src/device/device_handle.rs
+++ b/crates/buttplug_server/src/device/device_handle.rs
@@ -26,7 +26,7 @@ use buttplug_core::{
     OutputValue,
     StopCmdV4,
   },
-  util::{stream::convert_broadcast_receiver_to_stream, task::TaskScope},
+  util::{async_manager, stream::convert_broadcast_receiver_to_stream, task::TaskScope},
 };
 use buttplug_server_device_config::{
   DeviceConfigurationManager,
@@ -36,10 +36,13 @@ use buttplug_server_device_config::{
 };
 use dashmap::DashMap;
 use futures::future::{self, BoxFuture, FutureExt};
-use tokio::sync::{
-  broadcast,
-  mpsc::{Sender, channel},
-  oneshot,
+use tokio::{
+  select,
+  sync::{
+    broadcast,
+    mpsc::{Sender, channel},
+    oneshot,
+  },
 };
 use tokio_stream::StreamExt;
 use uuid::Uuid;
@@ -58,7 +61,7 @@ use crate::{
 use super::{
   InternalDeviceEvent,
   OutputObservation,
-  device_task::{DeviceTaskConfig, spawn_device_task},
+  device_task::{DeviceTaskConfig, DeviceTaskMessage, spawn_device_task},
   hardware::{Hardware, HardwareCommand, HardwareConnector, HardwareEvent},
   protocol::{ProtocolHandler, ProtocolKeepaliveStrategy, ProtocolSpecializer},
 };
@@ -109,7 +112,7 @@ pub struct DeviceHandle {
   legacy_attributes: ServerDeviceAttributes,
   last_output_command: Arc<DashMap<Uuid, CheckedOutputCmdV4>>,
   stop_commands: Arc<Vec<ButtplugDeviceCommandMessageUnionV4>>,
-  internal_hw_msg_sender: Sender<Vec<HardwareCommand>>,
+  internal_hw_msg_sender: Sender<DeviceTaskMessage>,
   output_observation_sender: Option<broadcast::Sender<OutputObservation>>,
   /// Scope owning this device's tasks. Rides in an Arc since DeviceHandle is
   /// Clone; the subtree cancels when the last clone drops.
@@ -127,7 +130,7 @@ impl DeviceHandle {
     definition: ServerDeviceDefinition,
     identifier: UserDeviceIdentifier,
     stop_commands: Vec<ButtplugDeviceCommandMessageUnionV4>,
-    internal_hw_msg_sender: Sender<Vec<HardwareCommand>>,
+    internal_hw_msg_sender: Sender<DeviceTaskMessage>,
     output_observation_sender: Option<broadcast::Sender<OutputObservation>>,
     task_scope: Arc<TaskScope>,
   ) -> Self {
@@ -272,11 +275,32 @@ impl DeviceHandle {
   // --- Private command handling methods ---
 
   fn handle_outputcmd_v4(&self, msg: &CheckedOutputCmdV4) -> ButtplugServerResultFuture {
+    match self.output_cmd_hardware_commands(msg) {
+      // Deduped (no-op) or empty: nothing to send, succeed immediately.
+      None => future::ready(Ok(message::OkV0::default().into())).boxed(),
+      Some(Err(err)) => future::ready(Err(err)).boxed(),
+      Some(Ok(commands)) => self.handle_hardware_commands(commands, false),
+    }
+  }
+
+  /// Run the bookkeeping for an output command (dedupe map, output-observation
+  /// emit) and ask the protocol handler for the hardware commands it produces.
+  ///
+  /// Returns `None` when the command is a deduped no-op (nothing to send),
+  /// `Some(Err(..))` when the handler errored, and `Some(Ok(cmds))` otherwise.
+  /// Shared by `handle_outputcmd_v4` (normal output) and the stop path, so both
+  /// keep identical dedupe-map and observation behaviour. The stop path
+  /// accumulates the returned commands across all stop messages and flushes them
+  /// in a single write-acknowledged batch.
+  fn output_cmd_hardware_commands(
+    &self,
+    msg: &CheckedOutputCmdV4,
+  ) -> Option<Result<Vec<HardwareCommand>, ButtplugError>> {
     if let Some(last_msg) = self.last_output_command.get(&msg.feature_id())
       && *last_msg == *msg
     {
       trace!("No commands generated for incoming device packet, skipping and returning success.");
-      return future::ready(Ok(message::OkV0::default().into())).boxed();
+      return None;
     }
     self
       .last_output_command
@@ -284,7 +308,6 @@ impl DeviceHandle {
 
     if let Some(sender) = &self.output_observation_sender {
       // OutputType derives Display via strum, producing clean names like "Vibrate", "Rotate".
-      // The design uses format!("{:?}") but to_string() is preferred for clean output.
       let _ = sender.send(OutputObservation {
         device_index: self.definition.index(),
         feature_index: msg.feature_index(),
@@ -293,37 +316,94 @@ impl DeviceHandle {
       });
     }
 
-    self.handle_generic_command_result(self.handler.handle_output_cmd(msg))
+    Some(self.handler.handle_output_cmd(msg).map_err(|e| e.into()))
   }
 
-  fn handle_hardware_commands(&self, commands: Vec<HardwareCommand>) -> ButtplugServerResultFuture {
+  /// Queue hardware commands for the device io task.
+  ///
+  /// When `wait_for_write` is false (all normal output paths) this is
+  /// fire-and-forget: the commands are dropped into the io channel and the
+  /// future resolves immediately, leaving any batching to the io task.
+  ///
+  /// When `wait_for_write` is true (the stop path) the commands carry a
+  /// write-acknowledgement channel: the io task flushes the batch to hardware
+  /// immediately and fires the ack, and this future only resolves once that
+  /// write has gone out (or the bounded wait elapses / the device goes away).
+  fn handle_hardware_commands(
+    &self,
+    commands: Vec<HardwareCommand>,
+    wait_for_write: bool,
+  ) -> ButtplugServerResultFuture {
     let sender = self.internal_hw_msg_sender.clone();
     async move {
-      let _ = sender.send(commands).await;
+      if wait_for_write {
+        let (ack_sender, ack_receiver) = oneshot::channel();
+        if sender
+          .send(DeviceTaskMessage {
+            commands,
+            ack: Some(ack_sender),
+          })
+          .await
+          .is_err()
+        {
+          debug!("Device io task gone, skipping write acknowledgement wait.");
+          return Ok(message::OkV0::default().into());
+        }
+        // Bounded wait for the write ack. There is no tokio::time on WASM, so
+        // race the receiver against a runtime-agnostic sleep instead of
+        // tokio::time::timeout. A wedged or dead device must not hang shutdown
+        // or error a stop, so any non-success outcome resolves Ok (matching the
+        // existing swallow semantics; scope cancellation handles stragglers).
+        select! {
+          ack = ack_receiver => {
+            if ack.is_err() {
+              debug!("Device io task dropped ack; device likely disconnecting.");
+            }
+          }
+          _ = async_manager::sleep(Duration::from_secs(1)) => {
+            warn!("Timed out waiting for stop command write acknowledgement.");
+          }
+        }
+      } else {
+        let _ = sender
+          .send(DeviceTaskMessage {
+            commands,
+            ack: None,
+          })
+          .await;
+      }
       Ok(message::OkV0::default().into())
     }
     .boxed()
   }
 
-  fn handle_generic_command_result(
-    &self,
-    command_result: Result<Vec<HardwareCommand>, ButtplugDeviceError>,
-  ) -> ButtplugServerResultFuture {
-    let hardware_commands = match command_result {
-      Ok(commands) => commands,
-      Err(err) => return future::ready(Err(err.into())).boxed(),
-    };
-
-    self.handle_hardware_commands(hardware_commands)
-  }
-
   fn handle_stop_device_cmd(&self, msg: &StopCmdV4) -> ButtplugServerResultFuture {
     let mut fut_vec = vec![];
     if msg.outputs() {
-      self
-        .stop_commands
-        .iter()
-        .for_each(|msg| fut_vec.push(self.parse_message(msg.clone())));
+      // Accumulate the hardware commands from every stop output into a single
+      // write-acknowledged batch. A multi-feature device produces one stop
+      // OutputCmd per feature, but the protocol handler folds them into one
+      // accumulated write; sending each individually with its own ack would
+      // emit one write per feature instead. Collecting them and issuing a
+      // single acked flush keeps the on-wire output identical to normal output
+      // batching while still resolving only once the stop write has landed.
+      //
+      // Stop commands are always OutputCmds (see build_device_handle); any
+      // other shape falls back to the normal fire-and-forget parse path.
+      let mut stop_hardware_commands = vec![];
+      for stop_msg in self.stop_commands.iter() {
+        match stop_msg {
+          ButtplugDeviceCommandMessageUnionV4::OutputCmd(output_msg) => {
+            match self.output_cmd_hardware_commands(output_msg) {
+              None => {}
+              Some(Ok(commands)) => stop_hardware_commands.extend(commands),
+              Some(Err(err)) => fut_vec.push(future::ready(Err(err)).boxed()),
+            }
+          }
+          other => fut_vec.push(self.parse_message(other.clone())),
+        }
+      }
+      fut_vec.push(self.handle_hardware_commands(stop_hardware_commands, true));
     }
     if msg.inputs() {
       self.definition.features().iter().for_each(|(i, f)| {
@@ -536,7 +616,7 @@ pub(super) async fn build_device_handle(
   let strategy = handler.keepalive_strategy();
 
   // Create the hardware command channel and spawn the device task
-  let (internal_hw_msg_sender, internal_hw_msg_recv) = channel::<Vec<HardwareCommand>>(1024);
+  let (internal_hw_msg_sender, internal_hw_msg_recv) = channel::<DeviceTaskMessage>(1024);
 
   let device_wait_duration = if let Some(gap) = definition.message_gap_ms() {
     Some(Duration::from_millis(gap as u64))
@@ -554,7 +634,7 @@ pub(super) async fn build_device_handle(
       keepalive_strategy: handler.keepalive_strategy(),
     },
     internal_hw_msg_recv,
-  );
+  )?;
 
   // Generate stop commands for this device
   let mut stop_commands: Vec<ButtplugDeviceCommandMessageUnionV4> = vec![];
@@ -637,51 +717,57 @@ pub(super) async fn build_device_handle(
   // to the device manager event loop via the provided sender.
   let event_stream = device_handle.event_stream();
   let identifier = device_handle.identifier().clone();
-  task_scope.spawn("event-forwarding", move |token| async move {
-    futures::pin_mut!(event_stream);
-    loop {
-      tokio::select! {
-        _ = token.cancelled() => {
-          info!("Event forwarding cancelled for device {:?}", identifier);
-          break;
-        }
-        event = futures::StreamExt::next(&mut event_stream) => {
-          match event {
-            Some(DeviceEvent::Disconnected(id)) => {
-              if device_event_sender
-                .send(InternalDeviceEvent::Disconnected(id))
-                .await
-                .is_err()
-              {
-                info!(
-                  "Device event sender closed for device {:?}, stopping event forwarding.",
-                  identifier
-                );
-                break;
+  task_scope
+    .spawn("event-forwarding", move |token| async move {
+      futures::pin_mut!(event_stream);
+      loop {
+        tokio::select! {
+          _ = token.cancelled() => {
+            info!("Event forwarding cancelled for device {:?}", identifier);
+            break;
+          }
+          event = futures::StreamExt::next(&mut event_stream) => {
+            match event {
+              Some(DeviceEvent::Disconnected(id)) => {
+                if device_event_sender
+                  .send(InternalDeviceEvent::Disconnected(id))
+                  .await
+                  .is_err()
+                {
+                  info!(
+                    "Device event sender closed for device {:?}, stopping event forwarding.",
+                    identifier
+                  );
+                  break;
+                }
               }
-            }
-            Some(DeviceEvent::Notification(_, msg)) => {
-              if device_event_sender
-                .send(InternalDeviceEvent::Notification(msg))
-                .await
-                .is_err()
-              {
-                info!(
-                  "Device event sender closed for device {:?}, stopping event forwarding.",
-                  identifier
-                );
+              Some(DeviceEvent::Notification(_, msg)) => {
+                if device_event_sender
+                  .send(InternalDeviceEvent::Notification(msg))
+                  .await
+                  .is_err()
+                {
+                  info!(
+                    "Device event sender closed for device {:?}, stopping event forwarding.",
+                    identifier
+                  );
+                  break;
+                }
+              }
+              None => {
+                // Stream ended (device likely disconnected)
                 break;
               }
             }
-            None => {
-              // Stream ended (device likely disconnected)
-              break;
-            }
           }
         }
       }
-    }
-  });
+    })
+    .map_err(|e| {
+      ButtplugDeviceError::DeviceConnectionError(format!(
+        "Could not spawn device event forwarding task: {e}"
+      ))
+    })?;
 
   Ok(device_handle)
 }
diff --git a/crates/buttplug_server/src/device/device_task.rs b/crates/buttplug_server/src/device/device_task.rs
index 8664ba35d..014bf351a 100644
--- a/crates/buttplug_server/src/device/device_task.rs
+++ b/crates/buttplug_server/src/device/device_task.rs
@@ -14,7 +14,10 @@
 
 use std::{collections::VecDeque, sync::Arc, time::Duration};
 
-use buttplug_core::util::{async_manager, task::TaskScope};
+use buttplug_core::{
+  errors::ButtplugDeviceError,
+  util::{async_manager, task::TaskScope},
+};
 use futures::future;
 use tokio::{select, sync::mpsc::Receiver, time::Instant};
 use tokio_util::sync::CancellationToken;
@@ -34,6 +37,15 @@ pub struct DeviceTaskConfig {
   pub keepalive_strategy: ProtocolKeepaliveStrategy,
 }
 
+/// A batch of hardware commands for the device io task, optionally carrying a
+/// write acknowledgement channel. When `ack` is present the batch is urgent:
+/// the io task flushes it (and any pending batched commands) to hardware
+/// immediately, then fires the ack.
+pub struct DeviceTaskMessage {
+  pub commands: Vec<HardwareCommand>,
+  pub ack: Option<tokio::sync::oneshot::Sender<()>>,
+}
+
 /// Spawn the device communication task.
 ///
 /// This task handles:
@@ -48,11 +60,15 @@ pub fn spawn_device_task(
   hardware: Arc<Hardware>,
   _handler: Arc<dyn ProtocolHandler>,
   config: DeviceTaskConfig,
-  mut command_receiver: Receiver<Vec<HardwareCommand>>,
-) {
-  task_scope.spawn("io", move |token| async move {
-    run_device_task(hardware, config, &mut command_receiver, token).await;
-  });
+  mut command_receiver: Receiver<DeviceTaskMessage>,
+) -> Result<(), ButtplugDeviceError> {
+  task_scope
+    .spawn("io", move |token| async move {
+      run_device_task(hardware, config, &mut command_receiver, token).await;
+    })
+    .map_err(|e| {
+      ButtplugDeviceError::DeviceConnectionError(format!("Could not spawn device I/O task: {e}"))
+    })
 }
 
 /// Run the device communication task (internal implementation).
@@ -62,7 +78,7 @@ pub fn spawn_device_task(
 async fn run_device_task(
   hardware: Arc<Hardware>,
   config: DeviceTaskConfig,
-  command_receiver: &mut Receiver<Vec<HardwareCommand>>,
+  command_receiver: &mut Receiver<DeviceTaskMessage>,
   token: CancellationToken,
 ) {
   let mut hardware_events = hardware.event_stream();
@@ -93,6 +109,23 @@ async fn run_device_task(
   let mut pending_commands: VecDeque<HardwareCommand> = VecDeque::new();
   let mut batch_deadline: Option<Instant> = None;
 
+  // Write every pending command to hardware in order, updating the keepalive
+  // packet as writes go out. Shared by the receive arm (urgent acked flush),
+  // the batch-deadline arm, and the flush-on-exit hardening below.
+  async fn flush_pending(
+    hardware: &Hardware,
+    pending_commands: &mut VecDeque<HardwareCommand>,
+    track_keepalive: bool,
+    keepalive_packet: &mut Option<HardwareWriteCmd>,
+  ) {
+    while let Some(cmd) = pending_commands.pop_front() {
+      let _ = hardware.parse_message(&cmd).await;
+      if track_keepalive && let HardwareCommand::Write(ref write_cmd) = cmd {
+        *keepalive_packet = Some(write_cmd.clone());
+      }
+    }
+  }
+
   loop {
     // Calculate keepalive timeout
     let keepalive_fut = async {
@@ -121,17 +154,54 @@ async fn run_device_task(
       // Priority 0: Cooperative cancellation - wins over new work.
       _ = token.cancelled() => {
         info!("Device task cancelled, shutting down");
+        // Best-effort flush of any batched writes (e.g. a stop command still
+        // sitting in the batch window) before exiting. The hardware is still
+        // present here, so the writes can land; write errors are swallowed.
+        flush_pending(
+          &hardware,
+          &mut pending_commands,
+          track_keepalive,
+          &mut keepalive_packet,
+        )
+        .await;
         return;
       }
 
       // Priority 1: Incoming commands
       msg = command_receiver.recv() => {
-        let Some(commands) = msg else {
+        let Some(DeviceTaskMessage { commands, ack }) = msg else {
           info!("No longer receiving messages from device parent, breaking");
+          // The command channel closed (all DeviceHandles dropped). Hardware is
+          // still present, so best-effort flush any batched writes before exit.
+          flush_pending(
+            &hardware,
+            &mut pending_commands,
+            track_keepalive,
+            &mut keepalive_packet,
+          )
+          .await;
           break;
         };
 
-        if let Some(device_wait_duration) = device_wait_duration {
+        if let Some(ack) = ack {
+          // Urgent write-acknowledged batch (stop path). Merge with any pending
+          // batch using the existing dedupe, flush everything to hardware now
+          // regardless of the batch deadline, then fire the ack so the caller
+          // only resolves once the writes have actually gone out.
+          for command in commands {
+            pending_commands.retain(|existing| !command.overlaps(existing));
+            pending_commands.push_back(command);
+          }
+          flush_pending(
+            &hardware,
+            &mut pending_commands,
+            track_keepalive,
+            &mut keepalive_packet,
+          )
+          .await;
+          batch_deadline = None;
+          let _ = ack.send(());
+        } else if let Some(device_wait_duration) = device_wait_duration {
           // Batching enabled
           if pending_commands.is_empty() {
             // First batch - add directly without deduplication (matches old behavior)
@@ -147,28 +217,27 @@ async fn run_device_task(
         } else {
           // No batching - send immediately
           trace!("No wait duration, sending commands immediately: {:?}", commands);
-          for cmd in commands {
-            let _ = hardware.parse_message(&cmd).await;
-            if track_keepalive
-              && let HardwareCommand::Write(ref write_cmd) = cmd
-            {
-              keepalive_packet = Some(write_cmd.clone());
-            }
-          }
+          pending_commands.extend(commands);
+          flush_pending(
+            &hardware,
+            &mut pending_commands,
+            track_keepalive,
+            &mut keepalive_packet,
+          )
+          .await;
         }
       }
 
       // Priority 2: Batch deadline reached - flush pending commands
       _ = batch_fut => {
         trace!("Batch deadline reached, sending {} commands", pending_commands.len());
-        while let Some(cmd) = pending_commands.pop_front() {
-          let _ = hardware.parse_message(&cmd).await;
-          if track_keepalive
-            && let HardwareCommand::Write(ref write_cmd) = cmd
-          {
-            keepalive_packet = Some(write_cmd.clone());
-          }
-        }
+        flush_pending(
+          &hardware,
+          &mut pending_commands,
+          track_keepalive,
+          &mut keepalive_packet,
+        )
+        .await;
         batch_deadline = None;
       }
 
@@ -208,6 +277,8 @@ async fn run_device_task(
       hw_event = hardware_events.recv() => {
         if matches!(hw_event, Ok(HardwareEvent::Disconnected(_))) || hw_event.is_err() {
           info!("Hardware disconnected, shutting down task");
+          // Do NOT flush pending_commands here: the hardware is gone, so writes
+          // would fail and any pending stop is moot.
           return;
         }
       }
diff --git a/crates/buttplug_server/src/device/protocol_impl/kgoal_boost.rs b/crates/buttplug_server/src/device/protocol_impl/kgoal_boost.rs
index c21f1a796..b2fbb1a4f 100644
--- a/crates/buttplug_server/src/device/protocol_impl/kgoal_boost.rs
+++ b/crates/buttplug_server/src/device/protocol_impl/kgoal_boost.rs
@@ -91,79 +91,89 @@ impl ProtocolHandler for KGoalBoost {
         let stream_sensors = stream_sensors.clone();
         info!("Starting Kgoal subscription");
         // If we subscribe successfully, we need to set up our event handler.
-        task_scope.spawn_and_hold("kgoal-events", move |token| async move {
-          let mut cached_values = vec![0u32, 0u32];
-          loop {
-            let info = tokio::select! {
-              biased;
-              _ = token.cancelled() => return,
-              info = hardware_stream.recv() => {
-                let Ok(info) = info else {
-                  return;
-                };
-                info
-              }
-            };
-            let subscribed_sensors = stream_sensors.load(Ordering::Relaxed);
-            // If we have no receivers, quit.
-            if sender.receiver_count() == 0 || subscribed_sensors == 0 {
-              return;
-            }
-            if let HardwareEvent::Notification(_, endpoint, data) = info
-              && endpoint == Endpoint::RxPressure
-            {
-              if data.len() < 7 {
-                // Not even sure how this would happen, error and continue on.
-                error!("KGoal Boost data not expected length!");
-                continue;
-              }
-              // Extract our two pressure values.
-              let normalized = (data[3] as u32) << 8 | data[4] as u32;
-              let unnormalized = (data[5] as u32) << 8 | data[6] as u32;
-              info!(
-                "Kgoal Reading {} {} {}",
-                subscribed_sensors, normalized, unnormalized
-              );
-              if (subscribed_sensors & (1 << 0)) > 0
-                && cached_values[0] != normalized
-                && sender
-                  .send(
-                    InputReadingV4::new(
-                      device_index,
-                      0,
-                      buttplug_core::message::InputTypeReading::Pressure(InputValue::new(
-                        normalized,
-                      )),
-                    )
-                    .into(),
-                  )
-                  .is_err()
-              {
-                debug!("Hardware device listener for KGoal Boost shut down, returning from task.");
+        task_scope
+          .spawn_and_hold("kgoal-events", move |token| async move {
+            let mut cached_values = vec![0u32, 0u32];
+            loop {
+              let info = tokio::select! {
+                biased;
+                _ = token.cancelled() => return,
+                info = hardware_stream.recv() => {
+                  let Ok(info) = info else {
+                    return;
+                  };
+                  info
+                }
+              };
+              let subscribed_sensors = stream_sensors.load(Ordering::Relaxed);
+              // If we have no receivers, quit.
+              if sender.receiver_count() == 0 || subscribed_sensors == 0 {
                 return;
               }
-              if (subscribed_sensors & (1 << 1)) > 0
-                && cached_values[1] != unnormalized
-                && sender
-                  .send(
-                    InputReadingV4::new(
-                      device_index,
-                      1,
-                      buttplug_core::message::InputTypeReading::Pressure(InputValue::new(
-                        unnormalized,
-                      )),
-                    )
-                    .into(),
-                  )
-                  .is_err()
+              if let HardwareEvent::Notification(_, endpoint, data) = info
+                && endpoint == Endpoint::RxPressure
               {
-                debug!("Hardware device listener for KGoal Boost shut down, returning from task.");
-                return;
+                if data.len() < 7 {
+                  // Not even sure how this would happen, error and continue on.
+                  error!("KGoal Boost data not expected length!");
+                  continue;
+                }
+                // Extract our two pressure values.
+                let normalized = (data[3] as u32) << 8 | data[4] as u32;
+                let unnormalized = (data[5] as u32) << 8 | data[6] as u32;
+                info!(
+                  "Kgoal Reading {} {} {}",
+                  subscribed_sensors, normalized, unnormalized
+                );
+                if (subscribed_sensors & (1 << 0)) > 0
+                  && cached_values[0] != normalized
+                  && sender
+                    .send(
+                      InputReadingV4::new(
+                        device_index,
+                        0,
+                        buttplug_core::message::InputTypeReading::Pressure(InputValue::new(
+                          normalized,
+                        )),
+                      )
+                      .into(),
+                    )
+                    .is_err()
+                {
+                  debug!(
+                    "Hardware device listener for KGoal Boost shut down, returning from task."
+                  );
+                  return;
+                }
+                if (subscribed_sensors & (1 << 1)) > 0
+                  && cached_values[1] != unnormalized
+                  && sender
+                    .send(
+                      InputReadingV4::new(
+                        device_index,
+                        1,
+                        buttplug_core::message::InputTypeReading::Pressure(InputValue::new(
+                          unnormalized,
+                        )),
+                      )
+                      .into(),
+                    )
+                    .is_err()
+                {
+                  debug!(
+                    "Hardware device listener for KGoal Boost shut down, returning from task."
+                  );
+                  return;
+                }
+                cached_values = vec![normalized, unnormalized];
               }
-              cached_values = vec![normalized, unnormalized];
             }
-          }
-        });
+          })
+          .map_err(|e| {
+            ButtplugDeviceError::DeviceConnectionError(format!(
+              "Could not start Kgoal event forwarding: {e}"
+            ))
+          })?;
       }
       stream_sensors.store(
         stream_sensors.load(Ordering::Relaxed) | (1 << feature_index),
diff --git a/crates/buttplug_server/src/device/server_device_manager.rs b/crates/buttplug_server/src/device/server_device_manager.rs
index ddb36fbf6..437f1cffd 100644
--- a/crates/buttplug_server/src/device/server_device_manager.rs
+++ b/crates/buttplug_server/src/device/server_device_manager.rs
@@ -190,20 +190,28 @@ impl ServerDeviceManagerBuilder {
     let devices_clone = devices.clone();
     let output_sender_clone = output_sender.clone();
     let output_observation_sender_clone = output_observation_sender.clone();
-    task_scope.spawn("event-loop", move |token| async move {
-      let mut event_loop = ServerDeviceManagerEventLoop::new(
-        comm_managers,
-        device_configuration_manager,
-        devices_clone,
-        token,
-        devices_scope,
-        output_sender_clone,
-        device_event_receiver,
-        device_command_receiver,
-        output_observation_sender_clone,
-      );
-      event_loop.run().await;
-    });
+    task_scope
+      .spawn("event-loop", move |token| async move {
+        let mut event_loop = ServerDeviceManagerEventLoop::new(
+          comm_managers,
+          device_configuration_manager,
+          devices_clone,
+          token,
+          devices_scope,
+          output_sender_clone,
+          device_event_receiver,
+          device_command_receiver,
+          output_observation_sender_clone,
+        );
+        event_loop.run().await;
+      })
+      .map_err(|e| {
+        ButtplugServerError::DeviceConfigurationManagerError(
+          ButtplugDeviceError::DeviceConnectionError(format!(
+            "Could not spawn device manager event loop: {e}"
+          )),
+        )
+      })?;
     Ok(ServerDeviceManager {
       device_configuration_manager: self.device_configuration_manager.clone(),
       devices,
diff --git a/crates/buttplug_server/src/device/server_device_manager_event_loop.rs b/crates/buttplug_server/src/device/server_device_manager_event_loop.rs
index 0635bffbc..0b909139f 100644
--- a/crates/buttplug_server/src/device/server_device_manager_event_loop.rs
+++ b/crates/buttplug_server/src/device/server_device_manager_event_loop.rs
@@ -330,9 +330,11 @@ impl ServerDeviceManagerEventLoop {
         // forever waiting for this task to deregister. We select on the token;
         // on cancellation we drop the build_device_handle future, which drops
         // device_scope and thereby cancels anything it has already spawned.
-        self.devices_scope.spawn(
+        let bringup_address = address.clone();
+        let bringup_result = self.devices_scope.spawn(
           &format!("bringup-{address}"),
           move |token| async move {
+            let address = bringup_address;
             tokio::select! {
               biased;
               _ = token.cancelled() => {
@@ -371,6 +373,12 @@ impl ServerDeviceManagerEventLoop {
             connecting_devices.remove(&address);
           },
         );
+        if let Err(e) = bringup_result {
+          debug!("Device bringup for {address} was not started because its scope is closed: {e}");
+          // Registration was rejected synchronously, so the closure did not run
+          // and cannot perform its normal cleanup.
+          self.connecting_devices.remove(&address);
+        }
       }
     }
   }
diff --git a/crates/buttplug_server/src/lib.rs b/crates/buttplug_server/src/lib.rs
index 37e4cb9a5..fa3d76521 100644
--- a/crates/buttplug_server/src/lib.rs
+++ b/crates/buttplug_server/src/lib.rs
@@ -82,6 +82,9 @@ pub type ButtplugServerResultFuture = BoxFuture<'static, ButtplugServerResult>;
 /// Error enum for Buttplug Server configuration errors.
 #[derive(Error, Debug)]
 pub enum ButtplugServerError {
+  /// A server task could not be spawned during construction.
+  #[error(transparent)]
+  TaskSpawnError(#[from] buttplug_core::util::task::TaskSpawnError),
   /// DeviceConfigurationManager could not be built.
   #[error("The DeviceConfigurationManager could not be built: {0}")]
   DeviceConfigurationManagerError(ButtplugDeviceError),
diff --git a/crates/buttplug_server/src/ping_timer.rs b/crates/buttplug_server/src/ping_timer.rs
index 2fece8dba..cc1a1c075 100644
--- a/crates/buttplug_server/src/ping_timer.rs
+++ b/crates/buttplug_server/src/ping_timer.rs
@@ -76,7 +76,11 @@ impl PingTimer {
   /// The callback is called once when the ping timer expires without receiving
   /// a ping message. If max_ping_time is 0, the timer is disabled and the
   /// callback will never be called.
-  pub fn new<F>(max_ping_time: u32, on_ping_timeout: Option<F>, task_scope: TaskScope) -> Self
+  pub fn new<F>(
+    max_ping_time: u32,
+    on_ping_timeout: Option<F>,
+    task_scope: TaskScope,
+  ) -> Result<Self, buttplug_core::util::task::TaskSpawnError>
   where
     F: FnOnce() + Send + 'static,
   {
@@ -85,13 +89,13 @@ impl PingTimer {
       let callback = Arc::new(Mutex::new(on_ping_timeout));
       task_scope.spawn("timer", move |token| {
         ping_timer(max_ping_time, receiver, callback, token)
-      });
+      })?;
     }
-    Self {
+    Ok(Self {
       max_ping_time,
       ping_msg_sender: sender,
       _task_scope: task_scope,
-    }
+    })
   }
 
   fn send_ping_msg(&self, msg: PingMessage) -> impl Future<Output = ()> + use<> {
diff --git a/crates/buttplug_server/src/server_builder.rs b/crates/buttplug_server/src/server_builder.rs
index 5763f384e..8ee71b313 100644
--- a/crates/buttplug_server/src/server_builder.rs
+++ b/crates/buttplug_server/src/server_builder.rs
@@ -120,14 +120,19 @@ impl ButtplugServerBuilder {
         // Stop all devices (spawn async task since callback is sync). The
         // callback is FnOnce, so the child scope moves in and is consumed by
         // spawn_and_hold, keeping it alive for the duration of the task.
-        ping_timeout_scope.spawn_and_hold("stop-devices", move |_token| async move {
+        match ping_timeout_scope.spawn_and_hold("stop-devices", move |_token| async move {
           if let Err(e) = device_manager_clone
             .stop_devices(&StopCmdV4::default())
             .await
           {
             error!("Could not stop devices on ping timeout: {:?}", e);
           }
-        });
+        }) {
+          Ok(()) => {}
+          Err(e) => {
+            debug!("Skipping ping-timeout device stop because teardown closed its scope: {e}");
+          }
+        }
         // Send error to output channel
         if output_sender_clone
           .send(ButtplugServerMessageV4::Error(message::ErrorV0::from(
@@ -146,7 +151,7 @@ impl ButtplugServerBuilder {
       ping_time,
       ping_timeout_callback,
       task_scope.child("ping-timer"),
-    ));
+    )?);
 
     // Assuming everything passed, return the server.
     Ok(ButtplugServer::new(
diff --git a/crates/buttplug_tests/tests/test_task_lifecycle.rs b/crates/buttplug_tests/tests/test_task_lifecycle.rs
index 4425c4443..d3b7c7d93 100644
--- a/crates/buttplug_tests/tests/test_task_lifecycle.rs
+++ b/crates/buttplug_tests/tests/test_task_lifecycle.rs
@@ -17,14 +17,78 @@ use buttplug_core::{
     OutputValue,
     RequestServerInfoV4,
     StartScanningV0,
+    StopCmdV4,
   },
   util::task::registry,
 };
-use buttplug_server::message::ButtplugClientMessageVariant;
+use buttplug_server::{device::hardware::HardwareCommand, message::ButtplugClientMessageVariant};
 use futures::{StreamExt, pin_mut};
 use std::time::Duration;
+use util::TestDeviceChannelHost;
 use util::stalling_device_communication_manager::StallingDeviceCommunicationManagerBuilder;
-use util::{test_server_with_comm_manager, test_server_with_device};
+use util::{
+  test_server_with_comm_manager,
+  test_server_with_device,
+  test_server_with_device_and_message_gap,
+};
+
+/// The Aneros "Massage Demo" stop write for feature 0: `[0xF1, 0x00]`. A vibrate
+/// on the same feature writes `[0xF1, 0x40]`; the stop resets it to zero. See
+/// `device_test_case/test_aneros_protocol.yaml` for the full sequence.
+const ANEROS_STOP_WRITE_FEATURE_0: [u8; 2] = [0xF1, 0x00];
+
+/// Non-blockingly drain every hardware write the test device has recorded so
+/// far and return whether any of them is the feature-0 stop write. We use
+/// `try_recv` so the check reflects exactly what was on the wire *at the moment
+/// stop/shutdown resolved* — a write still sitting in the device io task's batch
+/// window has not reached the host channel yet and will not be counted.
+fn recorded_a_stop_write(host: &mut TestDeviceChannelHost) -> bool {
+  let mut saw_stop = false;
+  while let Ok(command) = host.receiver.try_recv() {
+    if let HardwareCommand::Write(write) = command
+      && write.data().as_slice() == ANEROS_STOP_WRITE_FEATURE_0
+    {
+      saw_stop = true;
+    }
+  }
+  saw_stop
+}
+
+/// Bring a "Massage Demo" device online under `server`, returning its device
+/// index. Mirrors the handshake/scan/connect dance the other lifecycle tests do.
+async fn bring_device_online(server: &buttplug_server::ButtplugServer, client_name: &str) -> u32 {
+  let recv = server.server_version_event_stream();
+  pin_mut!(recv);
+  server
+    .parse_message(ButtplugClientMessageVariant::V4(
+      RequestServerInfoV4::new(
+        client_name,
+        BUTTPLUG_CURRENT_API_MAJOR_VERSION,
+        BUTTPLUG_CURRENT_API_MINOR_VERSION,
+      )
+      .into(),
+    ))
+    .await
+    .expect("server info request should succeed");
+  server
+    .parse_message(ButtplugClientMessageVariant::V4(
+      StartScanningV0::default().into(),
+    ))
+    .await
+    .expect("start scanning should succeed");
+  tokio::time::timeout(Duration::from_secs(5), async {
+    while let Some(msg) = recv.next().await {
+      if let ButtplugServerMessageV4::DeviceList(list) = msg
+        && let Some((&idx, _)) = list.devices().iter().next()
+      {
+        return idx;
+      }
+    }
+    panic!("device event stream ended before a device connected");
+  })
+  .await
+  .expect("timed out waiting for device to connect")
+}
 
 /// Bring a real (test-harness) device online and confirm that, once the server
 /// is shut down and dropped, no tasks remain registered under the scope tree.
@@ -165,6 +229,13 @@ async fn test_server_shutdown_leaves_no_tasks() {
 /// ordering bug, so a write-observation assertion is inherently flaky here. An
 /// instrumented-ordering variant was also attempted and found inherently flaky
 /// with this harness, so it is deliberately not pursued.
+///
+/// UPDATE: with the stop-command write acknowledgement (DeviceTaskMessage ack)
+/// and a configurable per-device message_gap, the write observation IS now
+/// possible — see `test_stop_resolves_only_after_stop_write_reaches_hardware`
+/// and `test_shutdown_writes_stop_before_resolving` below, which retire the
+/// limitation described in the NOTE above for the acked stop path. This test
+/// remains as the coarse hang/leak smoke test.
 #[tokio::test]
 async fn test_shutdown_under_load_drains_subtree() {
   // Hold the channel so the device stays connected through shutdown.
@@ -296,3 +367,165 @@ async fn test_shutdown_resolves_with_stalled_bringup() {
     .expect("shutdown hung with a stalled device bringup — bringup is not honoring its cancellation token")
     .expect("server shutdown errored");
 }
+
+/// A bring-up that has entered `connect()` is shut down deterministically. The
+/// retained device-manager scope handle cannot register a task after shutdown,
+/// and the rejected bring-up cannot add a device after shutdown resolves.
+#[tokio::test]
+async fn test_shutdown_rejects_bringup_and_drains_manager_scope() {
+  let builder = StallingDeviceCommunicationManagerBuilder::default();
+  let state = builder.state();
+  let server = test_server_with_comm_manager(builder);
+  let scope_prefix = server.device_manager().scope_path().to_owned();
+  let events = server.server_version_event_stream();
+  pin_mut!(events);
+
+  server
+    .parse_message(ButtplugClientMessageVariant::V4(
+      RequestServerInfoV4::new(
+        "Bringup Shutdown Race Test",
+        BUTTPLUG_CURRENT_API_MAJOR_VERSION,
+        BUTTPLUG_CURRENT_API_MINOR_VERSION,
+      )
+      .into(),
+    ))
+    .await
+    .expect("server info request should succeed");
+  server
+    .parse_message(ButtplugClientMessageVariant::V4(
+      StartScanningV0::default().into(),
+    ))
+    .await
+    .expect("start scanning should succeed");
+
+  tokio::time::timeout(Duration::from_secs(5), state.connect_started.notified())
+    .await
+    .expect("bring-up did not reach connect() in time");
+
+  tokio::time::timeout(Duration::from_secs(10), server.shutdown())
+    .await
+    .expect("shutdown did not resolve in time")
+    .expect("server shutdown errored");
+
+  assert_eq!(registry().live_count_under(&scope_prefix), 0);
+
+  // The stalling bring-up never reaches device construction, so shutdown must
+  // not be followed by a device-added event. Drain already-ready events only;
+  // do not sleep or loop to manufacture a race.
+  while let Ok(Some(message)) = tokio::time::timeout(Duration::from_millis(10), events.next()).await
+  {
+    if let ButtplugServerMessageV4::DeviceList(list) = message {
+      assert!(
+        list.devices().is_empty(),
+        "a device was added after shutdown resolved"
+      );
+    }
+  }
+  assert_eq!(registry().live_count_under(&scope_prefix), 0);
+}
+
+/// Test A — the stop-write-acknowledgement contract: a `StopCmd` must not
+/// resolve until the resulting stop write has actually reached the hardware,
+/// even when the device io task is batching commands over a long message gap.
+///
+/// This retires the limitation documented on `test_shutdown_under_load_drains_subtree`
+/// (the 1ms harness gap made write observation flaky). Here we deliberately give
+/// the device a 500ms message gap so the batching window is large and the race
+/// is deterministic: an active vibrate write lands, the stop write is queued
+/// into that 500ms batch, and we assert the stop write is on the wire *by the
+/// time the stop message resolves*.
+///
+/// RED (pre-fix, `git stash` Tasks 1-2): `handle_hardware_commands` fire-and-forgets
+/// the stop write into the io channel and `parse_message` for the StopCmd resolves
+/// immediately, while the write sits unflushed in the 500ms batch. `try_recv`
+/// finds no stop write and this assertion fails deterministically.
+///
+/// GREEN (with the ack-on-write fix): the stop path requests a write
+/// acknowledgement; the io task urgent-flushes the batch and fires the ack, so
+/// StopCmd only resolves after the write is on the wire.
+#[tokio::test]
+async fn test_stop_resolves_only_after_stop_write_reaches_hardware() {
+  // 500ms gap: large enough that, pre-fix, the stop write provably has not been
+  // flushed by the time StopCmd resolves (which is ~immediate).
+  let (server, mut device) =
+    test_server_with_device_and_message_gap("Massage Demo", Duration::from_millis(500));
+
+  let device_index = bring_device_online(&server, "Stop Write Ack Test").await;
+
+  // Put the device into an actively-running state so the stop has real work to
+  // flush. This vibrate write itself enters the batch window.
+  server
+    .parse_message(ButtplugClientMessageVariant::V4(
+      OutputCmdV4::new(
+        device_index,
+        0,
+        OutputCommand::Vibrate(OutputValue::new(50)),
+      )
+      .into(),
+    ))
+    .await
+    .expect("vibrate command should succeed");
+
+  // Stop. With ack-on-write, parse_message for StopCmd resolves only after the
+  // stop write has been flushed to hardware.
+  server
+    .parse_message(ButtplugClientMessageVariant::V4(
+      StopCmdV4::new(Some(device_index), None, true, true).into(),
+    ))
+    .await
+    .expect("stop command should succeed");
+
+  // The instant stop resolves, the stop write must already be on the wire. No
+  // sleep here: that is the whole point — we are asserting ordering, not
+  // eventual delivery.
+  assert!(
+    recorded_a_stop_write(&mut device),
+    "StopCmd resolved but the stop write was not yet recorded on the hardware channel — \
+     it is still sitting in the device io task's batch window"
+  );
+}
+
+/// Test B — the original incident: with a batched device in an active output
+/// state, `server.shutdown()` must drive the per-device stop write all the way
+/// to hardware before it resolves. This is the assertion the task-scope work
+/// could not make with the 1ms harness gap (the disconnect tore the io task down
+/// inside the batch window, dropping the pending stop write). With ack-on-write,
+/// stop_devices waits for the write before shutdown proceeds to disconnect.
+///
+/// RED (pre-fix): shutdown's stop_devices fire-and-forgets the stop write, then
+/// disconnect tears down the io task mid-batch and the write is dropped — no stop
+/// write is ever recorded.
+///
+/// GREEN (with the fix): the stop write is acknowledged before disconnect, so it
+/// is recorded before shutdown resolves.
+#[tokio::test]
+async fn test_shutdown_writes_stop_before_resolving() {
+  let (server, mut device) =
+    test_server_with_device_and_message_gap("Massage Demo", Duration::from_millis(500));
+
+  let device_index = bring_device_online(&server, "Shutdown Stop Write Test").await;
+
+  server
+    .parse_message(ButtplugClientMessageVariant::V4(
+      OutputCmdV4::new(
+        device_index,
+        0,
+        OutputCommand::Vibrate(OutputValue::new(50)),
+      )
+      .into(),
+    ))
+    .await
+    .expect("vibrate command should succeed");
+
+  tokio::time::timeout(Duration::from_secs(10), server.shutdown())
+    .await
+    .expect("shutdown did not resolve in time")
+    .expect("server shutdown errored");
+
+  // By the time shutdown resolved, the stop write must have reached hardware.
+  assert!(
+    recorded_a_stop_write(&mut device),
+    "shutdown() resolved but the device's stop write was never recorded — \
+     the pending stop write was dropped when the io task was torn down mid-batch"
+  );
+}
diff --git a/crates/buttplug_tests/tests/util/mod.rs b/crates/buttplug_tests/tests/util/mod.rs
index 9f4e07e8c..a82460353 100644
--- a/crates/buttplug_tests/tests/util/mod.rs
+++ b/crates/buttplug_tests/tests/util/mod.rs
@@ -170,6 +170,22 @@ pub fn test_server_with_device(device_type: &str) -> (ButtplugServer, TestDevice
   (test_server_with_comm_manager(builder), device)
 }
 
+/// Like `test_server_with_device`, but the device's io task batches commands
+/// over `message_gap`. A large gap widens the stop-write batching window so
+/// tests can deterministically observe whether stop commands are flushed to
+/// hardware before stop/shutdown resolve.
+#[allow(dead_code)]
+pub fn test_server_with_device_and_message_gap(
+  device_type: &str,
+  message_gap: std::time::Duration,
+) -> (ButtplugServer, TestDeviceChannelHost) {
+  let mut builder = TestDeviceCommunicationManagerBuilder::default();
+  let device = builder
+    .add_test_device_with_message_gap(&TestDeviceIdentifier::new(device_type, None), message_gap);
+
+  (test_server_with_comm_manager(builder), device)
+}
+
 #[allow(dead_code)]
 pub fn test_server_v4_with_device(device_type: &str) -> (ButtplugServer, TestDeviceChannelHost) {
   let mut builder = TestDeviceCommunicationManagerBuilder::default();
diff --git a/crates/buttplug_tests/tests/util/stalling_device_communication_manager.rs b/crates/buttplug_tests/tests/util/stalling_device_communication_manager.rs
index 7b21868c4..0d3540302 100644
--- a/crates/buttplug_tests/tests/util/stalling_device_communication_manager.rs
+++ b/crates/buttplug_tests/tests/util/stalling_device_communication_manager.rs
@@ -30,12 +30,19 @@ use buttplug_server_device_config::{BluetoothLESpecifier, ProtocolCommunicationS
 use futures::FutureExt;
 use log::error;
 use std::collections::HashMap;
-use tokio::sync::mpsc::Sender;
+use tokio::sync::{Notify, mpsc::Sender};
+
+/// Shared state used to deterministically observe that bring-up entered connect().
+#[derive(Clone, Debug, Default)]
+pub struct StallingDeviceState {
+  pub connect_started: std::sync::Arc<Notify>,
+}
 
 /// A `HardwareConnector` whose `connect()` future never resolves.
 #[derive(Debug)]
 struct StallingHardwareConnector {
   specifier: ProtocolCommunicationSpecifier,
+  state: StallingDeviceState,
 }
 
 #[async_trait]
@@ -47,13 +54,22 @@ impl HardwareConnector for StallingHardwareConnector {
   async fn connect(&mut self) -> Result<Box<dyn HardwareSpecializer>, ButtplugDeviceError> {
     // Block forever: this models a device connect that hangs and never returns.
     // The bringup task must drop this future when its cancellation token fires.
+    self.state.connect_started.notify_waiters();
     std::future::pending::<()>().await;
     unreachable!("stalling connector connect() should never resolve");
   }
 }
 
-#[derive(Default)]
-pub struct StallingDeviceCommunicationManagerBuilder {}
+#[derive(Clone, Default)]
+pub struct StallingDeviceCommunicationManagerBuilder {
+  state: StallingDeviceState,
+}
+
+impl StallingDeviceCommunicationManagerBuilder {
+  pub fn state(&self) -> StallingDeviceState {
+    self.state.clone()
+  }
+}
 
 impl HardwareCommunicationManagerBuilder for StallingDeviceCommunicationManagerBuilder {
   fn finish(
@@ -62,12 +78,14 @@ impl HardwareCommunicationManagerBuilder for StallingDeviceCommunicationManagerB
   ) -> Box<dyn HardwareCommunicationManager> {
     Box::new(StallingDeviceCommunicationManager {
       device_sender: sender,
+      state: self.state.clone(),
     })
   }
 }
 
 pub struct StallingDeviceCommunicationManager {
   device_sender: Sender<HardwareCommunicationManagerEvent>,
+  state: StallingDeviceState,
 }
 
 impl HardwareCommunicationManager for StallingDeviceCommunicationManager {
@@ -77,13 +95,17 @@ impl HardwareCommunicationManager for StallingDeviceCommunicationManager {
 
   fn start_scanning(&mut self) -> ButtplugResultFuture {
     let device_sender = self.device_sender.clone();
+    let state = self.state.clone();
     async move {
       // "Massage Demo" is a known test device name with a real protocol config,
       // so bringup proceeds into connect() (which then stalls).
       let specifier = ProtocolCommunicationSpecifier::BluetoothLE(
         BluetoothLESpecifier::new_from_device("Massage Demo", &HashMap::new(), &[]),
       );
-      let connector = StallingHardwareConnector { specifier };
+      let connector = StallingHardwareConnector {
+        specifier,
+        state: state.clone(),
+      };
       if device_sender
         .send(HardwareCommunicationManagerEvent::DeviceFound {
           name: "Massage Demo".to_owned(),
diff --git a/crates/buttplug_tests/tests/util/test_device_manager/test_device.rs b/crates/buttplug_tests/tests/util/test_device_manager/test_device.rs
index 91916c916..5e4c56654 100644
--- a/crates/buttplug_tests/tests/util/test_device_manager/test_device.rs
+++ b/crates/buttplug_tests/tests/util/test_device_manager/test_device.rs
@@ -115,12 +115,16 @@ impl HardwareSpecializer for TestHardwareSpecializer {
         }
       }
     }
+    // The device carries its own message gap (default 1ms for the slight delay
+    // multi-message protocols expect). Tests that need a large batching window
+    // to make the stop-write race deterministic override it via
+    // `add_test_device_with_message_gap`.
+    let message_gap = device.message_gap();
     let hardware = Hardware::new(
       &device.name(),
       &device.address(),
       &endpoints,
-      // Add slight delay for protocols with multiple messages.
-      &Some(Duration::from_millis(1)),
+      &Some(message_gap),
       false,
       Box::new(device),
     );
@@ -153,10 +157,15 @@ pub fn new_device_channel() -> (TestDeviceChannelHost, TestDeviceChannelDevice)
   )
 }
 
+/// Default inter-message gap for test hardware. A slight delay multi-message
+/// protocols expect; small enough that ordinary tests are unaffected.
+const DEFAULT_MESSAGE_GAP: Duration = Duration::from_millis(1);
+
 pub struct TestDevice {
   name: String,
   address: String,
   endpoints: HashSet<Endpoint>,
+  message_gap: Duration,
   test_device_channel: mpsc::Sender<HardwareCommand>,
   event_sender: broadcast::Sender<HardwareEvent>,
   subscribed_endpoints: Arc<DashSet<Endpoint>>,
@@ -166,6 +175,16 @@ pub struct TestDevice {
 impl TestDevice {
   #[allow(dead_code)]
   pub fn new(name: &str, address: &str, test_device_channel: TestDeviceChannelDevice) -> Self {
+    Self::new_with_message_gap(name, address, DEFAULT_MESSAGE_GAP, test_device_channel)
+  }
+
+  #[allow(dead_code)]
+  pub fn new_with_message_gap(
+    name: &str,
+    address: &str,
+    message_gap: Duration,
+    test_device_channel: TestDeviceChannelDevice,
+  ) -> Self {
     let (event_sender, _) = broadcast::channel(256);
 
     let event_sender_clone = event_sender.clone();
@@ -210,6 +229,7 @@ impl TestDevice {
       name: name.to_owned(),
       address: address.to_owned(),
       endpoints: HashSet::new(),
+      message_gap,
       test_device_channel: command_sender,
       event_sender,
       subscribed_endpoints,
@@ -221,6 +241,10 @@ impl TestDevice {
     self.endpoints.insert(*endpoint);
   }
 
+  pub fn message_gap(&self) -> Duration {
+    self.message_gap
+  }
+
   pub fn name(&self) -> String {
     self.name.clone()
   }
diff --git a/crates/buttplug_tests/tests/util/test_device_manager/test_device_comm_manager.rs b/crates/buttplug_tests/tests/util/test_device_manager/test_device_comm_manager.rs
index 6e0d18a70..81fe9c276 100644
--- a/crates/buttplug_tests/tests/util/test_device_manager/test_device_comm_manager.rs
+++ b/crates/buttplug_tests/tests/util/test_device_manager/test_device_comm_manager.rs
@@ -30,7 +30,7 @@ use std::{
     Arc,
     atomic::{AtomicBool, Ordering},
   },
-  time::{SystemTime, UNIX_EPOCH},
+  time::{Duration, SystemTime, UNIX_EPOCH},
 };
 use tokio::sync::mpsc::Sender;
 
@@ -74,8 +74,11 @@ impl TestDeviceIdentifier {
   }
 }
 
+/// Default inter-message gap for test hardware, matching `TestDevice`'s default.
+const DEFAULT_MESSAGE_GAP: Duration = Duration::from_millis(1);
+
 pub struct TestDeviceCommunicationManagerBuilder {
-  devices: Option<Vec<(TestDeviceIdentifier, TestDeviceChannelDevice)>>,
+  devices: Option<Vec<(TestDeviceIdentifier, Duration, TestDeviceChannelDevice)>>,
 }
 
 impl Default for TestDeviceCommunicationManagerBuilder {
@@ -88,12 +91,24 @@ impl Default for TestDeviceCommunicationManagerBuilder {
 
 impl TestDeviceCommunicationManagerBuilder {
   pub fn add_test_device(&mut self, device: &TestDeviceIdentifier) -> TestDeviceChannelHost {
+    self.add_test_device_with_message_gap(device, DEFAULT_MESSAGE_GAP)
+  }
+
+  /// Register a test device whose io task batches commands over `message_gap`.
+  /// A large gap (e.g. 500ms) makes the stop-write batching race deterministic
+  /// for tests asserting that stop commands are flushed before stop resolves.
+  #[allow(dead_code)]
+  pub fn add_test_device_with_message_gap(
+    &mut self,
+    device: &TestDeviceIdentifier,
+    message_gap: Duration,
+  ) -> TestDeviceChannelHost {
     let (host_channel, device_channel) = new_device_channel();
     self
       .devices
       .as_mut()
       .expect("Devices vec does not exist, is this running twice?")
-      .push((device.clone(), device_channel));
+      .push((device.clone(), message_gap, device_channel));
     host_channel
   }
 }
@@ -115,26 +130,28 @@ impl HardwareCommunicationManagerBuilder for TestDeviceCommunicationManagerBuild
 
 fn new_uninitialized_ble_test_device(
   identifier: &TestDeviceIdentifier,
+  message_gap: Duration,
   device_channel: TestDeviceChannelDevice,
 ) -> TestHardwareConnector {
   let address = identifier.address.clone();
   let specifier = ProtocolCommunicationSpecifier::BluetoothLE(
     BluetoothLESpecifier::new_from_device(&identifier.name, &HashMap::new(), &[]),
   );
-  let hardware = TestDevice::new(&identifier.name, &address, device_channel);
+  let hardware =
+    TestDevice::new_with_message_gap(&identifier.name, &address, message_gap, device_channel);
   TestHardwareConnector::new(specifier, hardware)
 }
 
 pub struct TestDeviceCommunicationManager {
   device_sender: Sender<HardwareCommunicationManagerEvent>,
-  devices: Vec<(TestDeviceIdentifier, TestDeviceChannelDevice)>,
+  devices: Vec<(TestDeviceIdentifier, Duration, TestDeviceChannelDevice)>,
   is_scanning: Arc<AtomicBool>,
 }
 
 impl TestDeviceCommunicationManager {
   pub fn new(
     device_sender: Sender<HardwareCommunicationManagerEvent>,
-    devices: Vec<(TestDeviceIdentifier, TestDeviceChannelDevice)>,
+    devices: Vec<(TestDeviceIdentifier, Duration, TestDeviceChannelDevice)>,
   ) -> Self {
     Self {
       device_sender,
@@ -156,8 +173,8 @@ impl HardwareCommunicationManager for TestDeviceCommunicationManager {
 
     let mut events = vec![];
 
-    while let Some((device, test_channel)) = self.devices.pop() {
-      let device_creator = new_uninitialized_ble_test_device(&device, test_channel);
+    while let Some((device, message_gap, test_channel)) = self.devices.pop() {
+      let device_creator = new_uninitialized_ble_test_device(&device, message_gap, test_channel);
 
       events.push(HardwareCommunicationManagerEvent::DeviceFound {
         name: device.name.clone(),
diff --git a/crates/intiface_engine/CHANGELOG.md b/crates/intiface_engine/CHANGELOG.md
index 4f5e5d67f..d764baad7 100644
--- a/crates/intiface_engine/CHANGELOG.md
+++ b/crates/intiface_engine/CHANGELOG.md
@@ -1,3 +1,19 @@
+# 5.0.0 (2026-07-19)
+
+## Breaking Changes
+
+- Added task lifecycle messages to the public engine frontend API:
+  - `EngineMessage::TaskStarted` reports a newly registered task.
+  - `EngineMessage::TaskEnded` reports task completion, cancellation, or panic.
+  - `EngineMessage::TaskList` returns a snapshot of live tasks.
+- Added `IntifaceMessage::RequestTaskList` for requesting a live-task snapshot.
+- Added the required `emit_task_events: bool` field to `EngineOptionsExternal`. Downstream code using struct literals must provide this field (or use `..Default::default()`).
+- Downstream exhaustive matches over `EngineMessage` or `IntifaceMessage` may need new arms for the task lifecycle variants.
+
+## Features
+
+- Task lifecycle event emission is opt-in. Set `emit_task_events` to `true` to receive `TaskStarted` and `TaskEnded` events; task events remain disabled by default.
+
 # 4.0.4 (2026-06-01)
 
 ## Features
diff --git a/crates/intiface_engine/Cargo.toml b/crates/intiface_engine/Cargo.toml
index c0c0754b5..6d4f5423f 100644
--- a/crates/intiface_engine/Cargo.toml
+++ b/crates/intiface_engine/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "intiface-engine"
-version = "4.0.4"
+version = "5.0.0"
 authors = ["Nonpolynomial Labs, LLC <kyle@nonpolynomial.com>"]
 description = "CLI and Library frontend for the Buttplug sex toy control library"
 license = "BSD-3-Clause"