diff --git a/src/bindings/loop_api/asyncio_cache.rs b/src/bindings/loop_api/asyncio_cache.rs
index 8dc529e..a32407b 100644
--- a/src/bindings/loop_api/asyncio_cache.rs
+++ b/src/bindings/loop_api/asyncio_cache.rs
@@ -39,6 +39,7 @@ cached_function!(
 );
 cached_function!(future_done_stop_fn, "rsloop._loop", "future_done_stop");
 
+#[cfg_attr(feature = "profile", hotpath::measure)]
 #[pyfunction]
 fn loop_reference_gone(_reference: &Bound<'_, PyAny>) {}
 
@@ -49,6 +50,7 @@ struct LoopBoundCallback {
 }
 #[pymethods]
 impl LoopBoundCallback {
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "LoopBoundCallback"))]
     fn __call__(&self, py: Python<'_>, arg: &Bound<'_, PyAny>) -> PyResult<()> {
         use pyo3::types::PyWeakrefMethods;
         if let Some(loop_obj) = self.loop_ref.bind(py).upgrade() {
@@ -58,6 +60,7 @@ impl LoopBoundCallback {
     }
 }
 
+#[cfg_attr(feature = "profile", hotpath::measure)]
 pub(super) fn loop_bound_callback(
     py: Python<'_>,
     core: &crate::engine::LoopCore,
@@ -108,6 +111,7 @@ pub(super) fn loop_bound_callback(
     Ok(callbacks[index].clone_ref(py))
 }
 
+#[cfg_attr(feature = "profile", hotpath::measure)]
 pub(super) fn ensure_future<'py>(
     py: Python<'py>,
     future: &Py<PyAny>,
diff --git a/src/bindings/loop_api/socket_operation.rs b/src/bindings/loop_api/socket_operation.rs
index 00574bc..e5f8e04 100644
--- a/src/bindings/loop_api/socket_operation.rs
+++ b/src/bindings/loop_api/socket_operation.rs
@@ -370,6 +370,10 @@ fn arm_wait(py: Python<'_>, operation: Py<SocketOperation>) -> PyResult<()> {
 /// when the service future is cancelled before its first poll.
 struct RequestCleanup(Arc<WakeState>);
 impl Drop for RequestCleanup {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<RequestCleanup as Drop>")
+    )]
     fn drop(&mut self) {
         let pending = self
             .0
diff --git a/src/engine/callbacks.rs b/src/engine/callbacks.rs
index 2c96b88..9fac2dc 100644
--- a/src/engine/callbacks.rs
+++ b/src/engine/callbacks.rs
@@ -53,6 +53,10 @@ struct CallbackResult<'py> {
     _python: std::marker::PhantomData<Python<'py>>,
 }
 impl Drop for CallbackResult<'_> {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<CallbackResult as Drop>")
+    )]
     fn drop(&mut self) {
         // SAFETY: this owns a nonnull result reference; the Python lifetime
         // prevents moving its destruction outside the attached invocation.
@@ -248,6 +252,7 @@ impl<const SOURCE_WORDS: usize> ReadyCallback<SOURCE_WORDS> {
         self.invoke_with(py, |ready| ready.invoke_direct(py))
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "ReadyCallback"))]
     /// Invoke a scheduled callback and release its result after context exit.
     pub fn invoke_discard(&self, py: Python<'_>) -> PyResult<()> {
         let result = self.invoke_with(py, |ready| ready.invoke_direct_raw(py))?;
@@ -255,6 +260,7 @@ impl<const SOURCE_WORDS: usize> ReadyCallback<SOURCE_WORDS> {
         Ok(())
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "ReadyCallback"))]
     fn invoke_with<R>(&self, py: Python<'_>, call: impl Fn(&Self) -> PyResult<R>) -> PyResult<R> {
         if !self.context_needs_run {
             return call(self);
@@ -291,6 +297,7 @@ impl<const SOURCE_WORDS: usize> ReadyCallback<SOURCE_WORDS> {
         }
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "ReadyCallback"))]
     fn invoke_direct_raw<'py>(&self, py: Python<'py>) -> PyResult<CallbackResult<'py>> {
         // SAFETY: the callback and stored arguments remain owned by self. The
         // inline arrays and tuple are valid for the synchronous call, which
