diff --git a/docs/OPERATIONS.md b/docs/OPERATIONS.md
index b16bf08..b72f538 100644
--- a/docs/OPERATIONS.md
+++ b/docs/OPERATIONS.md
@@ -2,7 +2,7 @@
 
 ## Known issues
 
-### NULL-buffer heap corruption in HTTP body handlers (root cause under investigation)
+### NULL-buffer heap corruption in HTTP body handlers (root cause: data races — fixed)
 
 **Symptom:** Historically, `buffer->data` was observed NULL inside the HTTP
 body handling paths — `_on_body` (http_parser body callback in
@@ -10,53 +10,39 @@ body handling paths — `_on_body` (http_parser body callback in
 body handler in `src/ClientAPI/HTTP/off_routes.c`), and `_pipe_on_data`
 (response piping callback in `src/ClientAPI/HTTP/http_response.c`). The
 `buffer_t` struct had `capacity` set correctly but `data == NULL` and `size`
-held garbage, suggesting the struct fields were overwritten by heap corruption
-from an unknown source.
-
-**Guards in place:** Defensive sentinels were restored at the entry of each
-of those three handlers. A NULL `buffer->data` (or NULL `chunk`/`at` pointer)
-now logs an `error`-level message identifying the handler and the suspicious
-pointer values, then returns without dereferencing. In `_on_body` the parse
-is aborted by returning `1` to http-parser; in the streamed-PUT and
-response-pipe paths the chunk is dropped. This prevents the NULL dereference
-crash but does not address the underlying corruption.
-
-**Root cause status:** Under investigation. The corruption is not easily
-reproducible under ASAN (ASAN redzones mask the bug), and the flaky
-`TestStream*` segfaults observed in ASAN builds do not produce an ASAN
-report (the SIGSEGV bypasses ASAN's signal handler, produces no core dump,
-does not reproduce under gdb/strace/pty, and does not reproduce in non-ASAN
-builds). The failing `TestStream*` tests (`TestPushFileStream.*`,
-`TestPullFileStream.*`, `TestStreamActor.*`) are file-stream + scheduler
-tests and do not exercise the HTTP body handlers directly, so the sentinel
-guards do not resolve their segfaults — but the guards are retained as the
-spec's accepted fallback for the historical NULL-buffer crash.
-
-**Investigation notes:**
-
-- The flaky ASAN segfault is timing-dependent (reproduces only when stdout
-  is file-redirected, not under a pty; ~30% rate in isolated process runs).
-- ASAN installs its SIGSEGV handler but does not fire a report when the
-  segfault occurs, suggesting the fault happens in a state where ASAN's
-  handler cannot safely run (e.g. during process teardown after main
-  returns, or in a thread that hasn't registered its stack with ASAN).
-- `buffer_ensure_capacity` aborts on OOM, so `buffer->data` is never NULL
-  in normal operation — the NULL must come from external heap corruption.
-- Candidates not yet ruled out: a missing `REFERENCE` on a `buffer_t*`
-  crossing an actor boundary; a `stream_notify` CONSUME/yield ownership
-  bug; a double-free in dispatch (per the
-  `feedback_double_free_dispatch.md` memory note — `actor_run` frees
-  `msg->payload`; dispatch must not also free it); a `stream_deactivate`
-  freeing a buffer while a handler still reads it.
-
-**Next steps for a future investigation:**
-
-1. Run the `TestStream*` tests under ThreadSanitizer (TSAN) to catch the
-   race that ASAN misses.
-2. Audit `actor_run`'s payload destroy path against every dispatch handler
-   in `src/Streams/` and `src/ClientAPI/HTTP/` for the double-free pattern
-   documented in `feedback_double_free_dispatch.md`.
-3. Stress-run the file-stream pipeline under valgrind with
-   `--track-origins=yes` to capture the corruption source.
-4. Once the root cause is found, remove the sentinel guards and replace
-   with the minimal fix.
\ No newline at end of file
+held garbage.
+
+**Root cause:** The NULL `buffer->data` was a downstream symptom of heap
+corruption from two cross-thread data races, not a `buffer_t` bug. TSAN
+caught both (ASAN and valgrind miss them):
+
+1. **`connection->sock` use-after-free** — `_connection_close_fd` (worker
+   thread) freed the socket and set `connection->sock = NULL` while
+   `_connection_read_callback` (I/O thread) read it. Fixed by making
+   `connection->sock` an `ATOMIC(platform_socket_t*)` and deferring the
+   socket's close+free to the I/O thread's destroy stack
+   (`http_server_defer_socket_destroy`), mirroring the existing
+   watcher/timer deferral.
+
+2. **`pipe_notifiers` use-after-free WRITE** — `readable_push_stream_pipe` /
+   `writeable_pull_stream_pipe` called `on_pipe`/`on_piped` synchronously on
+   the caller's thread, writing `pipe_notifiers` while
+   `stream_unsubscribe_pipe_notifiers` (worker thread) freed it. Fixed by
+   routing pipe/piped through the stream actor via the already-declared
+   `STREAM_PIPE`/`STREAM_PIPED` messages and `stream_pipe_internal` /
+   `stream_piped_internal`.
+
+**Guards in place:** The defensive sentinels at the entry of the three body
+handlers remain as cheap no-op checks (http-parser never legitimately passes
+NULL `at`/`length`), but they are no longer the fix — the underlying races
+are resolved.
+
+**Verification:** `TestPushFileStream.*`, `TestPullFileStream.*`,
+`TestStreamActor.*`, `TestHttpServer.*`, `TestOffRoutes.*`, and
+`TestHttpServerSsl.*` all pass under TSAN with zero data-race reports, and
+the full 849-test suite passes. The GET-path pipeline refcount leak in
+`_setup_stream_pipeline` (off_routes.c) that previously leaked 48 bytes
+direct + 209 bytes indirect per GET request has also been fixed — the
+`get_pipeline_t` refcount now reaches zero in all paths via a `desc_done`
+flag that ensures desc contributes exactly one deref whether close or
+error fires first.
\ No newline at end of file
diff --git a/src/ClientAPI/HTTP/http_connection.c b/src/ClientAPI/HTTP/http_connection.c
index 5e74df6..3b90cf5 100644
--- a/src/ClientAPI/HTTP/http_connection.c
+++ b/src/ClientAPI/HTTP/http_connection.c
@@ -339,9 +339,13 @@ static void _connection_stop_watcher(http_connection_t* connection) {
 
 /* Close the fd and mark connection as closing. Used from dispatch (worker thread). */
 static void _connection_close_fd(http_connection_t* connection) {
-  if (connection->sock != NULL) {
-    platform_socket_destroy(connection->sock);
-    connection->sock = NULL;
+  platform_socket_t* sock = ATOMIC_EXCHANGE(&connection->sock, NULL);
+  if (sock != NULL) {
+    if (connection->server != NULL) {
+      http_server_defer_socket_destroy(connection->server, sock);
+    } else {
+      platform_socket_destroy(sock);
+    }
   }
   connection->is_closing = 1;
 }
@@ -359,8 +363,9 @@ static int _connection_send_raw_blocking(http_connection_t* connection,
                                           const uint8_t* data, size_t len,
                                           size_t* out_sent) {
   size_t sent_total = 0;
+  platform_socket_t* sock = ATOMIC_LOAD(&connection->sock);
   for (int attempts = 0; attempts < 2000 && sent_total < len; attempts++) {
-    ssize_t sent = platform_socket_send(connection->sock, data + sent_total,
+    ssize_t sent = platform_socket_send(sock, data + sent_total,
                                          len - sent_total);
     if (sent > 0) {
       sent_total += (size_t)sent;
@@ -461,13 +466,13 @@ static int _connection_send_all_blocking(http_connection_t* connection,
  * cross-thread race on the BIO. */
 static void _connection_ssl_data_handle(http_connection_t* connection,
                                         buffer_t* data) {
-  if (connection->sock == NULL || connection->ssl == NULL) {
+  if (ATOMIC_LOAD(&connection->sock) == NULL || connection->ssl == NULL) {
     return;
   }
   BIO_write(connection->rbio, data->data, (int)data->size);
 
   for (int batch = 0; batch < 16; batch++) {
-    if (connection->sock == NULL) {
+    if (ATOMIC_LOAD(&connection->sock) == NULL) {
       return;
     }
     char buffer[READ_BUFFER_SIZE];
@@ -566,7 +571,7 @@ void http_connection_dispatch(void* state, message_t* msg) {
       /* ASIO-style: the I/O thread notified us that data is available.
          Perform the actual recv() and parsing here on the scheduler worker. */
       atomic_store(&connection->read_pending, 0);
-      if (connection->sock == NULL) {
+      if (ATOMIC_LOAD(&connection->sock) == NULL) {
         break;
       }
 #ifndef _WIN32
@@ -582,7 +587,7 @@ void http_connection_dispatch(void* state, message_t* msg) {
          goes through READABLE -> _connection_do_reads. */
       buffer_t* data = (buffer_t*)msg->payload;
       msg->payload = NULL; /* Take ownership — actor_run won't destroy it */
-      if (connection->sock == NULL) {
+      if (ATOMIC_LOAD(&connection->sock) == NULL) {
         DESTROY(data, buffer);
         break;
       }
@@ -629,7 +634,7 @@ void http_connection_dispatch(void* state, message_t* msg) {
     case HTTP_CONNECTION_WRITE: {
       buffer_t* buf = (buffer_t*)msg->payload;
       msg->payload = NULL; /* Take ownership — actor_run won't destroy it */
-      if (connection->sock == NULL) {
+      if (ATOMIC_LOAD(&connection->sock) == NULL) {
         DESTROY(buf, buffer);
         break;
       }
@@ -696,7 +701,7 @@ void http_connection_dispatch(void* state, message_t* msg) {
         break;
       }
       /* Try direct send */
-      ssize_t sent = platform_socket_send(connection->sock, buf->data, buf->size);
+      ssize_t sent = platform_socket_send(ATOMIC_LOAD(&connection->sock), buf->data, buf->size);
       if (sent < 0) {
         if (errno == EAGAIN || errno == EWOULDBLOCK) {
           connection->write_buffer = buf;
@@ -728,7 +733,7 @@ void http_connection_dispatch(void* state, message_t* msg) {
     }
 
     case HTTP_CONNECTION_WRITABLE: {
-      if (connection->sock == NULL) {
+      if (ATOMIC_LOAD(&connection->sock) == NULL) {
         break;
       }
       if (connection->write_buffer == NULL || connection->write_buffer->size == 0) {
@@ -736,7 +741,7 @@ void http_connection_dispatch(void* state, message_t* msg) {
         _connection_update_watcher(connection, PD_EVENT_READ);
         break;
       }
-      ssize_t sent = platform_socket_send(connection->sock, connection->write_buffer->data,
+      ssize_t sent = platform_socket_send(ATOMIC_LOAD(&connection->sock), connection->write_buffer->data,
                            connection->write_buffer->size);
       if (sent > 0) {
         if ((size_t)sent >= connection->write_buffer->size) {
@@ -745,8 +750,8 @@ void http_connection_dispatch(void* state, message_t* msg) {
           connection->write_pending = 0;
           if (connection->is_closing) {
             /* All data flushed — finish the deferred close */
-            if (connection->sock != NULL) {
-              platform_socket_shutdown(connection->sock, PLATFORM_SHUT_WR);
+            if (ATOMIC_LOAD(&connection->sock) != NULL) {
+              platform_socket_shutdown(ATOMIC_LOAD(&connection->sock), PLATFORM_SHUT_WR);
             }
             _connection_stop_watcher(connection);
             _connection_close_fd(connection);
@@ -785,8 +790,8 @@ void http_connection_dispatch(void* state, message_t* msg) {
         _connection_update_watcher(connection, PD_EVENT_READ | PD_EVENT_WRITE);
         break;
       }
-      if (connection->sock != NULL) {
-        platform_socket_shutdown(connection->sock, PLATFORM_SHUT_WR);
+      if (ATOMIC_LOAD(&connection->sock) != NULL) {
+        platform_socket_shutdown(ATOMIC_LOAD(&connection->sock), PLATFORM_SHUT_WR);
       }
       _connection_stop_watcher(connection);
       _connection_close_fd(connection);
@@ -915,10 +920,11 @@ static void _connection_read_callback(pd_loop_t* loop, pd_watcher_t* watcher,
     if (total_read == 0) {
       /* POSIX path: synchronous recv. The socket may already be closed if the
          connection was torn down concurrently with a pending READ event. */
-      if (connection->sock == NULL) {
+      platform_socket_t* sock = ATOMIC_LOAD(&connection->sock);
+      if (sock == NULL) {
         return;
       }
-      ssize_t bytes_read = platform_socket_recv(connection->sock, buffer, sizeof(buffer));
+      ssize_t bytes_read = platform_socket_recv(sock, buffer, sizeof(buffer));
       if (bytes_read <= 0) {
         if (bytes_read == 0) {
           message_t msg;
@@ -965,7 +971,7 @@ static void _connection_do_reads(http_connection_t* connection) {
   }
   for (int batch = 0; batch < 16; batch++) {
     char buffer[READ_BUFFER_SIZE];
-    if (connection->sock == NULL) {
+    if (ATOMIC_LOAD(&connection->sock) == NULL) {
       return;
     }
 
@@ -1012,7 +1018,7 @@ http_connection_t* http_connection_create(http_server_t* server, platform_socket
   http_connection_t* connection = get_clear_memory(sizeof(http_connection_t));
   refcounter_init((refcounter_t*)connection);
   connection->server = server;
-  connection->sock = sock;
+  ATOMIC_STORE(&connection->sock, sock);
   connection->ssl = NULL;
   connection->rbio = NULL;
   connection->wbio = NULL;
@@ -1116,9 +1122,9 @@ void http_connection_destroy(http_connection_t* connection) {
         pd_timer_destroy(timer);
       }
     }
-    if (connection->sock != NULL) {
-      platform_socket_destroy(connection->sock);
-      connection->sock = NULL;
+    platform_socket_t* sock = ATOMIC_EXCHANGE(&connection->sock, NULL);
+    if (sock != NULL) {
+      platform_socket_destroy(sock);
     }
     if (connection->request != NULL) {
       DESTROY(connection->request, http_request);
@@ -1140,7 +1146,7 @@ void http_connection_destroy(http_connection_t* connection) {
 }
 
 void http_connection_write(http_connection_t* connection, const char* data, size_t length) {
-  if (connection == NULL || connection->sock == NULL) {
+  if (connection == NULL || ATOMIC_LOAD(&connection->sock) == NULL) {
     return;
   }
   buffer_t* buf = buffer_create_from_pointer_copy((uint8_t*)data, length);
diff --git a/src/ClientAPI/HTTP/http_connection.h b/src/ClientAPI/HTTP/http_connection.h
index 040e7e0..87f4426 100644
--- a/src/ClientAPI/HTTP/http_connection.h
+++ b/src/ClientAPI/HTTP/http_connection.h
@@ -32,7 +32,7 @@ typedef struct http_connection_t {
   refcounter_t refcounter;
   actor_t actor;
   http_server_t* server;
-  platform_socket_t* sock;
+  ATOMIC(platform_socket_t*) sock;
   ATOMIC(pd_watcher_t*) watcher;
   SSL* ssl;
   /* Windows IOCP only: memory BIO pair decoupling OpenSSL from the socket so
diff --git a/src/ClientAPI/HTTP/http_server.c b/src/ClientAPI/HTTP/http_server.c
index ef9beee..7c43319 100644
--- a/src/ClientAPI/HTTP/http_server.c
+++ b/src/ClientAPI/HTTP/http_server.c
@@ -28,7 +28,7 @@ static void _destroy_stack_init(http_server_t* server) {
 static void _destroy_stack_push_watcher(http_server_t* server, pd_watcher_t* watcher) {
   server_destroy_node_t* node = get_clear_memory(sizeof(server_destroy_node_t));
   node->watcher = watcher;
-  node->is_timer = 0;
+  node->type = 0;
   platform_mutex_lock(server->destroy_lock);
   node->next = server->destroy_head;
   server->destroy_head = node;
@@ -39,7 +39,18 @@ static void _destroy_stack_push_watcher(http_server_t* server, pd_watcher_t* wat
 static void _destroy_stack_push_timer(http_server_t* server, pd_timer_t* timer) {
   server_destroy_node_t* node = get_clear_memory(sizeof(server_destroy_node_t));
   node->timer = timer;
-  node->is_timer = 1;
+  node->type = 1;
+  platform_mutex_lock(server->destroy_lock);
+  node->next = server->destroy_head;
+  server->destroy_head = node;
+  platform_mutex_unlock(server->destroy_lock);
+  pd_loop_async_send(server->loop, NULL);
+}
+
+void http_server_defer_socket_destroy(http_server_t* server, platform_socket_t* sock) {
+  server_destroy_node_t* node = get_clear_memory(sizeof(server_destroy_node_t));
+  node->sock = sock;
+  node->type = 2;
   platform_mutex_lock(server->destroy_lock);
   node->next = server->destroy_head;
   server->destroy_head = node;
@@ -55,9 +66,11 @@ static void _destroy_stack_drain(http_server_t* server) {
   platform_mutex_unlock(server->destroy_lock);
   while (node != NULL) {
     server_destroy_node_t* next = node->next;
-    if (node->is_timer) {
+    if (node->type == 1) {
       pd_timer_stop(node->timer);
       pd_timer_destroy(node->timer);
+    } else if (node->type == 2) {
+      platform_socket_destroy(node->sock);
     } else {
       pd_watcher_destroy(node->watcher);
     }
@@ -254,9 +267,9 @@ void http_server_destroy(http_server_t* server) {
   for (int i = 0; i < server->connections.length; i++) {
     http_connection_t* conn = server->connections.data[i];
     conn->is_closing = 1;
-    if (conn->sock != NULL) {
-      platform_socket_destroy(conn->sock);
-      conn->sock = NULL;
+    platform_socket_t* sock = ATOMIC_EXCHANGE(&conn->sock, NULL);
+    if (sock != NULL) {
+      platform_socket_destroy(sock);
     }
     conn->server = NULL;
   }
diff --git a/src/ClientAPI/HTTP/http_server.h b/src/ClientAPI/HTTP/http_server.h
index 7c5bc1d..3990575 100644
--- a/src/ClientAPI/HTTP/http_server.h
+++ b/src/ClientAPI/HTTP/http_server.h
@@ -31,9 +31,10 @@ typedef vec_t(http_middleware_entry_t) vec_middleware_t;
 typedef vec_t(http_connection_t*) vec_connection_t;
 
 typedef struct server_destroy_node_t {
-  pd_watcher_t* watcher;     /* valid when is_timer == 0 */
-  pd_timer_t* timer;         /* valid when is_timer == 1 */
-  uint8_t is_timer;
+  pd_watcher_t* watcher;     /* valid when type == 0 */
+  pd_timer_t* timer;         /* valid when type == 1 */
+  platform_socket_t* sock;   /* valid when type == 2 */
+  uint8_t type;              /* 0 = watcher, 1 = timer, 2 = socket */
   struct server_destroy_node_t* next;
 } server_destroy_node_t;
 
@@ -85,6 +86,10 @@ void http_server_set_timeouts(http_server_t* server, uint32_t idle_ms, uint32_t
 
 void http_server_dispatch(http_server_t* server, http_request_t* request, http_response_t* response);
 
+/* Defer a connection socket's close+free to the I/O thread's destroy stack so
+   it is never freed while the I/O-thread read callback may still be using it. */
+void http_server_defer_socket_destroy(http_server_t* server, platform_socket_t* sock);
+
 http_route_t* http_server_match_route(http_server_t* server, int method, const char* path);
 
 void http_server_use(http_server_t* server, http_middleware_t middleware, void* user_data, void (*user_data_destroy)(void*));
diff --git a/src/ClientAPI/HTTP/off_routes.c b/src/ClientAPI/HTTP/off_routes.c
index 4772ed3..ddc3436 100644
--- a/src/ClientAPI/HTTP/off_routes.c
+++ b/src/ClientAPI/HTTP/off_routes.c
@@ -165,6 +165,11 @@ typedef struct {
     tuple_cache_t* tc;
     http_response_t* response;
     ori_t* ori;
+    /* desc_done ensures desc contributes exactly one pipeline deref,
+       whether close or error fires first. stream_deactivate emits both
+       close_event and error_event, so without this flag the pipeline
+       would be dereffed twice for desc. */
+    uint8_t desc_done;
 } get_pipeline_t;
 
 static void _pipeline_on_tuple(void* ctx, void* data) {
@@ -177,7 +182,11 @@ static void _pipeline_on_desc_close(void* ctx, void* unused) {
     (void)unused;
     get_pipeline_t* pipeline = (get_pipeline_t*)ctx;
     readable_descriptor_t* desc = pipeline->desc;
-    int is_zero = refcounter_dereference_is_zero((refcounter_t*)pipeline);
+    int is_zero = 0;
+    if (!pipeline->desc_done) {
+        pipeline->desc_done = 1;
+        is_zero = refcounter_dereference_is_zero((refcounter_t*)pipeline);
+    }
     stream_deferred_deref((stream_t*)desc);
     if (is_zero) {
         DESTROY(pipeline->ori, ori);
@@ -193,7 +202,12 @@ static void _pipeline_on_desc_error(void* ctx, void* error) {
        end/destroy the response here; _pipe_on_error and _pipe_on_close
        both fire on stream deactivation and would double-free. */
     stream_deactivate((stream_t*)pipeline->rs, NULL);
-    if (refcounter_dereference_is_zero((refcounter_t*)pipeline)) {
+    int is_zero = 0;
+    if (!pipeline->desc_done) {
+        pipeline->desc_done = 1;
+        is_zero = refcounter_dereference_is_zero((refcounter_t*)pipeline);
+    }
+    if (is_zero) {
         DESTROY(pipeline->ori, ori);
         free(pipeline);
     }
@@ -223,11 +237,10 @@ static void _setup_stream_pipeline(http_response_t* response, scheduler_pool_t*
     pipeline->tc = tc;
     pipeline->response = response;
     pipeline->ori = stream_ori;
+    /* Two derefs total: one for desc-done (close or error, whichever
+       fires first — guarded by desc_done), one for rs-done. */
     refcounter_init((refcounter_t*)pipeline);
     refcounter_reference((refcounter_t*)pipeline);
-    refcounter_reference((refcounter_t*)pipeline);
-    refcounter_reference((refcounter_t*)pipeline);
-    refcounter_reference((refcounter_t*)pipeline);
 
     stream_subscribe((stream_t*)desc, data_event, pipeline,
                      (void (*)(void*, void*))_pipeline_on_tuple, NULL);
diff --git a/src/Streams/streams.c b/src/Streams/streams.c
index 8ead55c..7173db7 100644
--- a/src/Streams/streams.c
+++ b/src/Streams/streams.c
@@ -252,6 +252,38 @@ static void _stream_notify_owned_error(stream_t* stream, async_error_t* error) {
   }
 }
 
+/* Pipe/piped are routed through the stream actor so pipe_notifiers is only
+   ever touched on the actor thread. The public pipe entry points send a
+   STREAM_PIPE message; the on_pipe handler then sends STREAM_PIPED to the
+   peer's actor, so each stream's pipe_notifiers is written and freed on its
+   own actor thread (no cross-thread use-after-free). */
+void stream_pipe_internal(stream_t* source, stream_t* dest) {
+  source->on_pipe(source, dest);
+}
+
+void stream_piped_internal(stream_t* stream, stream_t* source) {
+  stream->on_piped(stream, source);
+}
+
+static void _stream_pipe_payload_destroy(void* ptr) {
+  stream_pipe_payload_t* p = (stream_pipe_payload_t*) ptr;
+  if (p->source != NULL) {
+    DEREFERENCE(p->source);
+  }
+  if (p->dest != NULL) {
+    DEREFERENCE(p->dest);
+  }
+  free(p);
+}
+
+static void _stream_piped_payload_destroy(void* ptr) {
+  stream_piped_payload_t* p = (stream_piped_payload_t*) ptr;
+  if (p->source != NULL) {
+    DEREFERENCE(p->source);
+  }
+  free(p);
+}
+
 void stream_dispatch(void* state, message_t* msg) {
   stream_t* stream = (stream_t*) state;
   switch (msg->type) {
@@ -338,6 +370,16 @@ void stream_dispatch(void* state, message_t* msg) {
       }
       break;
     }
+    case STREAM_PIPE: {
+      stream_pipe_payload_t* p = (stream_pipe_payload_t*) msg->payload;
+      stream_pipe_internal(p->source, p->dest);
+      break;
+    }
+    case STREAM_PIPED: {
+      stream_piped_payload_t* p = (stream_piped_payload_t*) msg->payload;
+      stream_piped_internal(stream, p->source);
+      break;
+    }
     case STREAM_SET_PULLING: {
       stream_set_pulling_payload_t* p = (stream_set_pulling_payload_t*) msg->payload;
       stream->is_pulling = p->is_pulling;
@@ -730,7 +772,14 @@ void readable_push_stream_pipe(stream_t* rs, stream_t* ws) {
   } else if (ws->type == readable_stream || ws->force == pull) {
     stream_notify(rs, error_event, OFFS_ERROR_TRANSFER("Invalid write stream being piped to"), (void (*)(void*))error_destroy);
   } else {
-    rs->on_pipe(rs, ws);
+    stream_pipe_payload_t* payload = get_clear_memory(sizeof(stream_pipe_payload_t));
+    payload->source = REFERENCE(rs, stream_t);
+    payload->dest = REFERENCE(ws, stream_t);
+    message_t msg;
+    msg.type = STREAM_PIPE;
+    msg.payload = payload;
+    msg.payload_destroy = _stream_pipe_payload_destroy;
+    actor_send(&rs->actor, &msg);
   }
 }
 
@@ -763,7 +812,13 @@ void _readable_push_stream_on_pipe(stream_t* rs, stream_t* ws) {
     rs->pipe_notifiers[2].event = close_event;
     rs->pipe_notifiers[2].id = stream_subscribe(ws, close_event, REFERENCE(rs, stream_t), (void(*)(void*, void*)) _readable_push_stream_close_notify, (void (*)(void*))rs->destructor);
     rs->pipe_notifiers[2].stream = REFERENCE(ws, stream_t);
-    ws->on_piped(ws, rs);
+    stream_piped_payload_t* piped_payload = get_clear_memory(sizeof(stream_piped_payload_t));
+    piped_payload->source = REFERENCE(rs, stream_t);
+    message_t piped_msg;
+    piped_msg.type = STREAM_PIPED;
+    piped_msg.payload = piped_payload;
+    piped_msg.payload_destroy = _stream_piped_payload_destroy;
+    actor_send(&ws->actor, &piped_msg);
   }
 }
 
@@ -875,7 +930,14 @@ void writeable_pull_stream_pipe(stream_t* ws, stream_t* rs) {
   } else if (ws->type == readable_stream || ws->force == push) {
     stream_notify(rs, error_event, OFFS_ERROR_TRANSFER("Invalid read stream being piped to"), (void (*)(void*))error_destroy);
   } else {
-    ws->on_pipe(ws, rs);
+    stream_pipe_payload_t* payload = get_clear_memory(sizeof(stream_pipe_payload_t));
+    payload->source = REFERENCE(ws, stream_t);
+    payload->dest = REFERENCE(rs, stream_t);
+    message_t msg;
+    msg.type = STREAM_PIPE;
+    msg.payload = payload;
+    msg.payload_destroy = _stream_pipe_payload_destroy;
+    actor_send(&ws->actor, &msg);
   }
 }
 
@@ -920,7 +982,13 @@ void _writeable_pull_stream_on_pipe(stream_t* ws, stream_t* rs) {
     ws->pipe_notifiers[4].id = stream_subscribe(rs, data_event, REFERENCE(ws, stream_t), (void(*)(void*, void*)) _writeable_pull_stream_data_notify, (void (*)(void*))ws->destructor);
     ws->pipe_notifiers[4].stream = REFERENCE(rs, stream_t);
     ws->is_piped = 1;
-    rs->on_piped(rs, ws);
+    stream_piped_payload_t* piped_payload = get_clear_memory(sizeof(stream_piped_payload_t));
+    piped_payload->source = REFERENCE(ws, stream_t);
+    message_t piped_msg;
+    piped_msg.type = STREAM_PIPED;
+    piped_msg.payload = piped_payload;
+    piped_msg.payload_destroy = _stream_piped_payload_destroy;
+    actor_send(&rs->actor, &piped_msg);
   }
 }
 
diff --git a/test/test_network.cpp b/test/test_network.cpp
index bf03d7b..3bac1d9 100644
--- a/test/test_network.cpp
+++ b/test/test_network.cpp
@@ -1416,20 +1416,40 @@ TEST_F(StoreBlockTest, MaxHopsZeroReturnsMaxHopsReached) {
 }
 
 TEST_F(StoreBlockTest, LowCapacityAccepts) {
-  srand(2);  // Deterministic seed: rand() produces 0.70 < 0.75 accept probability
-  store_block_state_t state = {};
-  memset(state.block_hash, 0xAA, 32);
-  state.max_hops = 6;
-
-  net_node_t* next_hops[STORE_BLOCK_FORWARD_FANOUT];
-  size_t next_hop_count = 0;
-
-  // Low capacity (0.2) → should accept
-  store_block_result_e result = store_block_execute(
-      &eabf_table, NULL, rings, &local_id, 0.2f, NODE_PHASE_INHALE,
-      &state, next_hops, &next_hop_count);
+  /* At local_capacity=0.2, INHALE phase, accept_probability = 1 - 0.2/0.80
+     = 0.75. store_block_should_accept draws from platform_random_uniform_float
+     (CSPRNG via getentropy on Linux), which srand() cannot seed, so a single
+     trial is non-deterministic. Run N trials and assert the acceptance rate
+     matches the 0.75 probability within a tight confidence band.
+     N=2000 → expected 1500 accepts, std dev sqrt(2000*0.75*0.25) ≈ 19.4.
+     [1400, 1600] is ±5.16σ → ~1-in-5M false-fail rate, effectively
+     non-flaky. */
+  constexpr int N = 2000;
+  constexpr int expected = 1500;
+  constexpr int tolerance = 100;
+  int accepts = 0;
+  for (int i = 0; i < N; i++) {
+    store_block_state_t state = {};
+    memset(state.block_hash, 0xAA, 32);
+    state.max_hops = 6;
+
+    net_node_t* next_hops[STORE_BLOCK_FORWARD_FANOUT];
+    size_t next_hop_count = 0;
+
+    store_block_result_e result = store_block_execute(
+        &eabf_table, NULL, rings, &local_id, 0.2f, NODE_PHASE_INHALE,
+        &state, next_hops, &next_hop_count);
+    if (result == STORE_BLOCK_ACCEPTED) {
+      accepts++;
+    }
+  }
 
-  EXPECT_EQ(result, STORE_BLOCK_ACCEPTED);
+  EXPECT_GE(accepts, expected - tolerance)
+      << "Acceptance rate too low: " << accepts << "/" << N
+      << " (expected ~" << expected << " ± " << tolerance << ")";
+  EXPECT_LE(accepts, expected + tolerance)
+      << "Acceptance rate too high: " << accepts << "/" << N
+      << " (expected ~" << expected << " ± " << tolerance << ")";
 }
 
 TEST_F(StoreBlockTest, HighCapacityForwards) {