diff --git a/src/io/lib.rs b/src/io/lib.rs
index 8a9776c9a637..5e1c65765204 100644
--- a/src/io/lib.rs
+++ b/src/io/lib.rs
@@ -910,15 +910,28 @@ impl IoRequestLoop {
         unsafe { (*LOOP.get()).assume_init_ref() }.tick();
     }
 
-    /// Enqueue `request` for the IO thread to pick up. Safe to call from any
+    /// Hand the owner of `request` over to the IO thread. Callable from any
     /// thread: only touches the lock-free `pending` queue and the
     /// async-signal-safe `waker`. This is the *only* cross-thread entry
     /// point — every other `IoRequestLoop` method is IO-thread-only.
-    pub fn schedule(request: &mut Request) {
+    ///
+    /// The push is the hand-over (the IO thread may be running the callback
+    /// before `wake()` returns), and the callback recovers the owner from this
+    /// very pointer by container-of, hence `*mut` projected from the owner
+    /// rather than `&mut`.
+    ///
+    /// # Safety
+    /// `request` is `&raw mut (*owner).io_request` of a live, unscheduled owner
+    /// the caller holds; the caller touches neither again until the IO thread
+    /// hands the owner back.
+    pub unsafe fn schedule(request: *mut Request) {
         Self::ensure_init();
-        debug_assert!(!request.scheduled);
-        request.scheduled = true;
-        let request = core::ptr::NonNull::from(request);
+        let request = core::ptr::NonNull::new(request).expect("io request of a live owner");
+        // SAFETY: fn contract — ours until the push below.
+        unsafe {
+            debug_assert!(!(*request.as_ptr()).scheduled);
+            (*request.as_ptr()).scheduled = true;
+        }
         // SAFETY: `ONCE` above established happens-before for `load()`'s
         // init of `pending`/`waker`. `get_unchecked` (no owner assert) and a
         // pointer cast of the `repr(transparent)` `MaybeUninit` (not
@@ -926,6 +939,7 @@ impl IoRequestLoop {
         // references formed are the `&pending` / `&waker` autorefs, which may
         // coexist with the IO thread's `&IoRequestLoop` held across `tick()`.
         // `pending.push` (lock-free MPSC) and `waker.wake` both take `&self`.
+        // Nothing here touches `*request` after the push.
         unsafe {
             let loop_p = LOOP.get_unchecked().cast::<IoRequestLoop>();
             (*core::ptr::addr_of!((*loop_p).pending)).push(request);
@@ -933,6 +947,19 @@ impl IoRequestLoop {
         }
     }
 
+    /// IO thread: runs the callback of a request just popped from `pending`,
+    /// passing the pointer through as `schedule` was given it (see [`Action`]
+    /// for how long the result may be used).
+    #[cfg(not(windows))]
+    fn take_request(request: *mut Request) -> Action {
+        // SAFETY: `pending` only holds pointers that went through `schedule`
+        // (live owner, owner provenance); popping one made it this thread's.
+        unsafe {
+            (*request).scheduled = false;
+            ((*request).callback)(request)
+        }
+    }
+
     #[cfg(not(windows))]
     pub(crate) fn tick(&self) {
         // SAFETY: literal is NUL-terminated; len excludes the NUL.
@@ -958,6 +985,18 @@ impl IoRequestLoop {
         }
     }
 
+    /// The `Readable` / `Writable` arm of [`tick_epoll`](Self::tick_epoll).
+    #[cfg(any(target_os = "linux", target_os = "android"))]
+    fn register_epoll(watcher_fd: Fd, file: &FileAction, flag: Flags) {
+        // SAFETY: see `Action` — `poll` is live and ours until `on_error` below.
+        let registered = unsafe {
+            Poll::register_for_epoll(file.poll, flag, file.tag, watcher_fd, true, file.fd)
+        };
+        if let Err(err) = registered {
+            (file.on_error)(file.ctx, &err);
+        }
+    }
+
     #[cfg(any(target_os = "linux", target_os = "android"))]
     pub(crate) fn tick_epoll(&self) {
         loop {
@@ -967,53 +1006,33 @@ impl IoRequestLoop {
                 let watcher_fd = self.pollfd();
 
                 loop {
-                    let request_ptr = pending.next();
-                    if request_ptr.is_null() {
+                    let request = pending.next();
+                    if request.is_null() {
                         break;
                     }
-                    // SAFETY: pop_batch yields live nodes pushed by `schedule()`.
-                    let request = unsafe { &mut *request_ptr };
-                    request.scheduled = false;
-                    match (request.callback)(request) {
-                        Action::Readable(readable) => {
-                            match readable.poll.register_for_epoll(
-                                Flags::PollReadable,
-                                readable.tag,
-                                watcher_fd,
-                                true,
-                                readable.fd,
-                            ) {
-                                Err(err) => {
-                                    (readable.on_error)(readable.ctx, &err);
-                                }
-                                Ok(()) => {}
-                            }
+                    match Self::take_request(request) {
+                        Action::Readable(file) => {
+                            Self::register_epoll(watcher_fd, &file, Flags::PollReadable)
                         }
-                        Action::Writable(writable) => {
-                            match writable.poll.register_for_epoll(
-                                Flags::PollWritable,
-                                writable.tag,
-                                watcher_fd,
-                                true,
-                                writable.fd,
-                            ) {
-                                Err(err) => {
-                                    (writable.on_error)(writable.ctx, &err);
-                                }
-                                Ok(()) => {}
-                            }
+                        Action::Writable(file) => {
+                            Self::register_epoll(watcher_fd, &file, Flags::PollWritable)
                         }
                         Action::Close(close) => {
-                            log!(
-                                "close({}, registered={})",
-                                close.fd,
-                                close.poll.flags.contains(Flags::Registered)
-                            );
-                            // Only remove from the interest list if it was previously registered.
-                            // Otherwise, epoll gets confused.
-                            // This state can happen if polling for readable/writable previously failed.
-                            if close.poll.flags.contains(Flags::WasEverRegistered) {
-                                close.poll.unregister_with_fd(watcher_fd, close.fd);
+                            // SAFETY: see `Action` — `poll` is live and ours until
+                            // `on_done` below; the reborrow ends with the block.
+                            unsafe {
+                                let poll = &mut *close.poll;
+                                log!(
+                                    "close({}, registered={})",
+                                    close.fd,
+                                    poll.flags.contains(Flags::Registered)
+                                );
+                                // Only remove from the interest list if it was previously registered.
+                                // Otherwise, epoll gets confused.
+                                // This state can happen if polling for readable/writable previously failed.
+                                if poll.flags.contains(Flags::WasEverRegistered) {
+                                    poll.unregister_with_fd(watcher_fd, close.fd);
+                                }
                             }
                             (close.on_done)(close.ctx);
                         }
@@ -1107,44 +1126,53 @@ impl IoRequestLoop {
                 }
 
                 loop {
-                    let request_ptr = pending.next();
-                    if request_ptr.is_null() {
+                    let request = pending.next();
+                    if request.is_null() {
                         break;
                     }
-                    // SAFETY: pop_batch yields live nodes pushed by `schedule()`.
-                    let request = unsafe { &mut *request_ptr };
-                    request.scheduled = false;
-                    match (request.callback)(request) {
+                    match Self::take_request(request) {
                         Action::Readable(readable) => {
-                            Poll::apply_kqueue(
-                                ApplyAction::Readable,
-                                readable.tag,
-                                readable.poll,
-                                readable.fd,
-                                add_one(&mut events_list),
-                            );
+                            // SAFETY: see `Action` — `poll` is live and ours
+                            // (nothing in this arm hands the owner on).
+                            unsafe {
+                                Poll::apply_kqueue(
+                                    ApplyAction::Readable,
+                                    readable.tag,
+                                    readable.poll,
+                                    readable.fd,
+                                    add_one(&mut events_list),
+                                );
+                            }
                         }
                         Action::Writable(writable) => {
-                            Poll::apply_kqueue(
-                                ApplyAction::Writable,
-                                writable.tag,
-                                writable.poll,
-                                writable.fd,
-                                add_one(&mut events_list),
-                            );
-                        }
-                        Action::Close(close) => {
-                            if close.poll.flags.contains(Flags::PollReadable)
-                                || close.poll.flags.contains(Flags::PollWritable)
-                            {
+                            // SAFETY: as for `Readable`.
+                            unsafe {
                                 Poll::apply_kqueue(
-                                    ApplyAction::Cancel,
-                                    close.tag,
-                                    close.poll,
-                                    close.fd,
+                                    ApplyAction::Writable,
+                                    writable.tag,
+                                    writable.poll,
+                                    writable.fd,
                                     add_one(&mut events_list),
                                 );
                             }
+                        }
+                        Action::Close(close) => {
+                            // SAFETY: see `Action` — `poll` is live and ours until
+                            // `on_done` below.
+                            unsafe {
+                                let flags = (*close.poll).flags;
+                                if flags.contains(Flags::PollReadable)
+                                    || flags.contains(Flags::PollWritable)
+                                {
+                                    Poll::apply_kqueue(
+                                        ApplyAction::Cancel,
+                                        close.tag,
+                                        close.poll,
+                                        close.fd,
+                                        add_one(&mut events_list),
+                                    );
+                                }
+                            }
                             (close.on_done)(close.ctx);
                         }
                     }
@@ -1190,15 +1218,23 @@ impl IoRequestLoop {
 
 // ─── Request ──────────────────────────────────────────────────────────────────
 
+/// Run by the IO thread with the pointer [`IoRequestLoop::schedule`] was given
+/// (`scheduled` already cleared), so it may recover the owner via
+/// [`IntrusiveIoRequest::from_io_request`].
+///
+/// # Safety
+/// IO thread only, with a pointer `schedule` was given and has not handed back.
+pub type RequestCallback = unsafe fn(*mut Request) -> Action;
+
 pub struct Request {
     pub next: bun_threading::Link<Request>,
-    pub callback: for<'a> fn(&'a mut Request) -> Action<'a>,
+    pub callback: RequestCallback,
     pub scheduled: bool,
 }
 
 impl Request {
     #[inline]
-    pub fn new(callback: for<'a> fn(&'a mut Request) -> Action<'a>) -> Self {
+    pub fn new(callback: RequestCallback) -> Self {
         Self {
             next: bun_threading::Link::new(),
             callback,
@@ -1216,7 +1252,7 @@ impl Request {
     /// followed by a full fence (matches the existing pattern in
     /// `webcore::blob::{read_file,write_file}`).
     #[inline]
-    pub fn store_callback_seq_cst(&mut self, cb: for<'a> fn(&'a mut Request) -> Action<'a>) {
+    pub fn store_callback_seq_cst(&mut self, cb: RequestCallback) {
         // SAFETY: `callback` is a plain pointer-sized field on `self`;
         // volatile write prevents the compiler from reordering or eliding it.
         unsafe { core::ptr::write_volatile(&raw mut self.callback, cb) };
@@ -1232,7 +1268,7 @@ impl Request {
 
 /// A type that embeds an intrusive `io_request: `[`Request`] field. Declares the
 /// byte offset once and provides the canonical container-of recovery used by
-/// every `fn(&mut Request) -> Action` io-loop trampoline.
+/// every [`RequestCallback`] io-loop trampoline.
 ///
 /// Implement via [`intrusive_io_request!`].
 ///
@@ -1250,7 +1286,8 @@ pub unsafe trait IntrusiveIoRequest: Sized {
     /// # Safety
     /// `req` must point to the [`Request`] field at `Self::IO_REQUEST_OFFSET`
     /// inside a live `Self` allocation that was scheduled via that field, and
-    /// the pointer's provenance must cover the whole allocation.
+    /// the pointer's provenance must cover the whole allocation (true of the
+    /// pointer a [`RequestCallback`] receives, per [`IoRequestLoop::schedule`]).
     #[inline(always)]
     unsafe fn from_io_request(req: *mut Request) -> *mut Self {
         // SAFETY: caller upholds the trait safety contract above.
@@ -1261,7 +1298,7 @@ pub unsafe trait IntrusiveIoRequest: Sized {
 /// Implements [`IntrusiveIoRequest`] for a struct that embeds an intrusive
 /// `io_request: `[`Request`] field. Brings
 /// [`IntrusiveIoRequest::from_io_request`] into scope for the type's
-/// `fn(&mut Request) -> Action` trampolines.
+/// [`RequestCallback`] trampolines.
 #[macro_export]
 macro_rules! intrusive_io_request {
     ($ty:ty, $field:ident) => {
@@ -1339,23 +1376,29 @@ pub(crate) type RequestQueue = bun_threading::UnboundedQueue<Request>;
 
 // ─── Action ───────────────────────────────────────────────────────────────────
 
-pub enum Action<'a> {
-    Readable(FileAction<'a>),
-    Writable(FileAction<'a>),
-    Close(CloseAction<'a>),
+/// Returned by a [`RequestCallback`]. `poll` / `ctx` point into the owner,
+/// which is the IO thread's only until `on_error` / `on_done` hands it on.
+/// `poll` is projected from the owner like the request: the kernel hands its
+/// address back and `__bun_io_pollable_on_ready` container-ofs the owner from it.
+pub enum Action {
+    Readable(FileAction),
+    Writable(FileAction),
+    Close(CloseAction),
 }
 
-pub struct FileAction<'a> {
+pub struct FileAction {
     pub fd: Fd,
-    pub poll: &'a mut Poll,
+    /// `&raw mut (*owner).io_poll`.
+    pub poll: *mut Poll,
     pub ctx: *mut (),
     pub tag: PollableTag,
     pub on_error: fn(*mut (), &sys::Error),
 }
 
-pub struct CloseAction<'a> {
+pub struct CloseAction {
     pub fd: Fd,
-    pub poll: &'a mut Poll,
+    /// `&raw mut (*owner).io_poll`.
+    pub poll: *mut Poll,
     pub ctx: *mut (),
     pub tag: PollableTag,
     pub on_done: fn(*mut ()),
@@ -1502,12 +1545,14 @@ enum ApplyAction {
 }
 
 impl Poll {
+    /// # Safety
+    /// `poll` is an [`Action`]'s `poll`, live and the IO thread's.
     #[cfg(any(target_os = "macos", target_os = "freebsd"))]
     #[inline]
-    pub(crate) fn apply_kqueue(
+    pub(crate) unsafe fn apply_kqueue(
         action: ApplyAction,
         tag: PollableTag,
-        poll: &mut Poll,
+        poll: *mut Poll,
         fd: Fd,
         kqueue_event: &mut KEvent,
     ) {
@@ -1522,7 +1567,10 @@ impl Poll {
         );
 
         let one_shot_flag = libc::EV_ONESHOT;
-        let udata: usize = Pollable::init(tag, std::ptr::from_mut::<Poll>(poll)).ptr() as usize;
+        // `udata` comes from the pointer itself (see `Action`), not the reborrow.
+        let udata: usize = Pollable::init(tag, poll).ptr() as usize;
+        // SAFETY: fn contract; the reborrow lasts for this call only.
+        let poll = unsafe { &mut *poll };
         let (filter, flags_): (i16, u16) = match action {
             ApplyAction::Readable => (libc::EVFILT_READ, libc::EV_ADD | one_shot_flag),
             ApplyAction::Writable => (libc::EVFILT_WRITE, libc::EV_ADD | one_shot_flag),
@@ -1675,11 +1723,13 @@ impl Poll {
         }
     }
 
+    /// # Safety
+    /// `poll` is an [`Action`]'s `poll`, live and the IO thread's.
     #[cfg(any(target_os = "linux", target_os = "android"))]
     // `enumset::EnumSetType` cannot be a const generic, so `flag` is a runtime
     // arg. The `match` below preserves the exhaustiveness check.
-    pub(crate) fn register_for_epoll(
-        &mut self,
+    pub(crate) unsafe fn register_for_epoll(
+        poll: *mut Poll,
         flag: Flags,
         tag: PollableTag,
         watcher_fd: Fd,
@@ -1690,11 +1740,16 @@ impl Poll {
 
         debug_assert!(fd != Fd::INVALID);
 
+        // `udata` comes from the pointer itself (see `Action`), not the reborrow.
+        let udata = Pollable::init(tag, poll).ptr();
+        // SAFETY: fn contract; the reborrow lasts for this call only.
+        let this = unsafe { &mut *poll };
+
         if one_shot {
-            self.flags.insert(Flags::OneShot);
+            this.flags.insert(Flags::OneShot);
         }
 
-        let one_shot_flag: u32 = if !self.flags.contains(Flags::OneShot) {
+        let one_shot_flag: u32 = if !this.flags.contains(Flags::OneShot) {
             0
         } else {
             linux::EPOLL_ONESHOT
@@ -1712,11 +1767,11 @@ impl Poll {
 
         let mut event = linux::epoll_event {
             events: flags,
-            u64: Pollable::init(tag, std::ptr::from_mut::<Poll>(self)).ptr(),
+            u64: udata,
         };
 
-        let op: i32 = if self.flags.contains(Flags::WasEverRegistered)
-            || self.flags.contains(Flags::NeedsRearm)
+        let op: i32 = if this.flags.contains(Flags::WasEverRegistered)
+            || this.flags.contains(Flags::NeedsRearm)
         {
             linux::EPOLL_CTL_MOD
         } else {
@@ -1740,10 +1795,10 @@ impl Poll {
         // Only mark if it successfully registered.
         // If it failed to register, we don't want to unregister it later if
         // it never had done so in the first place.
-        self.flags.insert(Flags::Registered);
-        self.flags.insert(Flags::WasEverRegistered);
+        this.flags.insert(Flags::Registered);
+        this.flags.insert(Flags::WasEverRegistered);
 
-        self.flags.insert(match flag {
+        this.flags.insert(match flag {
             Flags::PollReadable => Flags::PollReadable,
             Flags::PollProcess => {
                 if cfg!(any(target_os = "linux", target_os = "android")) {
@@ -1755,7 +1810,7 @@ impl Poll {
             Flags::PollWritable => Flags::PollWritable,
             _ => unreachable!(),
         });
-        self.flags.remove(Flags::NeedsRearm);
+        this.flags.remove(Flags::NeedsRearm);
 
         Ok(())
     }
diff --git a/src/runtime/webcore/Blob.rs b/src/runtime/webcore/Blob.rs
index 006cf841e312..4589e0e5543a 100644
--- a/src/runtime/webcore/Blob.rs
+++ b/src/runtime/webcore/Blob.rs
@@ -7037,17 +7037,24 @@ pub trait FileCloser: Sized {
     fn set_opened_fd(&mut self, fd: Fd);
     fn close_after_io(&self) -> bool;
     fn state(&self) -> &core::sync::atomic::AtomicU8;
-    fn io_request(&mut self) -> Option<&mut bun_io::Request>;
+    /// `&raw mut (*this).io_request`, as `bun_io::IoRequestLoop::schedule`
+    /// requires; `None` if the type never uses the io thread (`ReadFileUV`).
+    ///
+    /// # Safety
+    /// `this` points at a live `Self`.
+    unsafe fn io_request(this: *mut Self) -> Option<*mut bun_io::Request>;
     fn io_poll(&mut self) -> &mut bun_io::Poll;
     fn task(&mut self) -> &mut bun_jsc::WorkPoolTask;
-    fn update(&mut self);
     #[cfg(windows)]
     fn loop_(&self) -> *mut bun_libuv_sys::uv_loop_t;
 
-    /// Intrusive backref: Rust `offset_of!` cannot name
-    /// fields on a trait `Self`, so each concrete impl supplies its own
-    /// container_of recovery (no default body).
-    fn schedule_close(request: &mut bun_io::Request) -> bun_io::Action<'_>;
+    /// The `bun_io::RequestCallback` `do_close` installs. Intrusive backref:
+    /// Rust `offset_of!` cannot name fields on a trait `Self`, so each concrete
+    /// impl supplies its own container_of recovery (no default body).
+    ///
+    /// # Safety
+    /// `bun_io::RequestCallback`'s contract.
+    unsafe fn schedule_close(request: *mut bun_io::Request) -> bun_io::Action;
 
     fn on_io_request_closed(this: &mut Self) {
         this.io_poll()
@@ -7068,40 +7075,44 @@ pub trait FileCloser: Sized {
     /// `container_of` deref locally, so a fn-level qualifier is redundant.
     fn on_close_io_request(task: *mut bun_jsc::WorkPoolTask);
 
-    fn do_close(&mut self, is_allowed_to_close_fd: bool) -> bool {
-        // Check `close_after_io()` before `io_request()` so the immutable
-        // `self` reads finish before
-        // taking the `&mut self` borrow via `io_request()`.
-        if self.close_after_io() {
-            self.state().store(
-                ClosingState::Closing as u8,
-                core::sync::atomic::Ordering::SeqCst,
-            );
-            if let Some(io_request) = self.io_request() {
-                // The io thread reads `callback` after popping from its MPSC
-                // queue; a plain store here is a data race. `bun_io::Request::
-                // store_callback_seq_cst` lowers to a volatile write + SeqCst
-                // fence (Rust has no `AtomicFnPtr`).
-                io_request.store_callback_seq_cst(Self::schedule_close);
-                if !io_request.scheduled {
-                    bun_io::IoRequestLoop::schedule(io_request);
+    /// Returns `true` when `*this` was handed to the io thread instead (it comes
+    /// back through `on_close_io_request`); that hand-over is the last access.
+    ///
+    /// # Safety
+    /// `this` is a live `Self` the caller holds; after a `true` return the
+    /// caller must not touch it again.
+    unsafe fn do_close(this: *mut Self, is_allowed_to_close_fd: bool) -> bool {
+        // SAFETY: fn contract; every reborrow below ends with its statement.
+        unsafe {
+            if (*this).close_after_io() {
+                (*this).state().store(
+                    ClosingState::Closing as u8,
+                    core::sync::atomic::Ordering::SeqCst,
+                );
+                if let Some(io_request) = Self::io_request(this) {
+                    // The io thread reads `callback` after popping from its MPSC
+                    // queue; a plain store here is a data race. `bun_io::Request::
+                    // store_callback_seq_cst` lowers to a volatile write + SeqCst
+                    // fence (Rust has no `AtomicFnPtr`).
+                    (*io_request).store_callback_seq_cst(Self::schedule_close);
+                    if !(*io_request).scheduled {
+                        bun_io::IoRequestLoop::schedule(io_request);
+                    }
+                    return true;
                 }
-                return true;
             }
-        }
 
-        if is_allowed_to_close_fd
-            && self.opened_fd() != Fd::INVALID
-            && self.opened_fd().stdio_tag().is_none()
-        {
-            #[cfg(windows)]
-            bun_io::Closer::close(self.opened_fd(), self.loop_());
-            #[cfg(not(windows))]
-            {
-                use bun_sys::FdExt as _;
-                let _ = self.opened_fd().close_allowing_bad_file_descriptor(None);
+            let fd = (*this).opened_fd();
+            if is_allowed_to_close_fd && fd != Fd::INVALID && fd.stdio_tag().is_none() {
+                #[cfg(windows)]
+                bun_io::Closer::close(fd, (*this).loop_());
+                #[cfg(not(windows))]
+                {
+                    use bun_sys::FdExt as _;
+                    let _ = fd.close_allowing_bad_file_descriptor(None);
+                }
+                (*this).set_opened_fd(Fd::INVALID);
             }
-            self.set_opened_fd(Fd::INVALID);
         }
 
         false
@@ -7110,8 +7121,8 @@ pub trait FileCloser: Sized {
 
 /// Implements [`FileCloser`] for a task struct with the standard field set
 /// (`opened_fd`, `close_after_io`, `state`, `io_request`, `io_poll`, `task`),
-/// an inherent `update()`, and a [`bun_io::Tag`] variant named after the type.
-/// The type must also carry `bun_threading::intrusive_work_task!` and
+/// an inherent `unsafe fn update(this: *mut Self)`, and a [`bun_io::Tag`]
+/// variant named after the type. The type must also carry `bun_threading::intrusive_work_task!` and
 /// `bun_io::intrusive_io_request!`, which provide the parent-pointer recovery
 /// used by the two trampolines.
 macro_rules! impl_file_closer {
@@ -7130,8 +7141,9 @@ macro_rules! impl_file_closer {
             fn state(&self) -> &::core::sync::atomic::AtomicU8 {
                 &self.state
             }
-            fn io_request(&mut self) -> Option<&mut ::bun_io::Request> {
-                Some(&mut self.io_request)
+            unsafe fn io_request(this: *mut Self) -> Option<*mut ::bun_io::Request> {
+                // SAFETY: fn contract; a projection, not a reborrow.
+                Some(unsafe { &raw mut (*this).io_request })
             }
             fn io_poll(&mut self) -> &mut ::bun_io::Poll {
                 &mut self.io_poll
@@ -7139,27 +7151,22 @@ macro_rules! impl_file_closer {
             fn task(&mut self) -> &mut ::bun_jsc::WorkPoolTask {
                 &mut self.task
             }
-            fn update(&mut self) {
-                $T::update(self)
-            }
             #[cfg(windows)]
             fn loop_(&self) -> *mut ::bun_libuv_sys::uv_loop_t {
                 unreachable!()
             }
 
-            fn schedule_close(request: &mut ::bun_io::Request) -> ::bun_io::Action<'_> {
+            unsafe fn schedule_close(request: *mut ::bun_io::Request) -> ::bun_io::Action {
                 use ::bun_io::IntrusiveIoRequest as _;
-                // SAFETY: `request` is `&mut self.io_request` (intrusive); recover parent.
-                let this = unsafe { $T::from_io_request(::core::ptr::from_mut(request)) };
+                // SAFETY: fn contract — `request` is the projection `do_close` scheduled.
+                let this = unsafe { $T::from_io_request(request) };
                 fn on_done(ctx: *mut ()) {
                     // SAFETY: ctx is `self as *mut Self` set below.
                     let this = unsafe { ::bun_ptr::callback_ctx::<$T>(ctx.cast()) };
                     <$T as crate::webcore::blob::FileCloser>::on_io_request_closed(this);
                 }
-                // SAFETY: `request` is `&mut self.io_request` (intrusive), so `this` is the
-                // live parent; the `fd` copy and the `io_poll` field borrow are the only
-                // borrows formed.
-                let (fd, poll) = unsafe { ((*this).opened_fd, &mut (*this).io_poll) };
+                // SAFETY: as above; a field read and a projection, no reference escapes.
+                let (fd, poll) = unsafe { ((*this).opened_fd, &raw mut (*this).io_poll) };
                 ::bun_io::Action::Close(::bun_io::CloseAction {
                     fd,
                     poll,
@@ -7180,10 +7187,11 @@ macro_rules! impl_file_closer {
                 // `&mut self.task` (intrusive) registered in `on_io_request_closed`;
                 // recover parent.
                 let this = unsafe { $T::from_task_ptr(task) };
-                // SAFETY: `this` is the live parent (see above); scoped access.
-                unsafe { (*this).close_after_io = false };
-                // SAFETY: as above; exclusive borrow scoped to the call.
-                $T::update(unsafe { &mut *this });
+                // SAFETY: as above; `update` takes the pointer because it hands it on.
+                unsafe {
+                    (*this).close_after_io = false;
+                    $T::update(this);
+                }
             }
         }
     };
diff --git a/src/runtime/webcore/blob/read_file.rs b/src/runtime/webcore/blob/read_file.rs
index 8bc36a6b5805..49bf51ddc3ad 100644
--- a/src/runtime/webcore/blob/read_file.rs
+++ b/src/runtime/webcore/blob/read_file.rs
@@ -19,7 +19,7 @@ use bun_core;
 use bun_core::String as BunString;
 use bun_io as io;
 #[cfg(not(windows))]
-use bun_io::FileAction;
+use bun_io::{FileAction, IntrusiveIoRequest as _};
 #[cfg(windows)]
 // `bun_jsc::EventLoop` is the *module*; the struct is one level deeper.
 use bun_jsc::event_loop::EventLoop;
@@ -292,6 +292,15 @@ pub struct ReadFile {
 bun_threading::intrusive_work_task!(ReadFile, task);
 bun_io::intrusive_io_request!(ReadFile, io_request);
 
+/// What a pool-side step of a `ReadFile` ends in; see [`ReadFile::proceed`].
+#[cfg(not(windows))]
+#[derive(Clone, Copy)]
+enum Next {
+    ReadLoop,
+    WaitForReadable,
+    Finish,
+}
+
 // The default methods on the FileOpener/FileCloser traits provide the bodies.
 impl FileOpener for ReadFile {
     fn opened_fd(&self) -> Fd {
@@ -330,17 +339,49 @@ impl FileOpener for ReadFile {
 crate::webcore::blob::impl_file_closer!(ReadFile);
 
 impl ReadFile {
-    pub(crate) fn update(&mut self) {
+    /// Pool thread re-entry (`do_read_loop_task`, `on_close_io_request`). This
+    /// and everything below it end by handing `*this` to another thread (the io
+    /// thread, or the JS thread that frees it), so they carry the pointer the
+    /// pool gave us rather than a `&mut self` that would outlive the hand-over.
+    ///
+    /// # Safety
+    /// `this` is the live `ReadFile` this thread holds; not to be touched after.
+    pub(crate) unsafe fn update(this: *mut Self) {
         #[cfg(windows)]
         {
-            return; // why
+            let _ = this; // Reads go through ReadFileUV; never scheduled.
         }
         #[cfg(not(windows))]
         {
-            if self.state.load(Ordering::Relaxed) == ClosingState::Closing as u8 {
-                self.on_finish();
-            } else {
-                self.do_read_loop();
+            // SAFETY: fn contract.
+            let closing =
+                unsafe { (*this).state.load(Ordering::Relaxed) } == ClosingState::Closing as u8;
+            // SAFETY: fn contract, passed through.
+            unsafe {
+                Self::proceed(
+                    this,
+                    if closing {
+                        Next::Finish
+                    } else {
+                        Next::ReadLoop
+                    },
+                )
+            }
+        }
+    }
+
+    /// Performs the step a `&mut self` stage decided on, after that borrow ended.
+    ///
+    /// # Safety
+    /// As [`update`](Self::update).
+    #[cfg(not(windows))]
+    unsafe fn proceed(this: *mut Self, next: Next) {
+        // SAFETY: fn contract, passed through.
+        unsafe {
+            match next {
+                Next::ReadLoop => Self::do_read_loop(this),
+                Next::WaitForReadable => Self::wait_for_readable(this),
+                Next::Finish => Self::on_finish(this),
             }
         }
     }
@@ -433,35 +474,43 @@ impl ReadFile {
         unsafe { (*ctx.cast::<ReadFile>()).on_io_error(err) }
     }
 
+    /// The `io::RequestCallback` installed by `wait_for_readable`.
+    ///
+    /// # Safety
+    /// `io::RequestCallback`'s contract.
     #[cfg(not(windows))]
-    pub(crate) fn on_request_readable(request: &mut io::Request) -> io::Action<'_> {
+    pub(crate) unsafe fn on_request_readable(request: *mut io::Request) -> io::Action {
         bloblog!("ReadFile.onRequestReadable");
-        request.scheduled = false;
-        // SAFETY: request points to ReadFile.io_request (intrusive field); recover parent via offset_of.
-        let this: &mut ReadFile = unsafe {
-            &mut *(bun_core::from_field_ptr!(
-                ReadFile,
-                io_request,
-                std::ptr::from_mut::<io::Request>(request)
-            ))
-        };
+        // SAFETY: fn contract — `request` is the projection `wait_for_readable` scheduled.
+        let this = unsafe { ReadFile::from_io_request(request) };
+        // SAFETY: as above; a field read and a projection, no reference escapes.
+        let (fd, poll) = unsafe { ((*this).opened_fd, &raw mut (*this).io_poll) };
         io::Action::Readable(FileAction {
             on_error: Self::on_io_error_thunk,
-            ctx: std::ptr::from_mut::<ReadFile>(this).cast::<()>(),
-            fd: this.opened_fd,
-            poll: &mut this.io_poll,
+            ctx: this.cast::<()>(),
+            fd,
+            poll,
             tag: ReadFile::IO_TAG,
         })
     }
 
+    /// Hands `*this` to the io thread; it comes back through `on_ready` /
+    /// `on_io_error`.
+    ///
+    /// # Safety
+    /// As [`update`](Self::update).
     #[cfg(not(windows))]
-    pub(crate) fn wait_for_readable(&mut self) {
+    pub(crate) unsafe fn wait_for_readable(this: *mut Self) {
         bloblog!("ReadFile.waitForReadable");
-        self.close_after_io = true;
-        self.io_request
-            .store_callback_seq_cst(Self::on_request_readable);
-        if !self.io_request.scheduled {
-            io::IoRequestLoop::schedule(&mut self.io_request);
+        // SAFETY: fn contract; the schedule is the last access.
+        unsafe {
+            (*this).close_after_io = true;
+            (*this)
+                .io_request
+                .store_callback_seq_cst(Self::on_request_readable);
+            if !(*this).io_request.scheduled {
+                io::IoRequestLoop::schedule(&raw mut (*this).io_request);
+            }
         }
     }
 
@@ -627,20 +676,29 @@ impl ReadFile {
         self.file_store.pathlike.is_path()
     }
 
+    /// Hands `*this` to the io thread (`do_close`) or to the JS thread, which
+    /// frees it (`io_task.finish()`).
+    ///
+    /// # Safety
+    /// As [`update`](Self::update).
     #[cfg(not(windows))]
-    fn on_finish(&mut self) {
-        let close_after_io = self.close_after_io;
-        self.size = self.buffer.len() as SizeType;
+    unsafe fn on_finish(this: *mut Self) {
+        // SAFETY: fn contract; the reborrow ends with the block.
+        let (close_after_io, is_allowed_to_close_fd) = unsafe {
+            (*this).size = (*this).buffer.len() as SizeType;
+            ((*this).close_after_io, (*this).is_allowed_to_close())
+        };
 
-        {
-            if self.do_close(self.is_allowed_to_close()) {
-                bloblog!("ReadFile.onFinish() = deferred");
-                // we have to wait for the close to finish
-                return;
-            }
+        // SAFETY: fn contract, passed through.
+        if unsafe { Self::do_close(this, is_allowed_to_close_fd) } {
+            bloblog!("ReadFile.onFinish() = deferred");
+            // we have to wait for the close to finish
+            return;
         }
         if !close_after_io {
-            if let Some(io_task) = self.io_task.take() {
+            // SAFETY: `do_close` returned `false`, so `*this` is still ours;
+            // `finish` is the hand-over.
+            if let Some(io_task) = unsafe { (*this).io_task.take() } {
                 bloblog!("ReadFile.onFinish() = immediately");
                 io_task.finish();
             }
@@ -700,16 +758,27 @@ impl ReadFile {
         }
     }
 
+    /// `FileOpener::get_fd`'s callback. `get_fd` and `JobContext::run` still
+    /// pass `&mut self` down to here, so this frame's own use of it ends before
+    /// the hand-over.
     #[cfg(not(windows))]
     fn run_async_with_fd(&mut self, fd: Fd) {
+        let next = self.prepare_read(fd);
+        let this: *mut Self = self;
+        // SAFETY: live `ReadFile` from `get_fd`; neither binding is used afterwards.
+        unsafe { Self::proceed(this, next) }
+    }
+
+    /// Stat, buffer sizing and the initial readability check.
+    #[cfg(not(windows))]
+    fn prepare_read(&mut self, fd: Fd) -> Next {
         if self.errno.is_some() {
-            self.on_finish();
-            return;
+            return Next::Finish;
         }
 
         self.resolve_size_and_last_modified(fd);
         if self.errno.is_some() {
-            return self.on_finish();
+            return Next::Finish;
         }
 
         // Special files might report a size of > 0, and be wrong.
@@ -720,8 +789,7 @@ impl ReadFile {
             // default — `then()` reads `self.buffer` directly.
             self.byte_store = ByteStore::default();
 
-            self.on_finish();
-            return;
+            return Next::Finish;
         }
 
         // add an extra 16 bytes to the buffer to avoid having to resize it for trailing extra data
@@ -735,8 +803,7 @@ impl ReadFile {
                         .to_system_error()
                         .into(),
                 );
-                self.on_finish();
-                return;
+                return Next::Finish;
             }
             self.buffer = v;
         }
@@ -753,146 +820,147 @@ impl ReadFile {
         //
         // If we immediately call read(), it will block until stdin is
         // readable.
-        if self.could_block {
-            if bun_core::is_readable(fd) == bun_core::Pollable::NotReady {
-                self.wait_for_readable();
-                return;
-            }
+        if self.could_block && bun_core::is_readable(fd) == bun_core::Pollable::NotReady {
+            return Next::WaitForReadable;
         }
 
-        self.do_read_loop();
+        Next::ReadLoop
     }
 
     fn do_read_loop_task(task: *mut WorkPoolTask) {
-        // SAFETY: only reached via `WorkPoolTask::callback` with `task` =
-        // `&mut self.task` (intrusive) registered in `on_writable`/`init`;
-        // recover parent.
-        let this = unsafe { &mut *ReadFile::from_task_ptr(task) };
+        // SAFETY: `task` is the intrusive field `on_ready` / `on_io_error`
+        // scheduled; the parent is this thread's until `update` hands it on.
+        unsafe { Self::update(ReadFile::from_task_ptr(task)) }
+    }
 
-        this.update();
+    /// # Safety
+    /// As [`update`](Self::update).
+    #[cfg(not(windows))]
+    unsafe fn do_read_loop(this: *mut Self) {
+        // SAFETY: fn contract; the reborrow ends with the call.
+        let next = unsafe { (*this).read_until_blocked() };
+        // SAFETY: fn contract, passed through.
+        unsafe { Self::proceed(this, next) }
     }
 
+    /// Reads until done (`Finish`: EOF, `max_length` or `errno`) or the fd
+    /// would block (`WaitForReadable`).
     #[cfg(not(windows))]
-    fn do_read_loop(&mut self) {
-        #[cfg(not(windows))]
-        {
-            // we hold a 64 KB stack buffer incase the amount of data to
-            // be read is greater than the reported amount
-            //
-            // 64 KB is large, but since this is running in a thread
-            // with it's own stack, it should have sufficient space.
-            // hoisted out of the loop and zero-initialized once — the
-            // one-time 64 KB memset is negligible next to the per-iteration
-            // syscall, and avoids the `MaybeUninit<u8>` → `&mut [u8]` cast (uninit
-            // bytes behind a `&[u8]` is technically UB even when never read).
-            let mut stack_buffer = [0u8; 64 * 1024];
-            // `do_read` never touches `self.buffer`; move it out so the read
-            // target slice (which may point into its spare capacity) can be
-            // held as a safe `&mut [u8]` across the `&mut self` call.
-            let mut buffer = core::mem::take(&mut self.buffer);
-            while self.state.load(Ordering::Relaxed) == ClosingState::Running as u8 {
-                let (use_stack, buf) = Self::remaining_buffer(
-                    &mut buffer,
-                    &mut stack_buffer,
-                    self.max_length,
-                    self.read_off,
-                );
+    fn read_until_blocked(&mut self) -> Next {
+        // we hold a 64 KB stack buffer incase the amount of data to
+        // be read is greater than the reported amount
+        //
+        // 64 KB is large, but since this is running in a thread
+        // with it's own stack, it should have sufficient space.
+        // hoisted out of the loop and zero-initialized once — the
+        // one-time 64 KB memset is negligible next to the per-iteration
+        // syscall, and avoids the `MaybeUninit<u8>` → `&mut [u8]` cast (uninit
+        // bytes behind a `&[u8]` is technically UB even when never read).
+        let mut stack_buffer = [0u8; 64 * 1024];
+        // `do_read` never touches `self.buffer`; move it out so the read
+        // target slice (which may point into its spare capacity) can be
+        // held as a safe `&mut [u8]` across the `&mut self` call.
+        let mut buffer = core::mem::take(&mut self.buffer);
+        while self.state.load(Ordering::Relaxed) == ClosingState::Running as u8 {
+            let (use_stack, buf) = Self::remaining_buffer(
+                &mut buffer,
+                &mut stack_buffer,
+                self.max_length,
+                self.read_off,
+            );
 
-                if !buf.is_empty() && self.errno.is_none() && !self.read_eof {
-                    let mut read_amount: usize = 0;
-                    let mut retry = false;
-                    let continue_reading = self.do_read(buf, &mut read_amount, &mut retry);
-
-                    // We might read into the stack buffer, so we need to copy it into the heap.
-                    if use_stack {
-                        // `do_read` wrote `read_amount` initialized bytes at
-                        // `stack_buffer[..read_amount]`; the stack array is live
-                        // for this iteration.
-                        let read = &stack_buffer[..read_amount];
-                        if buffer.capacity() == 0 {
-                            // We need to allocate a new buffer
-                            // In this case, we want to use `ensureTotalCapacityPrecise` so that it's an exact amount
-                            // We want to avoid over-allocating incase it's a large amount of data sent in a single chunk followed by a 0 byte chunk.
-                            buffer.reserve_exact(read.len());
-                        } else {
-                            buffer.reserve(read.len());
-                        }
-                        buffer.extend_from_slice(read);
+            if !buf.is_empty() && self.errno.is_none() && !self.read_eof {
+                let mut read_amount: usize = 0;
+                let mut retry = false;
+                let continue_reading = self.do_read(buf, &mut read_amount, &mut retry);
+
+                // We might read into the stack buffer, so we need to copy it into the heap.
+                if use_stack {
+                    // `do_read` wrote `read_amount` initialized bytes at
+                    // `stack_buffer[..read_amount]`; the stack array is live
+                    // for this iteration.
+                    let read = &stack_buffer[..read_amount];
+                    if buffer.capacity() == 0 {
+                        // We need to allocate a new buffer
+                        // In this case, we want to use `ensureTotalCapacityPrecise` so that it's an exact amount
+                        // We want to avoid over-allocating incase it's a large amount of data sent in a single chunk followed by a 0 byte chunk.
+                        buffer.reserve_exact(read.len());
                     } else {
-                        // record the amount of data read
-                        // SAFETY: read() wrote `read_amount` initialized bytes into spare capacity.
-                        unsafe { bun_core::vec::commit_spare(&mut buffer, read_amount) };
-                    }
-                    // - If they DID set a max length, we should stop
-                    //   reading after that.
-                    //
-                    // - If they DID NOT set a max_length, then it will
-                    //   be Blob.max_size which is an impossibly large
-                    //   amount to read.
-                    if !self.read_eof && buffer.len() >= self.max_length as usize {
-                        break;
+                        buffer.reserve(read.len());
                     }
+                    buffer.extend_from_slice(read);
+                } else {
+                    // record the amount of data read
+                    // SAFETY: read() wrote `read_amount` initialized bytes into spare capacity.
+                    unsafe { bun_core::vec::commit_spare(&mut buffer, read_amount) };
+                }
+                // - If they DID set a max length, we should stop
+                //   reading after that.
+                //
+                // - If they DID NOT set a max_length, then it will
+                //   be Blob.max_size which is an impossibly large
+                //   amount to read.
+                if !self.read_eof && buffer.len() >= self.max_length as usize {
+                    break;
+                }
 
-                    if !continue_reading {
-                        // Stop reading, we errored
-                        break;
-                    }
+                if !continue_reading {
+                    // Stop reading, we errored
+                    break;
+                }
 
-                    // If it's not a regular file, it might be something
-                    // which would block on the next read. So we should
-                    // avoid immediately reading again until the next time
-                    // we're scheduled to read.
-                    //
-                    // An example of where this happens is stdin.
-                    //
-                    //    await Bun.stdin.text();
-                    //
-                    // If we immediately call read(), it will block until stdin is
-                    // readable.
-                    if retry
-                        || (self.could_block
-                        // If we received EOF, we can skip the poll() system
-                        // call. We already know it's done.
-                        && !self.read_eof)
+                // If it's not a regular file, it might be something
+                // which would block on the next read. So we should
+                // avoid immediately reading again until the next time
+                // we're scheduled to read.
+                //
+                // An example of where this happens is stdin.
+                //
+                //    await Bun.stdin.text();
+                //
+                // If we immediately call read(), it will block until stdin is
+                // readable.
+                if retry
+                    || (self.could_block
+                    // If we received EOF, we can skip the poll() system
+                    // call. We already know it's done.
+                    && !self.read_eof)
+                {
+                    if self.could_block
+                    // If we received EOF, we can skip the poll() system
+                    // call. We already know it's done.
+                    && !self.read_eof
                     {
-                        if self.could_block
-                        // If we received EOF, we can skip the poll() system
-                        // call. We already know it's done.
-                        && !self.read_eof
-                        {
-                            match bun_core::is_readable(self.opened_fd) {
-                                bun_core::Pollable::NotReady => {}
-                                bun_core::Pollable::Ready | bun_core::Pollable::Hup => continue,
-                            }
+                        match bun_core::is_readable(self.opened_fd) {
+                            bun_core::Pollable::NotReady => {}
+                            bun_core::Pollable::Ready | bun_core::Pollable::Hup => continue,
                         }
-                        self.read_eof = false;
-                        self.buffer = buffer;
-                        self.wait_for_readable();
-
-                        return;
                     }
-
-                    // There can be more to read
-                    continue;
+                    self.read_eof = false;
+                    self.buffer = buffer;
+                    return Next::WaitForReadable;
                 }
 
-                // -- We are done reading.
-                break;
+                // There can be more to read
+                continue;
             }
-            self.buffer = buffer;
 
-            if self.system_error.is_some() {
-                self.buffer = Vec::new(); // clearAndFree
-            }
+            // -- We are done reading.
+            break;
+        }
+        self.buffer = buffer;
 
-            // If we over-allocated by a lot, we should shrink the buffer to conserve memory.
-            if self.buffer.len() + 16_000 < self.buffer.capacity() {
-                self.buffer.shrink_to_fit();
-            }
-            // `Bytes` is owning, and `then()` delivers `self.buffer` directly,
-            // so do not also stash it in `byte_store` — that would double-free.
-            self.on_finish();
+        if self.system_error.is_some() {
+            self.buffer = Vec::new(); // clearAndFree
         }
+
+        // If we over-allocated by a lot, we should shrink the buffer to conserve memory.
+        if self.buffer.len() + 16_000 < self.buffer.capacity() {
+            self.buffer.shrink_to_fit();
+        }
+        // `Bytes` is owning, and `then()` delivers `self.buffer` directly,
+        // so do not also stash it in `byte_store` — that would double-free.
+        Next::Finish
     }
 }
 
@@ -981,7 +1049,7 @@ impl<'a> FileCloser for ReadFileUV<'a> {
     fn state(&self) -> &AtomicU8 {
         unreachable!("@hasField(ReadFileUV, \"io_request\") == false")
     }
-    fn io_request(&mut self) -> Option<&mut bun_io::Request> {
+    unsafe fn io_request(_: *mut Self) -> Option<*mut bun_io::Request> {
         None
     }
     fn io_poll(&mut self) -> &mut bun_io::Poll {
@@ -990,10 +1058,7 @@ impl<'a> FileCloser for ReadFileUV<'a> {
     fn task(&mut self) -> &mut bun_jsc::WorkPoolTask {
         unreachable!("@hasField(ReadFileUV, \"io_request\") == false")
     }
-    fn update(&mut self) {
-        unreachable!("@hasField(ReadFileUV, \"io_request\") == false")
-    }
-    fn schedule_close(_: &mut bun_io::Request) -> bun_io::Action<'_> {
+    unsafe fn schedule_close(_: *mut bun_io::Request) -> bun_io::Action {
         unreachable!("@hasField(ReadFileUV, \"io_request\") == false")
     }
     fn on_close_io_request(_: *mut bun_jsc::WorkPoolTask) {
@@ -1118,7 +1183,9 @@ impl<'a> ReadFileUV<'a> {
         self.total_size = self.total_size.max(self.size);
 
         if needs_close {
-            if self.do_close(self.is_allowed_to_close()) {
+            let is_allowed_to_close_fd = self.is_allowed_to_close();
+            // SAFETY: `self` is live; without an `io_request` this only closes the fd.
+            if unsafe { Self::do_close(self, is_allowed_to_close_fd) } {
                 // we have to wait for the close to finish
                 return;
             }
diff --git a/src/runtime/webcore/blob/write_file.rs b/src/runtime/webcore/blob/write_file.rs
index 293d064020a5..4b1bd20b0051 100644
--- a/src/runtime/webcore/blob/write_file.rs
+++ b/src/runtime/webcore/blob/write_file.rs
@@ -98,6 +98,25 @@ pub struct WriteFile {
 bun_threading::intrusive_work_task!(WriteFile, task);
 bun_io::intrusive_io_request!(WriteFile, io_request);
 
+/// What a pool-side step of a `WriteFile` ends in; see [`WriteFile::proceed`].
+#[cfg(not(windows))]
+#[derive(Clone, Copy)]
+enum Next {
+    WriteLoop,
+    WaitForWritable,
+    Finish,
+}
+
+/// Outcome of one `write(2)` in [`WriteFile::do_write`].
+#[cfg(not(windows))]
+enum WriteStep {
+    Wrote(usize),
+    /// `EAGAIN` on a pollable fd; the caller hands the file to the io thread.
+    WouldBlock,
+    /// `errno` / `system_error` are set.
+    Failed,
+}
+
 // ──────────────────────────────────────────────────────────────────────────
 // FileOpener / FileCloser
 // ──────────────────────────────────────────────────────────────────────────
@@ -201,15 +220,17 @@ impl WriteFile {
         WorkPool::schedule(&raw mut this.task);
     }
 
+    /// The `io::RequestCallback` installed by `wait_for_writable`.
+    ///
+    /// # Safety
+    /// `io::RequestCallback`'s contract.
     #[cfg(not(windows))]
-    pub(crate) fn on_request_writable(request: &mut io::Request) -> io::Action<'_> {
+    pub(crate) unsafe fn on_request_writable(request: *mut io::Request) -> io::Action {
         bun_output::scoped_log!(WriteFile, "WriteFile.onRequestWritable()");
-        request.scheduled = false;
-        // SAFETY: `request` points to WriteFile.io_request (intrusive); recover parent.
-        let this = unsafe { WriteFile::from_io_request(std::ptr::from_mut(request)) };
-        // SAFETY: `request` points to WriteFile.io_request (intrusive), so `this` is the
-        // live parent; `fd` copy and the `io_poll` field borrow are the only borrows formed.
-        let (fd, poll) = unsafe { ((*this).opened_fd, &mut (*this).io_poll) };
+        // SAFETY: fn contract — `request` is the projection `wait_for_writable` scheduled.
+        let this = unsafe { WriteFile::from_io_request(request) };
+        // SAFETY: as above; a field read and a projection, no reference escapes.
+        let (fd, poll) = unsafe { ((*this).opened_fd, &raw mut (*this).io_poll) };
         io::Action::Writable(io::FileAction {
             on_error: Self::on_io_error,
             ctx: this.cast::<()>(),
@@ -219,13 +240,22 @@ impl WriteFile {
         })
     }
 
+    /// Hands `*this` to the io thread; it comes back through `on_ready` /
+    /// `on_io_error`.
+    ///
+    /// # Safety
+    /// As [`update`](Self::update).
     #[cfg(not(windows))]
-    pub(crate) fn wait_for_writable(&mut self) {
-        self.close_after_io = true;
-        self.io_request
-            .store_callback_seq_cst(Self::on_request_writable);
-        if !self.io_request.scheduled {
-            io::IoRequestLoop::schedule(&mut self.io_request);
+    pub(crate) unsafe fn wait_for_writable(this: *mut Self) {
+        // SAFETY: fn contract; the schedule is the last access.
+        unsafe {
+            (*this).close_after_io = true;
+            (*this)
+                .io_request
+                .store_callback_seq_cst(Self::on_request_writable);
+            if !(*this).io_request.scheduled {
+                io::IoRequestLoop::schedule(&raw mut (*this).io_request);
+            }
         }
     }
 
@@ -286,8 +316,9 @@ impl WriteFile {
 
     // reshaped for borrowck — take (off, len) here and re-derive the slice
     // internally so callers don't hold a borrow of self across the &mut self call.
+    // Waiting is the caller's last step (a hand-over), hence `WouldBlock`.
     #[cfg(not(windows))]
-    pub(crate) fn do_write(&mut self, off: usize, len: usize, wrote: &mut usize) -> bool {
+    fn do_write(&mut self, off: usize, len: usize) -> WriteStep {
         let fd = self.opened_fd;
         debug_assert!(fd != Fd::INVALID);
 
@@ -302,8 +333,8 @@ impl WriteFile {
         loop {
             match &result {
                 bun_sys::Result::Ok(res) => {
-                    *wrote = *res;
                     self.total_written += *res;
+                    return WriteStep::Wrote(*res);
                 }
                 bun_sys::Result::Err(err) => {
                     if err.get_errno() == io::RETRY {
@@ -312,19 +343,15 @@ impl WriteFile {
                             // this is fine on kqueue, but not on epoll.
                             continue;
                         }
-                        self.wait_for_writable();
-                        return false;
+                        return WriteStep::WouldBlock;
                     } else {
                         self.errno = Some(bun_errno::from_errno(err.errno as i32).into());
                         self.system_error = Some(err.to_system_error().into());
-                        return false;
+                        return WriteStep::Failed;
                     }
                 }
             }
-            break;
         }
-
-        true
     }
 
     pub(crate) fn then(mut this: WriteFile, _global: &JSGlobalObject) -> Result<(), JsTerminated> {
@@ -385,26 +412,47 @@ impl WriteFile {
             .is_path()
     }
 
+    /// Hands `*this` to the io thread (`do_close`) or to the JS thread, which
+    /// frees it (`io_task.finish()`).
+    ///
+    /// # Safety
+    /// As [`update`](Self::update).
     #[cfg(not(windows))]
-    fn on_finish(&mut self) {
+    unsafe fn on_finish(this: *mut Self) {
         bun_output::scoped_log!(WriteFile, "WriteFile.onFinish()");
 
-        let close_after_io = self.close_after_io;
-        if self.do_close(self.is_allowed_to_close()) {
+        // SAFETY: fn contract; the reborrow ends with the statement.
+        let (close_after_io, is_allowed_to_close_fd) =
+            unsafe { ((*this).close_after_io, (*this).is_allowed_to_close()) };
+        // SAFETY: fn contract, passed through.
+        if unsafe { Self::do_close(this, is_allowed_to_close_fd) } {
             return;
         }
         if !close_after_io {
-            if let Some(io_task) = self.io_task.take() {
+            // SAFETY: `do_close` returned `false`, so `*this` is still ours;
+            // `finish` is the hand-over.
+            if let Some(io_task) = unsafe { (*this).io_task.take() } {
                 io_task.finish();
             }
         }
     }
 
+    /// `FileOpener::get_fd`'s callback. `get_fd` and `JobContext::run` still
+    /// pass `&mut self` down to here, so this frame's own use of it ends before
+    /// the hand-over.
     #[cfg(not(windows))]
-    fn run_with_fd(&mut self, fd_: Fd) {
+    fn run_with_fd(&mut self, fd: Fd) {
+        let next = self.prepare_write(fd);
+        let this: *mut Self = self;
+        // SAFETY: live `WriteFile` from `get_fd`; neither binding is used afterwards.
+        unsafe { Self::proceed(this, next) }
+    }
+
+    /// Blocking-ness, preallocation and the initial writability check.
+    #[cfg(not(windows))]
+    fn prepare_write(&mut self, fd_: Fd) -> Next {
         if fd_ == Fd::INVALID || self.errno.is_some() {
-            self.on_finish();
-            return;
+            return Next::Finish;
         }
 
         let fd = self.opened_fd;
@@ -448,8 +496,7 @@ impl WriteFile {
         // }
 
         if self.could_block && bun_core::is_writable(fd) == bun_core::Pollable::NotReady {
-            self.wait_for_writable();
-            return;
+            return Next::WaitForWritable;
         }
 
         #[cfg(any(target_os = "linux", target_os = "android"))]
@@ -469,38 +516,70 @@ impl WriteFile {
             }
         }
 
-        self.do_write_loop();
+        Next::WriteLoop
     }
 
     fn do_write_loop_task(task: *mut WorkPoolTask) {
-        // SAFETY: only reached via `WorkPoolTask::callback` with `task` = `&mut self.task`
-        // (intrusive) registered in `on_writable`/`init`; recover parent.
-        let this = unsafe { WriteFile::from_task_ptr(task) };
-        // On macOS, we use one-shot mode, so we don't need to unregister.
-        #[cfg(target_os = "macos")]
-        {
-            // SAFETY: `this` is the live parent (see above); scoped access.
-            unsafe { (*this).close_after_io = false };
+        // SAFETY: `task` is the intrusive field `on_ready` / `on_io_error`
+        // scheduled; the parent is this thread's until `update` hands it on.
+        unsafe {
+            let this = WriteFile::from_task_ptr(task);
+            // On macOS, we use one-shot mode, so we don't need to unregister.
+            #[cfg(target_os = "macos")]
+            {
+                (*this).close_after_io = false;
+            }
+            Self::update(this);
         }
-        // SAFETY: `this` is the live parent (see above); exclusive borrow scoped to the call.
-        unsafe { (*this).do_write_loop() };
-    }
-
-    pub(crate) fn update(&mut self) {
-        self.do_write_loop();
     }
 
-    fn do_write_loop(&mut self) {
+    /// Pool thread re-entry (`do_write_loop_task`, `on_close_io_request`);
+    /// pointer-shaped for the reason given on `ReadFile::update`.
+    ///
+    /// # Safety
+    /// `this` is the live `WriteFile` this thread holds; not to be touched after.
+    pub(crate) unsafe fn update(this: *mut Self) {
         #[cfg(windows)]
         {
-            return; // why
+            let _ = this; // Writes go through WriteFileWindows; never scheduled.
         }
         #[cfg(not(windows))]
-        self.do_write_loop_posix();
+        {
+            // SAFETY: fn contract, passed through.
+            unsafe { Self::do_write_loop(this) }
+        }
     }
 
+    /// Performs the step a `&mut self` stage decided on, after that borrow ended.
+    ///
+    /// # Safety
+    /// As [`update`](Self::update).
     #[cfg(not(windows))]
-    fn do_write_loop_posix(&mut self) {
+    unsafe fn proceed(this: *mut Self, next: Next) {
+        // SAFETY: fn contract, passed through.
+        unsafe {
+            match next {
+                Next::WriteLoop => Self::do_write_loop(this),
+                Next::WaitForWritable => Self::wait_for_writable(this),
+                Next::Finish => Self::on_finish(this),
+            }
+        }
+    }
+
+    /// # Safety
+    /// As [`update`](Self::update).
+    #[cfg(not(windows))]
+    unsafe fn do_write_loop(this: *mut Self) {
+        // SAFETY: fn contract; the reborrow ends with the call.
+        let next = unsafe { (*this).write_until_blocked() };
+        // SAFETY: fn contract, passed through.
+        unsafe { Self::proceed(this, next) }
+    }
+
+    /// Writes until done (`Finish`: all written, a zero-length write, closing,
+    /// or `errno`) or the fd would block (`WaitForWritable`).
+    #[cfg(not(windows))]
+    fn write_until_blocked(&mut self) -> Next {
         while self.state.load(Ordering::Relaxed) == ClosingState::Running as u8 {
             let remain_full = self.bytes_blob.shared_view();
             // reshaped for borrowck — capture len/offset before mut borrow
@@ -508,31 +587,22 @@ impl WriteFile {
             let remain_len = remain_full.len() - off;
 
             if remain_len > 0 && self.errno.is_none() {
-                let mut wrote: usize = 0;
-                let continue_writing = self.do_write(off, remain_len, &mut wrote);
-                if !continue_writing {
-                    // Stop writing, we errored
-                    if self.errno.is_some() {
-                        self.on_finish();
-                        return;
-                    }
-
-                    // Stop writing, we need to wait for it to become writable.
-                    return;
-                }
+                let wrote = match self.do_write(off, remain_len) {
+                    WriteStep::Wrote(wrote) => wrote,
+                    WriteStep::WouldBlock => return Next::WaitForWritable,
+                    WriteStep::Failed => return Next::Finish,
+                };
 
                 // Do not immediately attempt to write again if it's not a regular file.
                 if self.could_block
                     && bun_core::is_writable(self.opened_fd) == bun_core::Pollable::NotReady
                 {
-                    self.wait_for_writable();
-                    return;
+                    return Next::WaitForWritable;
                 }
 
                 if wrote == 0 {
                     // we are done, we received EOF
-                    self.on_finish();
-                    return;
+                    return Next::Finish;
                 }
 
                 continue;
@@ -541,7 +611,7 @@ impl WriteFile {
             break;
         }
 
-        self.on_finish();
+        Next::Finish
     }
 }
 
diff --git a/test/harness.ts b/test/harness.ts
index a6c07623ec17..67c941b60f9b 100644
--- a/test/harness.ts
+++ b/test/harness.ts
@@ -1832,6 +1832,20 @@ export function waitForFileToExist(path: string, interval_ms: number) {
   }
 }
 
+/**
+ * Resolves with the file's contents once they contain `needle` at least
+ * `occurrences` times. Polls without a deadline of its own (the test timeout is
+ * the deadline); race it against a failure signal, such as the exit of the
+ * process expected to write the file.
+ */
+export async function waitForFileToContain(path: string, needle: string, occurrences = 1): Promise<string> {
+  while (true) {
+    const text = fs.existsSync(path) ? fs.readFileSync(path, "utf8") : "";
+    if (text.split(needle).length - 1 >= occurrences) return text;
+    await Bun.sleep(5);
+  }
+}
+
 export function libcPathForDlopen() {
   switch (process.platform) {
     case "linux":
diff --git a/test/internal/source-lints/self-receiver-io-schedule.test.ts b/test/internal/source-lints/self-receiver-io-schedule.test.ts
new file mode 100644
index 000000000000..8a7deba2113a
--- /dev/null
+++ b/test/internal/source-lints/self-receiver-io-schedule.test.ts
@@ -0,0 +1,185 @@
+import { file } from "bun";
+import { expect, test } from "bun:test";
+import { realpathSync } from "fs";
+import path from "path";
+import { globAllSources } from "../../../scripts/glob-sources.ts";
+
+// `bun_io::IoRequestLoop::schedule` is the hand-over of a ReadFile/WriteFile
+// to the io thread. Two ways of spelling its argument are banned:
+//
+//   IoRequestLoop::schedule(&mut self.io_request)        // from `self`
+//   IoRequestLoop::schedule(&raw mut self.io_request)
+//   let r = self.io_request(); ... IoRequestLoop::schedule(r)
+//   IoRequestLoop::schedule(&mut (*this).io_request)     // a `&mut` reborrow
+//
+// The argument has to be the request *projected from the owner pointer the
+// caller holds* (`&raw mut (*this).io_request`, or a `*mut` returned by an
+// accessor that does that), for two reasons:
+//
+//   - The io thread hands that exact pointer back to the request's callback
+//     (`on_request_readable` / `on_request_writable` / `schedule_close`),
+//     which recovers the owner from it by container-of. A `&mut` reborrow of
+//     the field (which is what a `&mut Request` argument or `&mut x.io_request`
+//     coerced to `*mut` is) only has provenance over the field's bytes, so the
+//     container-of walk leaves its bounds; `bun_core::container_of`'s
+//     contract says as much ("a `&mut field` reborrow does not suffice").
+//   - The push is the moment the object changes threads: the io thread (and,
+//     through on_ready / on_error / on_done, a pool thread) may be writing to
+//     the request and its owner before `schedule` returns. A `&mut self`
+//     method publishing its own receiver therefore has a reference argument
+//     that stays protected, on this thread's stack, while another thread
+//     writes through the object, which both aliasing models (Tree Borrows is
+//     what `bun run rust:miri` uses) reject regardless of whether `self` is
+//     touched again. `schedule` takes `*mut Request` so that it does not do
+//     this itself; a caller spelling the argument from `self` reintroduces it
+//     one frame up.
+//
+// Before `schedule` took a pointer, all three callers had the shape:
+// `ReadFile::wait_for_readable` and `WriteFile::wait_for_writable`
+// (`schedule(&mut self.io_request)`) and `FileCloser::do_close`
+// (`schedule(io_request)` with `io_request` bound from `self.io_request()`).
+// The converted versions (`unsafe fn wait_for_readable(this: *mut Self)`,
+// `unsafe fn do_close(this: *mut Self, ..)` in src/runtime/webcore/Blob.rs)
+// are the templates; `do_write_loop_task` / `on_close_io_request` show how the
+// pointer reaches them without a `&mut` being formed on the way.
+//
+// Scope: calls to `..IoRequestLoop::schedule(` whose argument is spelled from
+// `self` (directly, or via a local bound from a `self.` expression earlier in
+// the same function), or is any `&mut` expression. An argument derived from a
+// reference other than `self` (`fn f(r: &mut ReadFile)` ... `&raw mut
+// r.io_request`) is the same bug but outside this lint; convert it on sight.
+//
+// Siblings: self-receiver-reclaim.test.ts (freeing the receiver),
+// fn-long-mut-reborrow.test.ts, frozen-nonnull-reborrow.test.ts.
+
+const root = path.resolve(import.meta.dir, "..", "..", "..");
+const rustSources = globAllSources().rust.filter(p => p.endsWith(".rs"));
+
+// Only scan files tracked in HEAD (a `git stash` round-trip can leave stray
+// `.rs` files in the working tree; CI runs on a clean checkout). Same guard as
+// dead-code-escapes.test.ts.
+const tracked: Set<string> | null = (() => {
+  const r = Bun.spawnSync({
+    cmd: ["git", "-C", root, "ls-tree", "-r", "--name-only", "-z", "HEAD"],
+    stdout: "pipe",
+    stderr: "ignore",
+  });
+  if (!r.success) return null;
+  return new Set(r.stdout.toString().split("\0").filter(Boolean));
+})();
+
+// `IoRequestLoop::schedule(`, however the path in front is spelled, with the
+// argument list open. `\s*` after the paren so a rustfmt-wrapped argument
+// still matches.
+const SCHEDULE = String.raw`\bIoRequestLoop::schedule\(\s*`;
+
+// Any `&mut ...` argument: a reborrow of the field, coerced to `*mut` at the
+// call. (`&raw mut` has `raw` in between and is not matched here; it is only
+// banned when its base is `self`, below.)
+const MUT_REBORROW = new RegExp(SCHEDULE + String.raw`&\s*mut\b`, "g");
+
+// An argument spelled from `self`: `self.` as the base of the argument,
+// optionally behind `&mut` / `&raw mut` / one wrapping call such as
+// `ptr::from_mut(`.
+const FROM_SELF = new RegExp(SCHEDULE + String.raw`(?:[\w:]+\(\s*)?(?:&\s*(?:raw\s+)?(?:mut|const)\s+)?self\s*\.`, "g");
+
+// A local bound from a `self.` expression: `let r = &mut self.io_request;`,
+// `let r = self.io_request();`, `if let Some(r) = self.io_request() {`. The
+// binding is then looked for as the schedule argument further down the same
+// function, which ends at the next `fn` item.
+const SELF_BINDING = new RegExp(
+  String.raw`let\s+(?:Some\(\s*)?(?:mut\s+)?(\w+)\s*\)?\s*(?::[^=;{]*)?=\s*(?:&\s*(?:raw\s+)?(?:mut|const)\s+)?self\s*\.`,
+  "g",
+);
+const FN_ITEM = /^[ \t]*(?:pub(?:\([^)]*\))?\s+)?(?:(?:const|async|unsafe|extern\s+"[^"]*")\s+)*fn\s/m;
+
+function scheduleOf(name: string): RegExp {
+  return new RegExp(SCHEDULE + name + String.raw`\s*,?\s*\)`);
+}
+
+/** Byte offsets (into `stripped`) of every banned schedule call in one file. */
+function findBanned(stripped: string): number[] {
+  const hits = new Set<number>();
+  for (const m of stripped.matchAll(MUT_REBORROW)) hits.add(m.index);
+  for (const m of stripped.matchAll(FROM_SELF)) hits.add(m.index);
+  for (const binding of stripped.matchAll(SELF_BINDING)) {
+    const start = binding.index + binding[0].length;
+    const rest = stripped.slice(start);
+    const fnEnd = rest.search(FN_ITEM);
+    const body = fnEnd === -1 ? rest : rest.slice(0, fnEnd);
+    const call = body.search(scheduleOf(binding[1]));
+    if (call !== -1) hits.add(start + call);
+  }
+  return [...hits].sort((a, b) => a - b);
+}
+
+function lineOf(text: string, offset: number): number {
+  return text.slice(0, offset).split("\n").length;
+}
+
+const offenders: string[] = [];
+let scanned = 0;
+for (const abs of rustSources) {
+  const source = path.relative(root, abs).replaceAll(path.sep, "/");
+  // `src/cli` is a symlink into `src/runtime/cli`; count each file once under
+  // its canonical path.
+  if (path.relative(root, realpathSync(abs)).replaceAll(path.sep, "/") !== source) continue;
+  if (tracked !== null && !tracked.has(source)) continue;
+  scanned++;
+  const content = await file(abs).text();
+  // Strip full-line comments so prose mentions (including the in-tree comments
+  // describing this hazard) don't count. `[ \t]*`, not `\s*`: `\s` crosses
+  // newlines and would swallow blank lines, shifting the reported line numbers.
+  const stripped = content.replace(/^[ \t]*\/\/.*$/gm, "");
+  for (const offset of findBanned(stripped)) {
+    offenders.push(`${source}:${lineOf(stripped, offset)}`);
+  }
+}
+
+test("scans a non-empty set of tracked Rust sources", () => {
+  // Guards against the tracked/realpath filters above over-firing and leaving
+  // nothing to scan, which would make the ban below pass vacuously.
+  expect(scanned).toBeGreaterThan(0);
+});
+
+test("the patterns match the banned spellings and nothing else", () => {
+  const banned = [
+    // `wait_for_readable` / `wait_for_writable` as they were.
+    "io::IoRequestLoop::schedule(&mut self.io_request);",
+    // `FileCloser::do_close` as it was.
+    "if let Some(io_request) = self.io_request() {\n    io_request.store_callback_seq_cst(Self::schedule_close);\n    if !io_request.scheduled {\n        bun_io::IoRequestLoop::schedule(io_request);\n    }\n}",
+    // The pointer-typed argument spelled from `self` anyway.
+    "io::IoRequestLoop::schedule(&raw mut self.io_request);",
+    "bun_io::IoRequestLoop::schedule(self.io_request_ptr());",
+    "IoRequestLoop::schedule(core::ptr::from_mut(&mut self.io_request));",
+    "let request = &raw mut self.io_request;\nrequest.scheduled;\nio::IoRequestLoop::schedule(request);",
+    "let mut request = self.io_request();\nio::IoRequestLoop::schedule(request);",
+    // A `&mut` reborrow, whatever it is based on: its provenance stops at the field.
+    "io::IoRequestLoop::schedule(&mut (*this).io_request);",
+    "bun_io::IoRequestLoop::schedule(&mut *io_request);",
+    // rustfmt-wrapped.
+    "::bun_io::IoRequestLoop::schedule(\n    &mut self.io_request,\n);",
+    "::bun_io::IoRequestLoop::schedule(\n    &raw mut self.io_request,\n);",
+  ];
+  const allowed = [
+    // Projected from the pointer the caller holds: the intended shape.
+    "io::IoRequestLoop::schedule(&raw mut (*this).io_request);",
+    "if let Some(io_request) = Self::io_request(this) {\n    (*io_request).store_callback_seq_cst(Self::schedule_close);\n    if !(*io_request).scheduled {\n        bun_io::IoRequestLoop::schedule(io_request);\n    }\n}",
+    "let request = unsafe { Self::io_request_of(this) };\nbun_io::IoRequestLoop::schedule(request);",
+    // A self-derived binding that is not what gets scheduled.
+    "let fd = self.opened_fd;\nio::IoRequestLoop::schedule(request);",
+    // The binding is scheduled, but in the next function, where it is a
+    // raw-pointer parameter of the same name.
+    "let request = &mut self.io_request;\nrequest.scheduled = false;\n}\n\nunsafe fn publish(request: *mut io::Request) {\n    io::IoRequestLoop::schedule(request);\n}",
+    // Other things called `schedule`.
+    "WorkPool::schedule(&raw mut self.task);",
+    "bun_jsc::Job::<ReadFile>::schedule(&global.js_thread(), this, completion);",
+    "self.schedule(&mut self.io_request);",
+  ];
+  expect(banned.map(s => findBanned(s).length)).toEqual(banned.map(() => 1));
+  expect(allowed.map(s => findBanned(s).length)).toEqual(allowed.map(() => 0));
+});
+
+test("nothing schedules an io request spelled from self or through a &mut reborrow", () => {
+  expect(offenders).toEqual([]);
+});
diff --git a/test/js/bun/io/bun-write.test.js b/test/js/bun/io/bun-write.test.js
index f02bdf4f8e17..a8b7e462c5a2 100644
--- a/test/js/bun/io/bun-write.test.js
+++ b/test/js/bun/io/bun-write.test.js
@@ -7,10 +7,15 @@ import {
   exampleSite,
   gcTick,
   isASAN,
+  isDebug,
+  isLinux,
+  isMacOS,
   isWindows,
   tempDir,
+  waitForFileToContain,
   withoutAggressiveGC,
 } from "harness";
+import { mkfifo } from "mkfifo";
 import path, { join } from "path";
 
 let i = 0;
@@ -537,6 +542,199 @@ const IS_UV_FS_COPYFILE_DISABLED =
     expect(exitCode).toBe(0);
   });
 
+  // Bun.write() into a non-blocking FIFO that is already full when it starts, so
+  // every WriteFile involved is handed to the io thread before its first
+  // write() (the pipe is not drained until that has happened), then again
+  // whenever the pipe fills up while the payload drains, and (except on macOS,
+  // whose one-shot kqueue registration needs no unregistering) once more at
+  // the end to unregister the fd. In debug builds the child's WriteFile trace
+  // proves those hand-overs; in release builds the logging is compiled out and
+  // only the outcome is checked. 256 KiB skips Bun.write()'s synchronous path
+  // (the full pipe would make it fall back to WriteFile anyway) and is several
+  // pipe buffers long.
+  describe.skipIf(isWindows)("Bun.write(Bun.file(fd)) into a full non-blocking FIFO", () => {
+    const size = 256 * 1024;
+    const count = (trace, marker) => trace.split(marker).length - 1;
+
+    // Runs `writes` concurrent Bun.write()s of the payload in a child, drains
+    // the FIFO once the child reports the pipe full (and, in debug builds, once
+    // `writes` WriteFiles have reached the io thread), and returns what came
+    // out of the pipe and how each write settled.
+    async function writeIntoFullFifo(writes) {
+      using dir = tempDir("bun-write-fifo", {});
+      const fifo = join(String(dir), "out.fifo");
+      const trace = join(String(dir), "trace.log");
+      mkfifo(fifo);
+      // A reader has to exist for the child's O_NONBLOCK open of the write end
+      // to succeed; this one never reads, the drain below does.
+      const holder = fs.openSync(fifo, fs.constants.O_RDONLY | fs.constants.O_NONBLOCK);
+      let opening;
+      try {
+        const script = `
+          const fs = require("fs");
+          const fd = fs.openSync(process.env.FIFO, fs.constants.O_WRONLY | fs.constants.O_NONBLOCK);
+          const dest = Bun.file(fd);
+          // fstat()s the fd: Bun.write() only polls a destination it knows is not a regular file.
+          dest.size;
+          const chunk = Buffer.alloc(4096, "P");
+          let filled = 0;
+          try {
+            for (;;) filled += fs.writeSync(fd, chunk);
+          } catch (e) {
+            if (e.code !== "EAGAIN") throw e;
+          }
+          const payload = Buffer.alloc(${size}, "W");
+          const settled = Promise.allSettled(Array.from({ length: ${writes} }, () => Bun.write(dest, payload)));
+          process.stderr.write("filled " + filled + "\\n");
+          const results = (await settled).map(r =>
+            r.status === "fulfilled" ? { written: r.value } : { code: r.reason.code, syscall: r.reason.syscall },
+          );
+          process.stdout.write(JSON.stringify(results));
+        `;
+        await using proc = Bun.spawn({
+          cmd: [bunExe(), "-e", script],
+          // BUN_DEBUG sends the scoped debug logs to a file instead of stdout.
+          env: { ...bunEnv, FIFO: fifo, ...(isDebug ? { BUN_DEBUG_WriteFile: "1", BUN_DEBUG: trace } : {}) },
+          stdout: "pipe",
+          stderr: "pipe",
+        });
+        const stdout = proc.stdout.text();
+        const filled = Promise.withResolvers();
+        const stderr = (async () => {
+          let text = "";
+          for await (const chunk of proc.stderr) {
+            text += Buffer.from(chunk).toString();
+            const m = /^filled (\d+)\n/m.exec(text);
+            if (m) filled.resolve(Number(m[1]));
+          }
+          return text;
+        })();
+        // The child cannot finish before the pipe is drained, so exiting before
+        // the drain has started is a failure. A blocking open of the read end
+        // would otherwise wait forever for a writer once the child is gone;
+        // after it, a dying child just ends the drain early (EOF) and the
+        // assertions report what arrived.
+        const childDied = proc.exited.then(async exitCode => {
+          throw new Error(`child exited with ${exitCode} before the FIFO was drained: ${await stderr}`);
+        });
+        const prefill = await Promise.race([filled.promise, childDied]);
+        if (isDebug) {
+          await Promise.race([waitForFileToContain(trace, "WriteFile.onRequestWritable()", writes), childDied]);
+        }
+        opening = fs.promises.open(fifo, "r");
+        const drain = await Promise.race([opening, childDied]);
+        opening = undefined;
+        let data;
+        try {
+          data = await drain.readFile();
+        } finally {
+          await drain.close();
+        }
+        const [stdoutText, stderrText, exitCode] = await Promise.all([stdout, stderr, proc.exited]);
+        let results;
+        try {
+          results = JSON.parse(stdoutText);
+        } catch {
+          results = stdoutText;
+        }
+        return {
+          prefill,
+          data,
+          results,
+          stderr: stderrText,
+          exitCode,
+          log: isDebug ? fs.readFileSync(trace, "utf8") : "",
+        };
+      } finally {
+        if (opening) {
+          // The open lost the race above and is still waiting for a writer:
+          // lend it one so it completes, then close what it opened.
+          opening.then(
+            handle => handle.close(),
+            () => {},
+          );
+          fs.closeSync(fs.openSync(fifo, fs.constants.O_WRONLY | fs.constants.O_NONBLOCK));
+        }
+        fs.closeSync(holder);
+      }
+    }
+
+    function pipeContents(prefill, data, payloads) {
+      return {
+        prefilled: prefill > 0,
+        length: data.length,
+        prefillIntact: data.subarray(0, prefill).equals(Buffer.alloc(prefill, "P")),
+        payloadsIntact: data.subarray(prefill).equals(Buffer.alloc(size * payloads, "W")),
+      };
+    }
+
+    it("waits for the reader and delivers the payload", async () => {
+      const { prefill, data, results, stderr, exitCode, log } = await writeIntoFullFifo(1);
+
+      expect({ ...pipeContents(prefill, data, 1), results, stderr }).toEqual({
+        prefilled: true,
+        length: prefill + size,
+        prefillIntact: true,
+        payloadsIntact: true,
+        results: [{ written: size }],
+        stderr: `filled ${prefill}\n`,
+      });
+      expect(exitCode).toBe(0);
+
+      if (isDebug) {
+        const requests = count(log, "WriteFile.onRequestWritable()");
+        expect(requests).toBeGreaterThanOrEqual(1);
+        // Every request the io thread popped came back as one readiness
+        // callback. Closing round-trips through the io thread as well, which
+        // makes on_finish run a second time; macOS clears that flag when the
+        // one-shot registration fires, so there it runs once.
+        expect({
+          readies: count(log, "WriteFile.onReady()"),
+          ioErrors: count(log, "WriteFile.onIOError()"),
+          finishes: count(log, "WriteFile.onFinish()"),
+        }).toEqual({ readies: requests, ioErrors: 0, finishes: isMacOS ? 1 : 2 });
+      }
+    });
+
+    // Two WriteFiles on one fd: the second epoll registration of the fd fails
+    // with EEXIST (two pollers per fd is not supported today; if that ever
+    // changes, this case becomes two successful writes), which is the one way
+    // to reach the registration-error hand-over: the io thread reports the
+    // error back, and the failed WriteFile still closes through the io thread
+    // and rejects, while the other one delivers its payload. Debug builds only,
+    // because only the trace makes the outcome deterministic: the pipe is not
+    // drained until both requests have reached the io thread, so the second
+    // registration always finds the first one in place. Linux only: on kqueue
+    // the second registration silently replaces the first instead of failing.
+    it.skipIf(!isLinux || !isDebug)(
+      "a second concurrent write is rejected with EEXIST and the first still completes",
+      async () => {
+        const { prefill, data, results, stderr, exitCode, log } = await writeIntoFullFifo(2);
+
+        expect({ ...pipeContents(prefill, data, 1), results, stderr }).toEqual({
+          prefilled: true,
+          length: prefill + size,
+          prefillIntact: true,
+          payloadsIntact: true,
+          results: expect.arrayContaining([{ written: size }, { code: "EEXIST", syscall: "epoll_ctl" }]),
+          stderr: `filled ${prefill}\n`,
+        });
+        expect(results).toHaveLength(2);
+        expect(exitCode).toBe(0);
+
+        const requests = count(log, "WriteFile.onRequestWritable()");
+        expect(requests).toBeGreaterThanOrEqual(2);
+        // One request ended in the error callback instead of a readiness
+        // callback; both WriteFiles still closed through the io thread.
+        expect({
+          readies: count(log, "WriteFile.onReady()"),
+          ioErrors: count(log, "WriteFile.onIOError()"),
+          finishes: count(log, "WriteFile.onFinish()"),
+        }).toEqual({ readies: requests - 1, ioErrors: 1, finishes: 4 });
+      },
+    );
+  });
+
   it("Bun.file(0) survives GC", async () => {
     for (let i = 0; i < 10; i++) {
       let f = Bun.file(0);
diff --git a/test/js/bun/util/bun-file-read.test.ts b/test/js/bun/util/bun-file-read.test.ts
index 40c6309a0c13..cb91f4e80f88 100644
--- a/test/js/bun/util/bun-file-read.test.ts
+++ b/test/js/bun/util/bun-file-read.test.ts
@@ -1,5 +1,8 @@
 import { describe, expect, it } from "bun:test";
-import { tempDir } from "harness";
+import { bunEnv, bunExe, isDebug, isMacOS, isWindows, tempDir, waitForFileToContain } from "harness";
+import { mkfifo } from "mkfifo";
+import { randomBytes } from "node:crypto";
+import { closeSync, constants, openSync, readFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import path from "node:path";
 
@@ -12,7 +15,7 @@ it("offset should work in Bun.file() #4963", async () => {
   expect(contents).toBe("ntents");
 });
 
-// do_read_loop picks its read target per iteration: the 64 KB stack buffer
+// ReadFile::read_until_blocked picks its read target per iteration: the 64 KB stack buffer
 // when self.buffer's spare capacity is smaller, otherwise the Vec's spare
 // capacity directly. Cover both branches plus the max_length cap so the
 // branch selection and the commit_spare path stay tied to the same decision.
@@ -54,3 +57,94 @@ describe("Bun.file read-loop target selection", () => {
     expect(Bun.hash(buf)).toBe(Bun.hash(bytes.subarray(start, end)));
   });
 });
+
+// Reading a FIFO by path. Nothing is written until the child has found the pipe
+// empty and handed its ReadFile to the io thread, so the read goes through at
+// least one io-thread round trip, and (on Linux) through one more at the end to
+// unregister the fd before the fd it opened is closed. In debug builds the
+// child's ReadFile trace is what proves those hand-overs happened; in release
+// builds the logging is compiled out and only the result is checked.
+//
+// macOS: the child never finishes this read (the test times out with the child
+// still alive) while the same setup completes on Linux; reading a FIFO by path
+// on macOS is also what "Bun.file() read text from pipe" in
+// test/js/web/streams/streams.test.js is todo'd for. The read-side hand-overs
+// are covered on macOS by the Bun.stdin pipe tests (test/regression/issue/07500,
+// bun-stdin-slice.test.ts); what this adds is the path-opened variant.
+describe.skipIf(isWindows || isMacOS)("Bun.file(fifo)", () => {
+  const count = (trace: string, marker: string) => trace.split(marker).length - 1;
+
+  it("bytes() waits for the writer, reads what it writes and ends when it closes", async () => {
+    const payload = randomBytes(256 * 1024);
+    using dir = tempDir("bun-file-read-fifo", {});
+    const fifo = path.join(String(dir), "in.fifo");
+    const trace = path.join(String(dir), "trace.log");
+    mkfifo(fifo);
+    // The write end can only be opened, and written to without EPIPE, while
+    // some reader has the FIFO open; `holder` is that reader until the child
+    // has opened its own. It never reads, so every byte goes to the child.
+    let holder = openSync(fifo, constants.O_RDONLY | constants.O_NONBLOCK);
+    const closeHolder = () => {
+      if (holder !== -1) closeSync(holder);
+      holder = -1;
+    };
+    let writer = openSync(fifo, "w");
+    try {
+      await using proc = Bun.spawn({
+        cmd: [
+          bunExe(),
+          "-e",
+          `const bytes = await Bun.file(process.env.FIFO).bytes(); process.stdout.write(bytes.length + " " + Bun.hash(bytes));`,
+        ],
+        // ReadFile logs under the WriteFile scope; BUN_DEBUG sends the scoped logs to a file.
+        env: { ...bunEnv, FIFO: fifo, ...(isDebug ? { BUN_DEBUG_WriteFile: "1", BUN_DEBUG: trace } : {}) },
+        stdout: "pipe",
+        stderr: "pipe",
+      });
+      const stderr = proc.stderr.text();
+      // The child cannot exit before the payload is written unless it failed;
+      // dropping `holder` then leaves the pipe without readers, so a write
+      // blocked on it fails with EPIPE instead of waiting forever.
+      const childDied = proc.exited.then(async exitCode => {
+        closeHolder();
+        throw new Error(`child exited with ${exitCode} before the payload was written: ${await stderr}`);
+      });
+      if (isDebug) {
+        // The io thread has run the child's request: it is waiting for data that does not exist yet.
+        await Promise.race([waitForFileToContain(trace, "ReadFile.onRequestReadable"), childDied]);
+      }
+      // The write end is blocking, so this completes as the child drains the
+      // pipe; closing it afterwards is what ends the child's read.
+      const written = await Promise.race([Bun.write(Bun.file(writer), payload), childDied]);
+      closeSync(writer);
+      writer = -1;
+      const [stdout, stderrText, exitCode] = await Promise.all([proc.stdout.text(), stderr, proc.exited]);
+
+      expect({ written, stdout, stderr: stderrText }).toEqual({
+        written: payload.length,
+        stdout: `${payload.length} ${Bun.hash(payload)}`,
+        stderr: "",
+      });
+      expect(exitCode).toBe(0);
+
+      if (isDebug) {
+        const log = readFileSync(trace, "utf8");
+        const waits = count(log, "ReadFile.waitForReadable");
+        expect(waits).toBeGreaterThanOrEqual(1);
+        // Every wait is one request popped by the io thread and one readiness
+        // callback; the close then round-trips through the io thread once
+        // (deferred) before the read completes (immediately).
+        expect({
+          requests: count(log, "ReadFile.onRequestReadable"),
+          readies: count(log, "ReadFile.onReady"),
+          ioErrors: count(log, "ReadFile.onIOError"),
+          deferredCloses: count(log, "ReadFile.onFinish() = deferred"),
+          completions: count(log, "ReadFile.onFinish() = immediately"),
+        }).toEqual({ requests: waits, readies: waits, ioErrors: 0, deferredCloses: 1, completions: 1 });
+      }
+    } finally {
+      if (writer !== -1) closeSync(writer);
+      closeHolder();
+    }
+  });
+});