diff --git a/src/engine/loop_core.rs b/src/engine/loop_core.rs
index 0fde1fa..16da2cb 100644
--- a/src/engine/loop_core.rs
+++ b/src/engine/loop_core.rs
@@ -104,6 +104,7 @@ struct WaitForWake<'a> {
     wake: &'a LoopWake,
 }
 impl<'a> WaitForWake<'a> {
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "WaitForWake"))]
     fn new(wake: &'a LoopWake) -> Self {
         Self { wake }
     }
@@ -338,6 +339,10 @@ pub struct LoopState {
 }
 
 impl crate::state_snapshot::SnapshotFlags for LoopState {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<LoopState as crate :: state_snapshot :: SnapshotFlags>")
+    )]
     fn snapshot_flags(&self) -> u8 {
         u8::from(self.running) | (u8::from(self.stopping) << 1) | (u8::from(self.closed) << 2)
     }
@@ -491,6 +496,7 @@ impl LoopCore {
         self.send_remote_command(LoopCommand::ScheduleReadyHandle(handle))
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "LoopCore"))]
     #[inline]
     pub(crate) fn current_turn(&self) -> u64 {
         self.turn_number.load(Ordering::Relaxed)
diff --git a/src/state_snapshot.rs b/src/state_snapshot.rs
index 9a86737..3208214 100644
--- a/src/state_snapshot.rs
+++ b/src/state_snapshot.rs
@@ -24,6 +24,7 @@ pub struct SnapshotGuard<'a, T: SnapshotFlags> {
 }
 
 impl<T: SnapshotFlags> SnapshotMutex<T> {
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "SnapshotMutex"))]
     pub fn new(value: T) -> Self {
         Self {
             flags: AtomicU8::new(value.snapshot_flags()),
@@ -31,11 +32,13 @@ impl<T: SnapshotFlags> SnapshotMutex<T> {
         }
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "SnapshotMutex"))]
     #[inline]
     pub fn flags(&self) -> u8 {
         self.flags.load(Ordering::Acquire)
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "SnapshotMutex"))]
     pub fn lock(&self) -> LockResult<SnapshotGuard<'_, T>> {
         self.inner
             .lock()
@@ -43,6 +46,7 @@ impl<T: SnapshotFlags> SnapshotMutex<T> {
             .map_err(|error| PoisonError::new(self.wrap(error.into_inner())))
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "SnapshotMutex"))]
     fn wrap<'a>(&'a self, guard: MutexGuard<'a, T>) -> SnapshotGuard<'a, T> {
         SnapshotGuard {
             owner: self,
@@ -51,6 +55,7 @@ impl<T: SnapshotFlags> SnapshotMutex<T> {
         }
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "SnapshotMutex"))]
     pub fn wait_timeout<'a>(
         &'a self,
         cv: &Condvar,
@@ -74,6 +79,7 @@ impl<T: SnapshotFlags> SnapshotMutex<T> {
 }
 
 impl<T: SnapshotFlags> SnapshotGuard<'_, T> {
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "SnapshotGuard"))]
     fn publish(&mut self) {
         if self.dirty {
             let flags = self
@@ -91,17 +97,29 @@ impl<T: SnapshotFlags> SnapshotGuard<'_, T> {
 
 impl<T: SnapshotFlags> Deref for SnapshotGuard<'_, T> {
     type Target = T;
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<SnapshotGuard as Deref>")
+    )]
     fn deref(&self) -> &T {
         self.guard.as_ref().expect("live snapshot guard")
     }
 }
 impl<T: SnapshotFlags> DerefMut for SnapshotGuard<'_, T> {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<SnapshotGuard as DerefMut>")
+    )]
     fn deref_mut(&mut self) -> &mut T {
         self.dirty = true;
         self.guard.as_mut().expect("live snapshot guard")
     }
 }
 impl<T: SnapshotFlags> Drop for SnapshotGuard<'_, T> {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<SnapshotGuard as Drop>")
+    )]
     fn drop(&mut self) {
         self.publish();
     }
