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) };
     }
 }
 
@@ -688,6 +706,8 @@
         let stdio1 = core::mem::replace(&mut stdio_guard[1], Stdio::Ignore);
         let stdio2 = core::mem::replace(&mut stdio_guard[2], Stdio::Ignore);
 
+        let stdin_is_stream = matches!(stdio0, Stdio::ReadableStream(_));
+
         // `to_process` consumes the result for pid/pidfd; pull the fd handles out first.
         let spawn_stdin = spawn_result.stdin.take();
         let spawn_stdout = spawn_result.stdout.take();
@@ -698,18 +718,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 +782,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
@@ -772,9 +808,9 @@
         // `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).
-        {
+        // life of the subprocess. Skipped for ReadableStream stdin, whose
+        // `source` was wired by `Writable::init`.
+        if !stdin_is_stream {
             // 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
@@ -863,6 +899,23 @@
 
     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 {
+            debug_assert!(!matches!(
+                *pipe.source.get(),
+                webcore::streams::SourceHandle::ShellWritable(_)
+            ));
+            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 +940,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 +957,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 {
@@ -948,29 +1008,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);
                 }
@@ -1066,11 +1116,11 @@
                     debug_assert!(memfd.is_valid());
                     let fd = *memfd;
                     // Ownership of the fd transfers to `Writable::Memfd`.
                     // Swap in `Ignore` and suppress the old value's destructor
                     // so `Stdio::Drop` doesn't close the fd we just took
                     // (`stdio = Stdio::Ignore` alone would drop+close the old
                     // `Stdio::Memfd`).
                     let _ =
                         core::mem::ManuallyDrop::new(core::mem::replace(&mut stdio, Stdio::Ignore));
                     Ok(Writable::Memfd(fd))
                 }
@@ -1078,9 +1128,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 +1155,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/spawn/process.rs

@@ -1473,6 +1473,19 @@ impl WindowsSpawnResult {
     pub fn to_process(&mut self, _event_loop: impl Sized) -> *mut Process {
         self.process.take().unwrap()
     }
+
+    /// Kill and release a child whose post-spawn stdio setup failed.
+    /// Consumes the result so `Drop` also closes stdio pipes still held.
+    pub fn dispose_failed_spawn(mut self, _event_loop: impl Sized) {
+        let proc = self.to_process(());
+        // SAFETY: `to_process` hands over the sole owned ref to the live
+        // heap-allocated `Process`; `deref` releases it after teardown.
+        unsafe {
+            let _ = (*proc).kill(bun_core::SignalCode::SIGKILL as u8);
+            (*proc).close();
+            Process::deref(proc);
+        }
+    }
 }
 
 #[cfg(windows)]
@@ -1632,13 +1645,32 @@ impl WindowsSpawnOptions {
 /// trait method so callers keep the `.to_process(loop_, sync)` spelling.
 pub trait SpawnResultExt {
     fn to_process(self, event_loop: EventLoopHandle) -> *mut Process;
+
+    /// Kill, reap, and release a child whose post-spawn stdio setup failed.
+    fn dispose_failed_spawn(self, event_loop: EventLoopHandle);
 }
 
 #[cfg(unix)]
 impl SpawnResultExt for PosixSpawnResult {
     fn to_process(self, event_loop: EventLoopHandle) -> *mut Process {
         Process::init_posix(&self, event_loop)
     }
+
+    fn dispose_failed_spawn(self, _event_loop: EventLoopHandle) {
+        // `Process::kill` no-ops while `Poller::Detached` (pre-`watch()`);
+        // signal the pid directly so the blocking reap terminates.
+        unsafe extern "C" {
+            #[link_name = "kill"]
+            safe fn libc_kill(pid: libc::pid_t, sig: c_int) -> c_int;
+        }
+        let _ = libc_kill(self.pid, bun_core::SignalCode::SIGKILL as c_int);
+        let _ = posix_spawn::wait4(self.pid, 0, None);
+        #[cfg(any(target_os = "linux", target_os = "android"))]
+        if let Some(pidfd) = self.pidfd {
+            use bun_sys::FdExt as _;
+            Fd::from_native(pidfd).close();
+        }
+    }
 }
 
 #[cfg(unix)]