shell: support ReadableStream stdin redirect via SinkHandle

src/runtime/shell/Builtin.rs

@@ -726,6 +726,13 @@ impl Builtin {
                     if redirect.stderr() {
                         me.stderr = BuiltinIO::ArrayBuf { buf: mk(), i: 0 };
                     }
+                } else if crate::webcore::ReadableStream::is_readable_stream(jsval) {
+                    let name = Self::of(interp, cmd).kind.as_str();
+                    let _ = global.throw(format_args!(
+                        "ReadableStream cannot be redirected to a builtin command ('{name}'). \
+                         Use an external command or buffer the stream first",
+                    ));
+                    return Some(Yield::failed());
                 } else if let Some(body) =
                     crate::webcore::body::Value::from_request_or_response(jsval)
                 {

src/runtime/shell/states/Cmd.rs

@@ -758,8 +758,32 @@ impl Cmd {
                             STDERR_NO as i32,
                         )?;
                     }
-                } else if crate::webcore::ReadableStream::from_js(jsval, global)?.is_some() {
-                    panic!("TODO SHELL READABLE STREAM");
+                } else if let Some(mut stream) =
+                    crate::webcore::ReadableStream::from_js(jsval, global)?
+                {
+                    if !flags.stdin() {
+                        return Err(global.throw(format_args!(
+                            "ReadableStream cannot be used for stdout or stderr; \
+                             only '< ${{...}}' (stdin) is supported"
+                        )));
+                    }
+                    if stream.is_locked(global) || stream.is_disturbed(global) {
+                        return Err(global
+                            .err(
+                                crate::jsc::ErrorCode::INVALID_STATE,
+                                format_args!(
+                                    "ReadableStream redirected to stdin has already been used"
+                                ),
+                            )
+                            .throw());
+                    }
+                    // Fully-buffered / not-yet-started file-backed streams collapse
+                    // to a blob and take the existing `StaticPipeWriter` path.
+                    if let Some(blob) = stream.to_any_blob(global) {
+                        stdio[STDIN_NO].extract_blob(global, blob, STDIN_NO as i32)?;
+                    } else {
+                        stdio[STDIN_NO] = Stdio::ReadableStream(stream);
+                    }
                 } else if let Some(req) = jsval.as_::<crate::webcore::Response>() {
                     // SAFETY: `as_` returns a live JSC-owned `*mut Response`;
                     // `get_body_value` is `&self`.

src/runtime/shell/subproc.rs

@@ -131,17 +131,35 @@
 pub struct FileSinkPtr(core::ptr::NonNull<FileSink>);
 
 impl FileSinkPtr {
-    /// Adopt the +1 ref returned by `FileSink::create*`.
-    ///
-    /// # Safety
-    /// `ptr` is non-null, points to a live `FileSink` from
-    /// `FileSink::create*`, and the caller transfers its single owned ref to
-    /// this handle.
+    /// Create a `FileSink` writing to `fd` and adopt the create-ref.
+    #[cfg(not(windows))]
+    fn create(event_loop: EventLoopHandle, fd: Fd) -> Self {
+        Self(
+            core::ptr::NonNull::new(FileSink::create(event_loop, fd))
+                .expect("FileSink::create returns non-null"),
+        )
+    }
+
+    /// Create a `FileSink` around `pipe` (ownership of the `Box<uv::Pipe>`
+    /// transfers to the sink's writer) and adopt the create-ref.
     #[cfg(windows)]
-    #[inline]
-    unsafe fn adopt(ptr: *mut FileSink) -> Self {
-        // SAFETY: caller contract — `ptr` is non-null.
-        Self(unsafe { core::ptr::NonNull::new_unchecked(ptr) })
+    fn create_with_pipe(
+        event_loop: EventLoopHandle,
+        pipe: *mut bun_sys::windows::libuv::Pipe,
+    ) -> Self {
+        Self(
+            core::ptr::NonNull::new(FileSink::create_with_pipe(event_loop, pipe))
+                .expect("FileSink::create_with_pipe returns non-null"),
+        )
+    }
+
+    /// The attached process exited: cancel the assigned stream and close the
+    /// writer.
+    fn on_attached_process_exit(&self, status: &bun_process::Status) {
+        // SAFETY: `self.0` is the canonical `FileSink::create*` pointer (full
+        // write+dealloc provenance, never laundered through a reference) and
+        // the ref this handle owns keeps the sink live across the call.
+        unsafe { FileSink::on_attached_process_exit(self.0.as_ptr(), status) };
     }
 }
 
@@ -698,18 +716,26 @@
         // (they store it on StaticPipeWriter / PipeReader as a backref).
         let mut slot = Box::<Subprocess>::new_uninit();
         let subprocess: *mut Subprocess = slot.as_mut_ptr();
-        // SAFETY: `out_subproc` points at the `SubprocExec.child` slot inside
-        // the heap-stable `Box<SubprocExec>` staged by the caller before this
-        // call; no `&` to that slot is live (the caller's `&mut Cmd` borrow
-        // ended before the call). Written *before* any callback below
-        // (`watch`/`start`/`read_all`) so re-entrant `Cmd` callbacks see a
-        // populated `exec.subproc.child`.
-        unsafe { *out_subproc = subprocess };
 
         let stdin = match Writable::init(stdio0, event_loop, subprocess, spawn_stdin) {
             Ok(w) => w,
-            Err(WritableInitError::UnexpectedCreatingStdin) => {
-                panic!("unexpected error while creating stdin");
+            Err(err) => {
+                #[cfg(not(windows))]
+                {
+                    let _ = spawn_stdout.map(bun_sys::Fd::close);
+                    let _ = spawn_stderr.map(bun_sys::Fd::close);
+                }
+                #[cfg(windows)]
+                {
+                    // `WindowsSpawnResult::drop` uv_closes handed-back slots.
+                    spawn_result.stdout = spawn_stdout;
+                    spawn_result.stderr = spawn_stderr;
+                }
+                spawn_result.dispose_failed_spawn(event_loop);
+                return Err(match err {
+                    WritableInitError::Sys(e) => ShellErr::Sys(e.to_shell_system_error()),
+                    WritableInitError::StreamAssign(msg) => ShellErr::Custom(msg),
+                });
             }
         };
         let stdout = Readable::init(
@@ -754,6 +780,14 @@
         // sound.
         // SAFETY: fully initialised by the `write` above.
         let _ = bun_core::heap::into_raw(unsafe { slot.assume_init() });
+        // SAFETY: `out_subproc` points at the `SubprocExec.child` slot inside
+        // the heap-stable `Box<SubprocExec>` staged by the caller before this
+        // call; no `&` to that slot is live (the caller's `&mut Cmd` borrow
+        // ended before the call). Published now that the Subprocess is fully
+        // initialised and before any callback below (`watch`/`start`/
+        // `read_all`) so re-entrant `Cmd` callbacks see a populated
+        // `exec.subproc.child`.
+        unsafe { *out_subproc = subprocess };
         // SAFETY: `subprocess` is the just-allocated `ShellSubprocess`; the
         // owning `Cmd` outlives the `Process` exit callback. All accesses
         // below are scoped so no borrow of the Subprocess spans the
@@ -768,37 +802,6 @@
         }
         let _ = scopeguard::ScopeGuard::into_inner(stdio_guard);
 
-        // Wire the FileSink's close-signal back to the enclosing `Writable` so
-        // `Writable::on_close` (drops the `Arc<FileSink>`) runs when the sink
-        // finishes. `stdin` lives inside the Box-allocated `Subprocess` at a
-        // stable address, so the self-referential raw pointer is sound for the
-        // life of the subprocess. Only reachable on Windows (POSIX
-        // `Writable::init` never returns `Pipe` for shell stdio).
-        {
-            // Derive `stdin_ptr` from the raw heap pointer (`subprocess`), not
-            // the local `subproc: &mut` reborrow — the pointer is stored
-            // long-term in `FileSink::source` and dereferenced from
-            // `Writable::on_close` after this frame returns. Under Stacked
-            // Borrows a child of `subproc`'s tag would be invalidated when
-            // that borrow ends; rooting in the allocation's provenance keeps
-            // it valid for the box's lifetime.
-            // SAFETY: `subprocess` is the live, fully-initialised heap alloc.
-            let stdin_ptr: *mut Writable = unsafe { &raw mut (*subprocess).stdin };
-            // SAFETY: reborrow as a child of `stdin_ptr` so it does not
-            // invalidate the sibling we store in `source`.
-            if let Writable::Pipe(pipe) = unsafe { &mut *stdin_ptr } {
-                // SAFETY: shell is single-threaded; the FileSink allocation is
-                // disjoint from `*stdin_ptr`. `stdin_ptr` outlives the sink —
-                // the Subprocess owns both and `Writable::on_close` is the only
-                // path that drops the FileSinkPtr.
-                pipe.source
-                    .set(webcore::streams::SourceHandle::ShellWritable(
-                        // SAFETY: `stdin_ptr` is the live `&raw mut` writable (write provenance).
-                        unsafe { bun_ptr::BackRef::from_raw_mut(stdin_ptr) },
-                    ));
-            }
-        }
-
         // SAFETY: scoped access; `watch` does not re-enter the subprocess.
         match unsafe { (*subprocess).proc().watch() } {
             bun_sys::Result::Ok(()) => {}
@@ -863,6 +866,19 @@
 
     pub(crate) fn on_process_exit(&mut self, _: &Process, status: &Status, _: &Rusage) {
         log!("onProcessExit({:x})", std::ptr::from_mut(self) as usize);
+
+        // ReadableStream stdin (the only shell `Writable::Pipe`): close the
+        // FileSink and mark stdin closed.
+        let stdin_is_stream_pipe = if let Writable::Pipe(pipe) = &self.stdin {
+            pipe.on_attached_process_exit(status);
+            true
+        } else {
+            false
+        };
+        if stdin_is_stream_pipe {
+            self.on_static_pipe_writer_done();
+        }
+
         let exit_code: Option<u8> = 'brk: {
             if let Status::Exited(exited) = &status {
                 break 'brk Some(exited.code);
@@ -887,8 +903,12 @@
             // through the node arena so it survives `Vec<Node>` reallocation.
             // `&mut self` is dead by NLL before `on_exit` re-enters interp.
             let cmd = unsafe { handle.cmd_mut() };
-            if cmd.exit_code.is_none() {
-                cmd.on_exit(code.into());
+            match cmd.exit_code {
+                None => cmd.on_exit(code.into()),
+                // Stdout/stderr closing set `exit_code` first; the stdin close
+                // above may have just satisfied `has_finished()`.
+                Some(prev) if cmd.has_finished() => cmd.on_exit(prev),
+                Some(_) => {}
             }
         }
     }
@@ -900,8 +920,11 @@
 
 #[derive(thiserror::Error, Debug, strum::IntoStaticStr)]
 pub enum WritableInitError {
-    #[error("UnexpectedCreatingStdin")]
-    UnexpectedCreatingStdin,
+    #[error("{0}")]
+    Sys(bun_sys::Error),
+    /// `assign_to_stream` threw synchronously; carries the formatted message.
+    #[error("Failed to pipe ReadableStream to stdin")]
+    StreamAssign(Box<[u8]>),
 }
 
 pub enum Writable {
@@ -913,20 +936,6 @@
     Ignore,
 }
 
-impl Writable {
-    // When the stream has closed we need to be notified to prevent a use-after-free
-    // We can test for this use-after-free by enabling hot module reloading on a file and then saving it twice
-    pub fn on_close(&mut self, _: Option<bun_sys::Error>) {
-        match self {
-            Writable::Buffer(_) | Writable::Pipe(_) => {
-                // Dropping the Arc on reassignment below derefs.
-            }
-            _ => {}
-        }
-        *self = Writable::Ignore;
-    }
-}
-
 impl Writable {
     pub(crate) fn init(
         stdio: Stdio,
@@ -948,29 +957,19 @@
                         // Ownership of the `Box<uv::Pipe>` transfers into the
                         // FileSink's writer.
                         let uv_pipe: *mut _ = bun_core::heap::into_raw(buf);
-                        let pipe_ptr = FileSink::create_with_pipe(event_loop, uv_pipe);
-
-                        // SAFETY: `create_with_pipe` returns a freshly-boxed
-                        // non-null FileSink with refcount 1; sole reference.
-                        match unsafe {
-                            (*pipe_ptr).writer.with_mut(|w| w.start_with_current_pipe())
-                        } {
-                            bun_sys::Result::Ok(()) => {}
-                            bun_sys::Result::Err(_err) => {
-                                // SAFETY: pipe_ptr is live with refcount 1;
-                                // deref frees it.
-                                unsafe { FileSink::deref(pipe_ptr) };
-                                return Err(WritableInitError::UnexpectedCreatingStdin);
-                            }
-                        }
+                        let pipe = FileSinkPtr::create_with_pipe(event_loop, uv_pipe);
 
-                        // TODO: uncoment this when is ready, commented because was not compiling
-                        // subprocess.weak_file_sink_stdin_ptr = pipe;
-                        // subprocess.flags.has_stdin_destructor_called = false;
+                        if let bun_sys::Result::Err(e) =
+                            pipe.writer.with_mut(|w| w.start_with_current_pipe())
+                        {
+                            // Dropping `pipe` derefs (and frees) the sink.
+                            return Err(WritableInitError::Sys(e));
+                        }
 
-                        // SAFETY: `create_with_pipe` returns non-null with one
-                        // owned ref; `adopt` takes it over.
-                        return Ok(Writable::Pipe(unsafe { FileSinkPtr::adopt(pipe_ptr) }));
+                        if let Stdio::ReadableStream(rs) = &mut stdio {
+                            return Self::assign_stream(pipe, rs, event_loop);
+                        }
+                        return Ok(Writable::Pipe(pipe));
                     }
                     return Ok(Writable::Inherit);
                 }
@@ -1078,9 +1077,24 @@
                 Stdio::Inherit => Ok(Writable::Inherit),
                 Stdio::Path(_) | Stdio::Ignore => Ok(Writable::Ignore),
                 Stdio::Ipc | Stdio::Capture(_) => Ok(Writable::Ignore),
-                Stdio::ReadableStream(_) => {
-                    // The shell never uses this
-                    panic!("Unimplemented stdin readable_stream");
+                Stdio::ReadableStream(rs) => {
+                    let fd = result.unwrap();
+                    if let bun_sys::Result::Err(e) = bun_sys::set_nonblocking(fd) {
+                        fd.close();
+                        return Err(WritableInitError::Sys(e));
+                    }
+                    let pipe = FileSinkPtr::create(event_loop, fd);
+                    if let bun_sys::Result::Err(e) = pipe.writer.with_mut(|w| w.start(fd, true)) {
+                        // Dropping `pipe` derefs (and frees) the sink.
+                        return Err(WritableInitError::Sys(e));
+                    }
+                    // The fd is a socketpair half, same as `Bun.spawn` stdin.
+                    pipe.writer.with_mut(|w| {
+                        if let Some(poll) = w.handle.get_poll() {
+                            poll.set_flag(bun_io::FilePollFlag::Socket);
+                        }
+                    });
+                    Self::assign_stream(pipe, rs, event_loop)
                 }
                 Stdio::SocketFd => {
                     // The shell never uses this; rejected at i < 3 anyway.
@@ -1090,6 +1104,32 @@
         }
     }
 
+    /// Wire `stream` into the stdin `FileSink` (native `SinkHandle` wire
+    /// first, JS pump fallback). A synchronous throw consumes the sink.
+    fn assign_stream(
+        mut pipe: FileSinkPtr,
+        stream: &mut webcore::ReadableStream,
+        event_loop: EventLoopHandle,
+    ) -> Result<Writable, WritableInitError> {
+        // ReadableStream stdin only exists for JS-origin shells.
+        let global_ptr = event_loop.global_object();
+        assert!(
+            !global_ptr.is_null(),
+            "ReadableStream stdin requires the JS event loop"
+        );
+        let global = jsc::JSGlobalObject::opaque_ref(global_ptr.cast());
+        if let Some(err) = pipe.assign_to_stream(stream, global).to_error() {
+            let msg = format!(
+                "Failed to pipe ReadableStream to stdin: {}",
+                err.fmt_string(global)
+            );
+            return Err(WritableInitError::StreamAssign(
+                msg.into_bytes().into_boxed_slice(),
+            ));
+        }
+        Ok(Writable::Pipe(pipe))
+    }
+
     // Note: there is intentionally no `Writable::toJS` here — the shell never
     // exposes its stdin Writable to JS.
 

src/runtime/webcore/streams.rs

@@ -907,7 +907,6 @@ pub enum SourceHandle {
     /// The `'static` bound erases the `&JSGlobalObject` borrow carried in
     /// `Subprocess<'a>`; the pointed-at allocation outlives this handle.
     Subprocess(BackRef<crate::api::bun::subprocess::Subprocess<'static>>),
-    ShellWritable(BackRef<crate::shell::subproc::Writable, bun_ptr::Mut>),
     FetchResponseBody(BackRef<crate::webcore::fetch::fetch_tasklet::FetchTasklet, bun_ptr::Mut>),
     ServerRequestBody(crate::server::AnyRequestContext),
     S3DownloadBody(BackRef<crate::webcore::s3::client::S3DownloadStreamWrapper, bun_ptr::Mut>),
@@ -947,8 +946,6 @@ impl SourceHandle {
             SourceHandle::FileReader(p) => p.on_close(err),
             SourceHandle::Subprocess(p) => p.on_close(err),
             // SAFETY: live backref; cleared before the pointee is freed.
-            SourceHandle::ShellWritable(mut p) => unsafe { p.get_mut() }.on_close(err),
-            // SAFETY: live backref; cleared before the pointee is freed.
             SourceHandle::FetchResponseBody(mut p) => unsafe { p.get_mut() }.on_stream_cancelled(),
             // SAFETY: live backref; cleared before the pointee is freed.
             SourceHandle::S3DownloadBody(mut p) => unsafe { p.get_mut() }.on_stream_cancelled(),
@@ -978,9 +975,7 @@ impl SourceHandle {
             SourceHandle::ServerRequestBody(any) => any.on_request_body_stream_drained(),
             SourceHandle::HTMLRewriter(p) => p.on_ready(),
             // Remaining variants leave `on_ready` at the trait default (no-op).
-            SourceHandle::Subprocess(_)
-            | SourceHandle::ShellWritable(_)
-            | SourceHandle::S3DownloadBody(_) => {}
+            SourceHandle::Subprocess(_) | SourceHandle::S3DownloadBody(_) => {}
         }
     }
 
@@ -994,7 +989,6 @@ impl SourceHandle {
             | SourceHandle::ByteStream(_)
             | SourceHandle::FileReader(_)
             | SourceHandle::Subprocess(_)
-            | SourceHandle::ShellWritable(_)
             | SourceHandle::S3DownloadBody(_)
             | SourceHandle::HTMLRewriter(_) => {}
         }