@@ -117,6 +135,57 @@ mod tests {
         }
     }
 
+    #[test]
+    fn timed_wait_publishes_flags_and_returns_a_live_guard() {
+        let state = SnapshotMutex::new(State(0));
+        let cv = Condvar::new();
+        let mut guard = state.lock().unwrap();
+        guard.0 = 3;
+
+        let (mut guard, timeout) = state.wait_timeout(&cv, guard, Duration::ZERO).unwrap();
+        assert!(timeout.timed_out());
+        assert_eq!(state.flags(), 3);
+        assert_eq!(guard.0, 3);
+
+        guard.0 = 5;
+        drop(guard);
+        assert_eq!(state.flags(), 5);
+        assert_eq!(state.lock().unwrap().0, 5);
+    }
+
+    #[test]
+    fn notified_wait_observes_other_thread_mutation_and_republishes() {
+        let state = SnapshotMutex::new(State(0));
+        let cv = Condvar::new();
+        let mut guard = state.lock().unwrap();
+        guard.0 = 3;
+
+        std::thread::scope(|scope| {
+            let notifier = scope.spawn(|| {
+                let mut other = state.lock().unwrap();
+                assert_eq!(state.flags(), 3);
+                assert_eq!(other.0, 3);
+                other.0 = 5;
+                drop(other);
+                cv.notify_one();
+            });
+            // Condvar waits may wake spuriously. Check the protected predicate.
+            while guard.0 != 5 {
+                let (next, timeout) = state
+                    .wait_timeout(&cv, guard, Duration::from_secs(5))
+                    .unwrap();
+                guard = next;
+                assert!(guard.0 == 5 || !timeout.timed_out());
+            }
+            notifier.join().unwrap();
+            assert_eq!(state.flags(), 5);
+            guard.0 = 7;
+            drop(guard);
+            assert_eq!(state.flags(), 7);
+            assert_eq!(state.lock().unwrap().0, 7);
+        });
+    }
+
     #[test]
     fn mutation_and_unwind_publish_flags_before_unlocking() {
         let state = SnapshotMutex::new(State(0));
diff --git a/src/transport/stream/buffers.rs b/src/transport/stream/buffers.rs
index e14aea9..f5d13f3 100644
--- a/src/transport/stream/buffers.rs
+++ b/src/transport/stream/buffers.rs
@@ -42,6 +42,7 @@ pub(super) struct OwnedReadBuffer {
 }
 
 impl OwnedReadBuffer {
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "OwnedReadBuffer"))]
     pub(super) fn is_pooled(&self) -> bool {
         self.pool.is_some()
     }
@@ -235,6 +236,7 @@ impl OwnedWriteBuffer {
         self.bytes.extend_from_slice(data);
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "OwnedWriteBuffer"))]
     /// Materialize Python-backed batches in the transport pool so their
     /// allocation is recycled when the staged write finishes.
     pub(super) fn extend_from_slice_pooled(&mut self, data: &[u8], pool: &Arc<WriteBufferPool>) {
diff --git a/src/transport/stream/connect.rs b/src/transport/stream/connect.rs
index 0bb988e..7a2ea72 100644
--- a/src/transport/stream/connect.rs
+++ b/src/transport/stream/connect.rs
@@ -29,6 +29,7 @@ struct ConnectDone {
 }
 #[pymethods]
 impl ConnectDone {
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "ConnectDone"))]
     fn __call__(&self, _future: &Bound<'_, PyAny>) {
         self.wake.done.store(true, Ordering::Release);
         self.wake.waker.wake();
@@ -39,6 +40,7 @@ impl ConnectDone {
         }
         Ok(())
     }
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "ConnectDone"))]
     fn __clear__(&mut self) {
         self.socket = None;
         self.wake.done.store(true, Ordering::Release);
@@ -46,6 +48,7 @@ impl ConnectDone {
     }
 }
 
+#[cfg_attr(feature = "profile", hotpath::measure)]
 /// Keep the original socket alive and wake the native wait on cancellation.
 pub(crate) fn connect_wait_state(
     py: Python<'_>,
diff --git a/src/transport/stream/core_events.rs b/src/transport/stream/core_events.rs
index 3a3edf7..77e9b0f 100644
--- a/src/transport/stream/core_events.rs
+++ b/src/transport/stream/core_events.rs
@@ -166,8 +166,12 @@ impl StreamTransportCore {
                 ));
             }
             let (guard, _) = self
