diff --git a/android/src/main/java/to/holepunch/bare/kit/IPC.java b/android/src/main/java/to/holepunch/bare/kit/IPC.java
index c847aef..321e2f1 100644
--- a/android/src/main/java/to/holepunch/bare/kit/IPC.java
+++ b/android/src/main/java/to/holepunch/bare/kit/IPC.java
@@ -1,6 +1,7 @@
 package to.holepunch.bare.kit;
 
 import java.io.Closeable;
+import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
 import java.util.concurrent.atomic.AtomicReference;
@@ -38,6 +39,8 @@ public IPC(Worklet worklet) {
     this.worklet = worklet;
 
     handle = init(worklet.handle);
+
+    worklet.ipc = this;
   }
 
   private native ByteBuffer
@@ -60,6 +63,8 @@ public IPC(Worklet worklet) {
 
   private void
   update() {
+    if (handle == null) return;
+
     update(handle, readable != null, writable != null);
   }
 
@@ -90,10 +95,13 @@ public IPC(Worklet worklet) {
   /**
    * Reads whatever is available without blocking. Returns null when nothing is
    * ready yet, and an empty buffer at end-of-stream, once the worklet has
-   * closed its end. Wait for readable() before retrying a null.
+   * closed its end or this IPC is closed. Wait for readable() before retrying
+   * a null.
    */
   public ByteBuffer
   read() {
+    if (handle == null) return ByteBuffer.allocateDirect(0);
+
     ByteBuffer buffer = read(handle);
 
     if (buffer == null) {
@@ -134,6 +142,8 @@ public IPC(Worklet worklet) {
 
   public int
   write(ByteBuffer data) {
+    if (handle == null) return -1;
+
     ByteBuffer buffer;
 
     if (data.isDirect()) {
@@ -149,6 +159,12 @@ public IPC(Worklet worklet) {
 
   public void
   write(ByteBuffer data, WriteCallback callback) {
+    if (handle == null) {
+      callback.apply(new IOException("IPC closed"));
+
+      return;
+    }
+
     ByteBuffer buffer;
 
     if (data.isDirect()) {
@@ -186,8 +202,18 @@ public IPC(Worklet worklet) {
     }
   }
 
+  /**
+   * Retires this IPC, which must happen before its worklet is destroyed.
+   * terminate() does it for you, and closing twice is a no-op.
+   */
   public void
   close() {
+    if (handle == null) return;
+
     destroy(handle);
+
+    handle = null;
+
+    worklet.ipc = null;
   }
 }
diff --git a/android/src/main/java/to/holepunch/bare/kit/Worklet.java b/android/src/main/java/to/holepunch/bare/kit/Worklet.java
index 40f1faa..65c3401 100644
--- a/android/src/main/java/to/holepunch/bare/kit/Worklet.java
+++ b/android/src/main/java/to/holepunch/bare/kit/Worklet.java
@@ -51,6 +51,9 @@ public static class Options {
 
   ByteBuffer handle;
 
+  // Unowned: the IPC holds this worklet and clears this when it closes.
+  IPC ipc;
+
   public Worklet(Options options) {
     if (options == null) options = new Options();
 
@@ -158,6 +161,8 @@ public Worklet(Options options) {
 
   public void
   terminate() {
+    if (ipc != null) ipc.close();
+
     terminate(handle);
 
     handle = null;
diff --git a/apple/BareKit/BareKit.h b/apple/BareKit/BareKit.h
index e8cc82a..c9edcf7 100644
--- a/apple/BareKit/BareKit.h
+++ b/apple/BareKit/BareKit.h
@@ -85,7 +85,7 @@
 /**
  * Reads whatever is available without blocking. Returns `nil` when nothing is
  * ready yet, and empty data at end-of-stream, once the worklet has closed its
- * end. Wait for `readable` before retrying a `nil`.
+ * end or this ipc is closed. Wait for `readable` before retrying a `nil`.
  */
 - (NSData *_Nullable)read;
 
@@ -95,11 +95,19 @@
  */
 - (void)read:(void (^_Nonnull)(NSData *_Nullable data, NSError *_Nullable error))completion;
 
+/**
+ * Writes whatever fits without blocking, returning the number of bytes taken,
+ * or -1 once this ipc is closed.
+ */
 - (NSInteger)write:(NSData *_Nonnull)data;
 
 - (void)write:(NSData *_Nonnull)data
    completion:(void (^_Nonnull)(NSError *_Nullable error))completion;
 
+/**
+ * Retires this ipc, which must happen before its worklet is destroyed.
+ * `terminate` does it for you, and closing twice is a no-op.
+ */
 - (void)close;
 
 @end
diff --git a/apple/BareKit/BareKit.m b/apple/BareKit/BareKit.m
index e3934bc..b491519 100644
--- a/apple/BareKit/BareKit.m
+++ b/apple/BareKit/BareKit.m
@@ -103,6 +103,9 @@ - (_Nullable instancetype)init {
 @implementation BareWorklet {
 @public
   bare_worklet_t _worklet;
+
+  // Unowned: the ipc retains this worklet and clears this when it closes.
+  BareIPC *_ipc;
 }
 
 + (void)optimizeForMemory:(BOOL)enabled {
@@ -234,6 +237,8 @@ - (void)resume {
 }
 
 - (void)terminate {
+  [_ipc close];
+
   int err;
   err = bare_worklet_terminate(&_worklet);
   assert(err == 0);
@@ -303,6 +308,8 @@ @implementation BareIPC {
 
   bare_ipc_t _ipc;
   bare_ipc_poll_t _poll;
+
+  bool _closed;
 }
 
 - (_Nullable instancetype)initWithWorklet:(BareWorklet *_Nonnull)worklet {
@@ -320,6 +327,8 @@ - (_Nullable instancetype)initWithWorklet:(BareWorklet *_Nonnull)worklet {
     assert(err == 0);
 
     _poll.data = (__bridge void *) self;
+
+    worklet->_ipc = self;
   }
 
   return self;
@@ -347,6 +356,8 @@ - (void)setWritable:(void (^)(BareIPC *_Nonnull))writable {
 }
 
 - (void)update {
+  if (_closed) return;
+
   int events = 0;
 
   if (_readable) events |= bare_ipc_readable;
@@ -369,6 +380,8 @@ - (void)poll:(int)events {
 }
 
 - (NSData *_Nullable)read {
+  if (_closed) return [NSData data];
+
   int err;
 
   void *data;
@@ -384,6 +397,8 @@ - (NSData *_Nullable)read {
 }
 
 - (void)read:(void (^_Nonnull)(NSData *_Nullable data, NSError *_Nullable error))completion {
+  if (_closed) return completion(nil, nil);
+
   int err;
 
   void *data;
@@ -411,6 +426,8 @@ - (void)read:(void (^_Nonnull)(NSData *_Nullable data, NSError *_Nullable error)
 }
 
 - (NSInteger)write:(NSData *_Nonnull)data {
+  if (_closed) return -1;
+
   int err;
   err = bare_ipc_write(&_ipc, data.bytes, data.length);
   assert(err >= 0 || err == bare_ipc_would_block);
@@ -424,6 +441,13 @@ - (NSInteger)write:(NSData *_Nonnull)data {
 
 - (void)write:(NSData *_Nonnull)data
    completion:(void (^_Nonnull)(NSError *_Nullable error))completion {
+  if (_closed) {
+    return completion([NSError
+      errorWithDomain:@"to.holepunch.bare.kit"
+                 code:-1
+             userInfo:@{NSLocalizedDescriptionKey : @"IPC closed"}]);
+  }
+
   NSInteger written = [self write:data];
 
   if (written == data.length) {
@@ -457,11 +481,19 @@ - (void)write:(NSData *_Nonnull)data
 }
 
 - (void)close {
+  if (_closed) return;
+
+  _closed = true;
+
+  _worklet->_ipc = nil;
+
   bare_ipc_poll_destroy(&_poll);
   bare_ipc_destroy(&_ipc);
 }
 
 - (void)dealloc {
+  [self close];
+
   [_worklet release];
 
   [super dealloc];
diff --git a/package-lock.json b/package-lock.json
index da83fa2..dc38df6 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -12,11 +12,11 @@
         "bare-crypto": "^1.2.8",
         "bare-events": "^2.4.2",
         "bare-fs": "^4.0.1",
-        "bare-ipc": "^1.0.1",
         "bare-logger": "^2.0.0",
         "bare-module": "^7.0.1",
         "bare-module-resolve": "^1.6.2",
         "bare-path": "^3.0.0",
+        "bare-stream": "^2.13.4",
         "bare-system-logger": "^1.0.1",
         "bare-unpack": "^1.0.2",
         "bare-url": "^2.0.1"
@@ -101,6 +101,7 @@
       "version": "3.7.1",
       "resolved": "https://registry.npmjs.org/bare-buffer/-/bare-buffer-3.7.1.tgz",
       "integrity": "sha512-cIjZnSO+y89ykVX9x96OVnA6tGORBn7vGJ/e3PCbBHMEyfMNG/ixzOS0lLa6saMlWcs69bSKjxFyDwjK152OMg==",
+      "devOptional": true,
       "license": "Apache-2.0",
       "engines": {
         "bare": ">=1.20.0"
@@ -302,17 +303,6 @@
         "bare": ">=1.18.0"
       }
     },
-    "node_modules/bare-ipc": {
-      "version": "1.1.2",
-      "resolved": "https://registry.npmjs.org/bare-ipc/-/bare-ipc-1.1.2.tgz",
-      "integrity": "sha512-z2VXX9nHkWTgtiOwDJpuZa3RJeUDqG1VpGJJttVHEdqX9ebX09Bf4K9/RwYeNKerpFJ9tJP0kjfTjk6bJC9wXw==",
-      "license": "Apache-2.0",
-      "dependencies": {
-        "bare-pipe": "^4.0.0",
-        "bare-stream": "^2.1.3",
-        "bare-structured-clone": "^1.4.0"
-      }
-    },
     "node_modules/bare-logger": {
       "version": "2.0.4",
       "resolved": "https://registry.npmjs.org/bare-logger/-/bare-logger-2.0.4.tgz",
@@ -469,7 +459,9 @@
       "version": "4.3.1",
       "resolved": "https://registry.npmjs.org/bare-pipe/-/bare-pipe-4.3.1.tgz",
       "integrity": "sha512-3P4MYTgOys1Bbz5dgWzwfroNOOl8x1zWJHTHVlwsojThhV/uJPRf+qFrYYBOnI0vQiJJLJtHlw8a4Osj/kcsJQ==",
+      "dev": true,
       "license": "Apache-2.0",
+      "optional": true,
       "dependencies": {
         "bare-events": "^2.5.4",
         "bare-stream": "^2.6.4"
@@ -576,7 +568,9 @@
       "version": "1.6.0",
       "resolved": "https://registry.npmjs.org/bare-structured-clone/-/bare-structured-clone-1.6.0.tgz",
       "integrity": "sha512-AZjEERyqF7kAuudlmgrweT2KwwWM0LsGG4Gx/bWyFwJrKU7E3wNG8c09z2DIrdw93p3vDtfOX8fyHOcXfy5m+g==",
+      "dev": true,
       "license": "Apache-2.0",
+      "optional": true,
       "dependencies": {
         "bare-buffer": "^3.6.0",
         "bare-type": "^1.1.0",
@@ -801,7 +795,9 @@
       "version": "3.5.0",
       "resolved": "https://registry.npmjs.org/compact-encoding/-/compact-encoding-3.5.0.tgz",
       "integrity": "sha512-X7yaWh0NNMVB2YXAWSTWjYU3WXvDTniKxbtQlisbXOVutIE6awd+zCQdNqScTjIXv9m9orqdlri+oaWWsKqTuA==",
+      "dev": true,
       "license": "Apache-2.0",
+      "optional": true,
       "dependencies": {
         "b4a": "^1.3.0"
       }
diff --git a/package.json b/package.json
index 419d34e..acac130 100644
--- a/package.json
+++ b/package.json
@@ -22,11 +22,11 @@
     "bare-crypto": "^1.2.8",
     "bare-events": "^2.4.2",
     "bare-fs": "^4.0.1",
-    "bare-ipc": "^1.0.1",
     "bare-logger": "^2.0.0",
     "bare-module": "^7.0.1",
     "bare-module-resolve": "^1.6.2",
     "bare-path": "^3.0.0",
+    "bare-stream": "^2.13.4",
     "bare-system-logger": "^1.0.1",
     "bare-unpack": "^1.0.2",
     "bare-url": "^2.0.1"
diff --git a/shared/CMakeLists.txt b/shared/CMakeLists.txt
index f421b15..29ccd9a 100644
--- a/shared/CMakeLists.txt
+++ b/shared/CMakeLists.txt
@@ -57,8 +57,6 @@ if(APPLE)
   target_sources(
     bare_worklet
     PRIVATE
-      posix/ipc.h
-      posix/ipc.c
       apple/ipc.h
       apple/ipc.m
       apple/suspension.h
@@ -90,8 +88,6 @@ if(ANDROID)
   target_sources(
     bare_worklet
     PRIVATE
-      posix/ipc.h
-      posix/ipc.c
       android/ipc.c
       android/ipc.h
       android/suspension.c
@@ -115,10 +111,8 @@ if(LINUX)
   target_sources(
     bare_worklet
     PRIVATE
-      posix/ipc.h
-      posix/ipc.c
-      linux/ipc.c
-      linux/ipc.h
+      uv/ipc.c
+      uv/ipc.h
       linux/suspension.c
       linux/suspension.h
   )
@@ -134,8 +128,8 @@ if(WIN32)
   target_sources(
     bare_worklet
     PRIVATE
-      win32/ipc.c
-      win32/ipc.h
+      uv/ipc.c
+      uv/ipc.h
       win32/suspension.c
       win32/suspension.h
   )
diff --git a/shared/android/ipc.c b/shared/android/ipc.c
index 370c937..7e0fbee 100644
--- a/shared/android/ipc.c
+++ b/shared/android/ipc.c
@@ -1,9 +1,58 @@
 #include <android/looper.h>
 #include <assert.h>
+#include <errno.h>
+#include <stdint.h>
 #include <stdlib.h>
+#include <sys/eventfd.h>
+#include <unistd.h>
 
 #include "ipc.h"
 
+// The looper wakes on a file descriptor, so the signal is an eventfd. The ipc
+// owns it because the worklet writes to it; the poll owns the registration,
+// which belongs to the host thread whose looper it is.
+typedef struct {
+  int wake;
+} bare_ipc_signal_t;
+
+int
+bare_ipc__signal_init(bare_ipc_t *ipc) {
+  bare_ipc_signal_t *signal = malloc(sizeof(bare_ipc_signal_t));
+
+  if (signal == NULL) return -1;
+
+  // The count stays up until a callback reads it, so a poll that starts after a
+  // signal still wakes.
+  signal->wake = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
+  assert(signal->wake >= 0);
+
+  ipc->signal = signal;
+
+  return 0;
+}
+
+void
+bare_ipc__signal(bare_ipc_t *ipc) {
+  bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal;
+
+  uint64_t value = 1;
+
+  ssize_t written = write(signal->wake, &value, sizeof(value));
+  assert(written == sizeof(value));
+}
+
+void
+bare_ipc__signal_destroy(bare_ipc_t *ipc) {
+  bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal;
+
+  int err = close(signal->wake);
+  assert(err == 0);
+
+  free(signal);
+
+  ipc->signal = NULL;
+}
+
 int
 bare_ipc_poll_alloc(bare_ipc_poll_t **result) {
   bare_ipc_poll_t *poll = malloc(sizeof(bare_ipc_poll_t));
@@ -20,10 +69,13 @@ bare_ipc_poll_init(bare_ipc_poll_t *poll, bare_ipc_t *ipc) {
   poll->ipc = ipc;
   poll->events = 0;
   poll->cb = NULL;
+  poll->data = NULL;
   poll->looper = ALooper_forThread();
 
   ALooper_acquire(poll->looper);
 
+  ipc->poll = poll;
+
   return 0;
 }
 
@@ -33,6 +85,8 @@ bare_ipc_poll_destroy(bare_ipc_poll_t *poll) {
   err = bare_ipc_poll_stop(poll);
   assert(err == 0);
 
+  poll->ipc->poll = NULL;
+
   ALooper_release(poll->looper);
 }
 
@@ -51,11 +105,22 @@ bare_ipc_poll_get_ipc(bare_ipc_poll_t *poll) {
   return poll->ipc;
 }
 
+// Runs on the looper's thread, which is also the thread that started the poll, so
+// the poll cannot go away underneath.
 static int
-bare_ipc__on_poll(int fd, int events, void *data) {
-  bare_ipc_poll_t *poll = data;
+bare_ipc__on_wake(int fd, int events, void *data) {
+  bare_ipc_poll_t *poll = (bare_ipc_poll_t *) data;
 
-  if (poll->cb) poll->cb(poll, (events & ALOOPER_EVENT_INPUT ? bare_ipc_readable : 0) | (events & ALOOPER_EVENT_OUTPUT ? bare_ipc_writable : 0));
+  uint64_t value;
+
+  ssize_t read_len = read(fd, &value, sizeof(value));
+  assert(read_len == sizeof(value) || errno == EAGAIN);
+
+  if (poll->cb != NULL) {
+    int ready = bare_ipc__ready(poll->ipc, poll->events);
+
+    if (ready != 0) poll->cb(poll, ready);
+  }
 
   return 1; // Don't deregister the file descriptor
 }
@@ -64,43 +129,29 @@ int
 bare_ipc_poll_start(bare_ipc_poll_t *poll, int events, bare_ipc_poll_cb cb) {
   if (events == 0) return bare_ipc_poll_stop(poll);
 
-  int err;
+  bare_ipc_t *ipc = poll->ipc;
 
-  if ((events & bare_ipc_readable) == 0) {
-    if ((poll->events & bare_ipc_readable) != 0) {
-      err = ALooper_removeFd(poll->looper, poll->ipc->incoming);
-      assert(err == 1);
-    }
-  } else {
-    if ((poll->events & bare_ipc_readable) == 0) {
-      err = ALooper_addFd(poll->looper, poll->ipc->incoming, ALOOPER_POLL_CALLBACK, ALOOPER_EVENT_INPUT, bare_ipc__on_poll, (void *) poll);
-      assert(err == 1);
-    }
-  }
+  bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal;
 
-  if ((events & bare_ipc_writable) == 0) {
-    if ((poll->events & bare_ipc_writable) != 0) {
-      err = ALooper_removeFd(poll->looper, poll->ipc->outgoing);
-      assert(err == 1);
-    }
-  } else {
-    if ((poll->events & bare_ipc_writable) == 0) {
-      err = ALooper_addFd(poll->looper, poll->ipc->outgoing, ALOOPER_POLL_CALLBACK, ALOOPER_EVENT_OUTPUT, bare_ipc__on_poll, (void *) poll);
-      assert(err == 1);
-    }
+  if (poll->events == 0) {
+    int err = ALooper_addFd(poll->looper, signal->wake, ALOOPER_POLL_CALLBACK, ALOOPER_EVENT_INPUT, bare_ipc__on_wake, (void *) poll);
+    assert(err == 1);
   }
 
   poll->events = events;
   poll->cb = cb;
 
+  // Level-triggered: re-check in case data or space is already there.
+  bare_ipc__signal(ipc);
+
   return 0;
 }
 
 int
 bare_ipc_poll_stop(bare_ipc_poll_t *poll) {
-  ALooper_removeFd(poll->looper, poll->ipc->incoming);
+  bare_ipc_signal_t *signal = (bare_ipc_signal_t *) poll->ipc->signal;
 
-  ALooper_removeFd(poll->looper, poll->ipc->outgoing);
+  if (poll->events != 0) ALooper_removeFd(poll->looper, signal->wake);
 
   poll->events = 0;
   poll->cb = NULL;
diff --git a/shared/apple/ipc.h b/shared/apple/ipc.h
index 742b882..30075bb 100644
--- a/shared/apple/ipc.h
+++ b/shared/apple/ipc.h
@@ -15,10 +15,6 @@ extern "C" {
 struct bare_ipc_poll_s {
   bare_ipc_t *ipc;
 
-  dispatch_queue_t queue;
-  dispatch_source_t reader;
-  dispatch_source_t writer;
-
   int events;
 
   _Atomic bare_ipc_poll_cb cb;
diff --git a/shared/apple/ipc.m b/shared/apple/ipc.m
index 75c052e..322d54a 100644
--- a/shared/apple/ipc.m
+++ b/shared/apple/ipc.m
@@ -5,63 +5,115 @@
 
 static const void *const bare_ipc_poll_queue = &bare_ipc_poll_queue;
 
-int
-bare_ipc_poll_alloc(bare_ipc_poll_t **result) {
-  bare_ipc_poll_t *poll = malloc(sizeof(bare_ipc_poll_t));
+// Waits for the serial queue to reach the current point, unless we are already
+// on it, in which case the handler is this call and there is nothing to wait for.
+static void
+bare_ipc__drain(dispatch_queue_t queue) {
+  if (dispatch_get_specific(bare_ipc_poll_queue) != NULL) return;
 
-  if (poll == NULL) return -1;
+  dispatch_semaphore_t done = dispatch_semaphore_create(0);
 
-  *result = poll;
+  dispatch_async(queue, ^{
+    dispatch_semaphore_signal(done);
+  });
 
-  return 0;
+  dispatch_semaphore_wait(done, DISPATCH_TIME_FOREVER);
 }
 
+typedef struct {
+  dispatch_queue_t queue;
+  dispatch_source_t source;
+} bare_ipc_signal_t;
+
 int
-bare_ipc_poll_init(bare_ipc_poll_t *poll, bare_ipc_t *ipc) {
+bare_ipc__signal_init(bare_ipc_t *ipc) {
+  bare_ipc_signal_t *signal = malloc(sizeof(bare_ipc_signal_t));
+
+  if (signal == NULL) return -1;
+
   dispatch_queue_t queue = dispatch_queue_create("to.holepunch.bare.kit.ipc", DISPATCH_QUEUE_SERIAL);
 
   dispatch_queue_set_specific(queue, bare_ipc_poll_queue, (void *) bare_ipc_poll_queue, NULL);
 
-  dispatch_source_t reader = dispatch_source_create(DISPATCH_SOURCE_TYPE_READ, ipc->incoming, 0, queue);
+  // A manual source the queue signal triggers; the handler reports whichever
+  // requested events are actually ready of whichever poll is subscribed.
+  dispatch_source_t source = dispatch_source_create(DISPATCH_SOURCE_TYPE_DATA_OR, 0, 0, queue);
 
-  dispatch_source_t writer = dispatch_source_create(DISPATCH_SOURCE_TYPE_WRITE, ipc->outgoing, 0, queue);
-
-  dispatch_source_set_cancel_handler(reader, ^{
-    dispatch_release(reader);
+  dispatch_source_set_cancel_handler(source, ^{
+    dispatch_release(source);
   });
 
-  dispatch_source_set_cancel_handler(writer, ^{
-    dispatch_release(writer);
-  });
+  dispatch_source_set_event_handler(source, ^{
+    bare_ipc_poll_t *poll = ipc->poll;
+
+    if (poll == NULL) return;
 
-  dispatch_source_set_event_handler(reader, ^{
     bare_ipc_poll_cb cb = atomic_load(&poll->cb);
 
     if (cb == NULL) return;
 
+    int ready = bare_ipc__ready(ipc, poll->events);
+
+    if (ready == 0) return;
+
     @autoreleasepool {
-      cb(poll, bare_ipc_readable);
+      cb(poll, ready);
     }
   });
 
-  dispatch_source_set_event_handler(writer, ^{
-    bare_ipc_poll_cb cb = atomic_load(&poll->cb);
+  signal->queue = queue;
+  signal->source = source;
 
-    if (cb == NULL) return;
+  ipc->signal = signal;
 
-    @autoreleasepool {
-      cb(poll, bare_ipc_writable);
-    }
-  });
+  dispatch_resume(source);
+
+  return 0;
+}
+
+void
+bare_ipc__signal(bare_ipc_t *ipc) {
+  bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal;
+
+  dispatch_source_merge_data(signal->source, 1);
+}
 
+void
+bare_ipc__signal_destroy(bare_ipc_t *ipc) {
+  bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal;
+
+  dispatch_source_cancel(signal->source);
+
+  bare_ipc__drain(signal->queue);
+
+  dispatch_release(signal->queue);
+
+  free(signal);
+
+  ipc->signal = NULL;
+}
+
+int
+bare_ipc_poll_alloc(bare_ipc_poll_t **result) {
+  bare_ipc_poll_t *poll = malloc(sizeof(bare_ipc_poll_t));
+
+  if (poll == NULL) return -1;
+
+  *result = poll;
+
+  return 0;
+}
+
+int
+bare_ipc_poll_init(bare_ipc_poll_t *poll, bare_ipc_t *ipc) {
   poll->ipc = ipc;
   poll->events = 0;
-  poll->queue = queue;
-  poll->reader = reader;
-  poll->writer = writer;
+  poll->data = NULL;
 
   atomic_init(&poll->cb, NULL);
 
+  ipc->poll = poll;
+
   return 0;
 }
 
@@ -71,23 +123,14 @@
   err = bare_ipc_poll_stop(poll);
   assert(err == 0);
 
-  dispatch_resume(poll->reader);
-  dispatch_source_cancel(poll->reader);
-
-  dispatch_resume(poll->writer);
-  dispatch_source_cancel(poll->writer);
+  bare_ipc_t *ipc = poll->ipc;
 
-  if (dispatch_get_specific(bare_ipc_poll_queue) == NULL) {
-    dispatch_semaphore_t done = dispatch_semaphore_create(0);
+  ipc->poll = NULL;
 
-    dispatch_async(poll->queue, ^{
-      dispatch_semaphore_signal(done);
-    });
+  bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal;
 
-    dispatch_semaphore_wait(done, DISPATCH_TIME_FOREVER);
-  }
-
-  dispatch_release(poll->queue);
+  // Let any handler already running finish before the caller frees the poll.
+  bare_ipc__drain(signal->queue);
 }
 
 void *
@@ -109,30 +152,18 @@
 bare_ipc_poll_start(bare_ipc_poll_t *poll, int events, bare_ipc_poll_cb cb) {
   if (events == 0) return bare_ipc_poll_stop(poll);
 
-  if ((events & bare_ipc_readable) == 0) {
-    if ((poll->events & bare_ipc_readable) != 0) dispatch_suspend(poll->reader);
-  } else {
-    if ((poll->events & bare_ipc_readable) == 0) dispatch_resume(poll->reader);
-  }
-
-  if ((events & bare_ipc_writable) == 0) {
-    if ((poll->events & bare_ipc_writable) != 0) dispatch_suspend(poll->writer);
-  } else {
-    if ((poll->events & bare_ipc_writable) == 0) dispatch_resume(poll->writer);
-  }
-
   poll->events = events;
 
   atomic_store(&poll->cb, cb);
 
+  // Level-trigger: re-check immediately in case data or space is already ready.
+  bare_ipc__signal(poll->ipc);
+
   return 0;
 }
 
 int
 bare_ipc_poll_stop(bare_ipc_poll_t *poll) {
-  if ((poll->events & bare_ipc_readable) != 0) dispatch_suspend(poll->reader);
-  if ((poll->events & bare_ipc_writable) != 0) dispatch_suspend(poll->writer);
-
   poll->events = 0;
 
   atomic_store(&poll->cb, NULL);
diff --git a/shared/builtins.json b/shared/builtins.json
index b8c3aa8..60278f2 100644
--- a/shared/builtins.json
+++ b/shared/builtins.json
@@ -3,12 +3,10 @@
   { "addon": "bare-fs" },
   { "addon": "bare-hrtime" },
   { "addon": "bare-inspect" },
-  { "addon": "bare-ipc" },
   { "addon": "bare-logger" },
   { "addon": "bare-module" },
   { "addon": "bare-module-lexer" },
   { "addon": "bare-path" },
-  { "addon": "bare-pipe" },
   { "addon": "bare-os" },
   { "addon": "bare-system-logger" },
   { "addon": "bare-type" },
diff --git a/shared/ipc.c b/shared/ipc.c
index 04629bf..9ad963c 100644
--- a/shared/ipc.c
+++ b/shared/ipc.c
@@ -1,5 +1,17 @@
+#include <stdlib.h>
+
 #include "ipc.h"
 
+// Runs on the worklet thread when the queue changes.
+static void
+bare_ipc__on_signal(bare_queue_port_t *port) {
+  bare_ipc_t *ipc = (bare_ipc_t *) port->data;
+
+  if (ipc == NULL) return;
+
+  bare_ipc__signal(ipc);
+}
+
 int
 bare_ipc_alloc(bare_ipc_t **result) {
   bare_ipc_t *ipc = malloc(sizeof(bare_ipc_t));
@@ -12,11 +24,58 @@ bare_ipc_alloc(bare_ipc_t **result) {
 }
 
 int
-bare_ipc_get_incoming(bare_ipc_t *ipc) {
-  return ipc->incoming;
+bare_ipc_init(bare_ipc_t *ipc, bare_worklet_t *worklet) {
+  ipc->poll = NULL;
+  ipc->signal = NULL;
+
+  int err = bare_ipc__signal_init(ipc);
+
+  if (err < 0) return err;
+
+  ipc->port = bare_queue_open_thread(worklet->queue, bare_ipc__on_signal);
+  ipc->port->data = ipc;
+
+  return 0;
+}
+
+void
+bare_ipc_destroy(bare_ipc_t *ipc) {
+  bare_queue_t *queue = ipc->port->queue;
+
+  // Order matters: close marks the end shut, retiring clears the callback under
+  // the queue's lock, and only then is nothing left that can reach the signal.
+  bare_queue_close(ipc->port);
+
+  bare_queue_shutdown_thread(queue);
+
+  bare_ipc__signal_destroy(ipc);
+}
+
+int
+bare_ipc__ready(bare_ipc_t *ipc, int events) {
+  int ready = 0;
+
+  if ((events & bare_ipc_readable) != 0 && bare_queue_readable(ipc->port)) ready |= bare_ipc_readable;
+  if ((events & bare_ipc_writable) != 0 && bare_queue_writable(ipc->port)) ready |= bare_ipc_writable;
+
+  return ready;
+}
+
+int
+bare_ipc_read(bare_ipc_t *ipc, void **data, size_t *len) {
+  int status = bare_queue_read(ipc->port, data, len);
+
+  if (status == bare_queue_would_block) return bare_ipc_would_block;
+
+  return 0;
 }
 
 int
-bare_ipc_get_outgoing(bare_ipc_t *ipc) {
-  return ipc->outgoing;
+bare_ipc_write(bare_ipc_t *ipc, const void *data, size_t len) {
+  int n = bare_queue_write(ipc->port, data, len);
+
+  if (n == bare_queue_would_block) return bare_ipc_would_block;
+  if (n == bare_queue_closed) return bare_ipc_error;
+
+  return n;
 }
diff --git a/shared/ipc.h b/shared/ipc.h
index 70fffdc..7fde258 100644
--- a/shared/ipc.h
+++ b/shared/ipc.h
@@ -7,10 +7,9 @@ extern "C" {
 
 #include <stddef.h>
 
+#include "queue.h"
 #include "worklet.h"
 
-#define BARE_IPC_READ_BUFFER_SIZE 64 * 1024
-
 typedef struct bare_ipc_s bare_ipc_t;
 typedef struct bare_ipc_poll_s bare_ipc_poll_t;
 
@@ -26,6 +25,32 @@ enum {
   bare_ipc_error = -2,
 };
 
+struct bare_ipc_s {
+  bare_queue_port_t *port;
+  bare_ipc_poll_t *poll;
+
+  // The primitive that hops a queue change onto the host run loop. Owned here,
+  // for the lifetime of the port, because the worklet signals through it: a poll
+  // comes and goes underneath.
+  void *signal;
+};
+
+// Per platform. `signal` runs on the worklet thread, so it may only touch state
+// owned by the ipc.
+int
+bare_ipc__signal_init(bare_ipc_t *ipc);
+
+void
+bare_ipc__signal_destroy(bare_ipc_t *ipc);
+
+void
+bare_ipc__signal(bare_ipc_t *ipc);
+
+// Which of the events a poll asked for are ready right now. A wake says the
+// queue changed, not how, so the platform asks after each one.
+int
+bare_ipc__ready(bare_ipc_t *ipc, int events);
+
 #if defined(BARE_KIT_DARWIN) || defined(BARE_KIT_IOS)
 #include "apple/ipc.h"
 #endif
@@ -34,14 +59,10 @@ enum {
 #include "android/ipc.h"
 #endif
 
-#if defined(BARE_KIT_LINUX)
-#include "linux/ipc.h"
-#endif
-
-#if defined(BARE_KIT_WINDOWS)
-#include "win32/ipc.h"
-#else
-#include "posix/ipc.h"
+// Neither Linux nor Windows hands us a host run loop to post to, so both wake on
+// a loop of their own.
+#if defined(BARE_KIT_LINUX) || defined(BARE_KIT_WINDOWS)
+#include "uv/ipc.h"
 #endif
 
 int
@@ -50,15 +71,12 @@ bare_ipc_alloc(bare_ipc_t **result);
 int
 bare_ipc_init(bare_ipc_t *ipc, bare_worklet_t *worklet);
 
+// Retires the host end. Must be called before `bare_worklet_destroy`: the queue
+// belongs to the worklet, which may free it as soon as the host has signalled it
+// to be destroyed. A host that keeps an ipc must therefore keep its worklet.
 void
 bare_ipc_destroy(bare_ipc_t *ipc);
 
-int
-bare_ipc_get_incoming(bare_ipc_t *ipc);
-
-int
-bare_ipc_get_outgoing(bare_ipc_t *ipc);
-
 int
 bare_ipc_read(bare_ipc_t *ipc, void **data, size_t *len);
 
diff --git a/shared/linux/ipc.c b/shared/linux/ipc.c
deleted file mode 100644
index 28d53f3..0000000
--- a/shared/linux/ipc.c
+++ /dev/null
@@ -1,184 +0,0 @@
-#include <assert.h>
-#include <pthread.h>
-#include <stdlib.h>
-#include <sys/epoll.h>
-#include <sys/eventfd.h>
-#include <unistd.h>
-
-#include "ipc.h"
-
-enum {
-  bare_ipc_poll_close,
-  bare_ipc_poll_incoming,
-  bare_ipc_poll_outging,
-};
-
-void *
-bare_ipc_poll__poller(void *data) {
-  bare_ipc_poll_t *poll = (bare_ipc_poll_t *) data;
-
-  while (true) {
-    struct epoll_event events[4];
-    int n = epoll_wait(poll->fd.poll, events, 4, -1);
-    assert(n >= 0 || errno == EINTR);
-
-    if (errno == EINTR) continue;
-
-    int flags = 0;
-
-    for (int i = 0; i < n; i++) {
-      struct epoll_event event = events[i];
-
-      if (event.data.fd == bare_ipc_poll_close) {
-        close(poll->fd.poll);
-        close(poll->fd.close);
-
-        pthread_exit(0);
-      }
-
-      if (event.data.fd == bare_ipc_poll_incoming && (event.events & EPOLLIN) != 0) {
-        flags |= bare_ipc_readable;
-      }
-
-      if (event.data.fd == bare_ipc_poll_outging && (event.events & EPOLLOUT) != 0) {
-        flags |= bare_ipc_writable;
-      }
-    }
-
-    if (poll->cb) poll->cb(poll, flags);
-  }
-}
-
-int
-bare_ipc_poll_alloc(bare_ipc_poll_t **result) {
-  bare_ipc_poll_t *poll = malloc(sizeof(bare_ipc_poll_t));
-
-  if (poll == NULL) return -1;
-
-  *result = poll;
-
-  return 0;
-}
-
-int
-bare_ipc_poll_init(bare_ipc_poll_t *poll, bare_ipc_t *ipc) {
-  int err;
-
-  poll->ipc = ipc;
-  poll->fd.poll = epoll_create(3);
-  poll->fd.close = eventfd(0, 0);
-  poll->events = 0;
-  poll->cb = NULL;
-
-  struct epoll_event signal = {
-    .events = EPOLLIN,
-    .data.fd = bare_ipc_poll_close,
-  };
-
-  err = epoll_ctl(poll->fd.poll, EPOLL_CTL_ADD, poll->fd.close, &signal);
-  assert(err == 0);
-
-  pthread_create(&poll->thread, NULL, &bare_ipc_poll__poller, (void *) poll);
-
-  return 0;
-}
-
-void
-bare_ipc_poll_destroy(bare_ipc_poll_t *poll) {
-  int err;
-  err = bare_ipc_poll_stop(poll);
-  assert(err == 0);
-
-  uint64_t signal = 1;
-
-  ssize_t written = write(poll->fd.close, (const void *) &signal, 8);
-  assert(written == 8);
-
-  err = pthread_join(poll->thread, NULL);
-  assert(err == 0);
-}
-
-void *
-bare_ipc_poll_get_data(bare_ipc_poll_t *poll) {
-  return poll->data;
-}
-
-void
-bare_ipc_poll_set_data(bare_ipc_poll_t *poll, void *data) {
-  poll->data = data;
-}
-
-bare_ipc_t *
-bare_ipc_poll_get_ipc(bare_ipc_poll_t *poll) {
-  return poll->ipc;
-}
-
-int
-bare_ipc_poll_start(bare_ipc_poll_t *poll, int events, bare_ipc_poll_cb cb) {
-  if (events == 0) return bare_ipc_poll_stop(poll);
-
-  int err;
-
-  struct epoll_event incoming = {
-    .events = EPOLLIN,
-    .data.fd = bare_ipc_poll_incoming,
-  };
-
-  if ((events & bare_ipc_readable) == 0) {
-    if ((poll->events & bare_ipc_readable) != 0) {
-      err = epoll_ctl(poll->fd.poll, EPOLL_CTL_DEL, poll->ipc->incoming, &incoming);
-      assert(err == 0);
-    }
-  } else {
-    if ((poll->events & bare_ipc_readable) == 0) {
-      err = epoll_ctl(poll->fd.poll, EPOLL_CTL_ADD, poll->ipc->incoming, &incoming);
-      assert(err == 0);
-    }
-  }
-
-  struct epoll_event outgoing = {
-    .events = EPOLLOUT,
-    .data.fd = bare_ipc_poll_outging,
-  };
-
-  if ((events & bare_ipc_writable) == 0) {
-    if ((poll->events & bare_ipc_writable) != 0) {
-      err = epoll_ctl(poll->fd.poll, EPOLL_CTL_DEL, poll->ipc->outgoing, &outgoing);
-      assert(err == 0);
-    }
-  } else {
-    if ((poll->events & bare_ipc_writable) == 0) {
-      err = epoll_ctl(poll->fd.poll, EPOLL_CTL_ADD, poll->ipc->outgoing, &outgoing);
-      assert(err == 0);
-    }
-  }
-
-  poll->events = events;
-  poll->cb = cb;
-
-  return 0;
-}
-
-int
-bare_ipc_poll_stop(bare_ipc_poll_t *poll) {
-  int err;
-
-  struct epoll_event incoming = {
-    .events = EPOLLIN,
-  };
-
-  err = epoll_ctl(poll->fd.poll, EPOLL_CTL_DEL, poll->ipc->incoming, &incoming);
-  assert(err == 0 || errno == ENOENT);
-
-  struct epoll_event outgoing = {
-    .events = EPOLLOUT,
-  };
-
-  err = epoll_ctl(poll->fd.poll, EPOLL_CTL_DEL, poll->ipc->outgoing, &outgoing);
-  assert(err == 0 || errno == ENOENT);
-
-  poll->events = 0;
-  poll->cb = NULL;
-
-  return 0;
-}
diff --git a/shared/linux/ipc.h b/shared/linux/ipc.h
deleted file mode 100644
index 81bf6c0..0000000
--- a/shared/linux/ipc.h
+++ /dev/null
@@ -1,34 +0,0 @@
-#ifndef BARE_KIT_LINUX_IPC_H
-#define BARE_KIT_LINUX_IPC_H
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-#include <pthread.h>
-#include <sys/epoll.h>
-
-#include "../ipc.h"
-
-struct bare_ipc_poll_s {
-  bare_ipc_t *ipc;
-
-  struct {
-    int poll;
-    int close;
-  } fd;
-
-  pthread_t thread;
-
-  int events;
-
-  bare_ipc_poll_cb cb;
-
-  void *data;
-};
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif // BARE_KIT_LINUX_IPC_H
diff --git a/shared/posix/ipc.c b/shared/posix/ipc.c
deleted file mode 100644
index 102999a..0000000
--- a/shared/posix/ipc.c
+++ /dev/null
@@ -1,42 +0,0 @@
-#include <unistd.h>
-
-#include "ipc.h"
-
-int
-bare_ipc_init(bare_ipc_t *ipc, bare_worklet_t *worklet) {
-  ipc->incoming = dup(worklet->incoming);
-  ipc->outgoing = dup(worklet->outgoing);
-
-  return 0;
-}
-
-void
-bare_ipc_destroy(bare_ipc_t *ipc) {
-  close(ipc->incoming);
-  close(ipc->outgoing);
-}
-
-int
-bare_ipc_read(bare_ipc_t *ipc, void **data, size_t *len) {
-  ssize_t res = read(ipc->incoming, ipc->data, BARE_IPC_READ_BUFFER_SIZE);
-
-  if (res < 0) {
-    return (errno == EAGAIN || errno == EWOULDBLOCK) ? bare_ipc_would_block : bare_ipc_error;
-  }
-
-  *len = res;
-  *data = ipc->data;
-
-  return 0;
-}
-
-int
-bare_ipc_write(bare_ipc_t *ipc, const void *data, size_t len) {
-  ssize_t res = write(ipc->outgoing, data, len);
-
-  if (res < 0) {
-    return (errno == EAGAIN || errno == EWOULDBLOCK) ? bare_ipc_would_block : bare_ipc_error;
-  }
-
-  return res;
-}
diff --git a/shared/posix/ipc.h b/shared/posix/ipc.h
deleted file mode 100644
index 8621f2b..0000000
--- a/shared/posix/ipc.h
+++ /dev/null
@@ -1,21 +0,0 @@
-#ifndef BARE_KIT_POSIX_IPC_H
-#define BARE_KIT_POSIX_IPC_H
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-#include "../ipc.h"
-
-struct bare_ipc_s {
-  int incoming;
-  int outgoing;
-
-  char data[BARE_IPC_READ_BUFFER_SIZE];
-};
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif // BARE_KIT_POSIX_IPC_H
diff --git a/shared/uv/ipc.c b/shared/uv/ipc.c
new file mode 100644
index 0000000..02b5f73
--- /dev/null
+++ b/shared/uv/ipc.c
@@ -0,0 +1,245 @@
+#include <assert.h>
+#include <stdbool.h>
+#include <stdlib.h>
+#include <uv.h>
+
+#include "ipc.h"
+
+// A loop of our own, on a thread of our own: the callback runs there, and hosts
+// marshal to their UI thread themselves. `running` is held across the callback so
+// a poll cannot be torn down underneath one, and the key marks the thread that
+// holds it, for a poll that stops or destroys itself from inside its callback.
+typedef struct {
+  uv_loop_t loop;
+
+  uv_async_t wake;
+  uv_async_t close;
+
+  uv_thread_t thread;
+  uv_key_t key;
+  uv_mutex_t running;
+} bare_ipc_signal_t;
+
+static inline bool
+bare_ipc__reentrant(bare_ipc_signal_t *signal) {
+  return uv_key_get(&signal->key) == (void *) signal;
+}
+
+static inline void
+bare_ipc__lock(bare_ipc_signal_t *signal) {
+  if (!bare_ipc__reentrant(signal)) uv_mutex_lock(&signal->running);
+}
+
+static inline void
+bare_ipc__unlock(bare_ipc_signal_t *signal) {
+  if (!bare_ipc__reentrant(signal)) uv_mutex_unlock(&signal->running);
+}
+
+static void
+bare_ipc__on_wake(uv_async_t *handle) {
+  bare_ipc_t *ipc = (bare_ipc_t *) uv_handle_get_data((uv_handle_t *) handle);
+
+  bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal;
+
+  uv_mutex_lock(&signal->running);
+
+  bare_ipc_poll_t *poll = ipc->poll;
+
+  if (poll != NULL && poll->cb != NULL) {
+    int ready = bare_ipc__ready(ipc, poll->events);
+
+    if (ready != 0) poll->cb(poll, ready);
+  }
+
+  uv_mutex_unlock(&signal->running);
+}
+
+static void
+bare_ipc__on_close(uv_async_t *handle) {
+  bare_ipc_t *ipc = (bare_ipc_t *) uv_handle_get_data((uv_handle_t *) handle);
+
+  bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal;
+
+  uv_close((uv_handle_t *) &signal->wake, NULL);
+  uv_close((uv_handle_t *) &signal->close, NULL);
+}
+
+static void
+bare_ipc__signal_thread(void *data) {
+  bare_ipc_t *ipc = (bare_ipc_t *) data;
+
+  bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal;
+
+  uv_key_set(&signal->key, (void *) signal);
+
+  int err;
+
+  err = uv_run(&signal->loop, UV_RUN_DEFAULT);
+  assert(err == 0);
+
+  err = uv_loop_close(&signal->loop);
+  assert(err == 0);
+}
+
+int
+bare_ipc__signal_init(bare_ipc_t *ipc) {
+  bare_ipc_signal_t *signal = malloc(sizeof(bare_ipc_signal_t));
+
+  if (signal == NULL) return -1;
+
+  int err;
+
+  err = uv_loop_init(&signal->loop);
+  assert(err == 0);
+
+  err = uv_async_init(&signal->loop, &signal->wake, bare_ipc__on_wake);
+  assert(err == 0);
+
+  uv_handle_set_data((uv_handle_t *) &signal->wake, (void *) ipc);
+
+  err = uv_async_init(&signal->loop, &signal->close, bare_ipc__on_close);
+  assert(err == 0);
+
+  uv_handle_set_data((uv_handle_t *) &signal->close, (void *) ipc);
+
+  err = uv_key_create(&signal->key);
+  assert(err == 0);
+
+  err = uv_mutex_init(&signal->running);
+  assert(err == 0);
+
+  ipc->signal = signal;
+
+  // The loop and its handles are up, so a signal arriving the moment the port
+  // opens has somewhere to land.
+  err = uv_thread_create(&signal->thread, bare_ipc__signal_thread, (void *) ipc);
+  assert(err == 0);
+
+  return 0;
+}
+
+void
+bare_ipc__signal(bare_ipc_t *ipc) {
+  bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal;
+
+  int err = uv_async_send(&signal->wake);
+  assert(err == 0);
+}
+
+void
+bare_ipc__signal_destroy(bare_ipc_t *ipc) {
+  bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal;
+
+  int err;
+
+  err = uv_async_send(&signal->close);
+  assert(err == 0);
+
+  err = uv_thread_join(&signal->thread);
+  assert(err == 0);
+
+  uv_mutex_destroy(&signal->running);
+
+  uv_key_delete(&signal->key);
+
+  free(signal);
+
+  ipc->signal = NULL;
+}
+
+int
+bare_ipc_poll_alloc(bare_ipc_poll_t **result) {
+  bare_ipc_poll_t *poll = malloc(sizeof(bare_ipc_poll_t));
+
+  if (poll == NULL) return -1;
+
+  *result = poll;
+
+  return 0;
+}
+
+int
+bare_ipc_poll_init(bare_ipc_poll_t *poll, bare_ipc_t *ipc) {
+  poll->ipc = ipc;
+  poll->events = 0;
+  poll->cb = NULL;
+  poll->data = NULL;
+
+  bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal;
+
+  bare_ipc__lock(signal);
+
+  ipc->poll = poll;
+
+  bare_ipc__unlock(signal);
+
+  return 0;
+}
+
+void
+bare_ipc_poll_destroy(bare_ipc_poll_t *poll) {
+  int err;
+  err = bare_ipc_poll_stop(poll);
+  assert(err == 0);
+
+  bare_ipc_t *ipc = poll->ipc;
+
+  bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal;
+
+  // Under the lock, so the caller may free the poll as soon as this returns.
+  bare_ipc__lock(signal);
+
+  ipc->poll = NULL;
+
+  bare_ipc__unlock(signal);
+}
+
+void *
+bare_ipc_poll_get_data(bare_ipc_poll_t *poll) {
+  return poll->data;
+}
+
+void
+bare_ipc_poll_set_data(bare_ipc_poll_t *poll, void *data) {
+  poll->data = data;
+}
+
+bare_ipc_t *
+bare_ipc_poll_get_ipc(bare_ipc_poll_t *poll) {
+  return poll->ipc;
+}
+
+int
+bare_ipc_poll_start(bare_ipc_poll_t *poll, int events, bare_ipc_poll_cb cb) {
+  if (events == 0) return bare_ipc_poll_stop(poll);
+
+  bare_ipc_t *ipc = poll->ipc;
+
+  bare_ipc_signal_t *signal = (bare_ipc_signal_t *) ipc->signal;
+
+  bare_ipc__lock(signal);
+
+  poll->events = events;
+  poll->cb = cb;
+
+  bare_ipc__unlock(signal);
+
+  // Level-triggered: re-check in case data or space is already there.
+  bare_ipc__signal(ipc);
+
+  return 0;
+}
+
+int
+bare_ipc_poll_stop(bare_ipc_poll_t *poll) {
+  bare_ipc_signal_t *signal = (bare_ipc_signal_t *) poll->ipc->signal;
+
+  bare_ipc__lock(signal);
+
+  poll->events = 0;
+  poll->cb = NULL;
+
+  bare_ipc__unlock(signal);
+
+  return 0;
+}
diff --git a/shared/uv/ipc.h b/shared/uv/ipc.h
new file mode 100644
index 0000000..1ede71d
--- /dev/null
+++ b/shared/uv/ipc.h
@@ -0,0 +1,24 @@
+#ifndef BARE_KIT_UV_IPC_H
+#define BARE_KIT_UV_IPC_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include "../ipc.h"
+
+struct bare_ipc_poll_s {
+  bare_ipc_t *ipc;
+
+  int events;
+
+  bare_ipc_poll_cb cb;
+
+  void *data;
+};
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif // BARE_KIT_UV_IPC_H
diff --git a/shared/win32/ipc.c b/shared/win32/ipc.c
deleted file mode 100644
index 099a517..0000000
--- a/shared/win32/ipc.c
+++ /dev/null
@@ -1,280 +0,0 @@
-#include <io.h>
-#include <uv.h>
-
-#include "ipc.h"
-
-void
-bare_ipc__close(uv_async_t *handle) {
-  bare_ipc_t *ipc = (bare_ipc_t *) uv_handle_get_data((uv_handle_t *) handle);
-
-  uv_close((uv_handle_t *) &ipc->pipe.incoming, NULL);
-  uv_close((uv_handle_t *) &ipc->pipe.outgoing, NULL);
-  uv_close((uv_handle_t *) &ipc->read, NULL);
-  uv_close((uv_handle_t *) &ipc->write, NULL);
-  uv_close((uv_handle_t *) &ipc->close, NULL);
-}
-
-void
-bare_ipc__on_alloc(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf) {
-  bare_ipc_t *ipc = (bare_ipc_t *) uv_handle_get_data(handle);
-  buf->base = ipc->read_tmp_buffer;
-  buf->len = BARE_IPC_READ_BUFFER_SIZE;
-}
-
-void
-bare_ipc__on_read(uv_stream_t *stream, ssize_t nread, const uv_buf_t *buf) {
-  // End-of-stream when the worklet closes its IPC. Stop reading rather than
-  // aborting; surfacing EOF to the host is not yet implemented for win32.
-  if (nread == UV_EOF) {
-    uv_read_stop(stream);
-    return;
-  }
-
-  assert(nread >= 0);
-
-  if (nread == 0) {
-    return;
-  }
-
-  int err;
-
-  err = uv_read_stop(stream);
-  assert(err == 0);
-
-  bare_ipc_t *ipc = (bare_ipc_t *) uv_handle_get_data((uv_handle_t *) stream);
-
-  uv_mutex_lock(&ipc->reading);
-
-  memcpy(ipc->read_buffer.base, buf->base, nread);
-  ipc->read_buffer.len = nread;
-
-  uv_mutex_unlock(&ipc->reading);
-
-  if (ipc->poll && ipc->poll->cb) ipc->poll->cb(ipc->poll, bare_ipc_readable);
-}
-
-void
-bare_ipc__read(uv_async_t *handle) {
-  bare_ipc_t *ipc = (bare_ipc_t *) uv_handle_get_data((uv_handle_t *) handle);
-
-  int err = uv_read_start((uv_stream_t *) &ipc->pipe.incoming, bare_ipc__on_alloc, bare_ipc__on_read);
-  assert(err == 0);
-}
-
-void
-bare_ipc__on_write(uv_write_t *req, int status) {
-  assert(status == 0);
-
-  bare_ipc_t *ipc = (bare_ipc_t *) uv_handle_get_data((uv_handle_t *) req);
-
-  free(ipc->write_buffer.base);
-
-  uv_mutex_unlock(&ipc->writing);
-
-  if (ipc->poll && ipc->poll->cb && (ipc->poll->events & bare_ipc_writable) != 0) ipc->poll->cb(ipc->poll, bare_ipc_writable);
-}
-
-void
-bare_ipc__write(uv_async_t *handle) {
-  int err;
-
-  bare_ipc_t *ipc = (bare_ipc_t *) uv_handle_get_data((uv_handle_t *) handle);
-
-  err = uv_write(&ipc->write_request, (uv_stream_t *) &ipc->pipe.outgoing, &ipc->write_buffer, 1, bare_ipc__on_write);
-  assert(err == 0);
-}
-
-void
-bare_ipc__loop(void *arg) {
-  bare_ipc_t *ipc = (bare_ipc_t *) arg;
-
-  int err;
-
-  err = uv_loop_init(&ipc->loop);
-  assert(err == 0);
-
-  uv_handle_set_data((uv_handle_t *) &ipc->write_request, (void *) ipc);
-
-  err = uv_async_init(&ipc->loop, &ipc->read, bare_ipc__read);
-  assert(err == 0);
-  uv_handle_set_data((uv_handle_t *) &ipc->read, (void *) ipc);
-
-  err = uv_async_init(&ipc->loop, &ipc->write, bare_ipc__write);
-  assert(err == 0);
-  uv_handle_set_data((uv_handle_t *) &ipc->write, (void *) ipc);
-
-  err = uv_async_init(&ipc->loop, &ipc->close, bare_ipc__close);
-  assert(err == 0);
-  uv_handle_set_data((uv_handle_t *) &ipc->close, (void *) ipc);
-
-  err = uv_pipe_init(&ipc->loop, &ipc->pipe.incoming, 0);
-  assert(err == 0);
-  err = uv_pipe_open(&ipc->pipe.incoming, ipc->incoming);
-  assert(err == 0);
-  uv_handle_set_data((uv_handle_t *) &ipc->pipe.incoming, (void *) ipc);
-
-  err = uv_pipe_init(&ipc->loop, &ipc->pipe.outgoing, 0);
-  assert(err == 0);
-  err = uv_pipe_open(&ipc->pipe.outgoing, ipc->outgoing);
-  assert(err == 0);
-  uv_handle_set_data((uv_handle_t *) &ipc->pipe.outgoing, (void *) ipc);
-
-  uv_barrier_wait(&ipc->ready);
-
-  err = uv_run(&ipc->loop, UV_RUN_DEFAULT);
-  assert(err == 0);
-
-  err = uv_loop_close(&ipc->loop);
-  assert(err == 0);
-}
-
-int
-bare_ipc_init(bare_ipc_t *ipc, bare_worklet_t *worklet) {
-  int err;
-
-  ipc->incoming = _dup(worklet->incoming);
-  ipc->outgoing = _dup(worklet->outgoing);
-
-  err = uv_mutex_init(&ipc->reading);
-  assert(err == 0);
-
-  err = uv_mutex_init(&ipc->writing);
-  assert(err == 0);
-
-  err = uv_barrier_init(&ipc->ready, 2);
-  assert(err == 0);
-
-  err = uv_thread_create(&ipc->thread, bare_ipc__loop, ipc);
-  assert(err == 0);
-
-  ipc->read_buffer.len = 0;
-  ipc->poll = NULL;
-
-  uv_barrier_wait(&ipc->ready);
-  uv_barrier_destroy(&ipc->ready);
-
-  return 0;
-}
-
-void
-bare_ipc_destroy(bare_ipc_t *ipc) {
-  int err;
-
-  err = uv_async_send(&ipc->close);
-  assert(err == 0);
-
-  err = uv_thread_join(&ipc->thread);
-  assert(err == 0);
-
-  uv_mutex_destroy(&ipc->reading);
-
-  uv_mutex_destroy(&ipc->writing);
-}
-
-int
-bare_ipc_read(bare_ipc_t *ipc, void **data, size_t *len) {
-  uv_mutex_lock(&ipc->reading);
-
-  if (ipc->read_buffer.len > 0) {
-    *data = ipc->read_buffer.base;
-    *len = ipc->read_buffer.len;
-    ipc->read_buffer.len = 0;
-
-    uv_mutex_unlock(&ipc->reading);
-
-    return 0;
-  }
-
-  uv_mutex_unlock(&ipc->reading);
-
-  return bare_ipc_would_block;
-}
-
-int
-bare_ipc_write(bare_ipc_t *ipc, const void *data, size_t len) {
-  int err;
-
-  if (uv_mutex_trylock(&ipc->writing) != 0) {
-    return bare_ipc_would_block;
-  }
-
-  ipc->write_buffer.base = malloc(len);
-  memcpy(ipc->write_buffer.base, data, len);
-  ipc->write_buffer.len = len;
-
-  err = uv_async_send(&ipc->write);
-  assert(err == 0);
-
-  return len;
-}
-
-int
-bare_ipc_poll_alloc(bare_ipc_poll_t **result) {
-  bare_ipc_poll_t *poll = malloc(sizeof(bare_ipc_poll_t));
-
-  if (poll == NULL) return -1;
-
-  *result = poll;
-
-  return 0;
-}
-
-int
-bare_ipc_poll_init(bare_ipc_poll_t *poll, bare_ipc_t *ipc) {
-  poll->ipc = ipc;
-  poll->events = 0;
-  poll->cb = NULL;
-  ipc->poll = poll;
-
-  return 0;
-}
-
-void
-bare_ipc_poll_destroy(bare_ipc_poll_t *poll) {
-  int err;
-
-  err = bare_ipc_poll_stop(poll);
-  assert(err == 0);
-}
-
-void *
-bare_ipc_poll_get_data(bare_ipc_poll_t *poll) {
-  return poll->data;
-}
-
-void
-bare_ipc_poll_set_data(bare_ipc_poll_t *poll, void *data) {
-  poll->data = data;
-}
-
-bare_ipc_t *
-bare_ipc_poll_get_ipc(bare_ipc_poll_t *poll) {
-  return poll->ipc;
-}
-
-int
-bare_ipc_poll_start(bare_ipc_poll_t *poll, int events, bare_ipc_poll_cb cb) {
-  if (events == 0) return bare_ipc_poll_stop(poll);
-
-  int err;
-
-  if ((events & bare_ipc_readable) != 0) {
-    if ((poll->events & bare_ipc_readable) == 0) {
-      err = uv_async_send(&poll->ipc->read);
-      assert(err == 0);
-    }
-  }
-
-  poll->events = events;
-  poll->cb = cb;
-
-  return 0;
-}
-
-int
-bare_ipc_poll_stop(bare_ipc_poll_t *poll) {
-  poll->events = 0;
-  poll->cb = NULL;
-
-  return 0;
-}
diff --git a/shared/win32/ipc.h b/shared/win32/ipc.h
deleted file mode 100644
index 0b73f26..0000000
--- a/shared/win32/ipc.h
+++ /dev/null
@@ -1,58 +0,0 @@
-#ifndef BARE_KIT_WIN32_IPC_H
-#define BARE_KIT_WIN32_IPC_H
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-#include <uv.h>
-
-#include "../ipc.h"
-
-struct bare_ipc_s {
-  int incoming;
-  int outgoing;
-
-  struct {
-    char base[BARE_IPC_READ_BUFFER_SIZE];
-    size_t len;
-  } read_buffer;
-
-  char read_tmp_buffer[BARE_IPC_READ_BUFFER_SIZE];
-
-  uv_write_t write_request;
-  uv_buf_t write_buffer;
-
-  struct {
-    uv_pipe_t incoming;
-    uv_pipe_t outgoing;
-  } pipe;
-
-  uv_loop_t loop;
-  uv_async_t close;
-  uv_async_t read;
-  uv_async_t write;
-
-  uv_barrier_t ready;
-  uv_mutex_t reading;
-  uv_mutex_t writing;
-  uv_thread_t thread;
-
-  bare_ipc_poll_t *poll;
-};
-
-struct bare_ipc_poll_s {
-  bare_ipc_t *ipc;
-
-  int events;
-
-  bare_ipc_poll_cb cb;
-
-  void *data;
-};
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif // BARE_KIT_WIN32_IPC_H
diff --git a/shared/worklet.c b/shared/worklet.c
index 606bab7..c9df2f3 100644
--- a/shared/worklet.c
+++ b/shared/worklet.c
@@ -80,6 +80,8 @@ bare_worklet_init(bare_worklet_t *worklet, const bare_worklet_options_t *options
 
   bare_worklet_state_t *state = malloc(sizeof(bare_worklet_state_t));
 
+  if (state == NULL) return -1;
+
   state->finished = false;
   state->suspended = false;
   state->resuming = false;
@@ -89,6 +91,16 @@ bare_worklet_init(bare_worklet_t *worklet, const bare_worklet_options_t *options
 
   memset(&state->callbacks, 0, sizeof(state->callbacks));
 
+  worklet->queue = bare_queue_create();
+
+  if (worklet->queue == NULL) {
+    uv_mutex_destroy(&state->lock);
+
+    free(state);
+
+    return -1;
+  }
+
   worklet->state = state;
 
   memset(&worklet->options, 0, sizeof(worklet->options));
@@ -98,25 +110,31 @@ bare_worklet_init(bare_worklet_t *worklet, const bare_worklet_options_t *options
     worklet->options.assets = options->assets == NULL ? NULL : strdup(options->assets);
   }
 
-  worklet->incoming = -1;
-  worklet->outgoing = -1;
-
   return 0;
 }
 
 void
 bare_worklet_destroy(bare_worklet_t *worklet) {
-  int err;
-
-  if (worklet->thread != 0) uv_sem_post(worklet->finished);
-  else {
+  // The host retires its ipc first: from here the queue is the worklet's and may
+  // be freed at any moment.
+  assert(!worklet->queue->thread_open || worklet->queue->thread_closed);
+
+  if (worklet->thread != 0) {
+    // Stop the worklet if it is still running and wake its loop so the stop
+    // takes effect, then unblock it. The thread is detached and tears itself
+    // down; this function must never block (blocking here causes ANR on
+    // Android). bare_terminate is a no-op if the worklet already exited.
+    bare_terminate(worklet->bare);
+    uv_async_send(&worklet->queue->async);
+    uv_sem_post(worklet->finished);
+  } else {
+    // Nothing ever took ownership of the queue, so it is ours to free.
     uv_mutex_destroy(&worklet->state->lock);
 
     free(worklet->state);
-  }
 
-  if (worklet->incoming >= 0) close(worklet->incoming);
-  if (worklet->outgoing >= 0) close(worklet->outgoing);
+    bare_queue_destroy(worklet->queue);
+  }
 
   free((char *) worklet->options.assets);
 }
@@ -376,6 +394,147 @@ bare_worklet__on_resume(bare_t *bare, void *data) {
   if (state->callbacks.resume) state->callbacks.resume(bare, state->callbacks.resume_data);
 }
 
+// native.read() -> ArrayBuffer (data) | null (EOF) | undefined (would block)
+static js_value_t *
+bare_worklet__on_ipc_read(js_env_t *env, js_callback_info_t *info) {
+  int err;
+
+  bare_queue_port_t *port;
+  err = js_get_callback_info(env, info, NULL, NULL, NULL, (void **) &port);
+  assert(err == 0);
+
+  void *data;
+  size_t len;
+  int status = bare_queue_read(port, &data, &len);
+
+  js_value_t *result;
+
+  if (status == bare_queue_would_block) {
+    err = js_get_undefined(env, &result);
+    assert(err == 0);
+  } else if (len == 0) {
+    err = js_get_null(env, &result);
+    assert(err == 0);
+  } else {
+    void *copy;
+    err = js_create_arraybuffer(env, len, &copy, &result);
+    assert(err == 0);
+
+    memcpy(copy, data, len);
+  }
+
+  return result;
+}
+
+// native.write(buffer) -> bytes written, -1 when the queue is full, or null
+// once the peer has closed
+static js_value_t *
+bare_worklet__on_ipc_write(js_env_t *env, js_callback_info_t *info) {
+  int err;
+
+  size_t argc = 1;
+  js_value_t *argv[1];
+
+  bare_queue_port_t *port;
+  err = js_get_callback_info(env, info, &argc, argv, NULL, (void **) &port);
+  assert(err == 0);
+  assert(argc == 1);
+
+  void *data;
+  size_t len;
+  err = js_get_typedarray_info(env, argv[0], NULL, &data, &len, NULL, NULL);
+  assert(err == 0);
+
+  int n = bare_queue_write(port, data, len);
+
+  js_value_t *result;
+
+  if (n == bare_queue_closed) {
+    err = js_get_null(env, &result);
+  } else {
+    err = js_create_int32(env, n, &result);
+  }
+  assert(err == 0);
+
+  return result;
+}
+
+static js_value_t *
+bare_worklet__on_ipc_close(js_env_t *env, js_callback_info_t *info) {
+  int err;
+
+  bare_queue_port_t *port;
+  err = js_get_callback_info(env, info, NULL, NULL, NULL, (void **) &port);
+  assert(err == 0);
+
+  bare_queue_close(port);
+
+  return NULL;
+}
+
+static js_value_t *
+bare_worklet__on_ipc_ref(js_env_t *env, js_callback_info_t *info) {
+  int err;
+
+  bare_queue_port_t *port;
+  err = js_get_callback_info(env, info, NULL, NULL, NULL, (void **) &port);
+  assert(err == 0);
+
+  uv_ref((uv_handle_t *) &port->queue->async);
+
+  return NULL;
+}
+
+static js_value_t *
+bare_worklet__on_ipc_unref(js_env_t *env, js_callback_info_t *info) {
+  int err;
+
+  bare_queue_port_t *port;
+  err = js_get_callback_info(env, info, NULL, NULL, NULL, (void **) &port);
+  assert(err == 0);
+
+  uv_unref((uv_handle_t *) &port->queue->async);
+
+  return NULL;
+}
+
+// Owned by the worklet thread, so the signal callback never dereferences the
+// host's worklet, which may be destroyed while the queue is still open.
+typedef struct {
+  js_env_t *env;
+  js_ref_t *signal;
+} bare_worklet__ipc_t;
+
+// Runs on the worklet loop when the host has touched the queue; drives the JS
+// duplex to re-check reads and any pending write.
+static void
+bare_worklet__on_ipc_signal(bare_queue_port_t *port) {
+  int err;
+
+  bare_worklet__ipc_t *ipc = (bare_worklet__ipc_t *) port->data;
+
+  assert(ipc->signal);
+
+  js_env_t *env = ipc->env;
+
+  js_handle_scope_t *scope;
+  err = js_open_handle_scope(env, &scope);
+  assert(err == 0);
+
+  js_value_t *fn;
+  err = js_get_reference_value(env, ipc->signal, &fn);
+  assert(err == 0);
+
+  js_value_t *global;
+  err = js_get_global(env, &global);
+  assert(err == 0);
+
+  js_call_function(env, global, fn, 0, NULL, NULL);
+
+  err = js_close_handle_scope(env, scope);
+  assert(err == 0);
+}
+
 static void
 bare_worklet__on_thread(void *opaque) {
   uv_once(&bare_worklet__platform_guard, bare_worklet__on_platform_init);
@@ -442,22 +601,50 @@ bare_worklet__on_thread(void *opaque) {
   err = js_get_named_property(env, module, "exports", &exports);
   assert(err == 0);
 
-  js_value_t *port;
-  err = js_get_named_property(env, exports, "port", &port);
-  assert(err == 0);
+  // Held for the rest of this function: the host may destroy its worklet, and
+  // with it every field read from it, the moment this thread hands back control.
+  bare_queue_t *queue = worklet->queue;
 
-  js_value_t *incoming;
-  err = js_get_named_property(env, port, "incoming", &incoming);
+  // Ours, not the host's: the signal callback runs until we shut the queue down,
+  // which is after the host is free to have destroyed its worklet.
+  bare_worklet__ipc_t *ipc = malloc(sizeof(bare_worklet__ipc_t));
+
+  ipc->env = env;
+  ipc->signal = NULL;
+
+  bare_queue_port_t *ipc_port = bare_queue_open_uv(queue, &loop, bare_worklet__on_ipc_signal);
+
+  ipc_port->data = ipc;
+
+  js_value_t *native;
+  err = js_create_object(env, &native);
   assert(err == 0);
 
-  err = js_get_value_int32(env, incoming, &worklet->incoming);
+#define V(name, cb) \
+  { \
+    js_value_t *fn; \
+    err = js_create_function(env, name, -1, cb, ipc_port, &fn); \
+    assert(err == 0); \
+    err = js_set_named_property(env, native, name, fn); \
+    assert(err == 0); \
+  }
+
+  V("read", bare_worklet__on_ipc_read);
+  V("write", bare_worklet__on_ipc_write);
+  V("close", bare_worklet__on_ipc_close);
+  V("ref", bare_worklet__on_ipc_ref);
+  V("unref", bare_worklet__on_ipc_unref);
+#undef V
+
+  js_value_t *open_ipc;
+  err = js_get_named_property(env, exports, "openIPC", &open_ipc);
   assert(err == 0);
 
-  js_value_t *outgoing;
-  err = js_get_named_property(env, port, "outgoing", &outgoing);
+  js_value_t *on_signal;
+  err = js_call_function(env, exports, open_ipc, 1, &native, &on_signal);
   assert(err == 0);
 
-  err = js_get_value_int32(env, outgoing, &worklet->outgoing);
+  err = js_create_reference(env, on_signal, 1, &ipc->signal);
   assert(err == 0);
 
   js_value_t *fn;
@@ -515,6 +702,11 @@ bare_worklet__on_thread(void *opaque) {
 
   state->finished = true;
 
+  // Signal EOF to the host by closing our end of the queue. This is in-process
+  // (a status flag plus a host wakeup), so it works on every exit path without
+  // pumping the loop or scheduling async work from the exit event.
+  bare_queue_close(ipc_port);
+
   uv_sem_wait(&finished);
 
   uv_sem_destroy(&finished);
@@ -522,6 +714,15 @@ bare_worklet__on_thread(void *opaque) {
   err = js_release_threadsafe_function(push, js_threadsafe_function_release);
   assert(err == 0);
 
+  // Closing the async here stops further signals; teardown's loop run flushes
+  // it. Do this before deleting the reference so no late signal dereferences it.
+  bare_queue_shutdown_uv(queue);
+
+  err = js_delete_reference(env, ipc->signal);
+  assert(err == 0);
+
+  free(ipc);
+
   int exit_code;
   err = bare_teardown(bare, UV_RUN_DEFAULT, &exit_code);
   assert(err == 0);
@@ -529,6 +730,10 @@ bare_worklet__on_thread(void *opaque) {
   err = uv_loop_close(&loop);
   assert(err == 0);
 
+  // Ours since the host started us, and nothing may reach it now: the host was
+  // required to retire its end before signalling us to be destroyed.
+  bare_queue_destroy(queue);
+
   uv_mutex_lock(&state->lock);
 
   err = bare_suspension_end(&state->suspension);
diff --git a/shared/worklet.h b/shared/worklet.h
index 65d0551..cc8fe61 100644
--- a/shared/worklet.h
+++ b/shared/worklet.h
@@ -10,6 +10,7 @@ extern "C" {
 #include <utf.h>
 #include <uv.h>
 
+#include "queue.h"
 #include "suspension.h"
 
 typedef struct bare_worklet_s bare_worklet_t;
@@ -66,8 +67,7 @@ struct bare_worklet_s {
   uv_barrier_t ready;
   uv_sem_t *finished;
 
-  uv_file incoming;
-  uv_file outgoing;
+  bare_queue_t *queue;
 
   js_threadsafe_function_t *push;
 
@@ -97,6 +97,9 @@ bare_worklet_alloc(bare_worklet_t **result);
 int
 bare_worklet_init(bare_worklet_t *worklet, const bare_worklet_options_t *options);
 
+// Signals the worklet to stop and returns immediately; it never blocks, and the
+// thread tears itself down. Any ipc opened on this worklet must already have been
+// destroyed, because from here the queue is the worklet's alone.
 void
 bare_worklet_destroy(bare_worklet_t *worklet);
 
diff --git a/shared/worklet.js b/shared/worklet.js
index 9d08665..23c4140 100644
--- a/shared/worklet.js
+++ b/shared/worklet.js
@@ -9,30 +9,106 @@ const Module = require('bare-module')
 const { startsWithWindowsDriveLetter } = require('bare-module-resolve')
 const SystemLog = require('bare-system-logger')
 const Console = require('bare-console')
-const IPC = require('bare-ipc')
+const { Duplex } = require('bare-stream')
 const unpack = require('bare-unpack')
 
 const { protocol } = module
 
 global.console = new Console(new SystemLog())
 
-const ports = IPC.open()
+const errors = {
+  IPC_CLOSED() {
+    const err = new Error('IPC stream closed')
+    err.code = 'IPC_CLOSED'
+    return err
+  }
+}
+
+// A Duplex over the native in-process queue. The native side provides
+// non-blocking read/write plus ref/unref; a single signal (fed back from the
+// worklet loop) means re-check both reads and a pending write.
+class IPCStream extends Duplex {
+  constructor(native) {
+    super()
+
+    this._native = native
+    this._reading = false
+    this._pendingWrite = null
+  }
+
+  _read() {
+    this._reading = true
+    this._drain()
+  }
+
+  _drain() {
+    if (!this._reading) return
+
+    let chunk
+    while ((chunk = this._native.read()) !== undefined) {
+      if (chunk === null) {
+        this._reading = false
+        this.push(null)
+        return
+      }
+
+      if (this.push(Buffer.from(chunk)) === false) {
+        this._reading = false
+        return
+      }
+    }
+  }
+
+  _write(chunk, encoding, cb) {
+    this._send(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk, encoding), cb)
+  }
 
-exports.port = ports[1]
+  _send(chunk, cb) {
+    const n = this._native.write(chunk)
 
-const ipc = new IPC(ports[0])
+    if (n === null) cb(errors.IPC_CLOSED())
+    else if (n < 0) this._pendingWrite = { chunk, cb }
+    else if (n < chunk.byteLength) this._pendingWrite = { chunk: chunk.subarray(n), cb }
+    else cb(null)
+  }
+
+  _retry() {
+    if (this._pendingWrite === null) return
 
-Bare.IPC = ipc
+    const { chunk, cb } = this._pendingWrite
+    this._pendingWrite = null
 
-Bare.on('suspend', () => ipc.unref())
-  .on('resume', () => ipc.ref())
-  .on('exit', () => ipc.destroy())
+    this._send(chunk, cb)
+  }
+
+  // end() closes our end of the queue so the host read reaches EOF. streamx
+  // only calls _final once outstanding writes have drained, so ordering holds.
+  _final(cb) {
+    this._native.close()
+    cb(null)
+  }
+
+  _onsignal() {
+    this._retry()
+    this._drain()
+  }
+
+  ref() {
+    this._native.ref()
+    return this
+  }
+
+  unref() {
+    this._native.unref()
+    return this
+  }
+}
 
 class BareKit extends EventEmitter {
   constructor() {
     super()
 
-    this.IPC = ipc
+    this.IPC = null
   }
 
   [Symbol.for('bare.inspect')]() {
@@ -51,6 +127,18 @@ Object.defineProperty(global, 'BareKit', {
   configurable: true
 })
 
+// Called by the host runtime with the native queue endpoint before `start`.
+// Returns the signal callback the worklet loop invokes when the queue changes.
+exports.openIPC = function openIPC(native) {
+  const ipc = new IPCStream(native)
+
+  Bare.IPC = exports.BareKit.IPC = ipc
+
+  Bare.on('suspend', () => ipc.unref()).on('resume', () => ipc.ref())
+
+  return () => ipc._onsignal()
+}
+
 exports.push = function push(payload, reply) {
   if (exports.BareKit.emit('push', Buffer.from(payload), replyOnce) === false) {
     replyOnce(null, null)
diff --git a/test/shared/CMakeLists.txt b/test/shared/CMakeLists.txt
index eaa1fc6..c8d4a4b 100644
--- a/test/shared/CMakeLists.txt
+++ b/test/shared/CMakeLists.txt
@@ -4,22 +4,16 @@ list(APPEND tests
   queue-signal-lock
   worklet-assets
   worklet-destroy-before-start
+  worklet-exit-eof
   worklet-ipc
   worklet-ipc-large-write
+  worklet-ipc-write-after-close
+  worklet-jsend-eof
   worklet-suspend-resume
   worklet-suspend-resume-suspend
   worklet-terminate-after-suspend
 )
 
-# The win32 IPC layer does not yet surface end-of-stream, so the EOF tests only
-# run on platforms with the posix read path.
-if(NOT WIN32)
-  list(APPEND tests
-    worklet-exit-eof
-    worklet-jsend-eof
-  )
-endif()
-
 execute_process(
   COMMAND node fixtures/generate.js
   WORKING_DIRECTORY ${CMAKE_CURRENT_LIST_DIR}
diff --git a/test/shared/worklet-ipc-write-after-close.c b/test/shared/worklet-ipc-write-after-close.c
new file mode 100644
index 0000000..a20303f
--- /dev/null
+++ b/test/shared/worklet-ipc-write-after-close.c
@@ -0,0 +1,70 @@
+#include <assert.h>
+#include <stdbool.h>
+#include <string.h>
+#include <uv.h>
+
+#include "../../shared/ipc.h"
+#include "../../shared/worklet.h"
+
+// The host retires its IPC while the worklet still has a write outstanding. The
+// worklet's queue end then reports closed, which must surface as a stream error
+// rather than a write that never completes; the worklet reports which it saw.
+
+static bool replied = false;
+static char reply_buf[64];
+
+static void
+on_push(bare_worklet_push_t *req, const char *error, const uv_buf_t *reply) {
+  assert(error == NULL);
+
+  memcpy(reply_buf, reply->base, reply->len);
+  reply_buf[reply->len] = '\0';
+
+  replied = true;
+}
+
+int
+main() {
+  int e;
+
+  bare_worklet_t worklet;
+  e = bare_worklet_init(&worklet, NULL);
+  assert(e == 0);
+
+  char *code =
+    "let errored = false\n"
+    "Bare.IPC.on('error', () => { errored = true })\n"
+    "setInterval(() => Bare.IPC.write(Buffer.alloc(64)), 5)\n"
+    "BareKit.on('push', (data, reply) => reply(null, errored ? 'errored' : 'wedged'))\n";
+
+  uv_buf_t source = uv_buf_init(code, strlen(code));
+
+  e = bare_worklet_start(&worklet, "app.js", &source, 0, NULL);
+  assert(e == 0);
+
+  bare_ipc_t ipc;
+  e = bare_ipc_init(&ipc, &worklet);
+  assert(e == 0);
+
+  // Never read, so the queue fills and the worklet is left with a pending write.
+  uv_sleep(100);
+
+  bare_ipc_destroy(&ipc);
+
+  // Give the worklet time to retry that write against the closed peer.
+  uv_sleep(300);
+
+  uv_buf_t payload = uv_buf_init("status", 6);
+
+  bare_worklet_push_t req;
+  e = bare_worklet_push(&worklet, &req, &payload, on_push);
+  assert(e == 0);
+
+  for (int i = 0; i < 200 && !replied; i++)
+    uv_sleep(10);
+
+  assert(replied);
+  assert(strcmp(reply_buf, "errored") == 0); // not "wedged": the write completed
+
+  bare_worklet_destroy(&worklet);
+}
diff --git a/test/shared/worklet-jsend-eof.c b/test/shared/worklet-jsend-eof.c
index 26fc3ac..622e475 100644
--- a/test/shared/worklet-jsend-eof.c
+++ b/test/shared/worklet-jsend-eof.c
@@ -3,9 +3,9 @@
 #include "../../shared/ipc.h"
 #include "../../shared/worklet.h"
 
-// Corroborates worklet-exit-no-eof: the host EOF comes from the JS side. When
-// the worklet ends its Bare.IPC (closing the b1 write end), the host read
-// returns 0 (EOF) - the signal the bindings turn into (nil, nil).
+// The host EOF comes from the JS side: when the worklet ends its Bare.IPC, its
+// end of the in-process queue closes and the host read returns 0 (EOF) - the
+// signal the bindings turn into (nil, nil).
 
 int
 main() {
@@ -29,7 +29,7 @@ main() {
   void *data;
   size_t len;
 
-  // Poll for up to ~2s for EOF. Expect read to return 0 once b1 closes.
+  // Poll for up to ~2s for EOF. Expect read to return 0 once the worklet ends.
   bool eof = false;
   for (int i = 0; i < 200; i++) {
     e = bare_ipc_read(&ipc, &data, &len);
diff --git a/test/shared/worklet-suspend-resume.c b/test/shared/worklet-suspend-resume.c
index 6e3ae14..913622e 100644
--- a/test/shared/worklet-suspend-resume.c
+++ b/test/shared/worklet-suspend-resume.c
@@ -31,7 +31,8 @@ main() {
   e = bare_worklet_on_resume(&worklet, on_resume, NULL);
   assert(e == 0);
 
-  char *code = "console.log('Hello world')";
+  // Keep Bare.IPC open so suspend/resume exercise the queue's ref/unref.
+  char *code = "BareKit.IPC.on('data', () => {})";
 
   uv_buf_t source = uv_buf_init(code, strlen(code));
 
diff --git a/win32/ipc.def b/win32/ipc.def
index 6085502..db25a59 100644
--- a/win32/ipc.def
+++ b/win32/ipc.def
@@ -2,8 +2,6 @@ EXPORTS
   bare_ipc_alloc
   bare_ipc_init
   bare_ipc_destroy
-  bare_ipc_get_incoming
-  bare_ipc_get_outgoing
   bare_ipc_read
   bare_ipc_write
   bare_ipc_poll_alloc