diff --git a/docs/developers-guide.md b/docs/developers-guide.md
index bf87cdfd6..93e03162f 100644
--- a/docs/developers-guide.md
+++ b/docs/developers-guide.md
@@ -1281,6 +1281,51 @@ client that finishes its work and drops its connection produces exactly one of
 those kinds. Every other I/O error is returned from the task instead, so it
 surfaces when the caller joins the handle.
 
+### Fallible assertions in integration tests
+
+Integration tests whose bodies return `Result` cannot use `assert_eq!` and
+friends without tripping `clippy::panic_in_result_fn`, which flags a panic
+inside a function that promises to report failure by returning `Err`. The
+`tests/common/fallible_assertions/` modules provide three small replacement
+checks. Each returns `Result<(), String>`: `Ok(())` when the check holds, and
+`Err` carrying the predicate-specific message when it does not. A test
+propagates a failure with `?` rather than panicking, so the diagnostic reaches
+the harness as an ordinary test failure.
+
+The three checks divide by what they compare:
+
+- `check` (in `check.rs`) takes a boolean predicate and the message to report
+  when it is false. Use it for conditions that are not plain equality, such as
+  a collection containing an expected element or a value lying in a range.
+- `check_eq` (in `check_eq.rs`) compares two `Copy` values and reports
+  `"{message}: expected {expected:?}, got {actual:?}"`. Use it for small scalar
+  values, such as a connection count or an expected byte.
+- `check_equal` (in `check_equal.rs`) compares two borrowed, possibly unsized
+  values (`A: PartialEq<E> + ?Sized`, `E: ?Sized`) and reports the same
+  expected/actual shape. Use it for string slices, structs compared by
+  reference, and other non-`Copy` values.
+
+Each helper lives in its own module because each integration-test crate pulls
+in only the checks it uses, with an explicit path attribute:
+
+```rust
+#[path = "common/fallible_assertions/check_equal.rs"]
+mod fallible_check_equal;
+
+use fallible_check_equal::check_equal;
+```
+
+The helpers are declared `pub(super)`, which resolves to crate visibility
+because the test file is the crate root. That is why the `#[path]` attribute
+must appear in the test file itself rather than inside a submodule: a
+declaration inside a submodule would be visible only to that module, not to the
+sibling test modules that need it.
+
+These helpers exist only for integration tests. Production code and
+non-`Result` unit tests continue to use the standard assertion macros, whose
+panic-on-failure behaviour is exactly right when the test has no result channel
+to report through.
+
 ### `LoggerHandle::Default` and `ObservabilityHandle::Default`
 
 Both `LoggerHandle` (in `wireframe_testing::logging`) and `ObservabilityHandle`
diff --git a/tests/client_lifecycle.rs b/tests/client_lifecycle.rs
index af73b7a11..6596ae894 100644
--- a/tests/client_lifecycle.rs
+++ b/tests/client_lifecycle.rs
@@ -10,33 +10,42 @@
 
 use std::sync::{
     Arc,
-    atomic::{AtomicUsize, Ordering},
+    atomic::{AtomicU32, AtomicUsize, Ordering},
 };
 
+#[path = "common/fallible_assertions/check.rs"]
+mod fallible_check;
+#[path = "common/fallible_assertions/check_eq.rs"]
+mod fallible_check_eq;
+
+use fallible_check::check;
+use fallible_check_eq::check_eq;
 use tokio::net::TcpListener;
-use wireframe::client::WireframeClient;
+use wireframe::{
+    client::WireframeClient,
+    preamble::{read_preamble, write_preamble},
+};
 use wireframe_testing::TestResult;
 
 /// Test that setup and teardown callbacks are both invoked for a full
 /// connection lifecycle.
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn client_setup_and_teardown_callbacks_run() -> TestResult<()> {
     let setup_count = Arc::new(AtomicUsize::new(0));
     let teardown_count = Arc::new(AtomicUsize::new(0));
+    let teardown_state = Arc::new(AtomicU32::new(0));
     let setup = setup_count.clone();
     let teardown = teardown_count.clone();
+    let observed_teardown_state = teardown_state.clone();
 
     let listener = TcpListener::bind("127.0.0.1:0").await?;
     let addr = listener.local_addr()?;
 
     // Server accepts and then closes
     let server = tokio::spawn(async move {
-        let (stream, _) = listener.accept().await.expect("accept");
+        let (stream, _) = listener.accept().await?;
         drop(stream); // Close immediately
+        Ok::<(), wireframe::testkit::TestError>(())
     });
 
     let client = WireframeClient::builder()
@@ -49,38 +58,40 @@ async fn client_setup_and_teardown_callbacks_run() -> TestResult<()> {
         })
         .on_connection_teardown(move |state: u32| {
             let teardown = teardown.clone();
+            let observed_teardown_state = observed_teardown_state.clone();
             async move {
-                assert_eq!(state, 42, "teardown should receive state from setup");
+                observed_teardown_state.store(state, Ordering::SeqCst);
                 teardown.fetch_add(1, Ordering::SeqCst);
             }
         })
         .connect(addr)
         .await?;
 
-    assert_eq!(
+    check_eq(
         setup_count.load(Ordering::SeqCst),
         1,
-        "setup callback should run exactly once"
-    );
+        "setup callback should run exactly once",
+    )?;
 
     client.close().await;
 
-    assert_eq!(
+    check_eq(
+        teardown_state.load(Ordering::SeqCst),
+        42,
+        "teardown should receive state from setup",
+    )?;
+    check_eq(
         teardown_count.load(Ordering::SeqCst),
         1,
-        "teardown callback should run exactly once"
-    );
+        "teardown callback should run exactly once",
+    )?;
 