-                .state_cv
-                .wait_timeout(state, deadline.saturating_duration_since(now))
+                .state
+                .wait_timeout(
+                    &self.state_cv,
+                    state,
+                    deadline.saturating_duration_since(now),
+                )
                 .expect("poisoned transport state");
             state = guard;
         }
diff --git a/src/transport/stream/core_write.rs b/src/transport/stream/core_write.rs
index 0e8c69f..a4af9d4 100644
--- a/src/transport/stream/core_write.rs
+++ b/src/transport/stream/core_write.rs
@@ -409,6 +409,10 @@ impl StreamTransportCore {
         self.try_write_bytes_with_status(bytes, self.write_backpressure_active())
     }
 
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "StreamTransportCore")
+    )]
     /// Learn whether this connection emits batches. The first isolated write
     /// stays direct; a repeated write stages the tail and predicts batching on
     /// the next turn, so recurring header/body pairs still coalesce together.
@@ -425,6 +429,10 @@ impl StreamTransportCore {
         }
     }
 
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "StreamTransportCore")
+    )]
     pub(super) fn try_write_bytes_with_status(
         self: &Arc<Self>,
         bytes: &Bound<'_, PyBytes>,
diff --git a/src/transport/stream/fast.rs b/src/transport/stream/fast.rs
index c82e24f..1836374 100644
--- a/src/transport/stream/fast.rs
+++ b/src/transport/stream/fast.rs
@@ -1209,6 +1209,10 @@ impl ExactReadAccumulator {
         buffer.consume(amount);
     }
 
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "ExactReadAccumulator")
+    )]
     fn fill_from_slice(&mut self, data: &[u8]) -> usize {
         let amount = exact_fill_amount(data.len(), self.filled, self.expected);
         if amount == 0 {
@@ -1721,6 +1725,10 @@ impl PyFastStreamReader {
         self.maybe_resume_transport(py)
     }
 
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "PyFastStreamReader")
+    )]
     /// Complete a waiting exact read directly from incoming bytes. Reconcile
     /// cancellation first, restoring its partial data before any new
     /// consumption.
diff --git a/src/transport/stream/mod.rs b/src/transport/stream/mod.rs
index 67f88b2..aa5c627 100644
--- a/src/transport/stream/mod.rs
+++ b/src/transport/stream/mod.rs
@@ -150,6 +150,12 @@ struct StreamTransportState {
 }
 
 impl crate::state_snapshot::SnapshotFlags for StreamTransportState {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(
+            impl_type = "<StreamTransportState as crate :: state_snapshot :: SnapshotFlags>"
+        )
+    )]
     fn snapshot_flags(&self) -> u8 {
         u8::from(self.closing)
             | (u8::from(self.writable) << 1)
diff --git a/src/transport/stream/py_transport.rs b/src/transport/stream/py_transport.rs
index c459f1d..5fb140c 100644
--- a/src/transport/stream/py_transport.rs
+++ b/src/transport/stream/py_transport.rs
@@ -28,16 +28,28 @@ struct InlineWriteSegments<'py> {
     len: usize,
 }
 impl<'py> InlineWriteSegments<'py> {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "InlineWriteSegments")
+    )]
     fn new() -> Self {
         Self {
             items: std::array::from_fn(|_| std::mem::MaybeUninit::uninit()),
             len: 0,
         }
     }
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "InlineWriteSegments")
+    )]
     fn push(&mut self, value: Bound<'py, PyBytes>) {
         self.items[self.len].write(value);
         self.len += 1;
     }
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "InlineWriteSegments")
+    )]
     fn as_slice(&self) -> &[Bound<'py, PyBytes>] {
         // SAFETY: push initializes each prefix element before increasing len;
         // MaybeUninit<T> has T's layout, and this immutable borrow retains it.
@@ -45,6 +57,10 @@ impl<'py> InlineWriteSegments<'py> {
     }
 }
 impl Drop for InlineWriteSegments<'_> {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<InlineWriteSegments as Drop>")
+    )]
     fn drop(&mut self) {
         for item in &mut self.items[..self.len] {
             // SAFETY: exactly the initialized prefix contains owned values.
diff --git a/src/vibeio/blocking/default.rs b/src/vibeio/blocking/default.rs
index 9402251..c46ee72 100644
--- a/src/vibeio/blocking/default.rs
+++ b/src/vibeio/blocking/default.rs
@@ -49,12 +49,24 @@ impl DefaultBlockingThreadPool {
 // Wrapping it in another closure would allocate a second Job box.
 struct PooledJob(Box<dyn FnOnce() + Send + 'static>);
 impl rusty_pool::Task<()> for PooledJob {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<PooledJob as rusty_pool :: Task < () >>")
+    )]
     fn run(self) {
         (self.0)();
     }
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<PooledJob as rusty_pool :: Task < () >>")
+    )]
     fn into_fn(self) -> Option<Box<dyn FnOnce() + Send + 'static>> {
         Some(self.0)
     }
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<PooledJob as rusty_pool :: Task < () >>")
+    )]
     fn is_fn(&self) -> bool {
         true
     }
diff --git a/src/vibeio/driver/mio.rs b/src/vibeio/driver/mio.rs
index 3adf390..04b8ae5 100644
--- a/src/vibeio/driver/mio.rs
+++ b/src/vibeio/driver/mio.rs
@@ -164,6 +164,7 @@ impl MioDriver {
         }
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "MioDriver"))]
     #[cfg(target_os = "linux")]
     pub(crate) fn take_unclaimed_readiness(&self, token: Token) -> bool {
         self.state
diff --git a/src/vibeio/driver/uring.rs b/src/vibeio/driver/uring.rs
index c5a82f1..3d5fe69 100644
--- a/src/vibeio/driver/uring.rs
+++ b/src/vibeio/driver/uring.rs
@@ -242,6 +242,7 @@ impl UringDriver {
         }
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "UringDriver"))]
     #[cfg(target_os = "linux")]
     pub(crate) fn take_unclaimed_readiness(&self, token: Token) -> bool {
         if let Some(HandleRegistration::Poll(registration)) =
diff --git a/src/vibeio/executor.rs b/src/vibeio/executor.rs
index 54dc46f..5a013e6 100644
--- a/src/vibeio/executor.rs
+++ b/src/vibeio/executor.rs
@@ -889,6 +889,7 @@ impl Runtime {
         self.drive(future, Some(deadline))
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "Runtime"))]
     fn drive<T>(
         &self,
         future: impl Future<Output = T>,
diff --git a/src/vibeio/fd_inner.rs b/src/vibeio/fd_inner.rs
index 5d0bf20..2d0cb9c 100644
--- a/src/vibeio/fd_inner.rs
+++ b/src/vibeio/fd_inner.rs
@@ -86,6 +86,7 @@ impl InnerRawHandle {
         }
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "InnerRawHandle"))]
     #[cfg(target_os = "linux")]
     pub(crate) fn take_unclaimed_readiness(&self) -> bool {
         match self.driver.as_ref() {
diff --git a/src/vibeio/fs/mod.rs b/src/vibeio/fs/mod.rs
index cf31104..8eee6fb 100644
--- a/src/vibeio/fs/mod.rs
+++ b/src/vibeio/fs/mod.rs
@@ -199,31 +199,55 @@ pub trait IntoFileBytes {
     fn into_file_bytes(self) -> Vec<u8>;
 }
 impl IntoFileBytes for Vec<u8> {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<Vec as IntoFileBytes>")
+    )]
     fn into_file_bytes(self) -> Vec<u8> {
         self
     }
 }
 impl IntoFileBytes for String {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<String as IntoFileBytes>")
+    )]
     fn into_file_bytes(self) -> Vec<u8> {
         self.into_bytes()
     }
 }
 impl IntoFileBytes for Box<[u8]> {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<Box as IntoFileBytes>")
+    )]
     fn into_file_bytes(self) -> Vec<u8> {
         self.into_vec()
     }
 }
 impl<T: AsRef<[u8]> + ?Sized> IntoFileBytes for &T {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<& T as IntoFileBytes>")
+    )]
     fn into_file_bytes(self) -> Vec<u8> {
         self.as_ref().to_vec()
     }
 }
 impl<const N: usize> IntoFileBytes for [u8; N] {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<[u8 ; N] as IntoFileBytes>")
+    )]
     fn into_file_bytes(self) -> Vec<u8> {
         Vec::from(self)
     }
 }
 impl<'a> IntoFileBytes for std::borrow::Cow<'a, [u8]> {
+    #[cfg_attr(
+        feature = "profile",
+        hotpath::measure(impl_type = "<Cow as IntoFileBytes>")
+    )]
     fn into_file_bytes(self) -> Vec<u8> {
         self.into_owned()
     }