-    server.await?;
+    server.await??;
     Ok(())
 }
 
 /// Test that the error hook is invoked when the server disconnects unexpectedly.
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn client_error_hook_invoked_on_disconnect() -> TestResult<()> {
     let error_count = Arc::new(AtomicUsize::new(0));
     let count = error_count.clone();
@@ -90,8 +101,9 @@ async fn client_error_hook_invoked_on_disconnect() -> TestResult<()> {
 
     // Server accepts and then closes immediately
     let server = tokio::spawn(async move {
-        let (stream, _) = listener.accept().await.expect("accept");
+        let (stream, _) = listener.accept().await?;
         drop(stream);
+        Ok::<(), wireframe::testkit::TestError>(())
     });
 
     let mut client = WireframeClient::builder()
@@ -104,17 +116,20 @@ async fn client_error_hook_invoked_on_disconnect() -> TestResult<()> {
         .connect(addr)
         .await?;
 
-    server.await?;
+    server.await??;
 
     // Try to receive - should fail and invoke error hook
     let result: Result<Vec<u8>, wireframe::client::ClientError> = client.receive().await;
-    assert!(result.is_err(), "receive should fail after disconnect");
+    check(
+        result.is_err(),
+        format!("receive should fail after disconnect, got {result:?}"),
+    )?;
 
-    assert_eq!(
+    check_eq(
         error_count.load(Ordering::SeqCst),
         1,
-        "error callback should be invoked on disconnect"
-    );
+        "error callback should be invoked on disconnect",
+    )?;
 
     Ok(())
 }
@@ -129,15 +144,18 @@ struct ServerAck {
     accepted: bool,
 }
 
+async fn serve_preamble_ack(listener: TcpListener) -> TestResult<()> {
+    let (mut stream, _) = listener.accept().await?;
+    let (_hello, _) = read_preamble::<_, ClientHello>(&mut stream).await?;
+    write_preamble(&mut stream, &ServerAck { accepted: true }).await?;
+    drop(stream);
+    Ok(())
+}
+
 /// Test that lifecycle hooks can be combined with preamble callbacks.
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn client_lifecycle_hooks_work_with_preamble() -> TestResult<()> {
     use futures::FutureExt;
-    use wireframe::preamble::{read_preamble, write_preamble};
 
     let setup_count = Arc::new(AtomicUsize::new(0));
     let teardown_count = Arc::new(AtomicUsize::new(0));
@@ -149,17 +167,7 @@ async fn client_lifecycle_hooks_work_with_preamble() -> TestResult<()> {
     let listener = TcpListener::bind("127.0.0.1:0").await?;
     let addr = listener.local_addr()?;
 
-    // Server reads preamble and sends ack
-    let server = tokio::spawn(async move {
-        let (mut stream, _) = listener.accept().await.expect("accept");
-        let (_hello, _) = read_preamble::<_, ClientHello>(&mut stream)
-            .await
-            .expect("read preamble");
-        write_preamble(&mut stream, &ServerAck { accepted: true })
-            .await
-            .expect("write ack");
-        drop(stream);
-    });
+    let server = tokio::spawn(serve_preamble_ack(listener));
 
     let client = WireframeClient::builder()
         .with_preamble(ClientHello { version: 1 })
@@ -170,7 +178,12 @@ async fn client_lifecycle_hooks_work_with_preamble() -> TestResult<()> {
                 let (ack, leftover) = read_preamble::<_, ServerAck>(stream).await.map_err(|e| {
                     std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string())
                 })?;
-                assert!(ack.accepted, "server should accept preamble");
+                if !ack.accepted {
+                    return Err(std::io::Error::new(
+                        std::io::ErrorKind::InvalidData,
+                        "server should accept preamble",
+                    ));
+                }
                 Ok(leftover)
             }
             .boxed()
@@ -191,26 +204,26 @@ async fn client_lifecycle_hooks_work_with_preamble() -> TestResult<()> {
         .connect(addr)
         .await?;
 
-    server.await?;
+    server.await??;
 
-    assert_eq!(
+    check_eq(
         preamble_count.load(Ordering::SeqCst),
         1,
-        "preamble callback should run"
-    );
-    assert_eq!(
+        "preamble callback should run",
+    )?;
+    check_eq(
         setup_count.load(Ordering::SeqCst),
         1,
-        "setup callback should run"
-    );
+        "setup callback should run",
+    )?;
 
     client.close().await;
 
-    assert_eq!(
+    check_eq(
         teardown_count.load(Ordering::SeqCst),
         1,
-        "teardown callback should run"
-    );
+        "teardown callback should run",
+    )?;
 
     Ok(())
 }
diff --git a/tests/common/fallible_assertions/check.rs b/tests/common/fallible_assertions/check.rs
new file mode 100644
index 000000000..f9bb6ec0e
--- /dev/null
+++ b/tests/common/fallible_assertions/check.rs
@@ -0,0 +1,23 @@
+//! Fallible predicate checks shared by integration-test crates.
+
+/// Return an error containing `message` when a predicate is false.
+///
+/// # Examples
+///
+/// ```rust,ignore
+/// let successful = check(true, "the route is registered");
+/// let failed = check(false, "the route is missing");
+/// // `successful` is `Ok(())`; `failed` contains the supplied message.
+/// ```
+///
+/// # Errors
+///
+/// Returns `Err` with the supplied message when `condition` is false.
+pub(super) fn check(condition: bool, message: impl Into<String>) -> Result<(), String> {
+    if condition {
+        Ok(())
+    } else {
+        let message = message.into();
+        Err(message)
+    }
+}
diff --git a/tests/common/fallible_assertions/check_eq.rs b/tests/common/fallible_assertions/check_eq.rs
new file mode 100644
index 000000000..89325fef0
--- /dev/null
+++ b/tests/common/fallible_assertions/check_eq.rs
@@ -0,0 +1,26 @@
+//! Fallible equality checks for copyable values in integration-test crates.
+
+/// Compare two values and report their context and debug representations.
+///
+/// # Examples
+///
+/// ```rust,ignore
+/// let matched = check_eq(3, 3, "item count");
+/// let mismatched = check_eq(2, 3, "item count");
+/// // `matched` is `Ok(())`; the error says `item count: expected 3, got 2`.
+/// ```
+///
+/// # Errors
+///
+/// Returns `Err` when the values differ, formatted as
+/// `"{message}: expected {expected:?}, got {actual:?}"`.
+pub(super) fn check_eq<T>(actual: T, expected: T, message: &str) -> Result<(), String>
+where
+    T: std::fmt::Debug + PartialEq + Copy,
+{
+    if actual == expected {
+        Ok(())
+    } else {
+        Err(format!("{message}: expected {expected:?}, got {actual:?}"))
+    }
+}
diff --git a/tests/common/fallible_assertions/check_equal.rs b/tests/common/fallible_assertions/check_equal.rs
new file mode 100644
index 000000000..49aff2e33
--- /dev/null
+++ b/tests/common/fallible_assertions/check_equal.rs
@@ -0,0 +1,27 @@
+//! Fallible equality checks for borrowed values in integration-test crates.
+
+/// Compare possibly unsized values and report their context on mismatch.
+///
+/// # Examples
+///
+/// ```rust,ignore
+/// let matched = check_equal(&"ready", &"ready", "state");
+/// let mismatched = check_equal(&"waiting", &"ready", "state");
+/// // `matched` is `Ok(())`; the error says `state: expected \"ready\", got \"waiting\"`.
+/// ```
+///
+/// # Errors
+///
+/// Returns `Err` when `actual` does not equal `expected`, formatted with their
+/// debug representations and `context`.
+pub(super) fn check_equal<A, E>(actual: &A, expected: &E, context: &str) -> Result<(), String>
+where
+    A: std::fmt::Debug + PartialEq<E> + ?Sized,
+    E: std::fmt::Debug + ?Sized,
+{
+    if actual == expected {
+        Ok(())
+    } else {
+        Err(format!("{context}: expected {expected:?}, got {actual:?}"))
+    }
+}
diff --git a/tests/connection.rs b/tests/connection.rs
index f8d70013e..996bcf32e 100644
--- a/tests/connection.rs
+++ b/tests/connection.rs
@@ -1,559 +1,17 @@
 //! Integration tests for connection actor queue helpers and multi-packet handling.
+//!
+//! Scenarios are grouped by concern under the `connection/` module directory:
+//! - Shared counters, harness configuration, and assertions (`support.rs`)
+//! - Multi-packet forwarding, closed handling, and drains (`multi_packet.rs`)
+//! - Closure-reason logging (`logs.rs`)
+//! - Queue polling and lifecycle state (`state.rs`)
 #![cfg(not(loom))]
 
-use std::sync::{
-    Arc,
-    atomic::{AtomicUsize, Ordering},
-};
-
-use log::Level;
-use rstest::{fixture, rstest};
-use serial_test::serial;
-use tokio::sync::mpsc;
-use wireframe::{
-    connection::test_support::{ActorHarness, ActorStateHarness, poll_queue_next},
-    hooks::{ConnectionContext, ProtocolHooks},
-};
-use wireframe_testing::{LoggerHandle, TestResult, logger};
-
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-struct HookCounts {
-    before: usize,
-    end: usize,
-}
-
-impl HookCounts {
-    /// Snapshot hook counters from shared state.
-    fn from_counters(before: &Arc<AtomicUsize>, end: Option<&Arc<AtomicUsize>>) -> Self {
-        let end = end.map_or(0, |counter| counter.load(Ordering::SeqCst));
-        Self {
-            before: before.load(Ordering::SeqCst),
-            end,
-        }
-    }
-}
-
-/// Shared counters for protocol hook configurations used in connection tests.
-#[derive(Clone)]
-struct HookCounters {
-    before_calls: Arc<AtomicUsize>,
-    end_calls: Arc<AtomicUsize>,
-}
-
-impl HookCounters {
-    /// Create counters tracking before-send and on-command-end hook invocations.
-    fn new() -> Self {
-        Self {
-            before_calls: Arc::new(AtomicUsize::new(0)),
-            end_calls: Arc::new(AtomicUsize::new(0)),
-        }
-    }
-
-    /// Construct protocol hooks that increment frames and record hook usage.
-    fn build_hooks_with_increment(
-        &self,
-        increment: u8,
-        stream_end_value: impl Fn(&mut ConnectionContext) -> Option<u8> + Send + Sync + 'static,
-    ) -> ProtocolHooks<u8, wireframe::NoProtocolError> {
-        let before_clone = Arc::clone(&self.before_calls);
-        let end_clone = Arc::clone(&self.end_calls);
-
-        ProtocolHooks {
-            before_send: Some(Box::new(
-                move |frame: &mut u8, _ctx: &mut ConnectionContext| {
-                    before_clone.fetch_add(1, Ordering::SeqCst);
-                    *frame += increment;
-                },
-            )),
-            on_command_end: Some(Box::new(move |_ctx: &mut ConnectionContext| {
-                end_clone.fetch_add(1, Ordering::SeqCst);
-            })
-                as Box<dyn FnMut(&mut ConnectionContext) + Send + 'static>),
-            stream_end: Some(Box::new(stream_end_value)),
-            ..ProtocolHooks::<u8, wireframe::NoProtocolError>::default()
-        }
-    }
-
-    /// Snapshot the recorded hook counts for verification.
-    fn get_counts(&self) -> HookCounts {
-        HookCounts::from_counters(&self.before_calls, Some(&self.end_calls))
-    }
-}
-
-type StreamEndHook = Box<dyn Fn(&mut ConnectionContext) -> Option<u8> + Send + Sync + 'static>;
-
-/// Configuration for building connection actor harnesses.
-#[derive(Default)]
-struct HarnessConfig {
-    has_response: bool,
-    has_multi_packet: bool,
-    increment: u8,
-    stream_end_value: Option<StreamEndHook>,
-}
-
-impl HarnessConfig {
-    /// Create a harness configuration with the default increment.
-    fn new() -> Self {
-        Self {
-            increment: 1,
-            ..Self::default()
-        }
-    }
-
-    /// Enable response tracking for the harness under construction.
-    #[expect(
-        dead_code,
-        reason = "Harness builders retain response toggles for future scenarios."
-    )]
-    fn with_response(mut self) -> Self {
-        self.has_response = true;
-        self
-    }
-
-    /// Enable multi-packet support for the harness under construction.
-    fn with_multi_packet(mut self) -> Self {
-        self.has_multi_packet = true;
-        self
-    }
-
-    /// Override the amount added to each forwarded frame.
-    fn with_increment(mut self, increment: u8) -> Self {
-        self.increment = increment;
-        self
-    }
-
-    /// Provide a stream-end hook for the harness under construction.
-    fn with_stream_end<F>(mut self, f: F) -> Self
-    where
-        F: Fn(&mut ConnectionContext) -> Option<u8> + Send + Sync + 'static,
-    {
-        self.stream_end_value = Some(Box::new(f));
-        self
-    }
-}
-
-#[derive(Clone)]
-struct HarnessFactory {
-    counters: HookCounters,
-}
-
-impl HarnessFactory {
-    /// Build a connection actor harness with shared hook counters.
-    fn create(
-        &self,
-        config: HarnessConfig,
-    ) -> Result<ActorHarness, wireframe::push::PushConfigError> {
-        let HarnessConfig {
-            has_response,
-            has_multi_packet,
-            increment,
-            stream_end_value,
-        } = config;
-        let stream_end_fn =
-            stream_end_value.unwrap_or_else(|| Box::new(|_: &mut ConnectionContext| None));
-        let hooks = self
-            .counters
-            .build_hooks_with_increment(increment, stream_end_fn);
-        ActorHarness::new_with_state(hooks, has_response, has_multi_packet)
-    }
-
-    /// Read the accumulated hook counters for the most recent harness.
-    fn counts(&self) -> HookCounts { self.counters.get_counts() }
-}
-
-#[fixture]
-fn hook_counters() -> HookCounters {
-    let counters = HookCounters::new();
-    debug_assert_eq!(counters.get_counts(), HookCounts { before: 0, end: 0 });
-    counters
-}
-
-#[fixture]
-fn harness_factory(hook_counters: HookCounters) -> HarnessFactory {
-    HarnessFactory {
-        counters: hook_counters,
-    }
-}
-
-fn assert_frame_processed(
-    out: &[u8],
-    expected: &[u8],
-    expected_counts: HookCounts,
-    actual_counts: HookCounts,
-) {
-    assert_eq!(out, expected, "frames should match expected output");
-    assert_eq!(actual_counts, expected_counts, "hook counts should match");
-}
-
-/// Helper to verify common multi-packet processing assertions.
-fn assert_multi_packet_processing_result(
-    harness: &ActorHarness,
-    harness_factory: &HarnessFactory,
-    expected_output: &[u8],
-    expected_counts: HookCounts,
-) {
-    let snapshot = harness.snapshot();
-    assert!(snapshot.is_active && !snapshot.is_shutting_down && !snapshot.is_done);
-    assert_frame_processed(
-        &harness.out,
-        expected_output,
-        expected_counts,
-        harness_factory.counts(),
-    );
-}
-
-fn assert_reason_logged(
-    logger: &mut LoggerHandle,
-    expected_level: Level,
-    expected_reason: &str,
-    expected_correlation: Option<u64>,
-) {
-    let expected_correlation = format!("correlation_id={expected_correlation:?}");
-    let mut found = false;
-    while let Some(record) = logger.pop() {
-        let message = record.args().to_string();
-        if !message.contains("multi-packet stream closed") {
-            continue;
-        }
-        if !message.contains(&expected_correlation) {
-            continue;
-        }
-        assert_eq!(
-            record.level(),
-            expected_level,
-            "unexpected log level for closure: message={message}",
-        );
-        assert!(
-            message.contains(&format!("reason={expected_reason}")),
-            "closure log missing reason: message={message}",
-        );
-        found = true;
-        break;
-    }
-    assert!(found, "multi-packet closure log not found");
-}
-
-#[rstest]
-fn process_multi_packet_forwards_frame(harness_factory: HarnessFactory) -> TestResult {
-    let mut harness = harness_factory.create(HarnessConfig::new())?;
-    harness.process_multi_packet(Some(5));
-
-    assert_multi_packet_processing_result(
-        &harness,
-        &harness_factory,
-        &[6],
-        HookCounts { before: 1, end: 0 },
-    );
-    Ok(())
-}
-
-#[rstest]
-fn process_multi_packet_none_emits_end_frame(harness_factory: HarnessFactory) -> TestResult {
-    let mut harness = harness_factory.create(
-        HarnessConfig::new()
-            .with_multi_packet()
-            .with_increment(2)
-            .with_stream_end(|_| Some(9)),
-    )?;
-    let (_tx, rx) = mpsc::channel(1);
-    harness.set_multi_queue(Some(rx))?;
-
-    harness.process_multi_packet(None);
-
-    assert_multi_packet_processing_result(
-        &harness,
-        &harness_factory,
-        &[11],
-        HookCounts { before: 1, end: 1 },
-    );
-    Ok(())
-}
-
-#[rstest(
-    terminator,
-    expected_output,
-    expected_before,
-    case::with_terminator(Some(5), vec![6], 1),
-    case::without_terminator(None, Vec::new(), 0),
-)]
-fn handle_multi_packet_closed_behaviour(
-    harness_factory: HarnessFactory,
-    terminator: Option<u8>,
-    expected_output: Vec<u8>,
-    expected_before: usize,
-) -> TestResult {
-    let mut harness = harness_factory.create(
-        HarnessConfig::new()
-            .with_multi_packet()
-            .with_stream_end(move |_| terminator),
-    )?;
-    let (_tx, rx) = mpsc::channel(1);
-    harness.set_multi_queue(Some(rx))?;
-
-    harness.handle_multi_packet_closed();
-
-    let snapshot = harness.snapshot();
-    if !snapshot.is_active {
-        return Err("connection should be active".into());
-    }
-    if snapshot.is_shutting_down {
-        return Err("connection should not be shutting down".into());
-    }
-    if snapshot.is_done {
-        return Err("connection should not be done".into());
-    }
-    if harness.has_multi_queue() {
-        return Err("multi-packet channel should be cleared".into());
-    }
-    assert_frame_processed(
-        &harness.out,
-        &expected_output,
-        HookCounts {
-            before: expected_before,
-            end: 1,
-        },
-        harness_factory.counts(),
-    );
-    Ok(())
-}
-
-#[rstest]
-fn try_opportunistic_drain_forwards_frame(harness_factory: HarnessFactory) -> TestResult {
-    let mut harness = harness_factory.create(HarnessConfig::new())?;
-    let (tx, rx) = mpsc::channel(1);
-    tx.try_send(9)?;
-    drop(tx);
-    harness.set_low_queue(Some(rx));
-
-    let drained = harness.try_drain_low();
-
-    if !drained {
-        return Err("queue should report a drained frame".into());
-    }
-    if !harness.has_low_queue() {
-        return Err("queue remains available".into());
-    }
-    assert_frame_processed(
-        &harness.out,
-        &[10],
-        HookCounts { before: 1, end: 0 },
-        harness_factory.counts(),
-    );
-    Ok(())
-}
-
-#[rstest]
-#[serial(connection_logs)]
-fn handle_multi_packet_closed_logs_reason(
-    harness_factory: HarnessFactory,
-    mut logger: LoggerHandle,
-) -> TestResult {
-    logger.clear();
-    let mut harness = harness_factory.create(
-        HarnessConfig::new()
-            .with_multi_packet()
-            .with_stream_end(|_| Some(5)),
-    )?;
-    let (_tx, rx) = mpsc::channel(1);
-    harness
-        .actor_mut()
-        .set_multi_packet_with_correlation(Some(rx), Some(11))?;
-    logger.clear();
-    harness.handle_multi_packet_closed();
-    assert_reason_logged(&mut logger, Level::Info, "drained", Some(11));
-    Ok(())
-}
-
-#[rstest]
-#[serial(connection_logs)]
-fn try_opportunistic_drain_multi_disconnect_logs_reason(
-    harness_factory: HarnessFactory,
-    mut logger: LoggerHandle,
-) -> TestResult {
-    logger.clear();
-    let mut harness = harness_factory.create(
-        HarnessConfig::new()
-            .with_multi_packet()
-            .with_stream_end(|_| Some(5)),
-    )?;
-    let (tx, rx) = mpsc::channel(1);
-    harness
-        .actor_mut()
-        .set_multi_packet_with_correlation(Some(rx), Some(12))?;
-    drop(tx);
-    logger.clear();
-    let drained = harness.try_drain_multi();
-    if drained {
-        return Err("disconnect should not report a drained frame".into());
-    }
-    assert_reason_logged(&mut logger, Level::Warn, "disconnected", Some(12));
-    Ok(())
-}
-
-#[rstest]
-#[serial(connection_logs)]
-fn start_shutdown_logs_reason(
-    harness_factory: HarnessFactory,
-    mut logger: LoggerHandle,
-) -> TestResult {
-    logger.clear();
-    let mut harness = harness_factory.create(
-        HarnessConfig::new()
-            .with_multi_packet()
-            .with_stream_end(|_| Some(5)),
-    )?;
-    let (_tx, rx) = mpsc::channel(1);
-    harness
-        .actor_mut()
-        .set_multi_packet_with_correlation(Some(rx), Some(13))?;
-    logger.clear();
-    harness.start_shutdown();
-    assert_reason_logged(&mut logger, Level::Info, "shutdown", Some(13));
-    if harness.has_multi_queue() {
-        return Err("multi-packet queue should be cleared after shutdown".into());
-    }
-    Ok(())
-}
-
-#[rstest]
-fn try_opportunistic_drain_multi_disconnect_emits_terminator(
-    harness_factory: HarnessFactory,
-) -> TestResult {
-    let mut harness = harness_factory.create(
-        HarnessConfig::new()
-            .with_multi_packet()
-            .with_stream_end(|_| Some(5)),
-    )?;
-    let (tx, rx) = mpsc::channel(1);
-    harness.set_multi_queue(Some(rx))?;
-    drop(tx);
-
-    let drained = harness.try_drain_multi();
-
-    if drained {
-        return Err("disconnect should not report a drained frame".into());
-    }
-    if harness.has_multi_queue() {
-        return Err("multi-packet queue should be cleared after disconnect".into());
-    }
-    assert_frame_processed(
-        &harness.out,
-        &[6],
-        HookCounts { before: 1, end: 1 },
-        harness_factory.counts(),
-    );
-    Ok(())
-}
-
-#[test]
-fn try_opportunistic_drain_returns_false_when_empty() -> TestResult {
-    let mut harness = ActorHarness::new()?;
-    let (_tx, rx) = mpsc::channel(1);
-    harness.set_low_queue(Some(rx));
-
-    let drained = harness.try_drain_low();
-
-    if drained {
-        return Err("no frame should be drained".into());
-    }
-    if !harness.has_low_queue() {
-        return Err("queue should remain available".into());
-    }
-    if !harness.out.is_empty() {
-        return Err("no frames should be emitted".into());
-    }
-    Ok(())
-}
-
-#[test]
-fn try_opportunistic_drain_handles_disconnect() -> TestResult {
-    let mut harness = ActorHarness::new()?;
-    let (tx, rx) = mpsc::channel(1);
-    harness.set_low_queue(Some(rx));
-    drop(tx);
-
-    let drained = harness.try_drain_low();
-
-    if drained {
-        return Err("disconnect should not produce a frame".into());
-    }
-    if harness.has_low_queue() {
-        return Err("queue should be cleared after disconnect".into());
-    }
-    let snapshot = harness.snapshot();
-    if !snapshot.is_active {
-        return Err("connection should be active".into());
-    }
-    if snapshot.is_shutting_down {
-        return Err("connection should not be shutting down".into());
-    }
-    if snapshot.is_done {
-        return Err("connection should not be done".into());
-    }
-    Ok(())
-}
-
-#[tokio::test]
-async fn poll_queue_reads_frame() {
-    let (tx, mut rx) = mpsc::channel(1);
-    tx.send(42).await.expect("send frame");
-
-    let value = poll_queue_next(Some(&mut rx)).await;
-
-    assert_eq!(value, Some(42));
-}
-
-#[tokio::test]
-async fn poll_queue_returns_none_for_absent_receiver() {
-    let value = poll_queue_next(None).await;
-    assert!(value.is_none());
-}
-
-#[tokio::test]
-async fn poll_queue_returns_none_after_close() {
-    let (tx, mut rx) = mpsc::channel(1);
-    drop(tx);
-
-    let value = poll_queue_next(Some(&mut rx)).await;
-
-    assert!(value.is_none());
-}
-
-#[test]
-fn actor_state_reports_shutting_down_after_start_shutdown() {
-    // `is_shutting_down` is only ever asserted false elsewhere; drive the
-    // transition and assert the positive polarity so a constant-`false`
-    // mutant cannot survive.
-    let mut harness = ActorStateHarness::new(false, false);
-    assert!(!harness.snapshot().is_shutting_down, "should start active");
-
-    harness.start_shutdown();
-    let snapshot = harness.snapshot();
-    assert!(snapshot.is_shutting_down, "should be shutting down");
-    assert!(!snapshot.is_active, "shutting down is not active");
-    assert!(!snapshot.is_done, "shutting down is not done");
-}
-
-#[rstest(
-    has_multi,
-    expected_marks,
-    case::with_multi(true, 3),
-    case::without_multi(false, 2)
-)]
-fn actor_state_tracks_sources(has_multi: bool, expected_marks: usize) {
-    let mut harness = ActorStateHarness::new(false, has_multi);
-    let snapshot = harness.snapshot();
-    assert!(snapshot.is_active && !snapshot.is_shutting_down && !snapshot.is_done);
-
-    for _ in 0..expected_marks.saturating_sub(1) {
-        harness.mark_closed();
-        let snapshot = harness.snapshot();
-        assert!(snapshot.is_active && !snapshot.is_shutting_down && !snapshot.is_done);
-    }
-
-    harness.mark_closed();
-    let final_snapshot = harness.snapshot();
-    assert!(
-        !final_snapshot.is_active && !final_snapshot.is_shutting_down && final_snapshot.is_done
-    );
-}
+#[path = "connection/logs.rs"]
+mod logs;
+#[path = "connection/multi_packet.rs"]
+mod multi_packet;
+#[path = "connection/state.rs"]
+mod state;
+#[path = "connection/support.rs"]
+mod support;
diff --git a/tests/connection/logs.rs b/tests/connection/logs.rs
new file mode 100644
index 000000000..b58d29dbe
--- /dev/null
+++ b/tests/connection/logs.rs
@@ -0,0 +1,82 @@
+//! Closure-reason logging tests for multi-packet queue drains and shutdown.
+
+use log::Level;
+use rstest::rstest;
+use serial_test::serial;
+use tokio::sync::mpsc;
+use wireframe_testing::{LoggerHandle, TestResult, logger};
+
+use crate::support::{HarnessConfig, HarnessFactory, assert_reason_logged, harness_factory};
+
+#[rstest]
+#[serial(connection_logs)]
+fn handle_multi_packet_closed_logs_reason(
+    harness_factory: HarnessFactory,
+    mut logger: LoggerHandle,
+) -> TestResult {
+    logger.clear();
+    let mut harness = harness_factory.create(
+        HarnessConfig::new()
+            .with_multi_packet()
+            .with_stream_end(|_| Some(5)),
+    )?;
+    let (_tx, rx) = mpsc::channel(1);
+    harness
+        .actor_mut()
+        .set_multi_packet_with_correlation(Some(rx), Some(11))?;
+    logger.clear();
+    harness.handle_multi_packet_closed();
+    assert_reason_logged(&mut logger, Level::Info, "drained", Some(11));
+    Ok(())
+}
+
+#[rstest]
+#[serial(connection_logs)]
+fn try_opportunistic_drain_multi_disconnect_logs_reason(
+    harness_factory: HarnessFactory,
+    mut logger: LoggerHandle,
+) -> TestResult {
+    logger.clear();
+    let mut harness = harness_factory.create(
+        HarnessConfig::new()
+            .with_multi_packet()
+            .with_stream_end(|_| Some(5)),
+    )?;
+    let (tx, rx) = mpsc::channel(1);
+    harness
+        .actor_mut()
+        .set_multi_packet_with_correlation(Some(rx), Some(12))?;
+    drop(tx);
+    logger.clear();
+    let drained = harness.try_drain_multi();
+    if drained {
+        return Err("disconnect should not report a drained frame".into());
+    }
+    assert_reason_logged(&mut logger, Level::Warn, "disconnected", Some(12));
+    Ok(())
+}
+
+#[rstest]
+#[serial(connection_logs)]
+fn start_shutdown_logs_reason(
+    harness_factory: HarnessFactory,
+    mut logger: LoggerHandle,
+) -> TestResult {
+    logger.clear();
+    let mut harness = harness_factory.create(
+        HarnessConfig::new()
+            .with_multi_packet()
+            .with_stream_end(|_| Some(5)),
+    )?;
+    let (_tx, rx) = mpsc::channel(1);
+    harness
+        .actor_mut()
+        .set_multi_packet_with_correlation(Some(rx), Some(13))?;
+    logger.clear();
+    harness.start_shutdown();
+    assert_reason_logged(&mut logger, Level::Info, "shutdown", Some(13));
+    if harness.has_multi_queue() {
+        return Err("multi-packet queue should be cleared after shutdown".into());
+    }
+    Ok(())
+}
diff --git a/tests/connection/multi_packet.rs b/tests/connection/multi_packet.rs
new file mode 100644
index 000000000..9fb8fc9d9
--- /dev/null
+++ b/tests/connection/multi_packet.rs
@@ -0,0 +1,201 @@
+//! Multi-packet queue tests: forwarding, closed handling, and drain behaviour.
+
+use rstest::rstest;
+use tokio::sync::mpsc;
+use wireframe_testing::TestResult;
+
+use crate::support::{
+    HarnessConfig,
+    HarnessFactory,
+    HookCounts,
+    assert_frame_processed,
+    assert_multi_packet_processing_result,
+    harness_factory,
+};
+
+#[rstest]
+fn process_multi_packet_forwards_frame(harness_factory: HarnessFactory) -> TestResult {
+    let mut harness = harness_factory.create(HarnessConfig::new())?;
+    harness.process_multi_packet(Some(5));
+
+    assert_multi_packet_processing_result(
+        &harness,
+        &harness_factory,
+        &[6],
+        HookCounts { before: 1, end: 0 },
+    );
+    Ok(())
+}
+
+#[rstest]
+fn process_multi_packet_none_emits_end_frame(harness_factory: HarnessFactory) -> TestResult {
+    let mut harness = harness_factory.create(
+        HarnessConfig::new()
+            .with_multi_packet()
+            .with_increment(2)
+            .with_stream_end(|_| Some(9)),
+    )?;
+    let (_tx, rx) = mpsc::channel(1);
+    harness.set_multi_queue(Some(rx))?;
+
+    harness.process_multi_packet(None);
+
+    assert_multi_packet_processing_result(
+        &harness,
+        &harness_factory,
+        &[11],
+        HookCounts { before: 1, end: 1 },
+    );
+    Ok(())
+}
+
+#[rstest(
+    terminator,
+    expected_output,
+    expected_before,
+    case::with_terminator(Some(5), vec![6], 1),
+    case::without_terminator(None, Vec::new(), 0),
+)]
+fn handle_multi_packet_closed_behaviour(
+    harness_factory: HarnessFactory,
+    terminator: Option<u8>,
+    expected_output: Vec<u8>,
+    expected_before: usize,
+) -> TestResult {
+    let mut harness = harness_factory.create(
+        HarnessConfig::new()
+            .with_multi_packet()
+            .with_stream_end(move |_| terminator),
+    )?;
+    let (_tx, rx) = mpsc::channel(1);
+    harness.set_multi_queue(Some(rx))?;
+
+    harness.handle_multi_packet_closed();
+
+    let snapshot = harness.snapshot();
+    if !snapshot.is_active {
+        return Err("connection should be active".into());
+    }
+    if snapshot.is_shutting_down {
+        return Err("connection should not be shutting down".into());
+    }
+    if snapshot.is_done {
+        return Err("connection should not be done".into());
+    }
+    if harness.has_multi_queue() {
+        return Err("multi-packet channel should be cleared".into());
+    }
+    assert_frame_processed(
+        &harness.out,
+        &expected_output,
+        HookCounts {
+            before: expected_before,
+            end: 1,
+        },
+        harness_factory.counts(),
+    );
+    Ok(())
+}
+
+#[rstest]
+fn try_opportunistic_drain_forwards_frame(harness_factory: HarnessFactory) -> TestResult {
+    let mut harness = harness_factory.create(HarnessConfig::new())?;
+    let (tx, rx) = mpsc::channel(1);
+    tx.try_send(9)?;
+    drop(tx);
+    harness.set_low_queue(Some(rx));
+
+    let drained = harness.try_drain_low();
+
+    if !drained {
+        return Err("queue should report a drained frame".into());
+    }
+    if !harness.has_low_queue() {
+        return Err("queue remains available".into());
+    }
+    assert_frame_processed(
+        &harness.out,
+        &[10],
+        HookCounts { before: 1, end: 0 },
+        harness_factory.counts(),
+    );
+    Ok(())
+}
+
+#[rstest]
+fn try_opportunistic_drain_multi_disconnect_emits_terminator(
+    harness_factory: HarnessFactory,
+) -> TestResult {
+    let mut harness = harness_factory.create(
+        HarnessConfig::new()
+            .with_multi_packet()
+            .with_stream_end(|_| Some(5)),
+    )?;
+    let (tx, rx) = mpsc::channel(1);
+    harness.set_multi_queue(Some(rx))?;
+    drop(tx);
+
+    let drained = harness.try_drain_multi();
+
+    if drained {
+        return Err("disconnect should not report a drained frame".into());
+    }
+    if harness.has_multi_queue() {
+        return Err("multi-packet queue should be cleared after disconnect".into());
+    }
+    assert_frame_processed(
+        &harness.out,
+        &[6],
+        HookCounts { before: 1, end: 1 },
+        harness_factory.counts(),
+    );
+    Ok(())
+}
+
+#[test]
+fn try_opportunistic_drain_returns_false_when_empty() -> TestResult {
+    let mut harness = wireframe::connection::test_support::ActorHarness::new()?;
+    let (_tx, rx) = mpsc::channel(1);
+    harness.set_low_queue(Some(rx));
+
+    let drained = harness.try_drain_low();
+
+    if drained {
+        return Err("no frame should be drained".into());
+    }
+    if !harness.has_low_queue() {
+        return Err("queue should remain available".into());
+    }
+    if !harness.out.is_empty() {
+        return Err("no frames should be emitted".into());
+    }
+    Ok(())
+}
+
+#[test]
+fn try_opportunistic_drain_handles_disconnect() -> TestResult {
+    let mut harness = wireframe::connection::test_support::ActorHarness::new()?;
+    let (tx, rx) = mpsc::channel(1);
+    harness.set_low_queue(Some(rx));
+    drop(tx);
+
+    let drained = harness.try_drain_low();
+
+    if drained {
+        return Err("disconnect should not produce a frame".into());
+    }
+    if harness.has_low_queue() {
+        return Err("queue should be cleared after disconnect".into());
+    }
+    let snapshot = harness.snapshot();
+    if !snapshot.is_active {
+        return Err("connection should be active".into());
+    }
+    if snapshot.is_shutting_down {
+        return Err("connection should not be shutting down".into());
+    }
+    if snapshot.is_done {
+        return Err("connection should not be done".into());
+    }
+    Ok(())
+}
diff --git a/tests/connection/state.rs b/tests/connection/state.rs
new file mode 100644
index 000000000..be9f196e3
--- /dev/null
+++ b/tests/connection/state.rs
@@ -0,0 +1,70 @@
+//! Queue polling and actor lifecycle-state tests.
+
+use rstest::rstest;
+use tokio::sync::mpsc;
+use wireframe::connection::test_support::{ActorStateHarness, poll_queue_next};
+
+#[tokio::test]
+async fn poll_queue_reads_frame() {
+    let (tx, mut rx) = mpsc::channel(1);
+    tx.send(42).await.expect("send frame");
+
+    let value = poll_queue_next(Some(&mut rx)).await;
+
+    assert_eq!(value, Some(42));
+}
+
+#[tokio::test]
+async fn poll_queue_returns_none_for_absent_receiver() {
+    let value = poll_queue_next(None).await;
+    assert!(value.is_none());
+}
+
+#[tokio::test]
+async fn poll_queue_returns_none_after_close() {
+    let (tx, mut rx) = mpsc::channel(1);
+    drop(tx);
+
+    let value = poll_queue_next(Some(&mut rx)).await;
+
+    assert!(value.is_none());
+}
+
+#[test]
+fn actor_state_reports_shutting_down_after_start_shutdown() {
+    // `is_shutting_down` is only ever asserted false elsewhere; drive the
+    // transition and assert the positive polarity so a constant-`false`
+    // mutant cannot survive.
+    let mut harness = ActorStateHarness::new(false, false);
+    assert!(!harness.snapshot().is_shutting_down, "should start active");
+
+    harness.start_shutdown();
+    let snapshot = harness.snapshot();
+    assert!(snapshot.is_shutting_down, "should be shutting down");
+    assert!(!snapshot.is_active, "shutting down is not active");
+    assert!(!snapshot.is_done, "shutting down is not done");
+}
+
+#[rstest(
+    has_multi,
+    expected_marks,
+    case::with_multi(true, 3),
+    case::without_multi(false, 2)
+)]
+fn actor_state_tracks_sources(has_multi: bool, expected_marks: usize) {
+    let mut harness = ActorStateHarness::new(false, has_multi);
+    let snapshot = harness.snapshot();
+    assert!(snapshot.is_active && !snapshot.is_shutting_down && !snapshot.is_done);
+
+    for _ in 0..expected_marks.saturating_sub(1) {
+        harness.mark_closed();
+        let snapshot = harness.snapshot();
+        assert!(snapshot.is_active && !snapshot.is_shutting_down && !snapshot.is_done);
+    }
+
+    harness.mark_closed();
+    let final_snapshot = harness.snapshot();
+    assert!(
+        !final_snapshot.is_active && !final_snapshot.is_shutting_down && final_snapshot.is_done
+    );
+}
diff --git a/tests/connection/support.rs b/tests/connection/support.rs
new file mode 100644
index 000000000..4cafd8e48
--- /dev/null
+++ b/tests/connection/support.rs
@@ -0,0 +1,221 @@
+//! Shared counters, harness configuration, and assertions for connection tests.
+
+use std::sync::{
+    Arc,
+    atomic::{AtomicUsize, Ordering},
+};
+
+use log::Level;
+use rstest::fixture;
+use wireframe::{
+    connection::test_support::ActorHarness,
+    hooks::{ConnectionContext, ProtocolHooks},
+};
+use wireframe_testing::LoggerHandle;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) struct HookCounts {
+    pub(crate) before: usize,
+    pub(crate) end: usize,
+}
+
+impl HookCounts {
+    /// Snapshot hook counters from shared state.
+    pub(crate) fn from_counters(before: &Arc<AtomicUsize>, end: Option<&Arc<AtomicUsize>>) -> Self {
+        let end = end.map_or(0, |counter| counter.load(Ordering::SeqCst));
+        Self {
+            before: before.load(Ordering::SeqCst),
+            end,
+        }
+    }
+}
+
+/// Shared counters for protocol hook configurations used in connection tests.
+#[derive(Clone)]
+pub(crate) struct HookCounters {
+    before_calls: Arc<AtomicUsize>,
+    end_calls: Arc<AtomicUsize>,
+}
+
+impl HookCounters {
+    /// Create counters tracking before-send and on-command-end hook invocations.
+    pub(crate) fn new() -> Self {
+        Self {
+            before_calls: Arc::new(AtomicUsize::new(0)),
+            end_calls: Arc::new(AtomicUsize::new(0)),
+        }
+    }
+
+    /// Construct protocol hooks that increment frames and record hook usage.
+    fn build_hooks_with_increment(
+        &self,
+        increment: u8,
+        stream_end_value: impl Fn(&mut ConnectionContext) -> Option<u8> + Send + Sync + 'static,
+    ) -> ProtocolHooks<u8, wireframe::NoProtocolError> {
+        let before_clone = Arc::clone(&self.before_calls);
+        let end_clone = Arc::clone(&self.end_calls);
+
+        ProtocolHooks {
+            before_send: Some(Box::new(
+                move |frame: &mut u8, _ctx: &mut ConnectionContext| {
+                    before_clone.fetch_add(1, Ordering::SeqCst);
+                    *frame += increment;
+                },
+            )),
+            on_command_end: Some(Box::new(move |_ctx: &mut ConnectionContext| {
+                end_clone.fetch_add(1, Ordering::SeqCst);
+            })
+                as Box<dyn FnMut(&mut ConnectionContext) + Send + 'static>),
+            stream_end: Some(Box::new(stream_end_value)),
+            ..ProtocolHooks::<u8, wireframe::NoProtocolError>::default()
+        }
+    }
+
+    /// Snapshot the recorded hook counts for verification.
+    pub(crate) fn get_counts(&self) -> HookCounts {
+        HookCounts::from_counters(&self.before_calls, Some(&self.end_calls))
+    }
+}
+
+type StreamEndHook = Box<dyn Fn(&mut ConnectionContext) -> Option<u8> + Send + Sync + 'static>;
+
+/// Configuration for building connection actor harnesses.
+#[derive(Default)]
+pub(crate) struct HarnessConfig {
+    has_response: bool,
+    has_multi_packet: bool,
+    increment: u8,
+    stream_end_value: Option<StreamEndHook>,
+}
+
+impl HarnessConfig {
+    /// Create a harness configuration with the default increment.
+    pub(crate) fn new() -> Self {
+        Self {
+            increment: 1,
+            ..Self::default()
+        }
+    }
+
+    /// Enable multi-packet support for the harness under construction.
+    pub(crate) fn with_multi_packet(mut self) -> Self {
+        self.has_multi_packet = true;
+        self
+    }
+
+    /// Override the amount added to each forwarded frame.
+    pub(crate) fn with_increment(mut self, increment: u8) -> Self {
+        self.increment = increment;
+        self
+    }
+
+    /// Provide a stream-end hook for the harness under construction.
+    pub(crate) fn with_stream_end<F>(mut self, f: F) -> Self
+    where
+        F: Fn(&mut ConnectionContext) -> Option<u8> + Send + Sync + 'static,
+    {
+        self.stream_end_value = Some(Box::new(f));
+        self
+    }
+}
+
+#[derive(Clone)]
+pub(crate) struct HarnessFactory {
+    counters: HookCounters,
+}
+
+impl HarnessFactory {
+    /// Build a connection actor harness with shared hook counters.
+    pub(crate) fn create(
+        &self,
+        config: HarnessConfig,
+    ) -> Result<ActorHarness, wireframe::push::PushConfigError> {
+        let HarnessConfig {
+            has_response,
+            has_multi_packet,
+            increment,
+            stream_end_value,
+        } = config;
+        let stream_end_fn =
+            stream_end_value.unwrap_or_else(|| Box::new(|_: &mut ConnectionContext| None));
+        let hooks = self
+            .counters
+            .build_hooks_with_increment(increment, stream_end_fn);
+        ActorHarness::new_with_state(hooks, has_response, has_multi_packet)
+    }
+
+    /// Read the accumulated hook counters for the most recent harness.
+    pub(crate) fn counts(&self) -> HookCounts { self.counters.get_counts() }
+}
+
+#[fixture]
+pub(crate) fn hook_counters() -> HookCounters {
+    let counters = HookCounters::new();
+    debug_assert_eq!(counters.get_counts(), HookCounts { before: 0, end: 0 });
+    counters
+}
+
+#[fixture]
+pub(crate) fn harness_factory(hook_counters: HookCounters) -> HarnessFactory {
+    HarnessFactory {
+        counters: hook_counters,
+    }
+}
+
+pub(crate) fn assert_frame_processed(
+    out: &[u8],
+    expected: &[u8],
+    expected_counts: HookCounts,
+    actual_counts: HookCounts,
+) {
+    assert_eq!(out, expected, "frames should match expected output");
+    assert_eq!(actual_counts, expected_counts, "hook counts should match");
+}
+
+/// Helper to verify common multi-packet processing assertions.
+pub(crate) fn assert_multi_packet_processing_result(
+    harness: &ActorHarness,
+    harness_factory: &HarnessFactory,
+    expected_output: &[u8],
+    expected_counts: HookCounts,
+) {
+    let snapshot = harness.snapshot();
+    assert!(snapshot.is_active && !snapshot.is_shutting_down && !snapshot.is_done);
+    assert_frame_processed(
+        &harness.out,
+        expected_output,
+        expected_counts,
+        harness_factory.counts(),
+    );
+}
+
+pub(crate) fn assert_reason_logged(
+    logger: &mut LoggerHandle,
+    expected_level: Level,
+    expected_reason: &str,
+    expected_correlation: Option<u64>,
+) {
+    let expected_correlation = format!("correlation_id={expected_correlation:?}");
+    let mut found = false;
+    while let Some(record) = logger.pop() {
+        let message = record.args().to_string();
+        if !message.contains("multi-packet stream closed") {
+            continue;
+        }
+        if !message.contains(&expected_correlation) {
+            continue;
+        }
+        assert_eq!(
+            record.level(),
+            expected_level,
+            "unexpected log level for closure: message={message}",
+        );
+        assert!(
+            message.contains(&format!("reason={expected_reason}")),
+            "closure log missing reason: message={message}",
+        );
+        found = true;
+        break;
+    }
+    assert!(found, "multi-packet closure log not found");
+}
diff --git a/tests/connection_fragmentation.rs b/tests/connection_fragmentation.rs
index e4908e47c..72f178a54 100644
--- a/tests/connection_fragmentation.rs
+++ b/tests/connection_fragmentation.rs
@@ -39,10 +39,6 @@ fn setup_fragmented_actor() -> TestResult<(
 }
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn connection_actor_fragments_outbound_frames() -> TestResult {
     let (mut actor, handle, cfg) = setup_fragmented_actor()?;
 
@@ -58,11 +54,13 @@ async fn connection_actor_fragments_outbound_frames() -> TestResult {
         .await
         .map_err(|err| io::Error::other(format!("actor run failed: {err:?}")))?;
 
-    assert!(
-        out.len() > 1,
-        "fragmentation should yield multiple frames, got {}",
-        out.len()
-    );
+    if out.len() <= 1 {
+        return Err(format!(
+            "fragmentation should yield multiple frames, got {}",
+            out.len()
+        )
+        .into());
+    }
 
     let mut reassembler = Reassembler::new(cfg.max_message_size, cfg.reassembly_timeout);
     let mut assembled: Option<Vec<u8>> = None;
@@ -79,15 +77,16 @@ async fn connection_actor_fragments_outbound_frames() -> TestResult {
     }
 
     let assembled = assembled.ok_or("missing reassembled payload")?;
-    assert_eq!(assembled, payload, "reassembled payload mismatch");
+    if assembled != payload {
+        return Err(format!(
+            "reassembled payload mismatch: expected {payload:?}, got {assembled:?}"
+        )
+        .into());
+    }
     Ok(())
 }
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn connection_actor_passes_through_small_outbound_frames_unfragmented() -> TestResult {
     let (mut actor, handle, cfg) = setup_fragmented_actor()?;
 
@@ -103,7 +102,9 @@ async fn connection_actor_passes_through_small_outbound_frames_unfragmented() ->
         .await
         .map_err(|err| io::Error::other(format!("actor run failed: {err:?}")))?;
 
-    assert_eq!(out.len(), 1, "expected unfragmented single frame");
+    if out.len() != 1 {
+        return Err(format!("expected unfragmented single frame, got {}", out.len()).into());
+    }
     let only = out
         .into_iter()
         .next()
@@ -113,6 +114,11 @@ async fn connection_actor_passes_through_small_outbound_frames_unfragmented() ->
         None => {}
         Some(_) => return Err("expected unfragmented payload".into()),
     }
-    assert_eq!(payload_out, payload, "payload mutated during round trip");
+    if payload_out != payload {
+        return Err(format!(
+            "payload mutated during round trip: expected {payload:?}, got {payload_out:?}"
+        )
+        .into());
+    }
     Ok(())
 }
diff --git a/tests/correlation_id.rs b/tests/correlation_id.rs
index 9335b7714..d7277ab9e 100644
--- a/tests/correlation_id.rs
+++ b/tests/correlation_id.rs
@@ -17,10 +17,6 @@ use wireframe::{
 use wireframe_testing::TestResult;
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn stream_frames_carry_request_correlation_id() -> TestResult {
     let cid = 42u64;
     let stream: FrameStream<Envelope> = Box::pin(try_stream! {
@@ -39,10 +35,15 @@ async fn stream_frames_carry_request_correlation_id() -> TestResult {
         .run(&mut out)
         .await
         .map_err(|e| io::Error::other(format!("actor run failed: {e:?}")))?;
-    assert!(
-        out.iter().all(|e| e.correlation_id() == Some(cid)),
-        "frames lost correlation id"
-    );
+    if !out.iter().all(|e| e.correlation_id() == Some(cid)) {
+        return Err(format!(
+            "frames lost correlation id: expected Some({cid}), got {:?}",
+            out.iter()
+                .map(CorrelatableFrame::correlation_id)
+                .collect::<Vec<_>>()
+        )
+        .into());
+    }
     Ok(())
 }
 
@@ -105,10 +106,12 @@ async fn multi_packet_frames_apply_expected_correlation(
         .iter()
         .map(CorrelatableFrame::correlation_id)
         .collect();
-    assert_eq!(
-        correlations, expected,
-        "unexpected correlation ids: {correlations:?}, expected {expected:?}"
-    );
+    if correlations != expected {
+        return Err(format!(
+            "unexpected correlation ids: expected {expected:?}, got {correlations:?}"
+        )
+        .into());
+    }
     Ok(())
 }
 
@@ -131,10 +134,12 @@ async fn multi_packet_terminator_applies_correlation(
     let [terminator] = frames.as_slice() else {
         return Err(io::Error::other("expected exactly one terminator frame").into());
     };
-    assert_eq!(
-        terminator.correlation_id(),
-        expected,
-        "unexpected terminator correlation"
-    );
+    let actual_correlation = terminator.correlation_id();
+    if actual_correlation != expected {
+        return Err(format!(
+            "unexpected terminator correlation: expected {expected:?}, got {actual_correlation:?}"
+        )
+        .into());
+    }
     Ok(())
 }
diff --git a/tests/fixtures/client_messaging.rs b/tests/fixtures/client_messaging.rs
index 1022df768..45aa32267 100644
--- a/tests/fixtures/client_messaging.rs
+++ b/tests/fixtures/client_messaging.rs
@@ -145,17 +145,15 @@ impl ClientMessagingWorld {
     /// Send multiple sequential envelopes and capture all correlation IDs.
     ///
     /// # Errors
-    /// Returns an error if the client is missing or communication fails.
-    #[expect(
-        clippy::cast_possible_truncation,
-        reason = "test helper with small count values"
-    )]
+    /// Returns an error if the client is missing, communication fails, or an
+    /// index does not fit the message ID.
     pub async fn send_multiple_envelopes(&mut self, count: usize) -> TestResult {
         let client = self.client.as_mut().ok_or("client not connected")?;
         self.sent_correlation_ids.clear();
 
-        for i in 0..count {
-            let envelope = Envelope::new(i as u32, None, vec![i as u8]);
+        for (i, payload_byte) in (0..count).zip((0_u8..=u8::MAX).cycle()) {
+            let message_id = u32::try_from(i)?;
+            let envelope = Envelope::new(message_id, None, vec![payload_byte]);
             let correlation_id = client.send_envelope(envelope).await?;
             self.sent_correlation_ids.push(correlation_id);
 
diff --git a/tests/fixtures/client_send_streaming.rs b/tests/fixtures/client_send_streaming.rs
index 9e620e1f0..d1dcc055c 100644
--- a/tests/fixtures/client_send_streaming.rs
+++ b/tests/fixtures/client_send_streaming.rs
@@ -49,15 +49,7 @@ pub fn client_send_streaming_world() -> ClientSendStreamingWorld {
 /// `src/client/tests/send_streaming_infra.rs`. The two cannot be shared
 /// directly because that helper is `pub(super)` inside the library
 /// crate.
-#[expect(
-    clippy::integer_division_remainder_used,
-    reason = "modulo generates a deterministic test byte pattern"
-)]
-#[expect(
-    clippy::cast_possible_truncation,
-    reason = "value is modulo 256, guaranteed to fit in u8"
-)]
-fn test_body(n: usize) -> Vec<u8> { (0..n).map(|i| (i % 256) as u8).collect() }
+fn test_body(n: usize) -> Vec<u8> { (0u8..=u8::MAX).cycle().take(n).collect() }
 
 impl ClientSendStreamingWorld {
     fn new() -> Self {
diff --git a/tests/fragment_transport/eviction.rs b/tests/fragment_transport/eviction.rs
index 0d6aa3a3b..2eddb4a18 100644
--- a/tests/fragment_transport/eviction.rs
+++ b/tests/fragment_transport/eviction.rs
@@ -25,10 +25,6 @@ use crate::fragment_helpers::{
 };
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn expired_fragments_are_evicted() -> TestResult {
     let buffer_capacity = 512;
     let timeout_ms = 10;
@@ -54,10 +50,11 @@ async fn expired_fragments_are_evicted() -> TestResult {
     tokio::io::AsyncWriteExt::shutdown(client.get_mut()).await?;
 
     let recv_result = timeout(Duration::from_millis(200), rx.recv()).await;
-    assert!(
-        recv_result.is_err(),
-        "handler should not receive after timeout eviction"
-    );
+    if recv_result.is_ok() {
+        return Err(TestError::Assertion(format!(
+            "handler should not receive after timeout eviction, got {recv_result:?}"
+        )));
+    }
 
     drop(client);
     server.await??;
diff --git a/tests/fragment_transport/rejection.rs b/tests/fragment_transport/rejection.rs
index a95486143..eec590c41 100644
--- a/tests/fragment_transport/rejection.rs
+++ b/tests/fragment_transport/rejection.rs
@@ -99,10 +99,6 @@ fn mutate_out_of_order(mut fragments: Vec<Envelope>) -> TestResult<Vec<Envelope>
 }
 
 /// Mutate fragments by truncating the header of the first fragment.
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 fn mutate_malformed_header(mut fragments: Vec<Envelope>) -> TestResult<Vec<Envelope>> {
     let parts = fragments
         .first()
@@ -112,10 +108,11 @@ fn mutate_malformed_header(mut fragments: Vec<Envelope>) -> TestResult<Vec<Envel
         ))?
         .into_parts();
     let mut payload = parts.clone().into_payload();
-    assert!(
-        payload.starts_with(FRAGMENT_MAGIC),
-        "expected fragment to start with marker"
-    );
+    if !payload.starts_with(FRAGMENT_MAGIC) {
+        return Err(TestError::Assertion(
+            "expected fragment to start with marker".into(),
+        ));
+    }
     let truncate_len = FRAGMENT_MAGIC.len() + 2;
     if payload.len() > truncate_len {
         payload.truncate(truncate_len);
diff --git a/tests/lifecycle.rs b/tests/lifecycle.rs
index 771ff2050..184514605 100644
--- a/tests/lifecycle.rs
+++ b/tests/lifecycle.rs
@@ -12,7 +12,14 @@ use std::{
     },
 };
 
+#[path = "common/fallible_assertions/check.rs"]
+mod fallible_check;
+#[path = "common/fallible_assertions/check_eq.rs"]
+mod fallible_check_eq;
+
 use bytes::BytesMut;
+use fallible_check::check;
+use fallible_check_eq::check_eq;
 use tokio_util::codec::Encoder;
 use wireframe::{
     app::{Envelope, Packet},
@@ -69,10 +76,6 @@ where
 }
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn setup_and_teardown_callbacks_run() -> TestResult<()> {
     let setup_count = Arc::new(AtomicUsize::new(0));
     let teardown_count = Arc::new(AtomicUsize::new(0));
@@ -82,24 +85,20 @@ async fn setup_and_teardown_callbacks_run() -> TestResult<()> {
 
     run_with_duplex_server(app).await;
 
-    assert_eq!(
+    check_eq(
         setup_count.load(Ordering::SeqCst),
         1,
-        "setup callback did not run exactly once"
-    );
-    assert_eq!(
+        "setup callback did not run exactly once",
+    )?;
+    check_eq(
         teardown_count.load(Ordering::SeqCst),
         1,
-        "teardown callback did not run exactly once"
-    );
+        "teardown callback did not run exactly once",
+    )?;
 
     Ok(())
 }
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn setup_without_teardown_runs() -> TestResult<()> {
     let counter = Arc::new(AtomicUsize::new(0));
     let cb = call_counting_callback(&counter, ());
@@ -108,20 +107,16 @@ async fn setup_without_teardown_runs() -> TestResult<()> {
 
     run_with_duplex_server(app).await;
 
-    assert_eq!(
+    check_eq(
         counter.load(Ordering::SeqCst),
         1,
-        "setup callback did not run"
-    );
+        "setup callback did not run",
+    )?;
 
     Ok(())
 }
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn teardown_without_setup_does_not_run() -> TestResult<()> {
     let counter = Arc::new(AtomicUsize::new(0));
     let cb = call_counting_callback(&counter, ());
@@ -130,20 +125,16 @@ async fn teardown_without_setup_does_not_run() -> TestResult<()> {
 
     run_with_duplex_server(app).await;
 
-    assert_eq!(
+    check_eq(
         counter.load(Ordering::SeqCst),
         0,
-        "teardown callback should not run"
-    );
+        "teardown callback should not run",
+    )?;
 
     Ok(())
 }
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn setup_after_teardown_clears_previous_teardown() -> TestResult<()> {
     let setup_count = Arc::new(AtomicUsize::new(0));
     let teardown_count = Arc::new(AtomicUsize::new(0));
@@ -156,25 +147,21 @@ async fn setup_after_teardown_clears_previous_teardown() -> TestResult<()> {
 
     run_with_duplex_server(app).await;
 
-    assert_eq!(
+    check_eq(
         setup_count.load(Ordering::SeqCst),
         1,
-        "setup callback did not run"
-    );
-    assert_eq!(
+        "setup callback did not run",
+    )?;
+    check_eq(
         teardown_count.load(Ordering::SeqCst),
         0,
-        "teardown callback for prior state type should be cleared"
-    );
+        "teardown callback for prior state type should be cleared",
+    )?;
 
     Ok(())
 }
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn helpers_preserve_correlation_id_and_run_callbacks() -> TestResult<()> {
     let setup = Arc::new(AtomicUsize::new(0));
     let teardown = Arc::new(AtomicUsize::new(0));
@@ -193,25 +180,25 @@ async fn helpers_preserve_correlation_id_and_run_callbacks() -> TestResult<()> {
     codec.encode(bytes.into(), &mut frame)?;
 
     let out = run_app(app, vec![frame.to_vec()], None).await?;
-    assert!(!out.is_empty(), "expected response frames");
+    check(!out.is_empty(), "expected response frames")?;
 
     let frames = decode_frames(&out)?;
     let [first] = frames.as_slice() else {
-        panic!("expected a single response frame");
+        return Err(format!("expected a single response frame, got {}", frames.len()).into());
     };
     let (resp, _) = BincodeSerializer.deserialize::<CommonTestEnvelope>(first)?;
-    assert_eq!(resp.correlation_id, Some(0), "correlation id not preserved");
+    check_eq(resp.correlation_id, Some(0), "correlation id not preserved")?;
 
-    assert_eq!(
+    check_eq(
         setup.load(Ordering::SeqCst),
         1,
-        "setup callback did not run exactly once"
-    );
-    assert_eq!(
+        "setup callback did not run exactly once",
+    )?;
+    check_eq(
         teardown.load(Ordering::SeqCst),
         1,
-        "teardown callback did not run exactly once"
-    );
+        "teardown callback did not run exactly once",
+    )?;
 
     Ok(())
 }
diff --git a/tests/metadata.rs b/tests/metadata.rs
index 29561f0a2..d99bb9b59 100644
--- a/tests/metadata.rs
+++ b/tests/metadata.rs
@@ -93,10 +93,6 @@ impl FrameMetadata for CountingSerializer {
 }
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn metadata_parser_invoked_before_deserialize() -> TestResult<()> {
     let counter = Arc::new(AtomicUsize::new(0));
     let deserialize_calls = Arc::new(AtomicUsize::new(0));
@@ -106,13 +102,20 @@ async fn metadata_parser_invoked_before_deserialize() -> TestResult<()> {
     let env = Envelope::new(1, Some(0), vec![42]);
 
     let out = drive_with_bincode(app, env).await?;
-    assert!(!out.is_empty(), "no frames emitted");
-    assert_eq!(counter.load(Ordering::Relaxed), 1, "expected 1 parse call");
-    assert_eq!(
-        deserialize_calls.load(Ordering::Relaxed),
-        1,
-        "expected 1 deserialize call with context"
-    );
+    if out.is_empty() {
+        return Err("no frames emitted".into());
+    }
+    let parse_calls = counter.load(Ordering::Relaxed);
+    if parse_calls != 1 {
+        return Err(format!("expected 1 parse call, got {parse_calls}").into());
+    }
+    let actual_deserialize_calls = deserialize_calls.load(Ordering::Relaxed);
+    if actual_deserialize_calls != 1 {
+        return Err(format!(
+            "expected 1 deserialize call with context, got {actual_deserialize_calls}"
+        )
+        .into());
+    }
     Ok(())
 }
 
@@ -136,10 +139,6 @@ impl FrameMetadata for FallbackSerializer {
 }
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn falls_back_to_deserialize_after_parse_error() -> TestResult<()> {
     let parse_calls = Arc::new(AtomicUsize::new(0));
     let serializer = FallbackSerializer(parse_calls.clone());
@@ -148,20 +147,17 @@ async fn falls_back_to_deserialize_after_parse_error() -> TestResult<()> {
     let env = Envelope::new(1, Some(0), vec![7]);
 
     let out = drive_with_bincode(app, env).await?;
-    assert!(!out.is_empty(), "no frames emitted");
-    assert_eq!(
-        parse_calls.load(Ordering::Relaxed),
-        1,
-        "expected 1 parse call"
-    );
+    if out.is_empty() {
+        return Err("no frames emitted".into());
+    }
+    let actual_parse_calls = parse_calls.load(Ordering::Relaxed);
+    if actual_parse_calls != 1 {
+        return Err(format!("expected 1 parse call, got {actual_parse_calls}").into());
+    }
     Ok(())
 }
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn metadata_is_forwarded_to_deserialize_context() -> TestResult<()> {
     let context_state = Arc::new(Mutex::new(None::<CapturedDeserializeContext>));
     let serializer = ContextCapturingSerializer::new(context_state.clone());
@@ -170,18 +166,38 @@ async fn metadata_is_forwarded_to_deserialize_context() -> TestResult<()> {
     let envelope = Envelope::new(1, Some(77), vec![1, 2, 3, 4]);
     let expected_parts = envelope.clone().into_parts();
     let output = drive_with_bincode(app, envelope.clone()).await?;
-    assert!(!output.is_empty(), "no frames emitted");
+    if output.is_empty() {
+        return Err("no frames emitted".into());
+    }
 
     let captured = (*context_state
         .lock()
         .map_err(|_| "mutex poisoned while locking context_state")?)
     .ok_or("expected captured deserialize context")?;
 
-    assert_eq!(captured.message_id, Some(expected_parts.id()));
-    assert_eq!(captured.correlation_id, expected_parts.correlation_id());
-    assert_eq!(
-        captured.frame_metadata_len, captured.metadata_bytes_consumed,
-        "metadata bytes consumed should match captured frame metadata slice length"
-    );
+    let expected_message_id = Some(expected_parts.id());
+    if captured.message_id != expected_message_id {
+        return Err(format!(
+            "message id mismatch: expected {expected_message_id:?}, got {:?}",
+            captured.message_id
+        )
+        .into());
+    }
+    let expected_correlation_id = expected_parts.correlation_id();
+    if captured.correlation_id != expected_correlation_id {
+        return Err(format!(
+            "correlation id mismatch: expected {expected_correlation_id:?}, got {:?}",
+            captured.correlation_id
+        )
+        .into());
+    }
+    if captured.frame_metadata_len != captured.metadata_bytes_consumed {
+        return Err(format!(
+            "metadata bytes consumed should match captured frame metadata slice length: metadata \
+             length {:?}, bytes consumed {:?}",
+            captured.frame_metadata_len, captured.metadata_bytes_consumed
+        )
+        .into());
+    }
     Ok(())
 }
diff --git a/tests/middleware_order.rs b/tests/middleware_order.rs
index 9fde29b56..662ad665f 100644
--- a/tests/middleware_order.rs
+++ b/tests/middleware_order.rs
@@ -3,7 +3,11 @@
 //! Verifies tags are applied in reverse to request and response bodies.
 #![cfg(not(loom))]
 
+#[path = "common/fallible_assertions/check_equal.rs"]
+mod fallible_check_equal;
+
 use async_trait::async_trait;
+use fallible_check_equal::check_equal;
 use tokio::io::{AsyncReadExt, AsyncWriteExt, duplex};
 use wireframe::{
     app::{Envelope, Handler},
@@ -53,21 +57,13 @@ impl Transform<HandlerService<Envelope>> for TagMiddleware {
 }
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 #[expect(deprecated, reason = "test covers the legacy builder connection API")]
 async fn middleware_applied_in_reverse_order() -> TestResult<()> {
     let handler: Handler<Envelope> = std::sync::Arc::new(|_env: &Envelope| Box::pin(async {}));
-    let app = TestApp::new()
-        .expect("failed to create app")
-        .route(1, handler)
-        .expect("route registration failed")
-        .wrap(TagMiddleware(b'A'))
-        .expect("wrap failed")
-        .wrap(TagMiddleware(b'B'))
-        .expect("wrap failed");
+    let app = TestApp::new()?
+        .route(1, handler)?
+        .wrap(TagMiddleware(b'A'))?
+        .wrap(TagMiddleware(b'B'))?;
 
     let (mut client, server) = duplex(256);
 
@@ -93,11 +89,7 @@ async fn middleware_applied_in_reverse_order() -> TestResult<()> {
     let parts = wireframe::app::Packet::into_parts(resp);
     let correlation_id = parts.correlation_id();
     let payload = parts.into_payload();
-    assert_eq!(
-        payload,
-        [b'X', b'A', b'B', b'B', b'A'],
-        "unexpected payload"
-    );
-    assert_eq!(correlation_id, Some(7), "unexpected correlation id");
+    check_equal(&payload, b"XABBA", "unexpected payload")?;
+    check_equal(&correlation_id, &Some(7), "unexpected correlation id")?;
     Ok(())
 }
diff --git a/tests/multi_packet.rs b/tests/multi_packet.rs
index d02f12ce8..6e5369bc0 100644
--- a/tests/multi_packet.rs
+++ b/tests/multi_packet.rs
@@ -67,7 +67,12 @@ async fn multi_packet_drains_all_messages(count: usize) -> TestResult {
         .into_iter()
         .map(TestMsg)
         .collect::<Vec<_>>();
-    assert_eq!(received, expected);
+    if received != expected {
+        return Err(format!(
+            "multi-packet frames mismatch: expected {expected:?}, got {received:?}"
+        )
+        .into());
+    }
     Ok(())
 }
 
@@ -103,7 +108,9 @@ async fn connection_actor_drains_multi_packet_channel(
         .await
         .map_err(|e| boxed_err("connection actor error", e))?;
 
-    assert_eq!(out, frames);
+    if out != frames {
+        return Err(format!("actor frames mismatch: expected {frames:?}, got {out:?}").into());
+    }
     Ok(())
 }
 
@@ -141,7 +148,12 @@ async fn connection_actor_interleaves_multi_packet_and_priority_frames(
         .await
         .map_err(|e| boxed_err("connection actor error", e))?;
 
-    assert_eq!(out, vec![10, 100, 11, 101, 1, 2, 3]);
+    let expected = vec![10, 100, 11, 101, 1, 2, 3];
+    if out != expected {
+        return Err(
+            format!("actor frame order mismatch: expected {expected:?}, got {out:?}").into(),
+        );
+    }
     Ok(())
 }
 
@@ -177,7 +189,9 @@ async fn shutdown_completes_multi_packet_channel(
         .map_err(|e| boxed_err("connection actor shutdown timeout", e))??;
     let out = join_result?;
 
-    assert!(out.is_empty());
+    if !out.is_empty() {
+        return Err(format!("expected no frames after shutdown, got {out:?}").into());
+    }
     drop(tx);
     Ok(())
 }
@@ -217,39 +231,34 @@ async fn shutdown_during_active_multi_packet_send(
         .await
         .map_err(|e| boxed_err("connection actor shutdown timeout", e))??;
     let out = join_result?;
-    assert!(out.is_empty() || out == vec![1, 2], "actor output: {out:?}");
+    if !out.is_empty() && out != vec![1, 2] {
+        return Err(format!(
+            "expected no frames or the two queued frames after shutdown, got {out:?}"
+        )
+        .into());
+    }
     drop(tx);
     Ok(())
 }
 
 /// Returns an empty stream for an empty vector response.
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn vec_empty_returns_empty_stream() -> TestResult {
     let resp: Response<TestMsg, ()> = Response::Vec(Vec::new());
     let received = drain_all(resp.into_stream()).await?;
-    assert!(
-        received.is_empty(),
-        "expected empty stream, got {received:?}"
-    );
+    if !received.is_empty() {
+        return Err(format!("expected empty stream, got {received:?}").into());
+    }
     Ok(())
 }
 
 /// `Response::Empty` yields no frames.
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn empty_returns_empty_stream() -> TestResult {
     let resp: Response<TestMsg, ()> = Response::Empty;
     let received = drain_all(resp.into_stream()).await?;
-    assert!(
-        received.is_empty(),
-        "expected empty stream, got {received:?}"
-    );
+    if !received.is_empty() {
+        return Err(format!("expected empty stream, got {received:?}").into());
+    }
     Ok(())
 }
diff --git a/tests/multi_packet_streaming.rs b/tests/multi_packet_streaming.rs
index 23bd7bbbe..fa8f7244e 100644
--- a/tests/multi_packet_streaming.rs
+++ b/tests/multi_packet_streaming.rs
@@ -108,10 +108,6 @@ impl ActorHarness {
 fn parts(frame: &Envelope) -> PacketParts { frame.clone().into_parts() }
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn client_receives_multi_packet_stream_with_terminator() -> TestResult<()> {
     let mut harness = ActorHarness::new()?;
     let (tx, rx) = mpsc::channel(4);
@@ -133,29 +129,45 @@ async fn client_receives_multi_packet_stream_with_terminator() -> TestResult<()>
 
     let out = harness.run().await?;
 
-    assert_eq!(out.len(), 3, "expected two frames plus terminator");
+    if out.len() != 3 {
+        return Err(format!("expected two frames plus terminator, got {}", out.len()).into());
+    }
     let payloads: Vec<Vec<u8>> = out
         .iter()
         .map(|frame| parts(frame).into_payload())
         .collect();
-    assert_eq!(payloads.first(), Some(&vec![1]), "first payload mismatch");
-    assert_eq!(
-        payloads.get(1),
-        Some(&vec![2, 3]),
-        "second payload mismatch"
-    );
-    assert_eq!(
-        payloads.get(2),
-        Some(&Vec::<u8>::new()),
-        "terminator payload should be empty"
-    );
+    if payloads.first() != Some(&vec![1]) {
+        return Err(format!(
+            "first payload mismatch: expected {:?}, got {:?}",
+            Some(&vec![1]),
+            payloads.first()
+        )
+        .into());
+    }
+    if payloads.get(1) != Some(&vec![2, 3]) {
+        return Err(format!(
+            "second payload mismatch: expected {:?}, got {:?}",
+            Some(&vec![2, 3]),
+            payloads.get(1)
+        )
+        .into());
+    }
+    if payloads.get(2) != Some(&Vec::<u8>::new()) {
+        return Err(format!(
+            "terminator payload should be empty: got {:?}",
+            payloads.get(2)
+        )
+        .into());
+    }
 
     for frame in &out {
-        assert_eq!(
-            parts(frame).correlation_id(),
-            correlation,
-            "correlation id mismatch"
-        );
+        let actual_correlation = parts(frame).correlation_id();
+        if actual_correlation != correlation {
+            return Err(format!(
+                "correlation id mismatch: expected {correlation:?}, got {actual_correlation:?}"
+            )
+            .into());
+        }
     }
     Ok(())
 }
@@ -193,13 +205,22 @@ async fn multi_packet_logs_disconnected_when_sender_dropped(
 
     let out = harness.run().await?;
 
-    assert_eq!(out.len(), 2, "expected push frame followed by terminator");
-    let last = out.last().expect("terminator missing");
-    assert_eq!(
-        parts(last).correlation_id(),
-        correlation,
-        "terminator correlation mismatch"
-    );
+    if out.len() != 2 {
+        return Err(format!(
+            "expected push frame followed by terminator, got {} frames",
+            out.len()
+        )
+        .into());
+    }
+    let last = out.last().ok_or("terminator missing")?;
+    let terminator_correlation = parts(last).correlation_id();
+    if terminator_correlation != correlation {
+        return Err(format!(
+            "terminator correlation mismatch: expected {correlation:?}, got \
+             {terminator_correlation:?}"
+        )
+        .into());
+    }
 
     let mut saw_disconnect = false;
     while let Some(record) = logger.pop() {
@@ -208,7 +229,9 @@ async fn multi_packet_logs_disconnected_when_sender_dropped(
             break;
         }
     }
-    assert!(saw_disconnect, "missing disconnect log");
+    if !saw_disconnect {
+        return Err("missing disconnect log".into());
+    }
     Ok(())
 }
 
diff --git a/tests/preamble/basic.rs b/tests/preamble/basic.rs
index 9ba6a7bc4..147bb1263 100644
--- a/tests/preamble/basic.rs
+++ b/tests/preamble/basic.rs
@@ -7,10 +7,6 @@ use wireframe_testing::TestResult;
 use crate::support::HotlinePreamble;
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn parse_valid_preamble() -> TestResult {
     let (mut client, mut server) = duplex(64);
     let bytes = b"TRTPHOTL\x00\x01\x00\x02";
@@ -18,26 +14,40 @@ async fn parse_valid_preamble() -> TestResult {
     client.shutdown().await?;
     let (p, _) = read_preamble::<_, HotlinePreamble>(&mut server).await?;
     p.validate()?;
-    assert_eq!(p.magic, HotlinePreamble::MAGIC, "preamble magic mismatch");
-    assert_eq!(p.min_version, 1, "preamble minimum version mismatch");
-    assert_eq!(p.client_version, 2, "preamble client version mismatch");
+    if p.magic != HotlinePreamble::MAGIC {
+        return Err(format!(
+            "preamble magic mismatch: expected {:?}, got {:?}",
+            HotlinePreamble::MAGIC,
+            p.magic
+        )
+        .into());
+    }
+    if p.min_version != 1 {
+        return Err(format!(
+            "preamble minimum version mismatch: expected 1, got {}",
+            p.min_version
+        )
+        .into());
+    }
+    if p.client_version != 2 {
+        return Err(format!(
+            "preamble client version mismatch: expected 2, got {}",
+            p.client_version
+        )
+        .into());
+    }
     Ok(())
 }
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn invalid_magic_is_error() -> TestResult {
     let (mut client, mut server) = duplex(64);
     let bytes = b"WRONGMAG\x00\x01\x00\x02";
     client.write_all(bytes).await?;
     client.shutdown().await?;
     let (preamble, _) = read_preamble::<_, HotlinePreamble>(&mut server).await?;
-    assert!(
-        preamble.validate().is_err(),
-        "invalid magic should fail validation"
-    );
+    if preamble.validate().is_ok() {
+        return Err("invalid magic should fail validation".into());
+    }
     Ok(())
 }
diff --git a/tests/preamble/responses.rs b/tests/preamble/responses.rs
index b371a09ea..3d7b8b1e6 100644
--- a/tests/preamble/responses.rs
+++ b/tests/preamble/responses.rs
@@ -21,10 +21,6 @@ use crate::support::{
 };
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn success_callback_can_write_response() -> TestResult {
     let factory = factory();
     let (response_tx, response_rx) = oneshot::channel();
@@ -51,15 +47,13 @@ async fn success_callback_can_write_response() -> TestResult {
     })
     .await?;
     let buf = recv_within(Duration::from_secs(1), response_rx).await?;
-    assert_eq!(&buf, b"ACK");
+    if buf.as_slice() != b"ACK" {
+        return Err(format!("response mismatch: expected b\"ACK\", got {buf:?}").into());
+    }
     Ok(())
 }
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn failure_callback_can_write_response() -> TestResult {
     let factory = factory();
     let (failure_holder, failure_rx) = channel_holder();
@@ -89,7 +83,9 @@ async fn failure_callback_can_write_response() -> TestResult {
     })
     .await?;
     let buf = recv_within(Duration::from_secs(1), response_rx).await?;
-    assert_eq!(&buf, b"ERR");
+    if buf.as_slice() != b"ERR" {
+        return Err(format!("response mismatch: expected b\"ERR\", got {buf:?}").into());
+    }
     recv_within(Duration::from_millis(200), failure_rx).await?;
     Ok(())
 }
diff --git a/tests/preamble/timeouts.rs b/tests/preamble/timeouts.rs
index 6c84f17d9..465db3a0c 100644
--- a/tests/preamble/timeouts.rs
+++ b/tests/preamble/timeouts.rs
@@ -43,6 +43,20 @@ fn is_connection_closed(read_res: std::io::Result<usize>) -> TestResult<bool> {
     }
 }
 
+fn ensure_timed_out_decode_error(err: &DecodeError) -> io::Result<()> {
+    if matches!(
+        err,
+        DecodeError::Io { inner, .. }
+            if inner.kind() == io::ErrorKind::TimedOut
+    ) {
+        Ok(())
+    } else {
+        Err(io::Error::other(format!(
+            "expected timed out error, got {err:?}"
+        )))
+    }
+}
+
 fn timeout_success_handler(
     holder: Holder,
 ) -> impl for<'a> Fn(&'a HotlinePreamble, &'a mut TcpStream) -> BoxFuture<'a, io::Result<()>>
@@ -57,17 +71,19 @@ fn timeout_success_handler(
             stream.write_all(b"OK").await?;
             stream.flush().await?;
             // keep connection open by not shutting down here
-            assert_eq!(clone.magic, HotlinePreamble::MAGIC);
+            if clone.magic != HotlinePreamble::MAGIC {
+                return Err(io::Error::other(format!(
+                    "preamble magic mismatch: expected {:?}, got {:?}",
+                    HotlinePreamble::MAGIC,
+                    clone.magic
+                )));
+            }
             Ok::<(), io::Error>(())
         })
     }
 }
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn preamble_timeout_invokes_failure_handler_and_closes_connection() -> TestResult {
     let factory = factory();
     let (failure_holder, failure_rx) = channel_holder();
@@ -78,14 +94,7 @@ async fn preamble_timeout_invokes_failure_handler_and_closes_connection() -> Tes
         .on_preamble_decode_failure(move |err, stream| {
             let failure_holder = failure_holder.clone();
             Box::pin(async move {
-                assert!(
-                    matches!(
-                        err,
-                        DecodeError::Io { inner, .. }
-                            if inner.kind() == io::ErrorKind::TimedOut
-                    ),
-                    "expected timed out error, got {err:?}"
-                );
+                ensure_timed_out_decode_error(err)?;
                 stream.write_all(b"ERR").await?;
                 stream.flush().await?;
                 stream.shutdown().await?;
@@ -111,16 +120,16 @@ async fn preamble_timeout_invokes_failure_handler_and_closes_connection() -> Tes
     })
     .await?;
     let (buf, closed) = recv_within(Duration::from_secs(1), result_rx).await?;
-    assert_eq!(&buf, b"ERR");
-    assert!(closed, "expected connection to close");
+    if buf.as_slice() != b"ERR" {
+        return Err(format!("response mismatch: expected b\"ERR\", got {buf:?}").into());
+    }
+    if !closed {
+        return Err("expected connection to close".into());
+    }
     Ok(())
 }
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn preamble_timeout_allows_timely_preamble() -> TestResult {
     let factory = factory();
     let (success_holder, success_rx) = channel_holder();
@@ -153,10 +162,11 @@ async fn preamble_timeout_allows_timely_preamble() -> TestResult {
     })
     .await?;
     let (buf, failure_fired) = recv_within(Duration::from_secs(1), result_rx).await?;
-    assert_eq!(&buf, b"OK");
-    assert!(
-        !failure_fired,
-        "failure handler should not fire for timely preamble"
-    );
+    if buf.as_slice() != b"OK" {
+        return Err(format!("response mismatch: expected b\"OK\", got {buf:?}").into());
+    }
+    if failure_fired {
+        return Err("failure handler should not fire for timely preamble".into());
+    }
     Ok(())
 }
diff --git a/tests/prepared_app.rs b/tests/prepared_app.rs
index 298f2e471..18a728d2d 100644
--- a/tests/prepared_app.rs
+++ b/tests/prepared_app.rs
@@ -1,9 +1,14 @@
 //! Integration coverage for one-time application preparation.
+//!
+//! Real-socket scenarios live in `prepared_app/tcp.rs`; the scenarios here
+//! cover startup instrumentation, teardown, and shared-route reuse.
 
 mod common;
 
+#[path = "prepared_app/tcp.rs"]
+mod tcp;
+
 use std::{
-    net::SocketAddr,
     sync::{
         Arc,
         atomic::{AtomicUsize, Ordering},
@@ -11,6 +16,9 @@ use std::{
     time::Duration,
 };
 
+#[path = "common/fallible_assertions/check_equal.rs"]
+mod fallible_check_equal;
+
 use common::prepared_app::{
     TestApp,
     TransformCountingMiddleware,
@@ -18,9 +26,10 @@ use common::prepared_app::{
     handler,
     response_payload,
 };
+use fallible_check_equal::check_equal;
 use tokio::{
-    io::{AsyncReadExt, AsyncWriteExt},
-    net::{TcpListener, TcpStream},
+    io::AsyncWriteExt,
+    net::TcpStream,
     sync::{Barrier, oneshot},
     time::{sleep, timeout},
 };
@@ -160,28 +169,29 @@ async fn run_server_connections(
 }
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "assertions make transform counts and middleware order failures explicit"
-)]
 async fn connection_startup_records_counts_before_and_after_preparation() -> TestResult<()> {
     let instrumentation = ConnectionStartupInstrumentation::new();
     let app_factory = counted_app_factory(instrumentation.clone());
 
-    assert_eq!(
-        instrumentation.snapshot(),
-        ConnectionStartupCounts {
+    check_equal(
+        &instrumentation.snapshot(),
+        &ConnectionStartupCounts {
             factory_calls: 0,
             transforms: 0,
-        }
-    );
+        },
+        "connection counters should start at zero",
+    )?;
 
     let server_counts = ConnectionStartupCounts {
         factory_calls: 1,
         transforms: ROUTES * MIDDLEWARE_LAYERS,
     };
     run_server_connections(app_factory.clone(), &instrumentation, &server_counts).await?;
-    assert_eq!(instrumentation.snapshot(), server_counts);
+    check_equal(
+        &instrumentation.snapshot(),
+        &server_counts,
+        "server startup should prepare the application once",
+    )?;
 
     let prepared: TestPreparedApp = app_factory()?
         .prepare()
@@ -191,22 +201,34 @@ async fn connection_startup_records_counts_before_and_after_preparation() -> Tes
         factory_calls: 2,
         transforms: 2 * ROUTES * MIDDLEWARE_LAYERS,
     };
-    assert_eq!(instrumentation.snapshot(), prepared_counts);
+    check_equal(
+        &instrumentation.snapshot(),
+        &prepared_counts,
+        "preparation should transform each route once",
+    )?;
 
     let first = drive_prepared_with_frames(&prepared, vec![build_frame(1, vec![b'X'])?]).await?;
     let second = drive_prepared_with_frames(&prepared, vec![build_frame(2, vec![b'Y'])?]).await?;
 
-    assert_eq!(instrumentation.snapshot(), prepared_counts);
-    assert_eq!(response_payload(&first)?, [b'X', b'A', b'B', b'B', b'A']);
-    assert_eq!(response_payload(&second)?, [b'Y', b'A', b'B', b'B', b'A']);
+    check_equal(
+        &instrumentation.snapshot(),
+        &prepared_counts,
+        "serving prepared frames should not add transforms",
+    )?;
+    check_equal(
+        &response_payload(&first)?,
+        b"XABBA",
+        "the first response should follow middleware order",
+    )?;
+    check_equal(
+        &response_payload(&second)?,
+        b"YABBA",
+        "the second response should follow middleware order",
+    )?;
     Ok(())
 }
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "assertions make prepared-connection failure behaviour explicit"
-)]
 async fn prepared_app_runs_teardown_after_processing_error() -> TestResult<()> {
     let teardown_calls = Arc::new(AtomicUsize::new(0));
     let teardown_counter = Arc::clone(&teardown_calls);
@@ -222,25 +244,39 @@ async fn prepared_app_runs_teardown_after_processing_error() -> TestResult<()> {
         .await
         .map_err(|error| -> Box<dyn std::error::Error + Send + Sync> { Box::new(error) })?;
 
-    let error = drive_prepared_with_frames(&prepared, vec![vec![0, 0, 0, 2, 1]])
-        .await
-        .expect_err("truncated frame should fail processing");
-    assert_eq!(error.kind(), std::io::ErrorKind::UnexpectedEof);
-    assert_eq!(teardown_calls.load(Ordering::SeqCst), 1);
+    let error = match drive_prepared_with_frames(&prepared, vec![vec![0, 0, 0, 2, 1]]).await {
+        Err(error) => error,
+        Ok(response) => {
+            return Err(format!(
+                "truncated frame should fail processing, got response {response:?}"
+            )
+            .into());
+        }
+    };
+    check_equal(
+        &error.kind(),
+        &std::io::ErrorKind::UnexpectedEof,
+        "truncated frame should return unexpected EOF",
+    )?;
+    check_equal(
+        &teardown_calls.load(Ordering::SeqCst),
+        &1,
+        "teardown should run after the in-process processing error",
+    )?;
 
     let (mut client, server) = tokio::io::duplex(64);
     client.write_all(&[0, 0, 0, 2, 1]).await?;
     client.shutdown().await?;
     prepared.handle_connection(server).await;
-    assert_eq!(teardown_calls.load(Ordering::SeqCst), 2);
+    check_equal(
+        &teardown_calls.load(Ordering::SeqCst),
+        &2,
+        "teardown should run after the duplex processing error",
+    )?;
     Ok(())
 }
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "assertions make concurrent prepared-service reuse explicit"
-)]
 async fn prepared_app_reuses_services_across_overlapping_connections() -> TestResult<()> {
     let transforms = Arc::new(AtomicUsize::new(0));
     let barrier = Arc::new(Barrier::new(CONNECTIONS));
@@ -260,7 +296,11 @@ async fn prepared_app_reuses_services_across_overlapping_connections() -> TestRe
         .prepare()
         .await
         .map_err(|error| -> Box<dyn std::error::Error + Send + Sync> { Box::new(error) })?;
-    assert_eq!(transforms.load(Ordering::SeqCst), 1);
+    check_equal(
+        &transforms.load(Ordering::SeqCst),
+        &1,
+        "preparation should transform the shared route once",
+    )?;
 
     let first_frame = build_frame(1, vec![b'X'])?;
     let second_frame = build_frame(1, vec![b'Y'])?;
@@ -272,78 +312,20 @@ async fn prepared_app_reuses_services_across_overlapping_connections() -> TestRe
     })
     .await
     .map_err(|_| "prepared connections did not overlap")?;
-    assert_eq!(response_payload(&first?)?, [b'X', b'A', b'A']);
-    assert_eq!(response_payload(&second?)?, [b'Y', b'A', b'A']);
-    assert_eq!(transforms.load(Ordering::SeqCst), 1);
-    Ok(())
-}
-
-/// Bounded waits keep the real-socket test from hanging on a stalled peer.
-const TCP_OPERATION_TIMEOUT: Duration = Duration::from_secs(5);
-
-/// Accepts one TCP connection and processes it with the prepared application.
-async fn serve_one_prepared_tcp_connection(
-    listener: TcpListener,
-    prepared: TestPreparedApp,
-) -> TestResult<()> {
-    let (stream, _peer) = timeout(TCP_OPERATION_TIMEOUT, listener.accept())
-        .await
-        .map_err(|_| "prepared TCP accept timed out".to_string())??;
-    timeout(
-        TCP_OPERATION_TIMEOUT,
-        prepared.handle_connection_result(stream),
-    )
-    .await
-    .map_err(|_| "prepared TCP connection processing timed out".to_string())??;
-    Ok(())
-}
-
-/// Sends one encoded frame to the server and returns the decoded response.
-async fn exchange_one_tcp_frame(address: SocketAddr, frame: Vec<u8>) -> TestResult<Vec<u8>> {
-    let mut client = timeout(TCP_OPERATION_TIMEOUT, TcpStream::connect(address))
-        .await
-        .map_err(|_| "prepared TCP client connect timed out".to_string())??;
-    client.write_all(&frame).await?;
-    // Signal end-of-input so the server can finish the request-response cycle.
-    client.shutdown().await?;
-    let mut response = Vec::new();
-    timeout(TCP_OPERATION_TIMEOUT, client.read_to_end(&mut response))
-        .await
-        .map_err(|_| "prepared TCP response read timed out".to_string())??;
-    response_payload(&response)
-}
-
-#[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "assertions make real-socket prepared dispatch and transform reuse explicit"
-)]
-async fn prepared_app_serves_real_tcp_connection_without_retransforming() -> TestResult<()> {
-    let instrumentation = ConnectionStartupInstrumentation::new();
-    let prepared: TestPreparedApp = counted_app_factory(instrumentation.clone())()?
-        .prepare()
-        .await
-        .map_err(|error| -> Box<dyn std::error::Error + Send + Sync> { Box::new(error) })?;
-    let prepared_counts = ConnectionStartupCounts {
-        factory_calls: 1,
-        transforms: ROUTES * MIDDLEWARE_LAYERS,
-    };
-    assert_eq!(instrumentation.snapshot(), prepared_counts);
-
-    let listener = TcpListener::bind("127.0.0.1:0").await?;
-    let address = listener.local_addr()?;
-    let server = tokio::spawn(serve_one_prepared_tcp_connection(listener, prepared));
-
-    let payload = exchange_one_tcp_frame(address, build_frame(1, vec![b'X'])?).await?;
-    // The request gains tags A then B in wrap order; the response unwinds them.
-    assert_eq!(payload, [b'X', b'A', b'B', b'B', b'A']);
-
-    let joined = timeout(TCP_OPERATION_TIMEOUT, server)
-        .await
-        .map_err(|_| "prepared TCP accept task did not finish".to_string())?;
-    // Surfaces both a failed join and a failed connection-processing result.
-    joined??;
-    // Preparation transformed each route once; the connection added none.
-    assert_eq!(instrumentation.snapshot(), prepared_counts);
+    check_equal(
+        &response_payload(&first?)?,
+        b"XAA",
+        "the first overlapping response should carry both middleware tags",
+    )?;
+    check_equal(
+        &response_payload(&second?)?,
+        b"YAA",
+        "the second overlapping response should carry both middleware tags",
+    )?;
+    check_equal(
+        &transforms.load(Ordering::SeqCst),
+        &1,
+        "overlapping connections should not rebuild the shared route",
+    )?;
     Ok(())
 }
diff --git a/tests/prepared_app/tcp.rs b/tests/prepared_app/tcp.rs
new file mode 100644
index 000000000..df7ee7df8
--- /dev/null
+++ b/tests/prepared_app/tcp.rs
@@ -0,0 +1,104 @@
+//! Real-socket coverage for prepared applications.
+//!
+//! These scenarios drive a prepared application over a loopback listener so
+//! that dispatch, middleware ordering, and transport reuse are exercised
+//! through the socket path rather than the in-process frame drivers.
+
+use std::{net::SocketAddr, time::Duration};
+
+use tokio::{
+    io::{AsyncReadExt, AsyncWriteExt},
+    net::{TcpListener, TcpStream},
+    time::timeout,
+};
+use wireframe_testing::TestResult;
+
+use super::{
+    ConnectionStartupCounts,
+    ConnectionStartupInstrumentation,
+    MIDDLEWARE_LAYERS,
+    ROUTES,
+    TestPreparedApp,
+    build_frame,
+    counted_app_factory,
+    fallible_check_equal::check_equal,
+    response_payload,
+};
+
+/// Bounded waits keep the real-socket test from hanging on a stalled peer.
+const TCP_OPERATION_TIMEOUT: Duration = Duration::from_secs(5);
+
+/// Accepts one TCP connection and processes it with the prepared application.
+async fn serve_one_prepared_tcp_connection(
+    listener: TcpListener,
+    prepared: TestPreparedApp,
+) -> TestResult<()> {
+    let (stream, _peer) = timeout(TCP_OPERATION_TIMEOUT, listener.accept())
+        .await
+        .map_err(|_| "prepared TCP accept timed out".to_string())??;
+    timeout(
+        TCP_OPERATION_TIMEOUT,
+        prepared.handle_connection_result(stream),
+    )
+    .await
+    .map_err(|_| "prepared TCP connection processing timed out".to_string())??;
+    Ok(())
+}
+
+/// Sends one encoded frame to the server and returns the decoded response.
+async fn exchange_one_tcp_frame(address: SocketAddr, frame: Vec<u8>) -> TestResult<Vec<u8>> {
+    let mut client = timeout(TCP_OPERATION_TIMEOUT, TcpStream::connect(address))
+        .await
+        .map_err(|_| "prepared TCP client connect timed out".to_string())??;
+    client.write_all(&frame).await?;
+    // Signal end-of-input so the server can finish the request-response cycle.
+    client.shutdown().await?;
+    let mut response = Vec::new();
+    timeout(TCP_OPERATION_TIMEOUT, client.read_to_end(&mut response))
+        .await
+        .map_err(|_| "prepared TCP response read timed out".to_string())??;
+    response_payload(&response)
+}
+
+#[tokio::test]
+async fn prepared_app_serves_real_tcp_connection_without_retransforming() -> TestResult<()> {
+    let instrumentation = ConnectionStartupInstrumentation::new();
+    let prepared: TestPreparedApp = counted_app_factory(instrumentation.clone())()?
+        .prepare()
+        .await
+        .map_err(|error| -> Box<dyn std::error::Error + Send + Sync> { Box::new(error) })?;
+    let prepared_counts = ConnectionStartupCounts {
+        factory_calls: 1,
+        transforms: ROUTES * MIDDLEWARE_LAYERS,
+    };
+    check_equal(
+        &instrumentation.snapshot(),
+        &prepared_counts,
+        "preparation should transform each route once",
+    )?;
+
+    let listener = TcpListener::bind("127.0.0.1:0").await?;
+    let address = listener.local_addr()?;
+    let server = tokio::spawn(serve_one_prepared_tcp_connection(listener, prepared));
+
+    let payload = exchange_one_tcp_frame(address, build_frame(1, vec![b'X'])?).await?;
+    // The request gains tags A then B in wrap order; the response unwinds them.
+    check_equal(
+        &payload,
+        b"XABBA",
+        "the real TCP response should follow middleware order",
+    )?;
+
+    let joined = timeout(TCP_OPERATION_TIMEOUT, server)
+        .await
+        .map_err(|_| "prepared TCP accept task did not finish".to_string())?;
+    // Surfaces both a failed join and a failed connection-processing result.
+    joined??;
+    // Preparation transformed each route once; the connection added none.
+    check_equal(
+        &instrumentation.snapshot(),
+        &prepared_counts,
+        "the real TCP connection should not add transforms",
+    )?;
+    Ok(())
+}
diff --git a/tests/prepared_app_tcp.rs b/tests/prepared_app_tcp.rs
index e70b44907..db5c6ece7 100644
--- a/tests/prepared_app_tcp.rs
+++ b/tests/prepared_app_tcp.rs
@@ -7,6 +7,9 @@ use std::sync::{
     atomic::{AtomicUsize, Ordering},
 };
 
+#[path = "common/fallible_assertions/check_equal.rs"]
+mod fallible_check_equal;
+
 use common::prepared_app::{
     TestApp,
     TransformCountingMiddleware,
@@ -14,6 +17,7 @@ use common::prepared_app::{
     handler,
     response_payload,
 };
+use fallible_check_equal::check_equal;
 use tokio::{
     io::{AsyncReadExt, AsyncWriteExt},
     net::{TcpListener, TcpStream},
@@ -28,10 +32,6 @@ use wireframe_testing::{
 
 /// Prepared applications serve TCP connections without rebuilding middleware.
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "assertions make prepared TCP dispatch and transform reuse explicit"
-)]
 async fn prepared_app_serves_tcp_connection_without_retransforming() -> TestResult<()> {
     let transforms = Arc::new(AtomicUsize::new(0));
     let prepared = TestApp::new()?
@@ -43,7 +43,11 @@ async fn prepared_app_serves_tcp_connection_without_retransforming() -> TestResu
         .prepare()
         .await
         .map_err(|error| -> Box<dyn std::error::Error + Send + Sync> { Box::new(error) })?;
-    assert_eq!(transforms.load(Ordering::SeqCst), 1);
+    check_equal(
+        &transforms.load(Ordering::SeqCst),
+        &1,
+        "preparation should transform the route once",
+    )?;
 
     let listener = TcpListener::bind("127.0.0.1:0").await?;
     let address = listener.local_addr()?;
@@ -59,8 +63,16 @@ async fn prepared_app_serves_tcp_connection_without_retransforming() -> TestResu
     client.read_to_end(&mut response).await?;
     server.await??;
 
-    assert_eq!(response_payload(&response)?, [b'X', b'A', b'A']);
-    assert_eq!(transforms.load(Ordering::SeqCst), 1);
+    check_equal(
+        &response_payload(&response)?,
+        b"XAA",
+        "the TCP response should carry both middleware tags",
+    )?;
+    check_equal(
+        &transforms.load(Ordering::SeqCst),
+        &1,
+        "serving the TCP connection should not rebuild the route",
+    )?;
     Ok(())
 }
 
@@ -76,10 +88,6 @@ async fn exchange_frame(address: std::net::SocketAddr, frame: Vec<u8>) -> TestRe
 
 /// `from_app` prepares its supplied application once and shares it over TCP.
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "assertions make public server preparation and route reuse explicit"
-)]
 async fn from_app_serves_concurrent_tcp_connections_from_one_prepared_root() -> TestResult<()> {
     let transforms = Arc::new(AtomicUsize::new(0));
     let app = TestApp::new()?
@@ -111,9 +119,21 @@ async fn from_app_serves_concurrent_tcp_connections_from_one_prepared_root() ->
         exchange_frame(address, frame.clone()),
         exchange_frame(address, frame)
     );
-    assert_eq!(first?, [b'X', b'A', b'A']);
-    assert_eq!(second?, [b'X', b'A', b'A']);
-    assert_eq!(transforms.load(Ordering::SeqCst), 1);
+    check_equal(
+        &first?,
+        b"XAA",
+        "the first concurrent connection should follow middleware order",
+    )?;
+    check_equal(
+        &second?,
+        b"XAA",
+        "the second concurrent connection should follow middleware order",
+    )?;
+    check_equal(
+        &transforms.load(Ordering::SeqCst),
+        &1,
+        "from_app should prepare the application once for both workers",
+    )?;
 
     shutdown_tx
         .send(())
diff --git a/tests/push.rs b/tests/push.rs
index 0c053baca..3dbc2e38f 100644
--- a/tests/push.rs
+++ b/tests/push.rs
@@ -5,6 +5,13 @@
 
 mod support;
 
+#[path = "common/fallible_assertions/check.rs"]
+mod fallible_check;
+#[path = "common/fallible_assertions/check_eq.rs"]
+mod fallible_check_eq;
+
+use fallible_check::check;
+use fallible_check_eq::check_eq;
 use futures::FutureExt;
 use rstest::{fixture, rstest};
 use tokio::time::{self, Duration};
@@ -58,10 +65,6 @@ fn builder_accepts_max_rate() {
 
 /// Disabling throttling allows rapid bursts to succeed.
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn disables_throttling_allows_burst_pushes() -> TestResult<()> {
     time::pause();
     let (_queues, handle) = support::builder::<u8>()
@@ -73,12 +76,11 @@ async fn disables_throttling_allows_burst_pushes() -> TestResult<()> {
         push_expect!(handle.push_high_priority(i))?;
         push_expect!(handle.push_low_priority(i))?;
     }
-    let res = time::timeout(Duration::from_millis(10), handle.push_high_priority(99)).await;
-    let push_res = res.expect("push should not block when throttling disabled");
-    assert!(
-        push_res.is_ok(),
-        "push should not error when throttling disabled"
-    );
+    let push_res = time::timeout(Duration::from_millis(10), handle.push_high_priority(99))
+        .await
+        .map_err(|error| format!("push should not block when throttling disabled: {error}"))?;
+    push_res
+        .map_err(|error| format!("push should not error when throttling is disabled: {error}"))?;
     Ok(())
 }
 
@@ -108,10 +110,6 @@ fn builder_rejects_zero_capacity() {
 
 /// Frames are delivered to queues matching their push priority.
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn frames_routed_to_correct_priority_queues() -> TestResult<()> {
     let (mut queues, handle) = small_queues()?;
 
@@ -121,18 +119,18 @@ async fn frames_routed_to_correct_priority_queues() -> TestResult<()> {
     let (prio1, frame1) = recv_expect!(queues.recv())?;
     let (prio2, frame2) = recv_expect!(queues.recv())?;
 
-    assert_eq!(
+    check_eq(
         prio1,
         PushPriority::High,
-        "first frame should be high priority"
-    );
-    assert_eq!(frame1, 2, "unexpected first frame value");
-    assert_eq!(
+        "first frame should be high priority",
+    )?;
+    check_eq(frame1, 2, "unexpected first frame value")?;
+    check_eq(
         prio2,
         PushPriority::Low,
-        "second frame should be low priority"
-    );
-    assert_eq!(frame2, 1, "unexpected second frame value");
+        "second frame should be low priority",
+    )?;
+    check_eq(frame2, 1, "unexpected second frame value")?;
     Ok(())
 }
 
@@ -141,48 +139,40 @@ async fn frames_routed_to_correct_priority_queues() -> TestResult<()> {
 /// Using [`PushPolicy::ReturnErrorIfFull`] causes `try_push` to
 /// return [`PushError::QueueFull`] once the queue is at capacity.
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn try_push_respects_policy() -> TestResult<()> {
     let (mut queues, handle) = small_queues()?;
 
     push_expect!(handle.push_high_priority(1u8))?;
     let result = handle.try_push(2u8, PushPriority::High, PushPolicy::ReturnErrorIfFull);
-    assert!(
-        matches!(result, Err(PushError::QueueFull)),
-        "expected queue full error"
-    );
+    check(
+        matches!(&result, Err(PushError::QueueFull)),
+        format!("expected queue full error, got {result:?}"),
+    )?;
 
     // drain queue to allow new push
     let _ = queues.recv().await;
     push_expect!(handle.push_high_priority(3u8))?;
     let (_, last) = recv_expect!(queues.recv())?;
-    assert_eq!(last, 3, "unexpected drained frame");
+    check_eq(last, 3, "unexpected drained frame")?;
     Ok(())
 }
 
 /// Push attempts return `Closed` when all queues have been shut down.
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn push_queues_error_on_closed() -> TestResult<()> {
     let (mut queues, handle) = small_queues()?;
     queues.close();
     let res = handle.push_high_priority(42u8).await;
-    assert!(
-        matches!(res, Err(PushError::Closed)),
-        "expected closed error on high priority push"
-    );
+    check(
+        matches!(&res, Err(PushError::Closed)),
+        format!("expected closed error on high priority push, got {res:?}"),
+    )?;
 
     let res = handle.push_low_priority(24u8).await;
-    assert!(
-        matches!(res, Err(PushError::Closed)),
-        "expected closed error on low priority push"
-    );
+    check(
+        matches!(&res, Err(PushError::Closed)),
+        format!("expected closed error on low priority push, got {res:?}"),
+    )?;
     Ok(())
 }
 
@@ -207,10 +197,11 @@ async fn rate_limiter_blocks_when_exceeded(#[case] priority: PushPriority) -> Te
         PushPriority::Low => handle.push_low_priority(2u8).boxed(),
     };
     tokio::task::yield_now().await; // register w/ scheduler
-    assert!(
-        fut.as_mut().now_or_never().is_none(),
-        "second push should be pending under rate limit"
-    );
+    let completed = fut.as_mut().now_or_never();
+    check(
+        completed.is_none(),
+        format!("second push should be pending under rate limit, got {completed:?}"),
+    )?;
 
     time::advance(Duration::from_secs(1)).await;
     match priority {
@@ -220,20 +211,16 @@ async fn rate_limiter_blocks_when_exceeded(#[case] priority: PushPriority) -> Te
 
     let (_, first) = recv_expect!(queues.recv())?;
     let (_, second) = recv_expect!(queues.recv())?;
-    assert_eq!(
+    check_eq(
         (first, second),
         (1, 3),
-        "unexpected drained frames under rate limit"
-    );
+        "unexpected drained frames under rate limit",
+    )?;
     Ok(())
 }
 
 /// Exceeding the rate limit succeeds after the window has passed.
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn rate_limiter_allows_after_wait() -> TestResult<()> {
     time::pause();
     let (mut queues, handle) = queues()?;
@@ -243,7 +230,7 @@ async fn rate_limiter_allows_after_wait() -> TestResult<()> {
 
     let (_, a) = recv_expect!(queues.recv())?;
     let (_, b) = recv_expect!(queues.recv())?;
-    assert_eq!((a, b), (1, 2), "unexpected frame ordering after wait");
+    check_eq((a, b), (1, 2), "unexpected frame ordering after wait")?;
     Ok(())
 }
 
@@ -251,10 +238,6 @@ async fn rate_limiter_allows_after_wait() -> TestResult<()> {
 /// The token bucket is shared, so pushes from one priority reduce
 /// the allowance for the other.
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn rate_limiter_shared_across_priorities() -> TestResult<()> {
     time::pause();
     let (mut queues, handle) = queues()?;
@@ -262,49 +245,43 @@ async fn rate_limiter_shared_across_priorities() -> TestResult<()> {
 
     let mut fut = handle.push_low_priority(2u8).boxed();
     tokio::task::yield_now().await;
-    assert!(
-        fut.as_mut().now_or_never().is_none(),
-        "second push should be pending across queues"
-    );
+    let completed = fut.as_mut().now_or_never();
+    check(
+        completed.is_none(),
+        format!("second push should be pending across queues, got {completed:?}"),
+    )?;
 
     time::advance(Duration::from_secs(1)).await;
     push_expect!(handle.push_low_priority(2u8))?;
 
     let (prio1, frame1) = recv_expect!(queues.recv())?;
     let (prio2, frame2) = recv_expect!(queues.recv())?;
-    assert_eq!(prio1, PushPriority::High, "first priority should be high");
-    assert_eq!(frame1, 1, "unexpected first frame value");
-    assert_eq!(prio2, PushPriority::Low, "second priority should be low");
-    assert_eq!(frame2, 2, "unexpected second frame value");
+    check_eq(prio1, PushPriority::High, "first priority should be high")?;
+    check_eq(frame1, 1, "unexpected first frame value")?;
+    check_eq(prio2, PushPriority::Low, "second priority should be low")?;
+    check_eq(frame2, 2, "unexpected second frame value")?;
     Ok(())
 }
 
 /// Unlimited queues never block pushes.
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn unlimited_queues_do_not_block() -> TestResult<()> {
     time::pause();
     let (mut queues, handle) = support::builder::<u8>().unlimited().build()?;
     push_expect!(handle.push_high_priority(1u8))?;
-    let res = time::timeout(Duration::from_millis(10), handle.push_low_priority(2u8)).await;
-    assert!(res.is_ok(), "pushes should not block when unlimited");
+    let _ = time::timeout(Duration::from_millis(10), handle.push_low_priority(2u8))
+        .await
+        .map_err(|error| format!("pushes should not block when unlimited: {error}"))?;
 
     let (_, a) = recv_expect!(queues.recv())?;
     let (_, b) = recv_expect!(queues.recv())?;
-    assert_eq!((a, b), (1, 2), "unexpected ordering for unlimited queues");
+    check_eq((a, b), (1, 2), "unexpected ordering for unlimited queues")?;
     Ok(())
 }
 
 /// A burst up to capacity succeeds and further pushes are blocked.
 /// The maximum burst size equals the configured `capacity` parameter.
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn rate_limiter_allows_burst_within_capacity_and_blocks_excess() -> TestResult<()> {
     time::pause();
     let (mut queues, handle) = support::builder::<u8>()
@@ -319,20 +296,18 @@ async fn rate_limiter_allows_burst_within_capacity_and_blocks_excess() -> TestRe
 
     let mut fut = handle.push_high_priority(99).boxed();
     tokio::task::yield_now().await;
-    assert!(
-        fut.as_mut().now_or_never().is_none(),
-        "push exceeding burst capacity should be pending"
-    );
+    let completed = fut.as_mut().now_or_never();
+    check(
+        completed.is_none(),
+        format!("push exceeding burst capacity should be pending, got {completed:?}"),
+    )?;
 
     time::advance(Duration::from_secs(1)).await;
     push_expect!(handle.push_high_priority(100))?;
 
     for expected in [0u8, 1u8, 2u8, 100u8] {
         let (_, frame) = recv_expect!(queues.recv())?;
-        assert_eq!(
-            frame, expected,
-            "frames drained in unexpected order: expected {expected}, got {frame}"
-        );
+        check_eq(frame, expected, "frames drained in unexpected order")?;
     }
     Ok(())
 }
diff --git a/tests/routes.rs b/tests/routes.rs
index aba0d8dd5..d2b9f0932 100644
--- a/tests/routes.rs
+++ b/tests/routes.rs
@@ -32,10 +32,6 @@ type TestApp = wireframe::app::WireframeApp<BincodeSerializer, (), CommonTestEnv
 struct Echo(u8);
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn handler_receives_message_and_echoes_response() -> TestResult<()> {
     let called = Arc::new(AtomicUsize::new(0));
     let called_clone = called.clone();
@@ -63,22 +59,25 @@ async fn handler_receives_message_and_echoes_response() -> TestResult<()> {
         return Err("expected a single response frame".into());
     };
     let (resp_env, _) = BincodeSerializer.deserialize::<CommonTestEnvelope>(first)?;
-    assert_eq!(resp_env.correlation_id, Some(99), "correlation id mismatch");
+    if resp_env.correlation_id != Some(99) {
+        return Err(format!(
+            "correlation id mismatch: expected Some(99), got {:?}",
+            resp_env.correlation_id
+        )
+        .into());
+    }
     let (echo, _) = Echo::from_bytes(&resp_env.payload)?;
-    assert_eq!(echo, Echo(42), "echo payload mismatch");
-    assert_eq!(
-        called.load(Ordering::SeqCst),
-        1,
-        "route not invoked exactly once"
-    );
+    if echo != Echo(42) {
+        return Err(format!("echo payload mismatch: expected Echo(42), got {echo:?}").into());
+    }
+    let call_count = called.load(Ordering::SeqCst);
+    if call_count != 1 {
+        return Err(format!("route not invoked exactly once: got {call_count} calls").into());
+    }
     Ok(())
 }
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn handler_echoes_with_none_correlation_id() -> TestResult<()> {
     let app = TestApp::new()?.route(
         1,
@@ -99,20 +98,21 @@ async fn handler_echoes_with_none_correlation_id() -> TestResult<()> {
     };
     let (resp_env, _) = BincodeSerializer.deserialize::<CommonTestEnvelope>(first)?;
 
-    assert!(
-        resp_env.correlation_id.is_none(),
-        "unexpected correlation id"
-    );
+    if resp_env.correlation_id.is_some() {
+        return Err(format!(
+            "unexpected correlation id: got {:?}",
+            resp_env.correlation_id
+        )
+        .into());
+    }
     let (echo, _) = Echo::from_bytes(&resp_env.payload)?;
-    assert_eq!(echo, Echo(7), "echo payload mismatch");
+    if echo != Echo(7) {
+        return Err(format!("echo payload mismatch: expected Echo(7), got {echo:?}").into());
+    }
     Ok(())
 }
 
 #[tokio::test]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "asserts provide clearer diagnostics in tests"
-)]
 async fn multiple_frames_processed_in_sequence() -> TestResult<()> {
     let app = TestApp::new()?.route(
         1,
@@ -144,18 +144,28 @@ async fn multiple_frames_processed_in_sequence() -> TestResult<()> {
     let (echo1, _) = Echo::from_bytes(&env1.payload)?;
     let (env2, _) = BincodeSerializer.deserialize::<CommonTestEnvelope>(second)?;
     let (echo2, _) = Echo::from_bytes(&env2.payload)?;
-    assert_eq!(
-        env1.correlation_id,
-        Some(1),
-        "first correlation id mismatch"
-    );
-    assert_eq!(
-        env2.correlation_id,
-        Some(2),
-        "second correlation id mismatch"
-    );
-    assert_eq!(echo1, Echo(1), "first echo payload mismatch");
-    assert_eq!(echo2, Echo(2), "second echo payload mismatch");
+    if env1.correlation_id != Some(1) {
+        return Err(format!(
+            "first correlation id mismatch: expected Some(1), got {:?}",
+            env1.correlation_id
+        )
+        .into());
+    }
+    if env2.correlation_id != Some(2) {
+        return Err(format!(
+            "second correlation id mismatch: expected Some(2), got {:?}",
+            env2.correlation_id
+        )
+        .into());
+    }
+    if echo1 != Echo(1) {
+        return Err(format!("first echo payload mismatch: expected Echo(1), got {echo1:?}").into());
+    }
+    if echo2 != Echo(2) {
+        return Err(
+            format!("second echo payload mismatch: expected Echo(2), got {echo2:?}").into(),
+        );
+    }
     Ok(())
 }
 
@@ -189,7 +199,13 @@ async fn single_frame_propagates_correlation_id(#[case] cid: Option<u64>) -> Tes
     };
     let (resp, _) = BincodeSerializer.deserialize::<CommonTestEnvelope>(first)?;
 
-    assert_eq!(resp.correlation_id, cid, "correlation id mismatch");
+    if resp.correlation_id != cid {
+        return Err(format!(
+            "correlation id mismatch: expected {cid:?}, got {:?}",
+            resp.correlation_id
+        )
+        .into());
+    }
     Ok(())
 }
 
diff --git a/tests/workspace_manifest.rs b/tests/workspace_manifest.rs
index a9770c81c..061182527 100644
--- a/tests/workspace_manifest.rs
+++ b/tests/workspace_manifest.rs
@@ -11,6 +11,13 @@ mod workspace_manifest_support;
 #[path = "common/repo_access.rs"]
 mod repo_access;
 
+#[path = "common/fallible_assertions/check.rs"]
+mod fallible_check;
+#[path = "common/fallible_assertions/check_equal.rs"]
+mod fallible_check_equal;
+
+use fallible_check::check;
+use fallible_check_equal::check_equal;
 use repo_access::{read_repo_file, repo_root};
 use rstest::rstest;
 use serde_json::Value;
@@ -33,64 +40,82 @@ fn contains_json_string_field(json: &str, field: &str, value: &str) -> bool {
     json.contains(&format!("\"{field}\":\"{escaped}\""))
 }
 
+/// Extract a field that `cargo metadata` should expose as a JSON array.
+fn metadata_array<'a>(metadata_json: &'a Value, field: &str) -> Result<&'a [Value], String> {
+    metadata_json
+        .get(field)
+        .and_then(Value::as_array)
+        .map(Vec::as_slice)
+        .ok_or_else(|| {
+            format!(
+                "cargo metadata should expose {field} as an array, got {:?}",
+                metadata_json.get(field)
+            )
+        })
+}
+
+fn check_workspace_members(members: &[Value], expected: &[(&str, &str)]) -> TestResult {
+    for (package_id, description) in expected {
+        check(
+            members
+                .iter()
+                .any(|member| member.as_str() == Some(*package_id)),
+            format!("workspace_members should include the id of {description}"),
+        )?;
+    }
+    Ok(())
+}
+
 #[rstest]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "assertions provide clearer diagnostics in integration tests"
-)]
 fn root_manifest_declares_explicit_workspace_section() -> TestResult {
     let manifest = root_manifest()?;
-    assert!(
+    check(
         has_manifest_table(&manifest, "[workspace]"),
-        "root Cargo.toml should declare an explicit [workspace] section"
-    );
-    assert!(
+        "root Cargo.toml should declare an explicit [workspace] section",
+    )?;
+    check(
         has_manifest_line(
             &manifest,
             "[workspace]",
             "members = [\".\", \"crates/wireframe-loom\", \"crates/wireframe-verification\", \
-             \"wireframe_testing\"]"
+             \"wireframe_testing\"]",
         ),
-        "the workspace should explicitly list the root, Loom, verification, and testing crates"
-    );
-    assert!(
+        "the workspace should explicitly list the root, Loom, verification, and testing crates",
+    )?;
+    check(
         has_manifest_line(&manifest, "[workspace]", "default-members = [\".\"]"),
-        "15.1.2 should keep the root package as the only default workspace member"
-    );
-    assert!(
+        "15.1.2 should keep the root package as the only default workspace member",
+    )?;
+    check(
         has_manifest_line(&manifest, "[workspace]", "resolver = \"3\""),
-        "the hybrid workspace should opt into the edition-2024 resolver"
-    );
+        "the hybrid workspace should opt into the edition-2024 resolver",
+    )?;
     Ok(())
 }
 
 #[rstest]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "assertions provide clearer diagnostics in integration tests"
-)]
 fn companion_crates_inherit_private_documentation_clippy_policy() -> TestResult {
     let root_manifest = root_manifest()?;
-    assert!(
+    check(
         has_manifest_table(&root_manifest, "[workspace.lints.clippy]"),
-        "the workspace must define its shared Clippy policy"
-    );
-    assert!(
+        "the workspace must define its shared Clippy policy",
+    )?;
+    check(
         has_manifest_line(
             &root_manifest,
             "[workspace.lints.clippy]",
-            "missing_docs_in_private_items = \"deny\""
+            "missing_docs_in_private_items = \"deny\"",
         ),
-        "the shared Clippy policy must deny undocumented private implementation items"
-    );
-    assert!(
+        "the shared Clippy policy must deny undocumented private implementation items",
+    )?;
+    check(
         has_manifest_table(&root_manifest, "[lints]"),
-        "the root package must opt into workspace lint inheritance"
-    );
-    assert!(
+        "the root package must opt into workspace lint inheritance",
+    )?;
+    check(
         has_manifest_line(&root_manifest, "[lints]", "workspace = true"),
-        "the root package must inherit the shared Clippy policy"
-    );
+        "the root package must inherit the shared Clippy policy",
+    )?;
 
     for (package_name, manifest_path) in [
         ("wireframe_testing", "wireframe_testing/Cargo.toml"),
@@ -101,23 +126,19 @@ fn companion_crates_inherit_private_documentation_clippy_policy() -> TestResult
         ("wireframe-loom", "crates/wireframe-loom/Cargo.toml"),
     ] {
         let manifest = read_repo_file(manifest_path)?;
-        assert!(
+        check(
             has_manifest_table(&manifest, "[lints]"),
-            "{package_name} must opt into workspace lint inheritance"
-        );
-        assert!(
+            format!("{package_name} must opt into workspace lint inheritance"),
+        )?;
+        check(
             has_manifest_line(&manifest, "[lints]", "workspace = true"),
-            "{package_name} must inherit the shared Clippy policy"
-        );
+            format!("{package_name} must inherit the shared Clippy policy"),
+        )?;
     }
     Ok(())
 }
 
 #[rstest]
-#[expect(
-    clippy::panic_in_result_fn,
-    reason = "assertions provide clearer diagnostics in integration tests"
-)]
 fn cargo_metadata_reports_explicit_members_without_widening_default_members() -> TestResult {
     let repo_root = repo_root()?;
     let repo_root_str = repo_root.as_str();
@@ -130,56 +151,46 @@ fn cargo_metadata_reports_explicit_members_without_widening_default_members() ->
     let metadata = cargo_metadata()?;
     let metadata_json = parse_metadata_json(&metadata)?;
 
-    assert!(
+    check(
         contains_json_string_field(&metadata, "workspace_root", repo_root_str),
-        "workspace_root should be the repository root"
-    );
-    assert!(
+        "workspace_root should be the repository root",
+    )?;
+    check(
         contains_json_string_field(&metadata, "manifest_path", manifest_path_str),
-        "metadata should continue to resolve the root package manifest"
-    );
-    assert!(
+        "metadata should continue to resolve the root package manifest",
+    )?;
+    check(
         metadata.contains(&root_package_id),
-        "workspace metadata should include the root package"
-    );
-    let workspace_members = metadata_json
-        .get("workspace_members")
-        .and_then(Value::as_array)
-        .expect("cargo metadata should expose workspace_members as an array");
-    for (package_id, description) in [
-        (&root_package_id, "the root package"),
-        (&verification_package_id, "the verification crate"),
-        (&loom_package_id, "the wireframe-loom crate"),
-        (&helper_package_id, "the wireframe_testing crate"),
-    ] {
-        assert!(
-            workspace_members
-                .iter()
-                .any(|member| member.as_str() == Some(package_id.as_str())),
-            "workspace_members should include the id of {description}"
-        );
-    }
-    assert!(
+        "workspace metadata should include the root package",
+    )?;
+    let workspace_members = metadata_array(&metadata_json, "workspace_members")?;
+    check_workspace_members(
+        workspace_members,
+        &[
+            (&root_package_id, "the root package"),
+            (&verification_package_id, "the verification crate"),
+            (&loom_package_id, "the wireframe-loom crate"),
+            (&helper_package_id, "the wireframe_testing crate"),
+        ],
+    )?;
+    check(
         metadata.contains("wireframe-verification"),
-        "15.1.2 should add the verification crate to cargo metadata"
-    );
-    assert!(
+        "15.1.2 should add the verification crate to cargo metadata",
+    )?;
+    check(
         metadata.contains("wireframe_testing"),
-        "workspace metadata should include the test helper crate"
-    );
-    let workspace_default_members = metadata_json
-        .get("workspace_default_members")
-        .and_then(Value::as_array)
-        .expect("cargo metadata should expose workspace_default_members as an array");
-    assert_eq!(
-        workspace_default_members.len(),
-        1,
-        "15.1.2 should keep exactly one default workspace member"
-    );
-    assert_eq!(
-        workspace_default_members.first().and_then(Value::as_str),
-        Some(root_package_id.as_str()),
-        "15.1.2 should keep the root package as the only default workspace member"
-    );
+        "workspace metadata should include the test helper crate",
+    )?;
+    let workspace_default_members = metadata_array(&metadata_json, "workspace_default_members")?;
+    check_equal(
+        &workspace_default_members.len(),
+        &1_usize,
+        "15.1.2 should keep exactly one default workspace member",
+    )?;
+    check_equal(
+        &workspace_default_members.first().and_then(Value::as_str),
+        &Some(root_package_id.as_str()),
+        "15.1.2 should keep the root package as the only default workspace member",
+    )?;
     Ok(())
 }