diff --git a/src/vibeio/net/tcp/listener.rs b/src/vibeio/net/tcp/listener.rs
index af62fe1..4593ea3 100644
--- a/src/vibeio/net/tcp/listener.rs
+++ b/src/vibeio/net/tcp/listener.rs
@@ -102,6 +102,7 @@ impl TcpListener {
         }
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "TcpListener"))]
     /// Cached address family for accepted socket handoffs.
     pub fn is_ipv6(&self) -> bool {
         self.ipv6
diff --git a/src/vibeio/net/tcp/stream.rs b/src/vibeio/net/tcp/stream.rs
index 3e710c0..077d94a 100644
--- a/src/vibeio/net/tcp/stream.rs
+++ b/src/vibeio/net/tcp/stream.rs
@@ -383,6 +383,7 @@ impl PollTcpStream {
         })
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "PollTcpStream"))]
     /// Register an endpoint whose nonblocking mode the transport established.
     #[allow(dead_code, reason = "Used by the Python transport embedding")]
     pub(crate) fn from_shared_nonblocking(
diff --git a/src/vibeio/net/unix/stream.rs b/src/vibeio/net/unix/stream.rs
index 0c40feb..566462f 100644
--- a/src/vibeio/net/unix/stream.rs
+++ b/src/vibeio/net/unix/stream.rs
@@ -242,6 +242,7 @@ impl UnixStream {
         Self::from_shared_with_mode(std::sync::Arc::new(inner), mode)
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "UnixStream"))]
     fn from_shared_with_mode(
         inner: std::sync::Arc<StdUnixStream>,
         mode: RegistrationMode,
@@ -303,6 +304,7 @@ impl PollUnixStream {
         })
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "PollUnixStream"))]
     /// Register a reader sharing the transport's socket allocation.
     pub fn from_shared(inner: std::sync::Arc<StdUnixStream>) -> Result<Self, io::Error> {
         Ok(Self {
@@ -312,6 +314,7 @@ impl PollUnixStream {
         })
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "PollUnixStream"))]
     /// Register an endpoint whose nonblocking mode the transport established.
     #[allow(dead_code, reason = "Used by the Python transport embedding")]
     pub(crate) fn from_shared_nonblocking(
diff --git a/src/vibeio/op/io_util.rs b/src/vibeio/op/io_util.rs
index 2cdf91b..700631d 100644
--- a/src/vibeio/op/io_util.rs
+++ b/src/vibeio/op/io_util.rs
@@ -88,6 +88,7 @@ pub(super) struct NativeIoVecs {
 }
 
 impl NativeIoVecs {
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "NativeIoVecs"))]
     pub(super) fn new(bufs: &[crate::vibeio::io::IoVec]) -> io::Result<Self> {
         #[cfg(unix)]
         let empty = libc::iovec {
@@ -128,10 +129,12 @@ impl NativeIoVecs {
         Ok(storage)
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "NativeIoVecs"))]
     pub(super) fn len(&self) -> usize {
         self.len
     }
 
+    #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "NativeIoVecs"))]
     pub(super) fn as_mut_ptr(&mut self) -> *mut NativeIoVec {
         if self.len <= 16 {
             self.inline.as_mut_ptr()
diff --git a/tests/test_local_sockets.py b/tests/test_local_sockets.py
index b7cb7a6..78bddbd 100644
--- a/tests/test_local_sockets.py
+++ b/tests/test_local_sockets.py
@@ -291,7 +291,7 @@ def test_loop_close_releases_socket_wait_cancelled_before_native_poll():
     loop = rsloop.new_event_loop()
     reference = weakref.ref(receiver)
     try:
-        future = loop.sock_recv(receiver, 1)
+        future = cast(asyncio.Future[bytes], loop.sock_recv(receiver, 1))
         # Start the service, then end the run before it needs a readiness event.
         loop.call_soon(loop.stop)
         loop.run_forever()
diff --git a/tests/test_write_reentrancy.py b/tests/test_write_reentrancy.py
index ab5eb3d..a844ad3 100644
--- a/tests/test_write_reentrancy.py
+++ b/tests/test_write_reentrancy.py
@@ -16,17 +16,31 @@
 @pytest.mark.parametrize("method", ["write", "writelines"])
 @pytest.mark.parametrize("callback", ["pause", "resume"])
 @pytest.mark.parametrize("finish", ["close", "write_eof"])
-def test_flow_control_callback_preserves_unsent_data(callback, finish, method):
-    # Leave enough queued data to cross high water even after the first
-    # small write takes the immediate socket path.
-    payload = bytes(range(256)) * (1024 if method == "writelines" else 280)
+@pytest.mark.parametrize("send_buffer", [4096, 131072])
+@pytest.mark.parametrize("prefill", [False, True])
+def test_flow_control_callback_preserves_unsent_data(
+    callback, finish, method, send_buffer, prefill
+):
+    # Queue enough staged data to cross high water independently of how
+    # much the OS accepts on the first immediate write.
+    payload = bytes(range(256)) * 4096
 
     async def main():
         loop = asyncio.get_running_loop()
         sender, receiver = socket.socketpair()
-        sender.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 4096)
+        sender.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, send_buffer)
         sender.setblocking(False)
         receiver.setblocking(False)
+        prefix = bytearray()
+        if prefill:
+            # A full kernel buffer reproduces the macOS case where the
+            # staged flush cannot reach low water until the peer reads.
+            try:
+                while True:
+                    prefix.extend(b"p" * sender.send(b"p" * 4096))
+            except BlockingIOError:
+                pass
+        paused = loop.create_future()
         finished = loop.create_future()
         transport = None
 
@@ -56,6 +70,8 @@ def send(self):
                         transport.write(payload)
 
             def pause_writing(self):
+                if not paused.done():
+                    paused.set_result(self.transport.get_write_buffer_size())
                 if callback == "pause":
                     self.finish()
 
@@ -66,7 +82,7 @@ def resume_writing(self):
             def finish(self):
                 if not finished.done():
                     # Capture progress as the callback re-enters shutdown.
-                    # Large batches can already be draining on resume.
+                    # The writer may already have drained on resume.
                     remaining = self.transport.get_write_buffer_size()
                     finished.set_result(remaining)
                     getattr(self.transport, finish)()
@@ -79,18 +95,20 @@ async def receive_all():
 
         try:
             transport, _ = await loop.create_unix_connection(Sender, sock=sender)
-            if method == "writelines" and callback == "resume":
-                # A large batch must make progress before it crosses low water.
+            assert await asyncio.wait_for(paused, 5) > 0
+            if callback == "resume":
+                # Every resume case needs a reader: kernel buffer sizes and
+                # partial-write boundaries differ between Linux and macOS.
                 receiving = asyncio.create_task(receive_all())
                 try:
                     await asyncio.wait_for(finished, 5)
-                    assert await receiving == payload
+                    assert await receiving == prefix + payload
                 finally:
                     receiving.cancel()
                     await asyncio.gather(receiving, return_exceptions=True)
             else:
                 assert await asyncio.wait_for(finished, 5) > 0
-                assert await receive_all() == payload
+                assert await receive_all() == prefix + payload
         finally:
             if transport is not None:
                 transport.close()
@@ -112,6 +130,7 @@ async def main():
 
         class Sender(asyncio.Protocol):
             def connection_made(self, transport):
+                transport = cast(asyncio.Transport, transport)
                 try:
                     transport.write(payload)
                     # An isolated write must reach the socket in this callback,