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

@@ -137,12 +137,16 @@
     /// `ptr` is non-null, points to a live `FileSink` from
     /// `FileSink::create*`, and the caller transfers its single owned ref to
     /// this handle.
-    #[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) })
     }
+
+    #[inline]
+    fn as_ptr(&self) -> *mut FileSink {
+        self.0.as_ptr()
+    }
 }
 
 impl core::ops::Deref for FileSinkPtr {
@@ -688,6 +692,11 @@
         let stdio1 = core::mem::replace(&mut stdio_guard[1], Stdio::Ignore);
         let stdio2 = core::mem::replace(&mut stdio_guard[2], Stdio::Ignore);
 
+        let stdin_stream: Option<webcore::ReadableStream> = match &stdio0 {
+            Stdio::ReadableStream(s) => Some(*s),
+            _ => None,
+        };
+
         // `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();
@@ -708,9 +717,26 @@
 
         let stdin = match Writable::init(stdio0, event_loop, subprocess, spawn_stdin) {
             Ok(w) => w,
             Err(WritableInitError::UnexpectedCreatingStdin) => {
                 panic!("unexpected error while creating stdin");
             }
+            Err(WritableInitError::Sys(e)) => {
+                #[cfg(not(windows))]
+                {
+                    let _ = spawn_stdout.map(bun_sys::Fd::close);
+                    let _ = spawn_stderr.map(bun_sys::Fd::close);
+                }
+                let proc = spawn_result.to_process(event_loop);
+                // SAFETY: `to_process` returns a live heap-allocated `Process`
+                // with one intrusive ref; sole reference.
+                unsafe {
+                    let _ = (*proc).kill(SignalCode::SIGKILL as u8);
+                    (*proc).wait(true);
+                    (*proc).close();
+                    bun_ptr::ThreadSafeRefCount::<Process>::deref(proc);
+                }
+                return Err(ShellErr::Sys(e.to_shell_system_error()));
+            }
         };
         let stdout = Readable::init(
             OutKind::Stdout,
@@ -772,9 +798,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, which sets
+        // `source` to the upstream ByteStream/FileReader below.
+        if stdin_stream.is_none() {
             // 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
@@ -799,6 +825,38 @@
             }
         }
 
+        // ReadableStream stdin: `assign_to_stream` tries `wire_native_sink`
+        // (`SinkHandle::FileSink`) first, then falls back to the JS pump.
+        if let Some(mut stream) = stdin_stream {
+            let global = cmd_parent
+                .interp
+                .get()
+                .global_this_ref()
+                .expect("ReadableStream redirect implies JS event loop");
+            // SAFETY: borrow of the stdin slot scoped to this block; single-threaded.
+            let Writable::Pipe(pipe) = (unsafe { &mut (*subprocess).stdin }) else {
+                unreachable!("Writable::init returns Pipe for Stdio::ReadableStream")
+            };
+            let assign_err = pipe
+                .assign_to_stream(&mut stream, global)
+                .to_error()
+                .map(|err| {
+                    format!(
+                        "Failed to pipe ReadableStream to stdin: {}",
+                        err.fmt_string(global)
+                    )
+                    .into_bytes()
+                    .into_boxed_slice()
+                });
+            if let Some(msg) = assign_err {
+                // SAFETY: scoped `&mut` for the kill; `abort_after_failed_start`
+                // then consumes the allocation.
+                let _ = unsafe { (*subprocess).try_kill(SignalCode::SIGTERM as i32) };
+                Self::abort_after_failed_start(subprocess);
+                return Err(ShellErr::Custom(msg));
+            }
+        }
+
         // SAFETY: scoped access; `watch` does not re-enter the subprocess.
         match unsafe { (*subprocess).proc().watch() } {
             bun_sys::Result::Ok(()) => {}
@@ -863,6 +921,25 @@
 
     pub(crate) fn on_process_exit(&mut self, _: &Process, status: &Status, _: &Rusage) {
         log!("onProcessExit({:x})", std::ptr::from_mut(self) as usize);
+
+        // ReadableStream stdin: close the FileSink and mark stdin closed.
+        // Shell stdin is never `Stdio::Pipe`, so `Writable::Pipe` here is the
+        // ReadableStream case and `source` is the upstream source or `None`.
+        if let Writable::Pipe(pipe) = &self.stdin {
+            debug_assert!(!matches!(
+                *pipe.source.get(),
+                webcore::streams::SourceHandle::ShellWritable(_)
+            ));
+            let raw = pipe.as_ptr();
+            // SAFETY: `raw` is the canonical `*mut FileSink` (from
+            // `FileSink::create*` via `FileSinkPtr::adopt`); kept alive by the
+            // `FileSinkPtr` held in `self.stdin` for the duration.
+            unsafe { FileSink::on_attached_process_exit(raw, status) };
+            // SAFETY: cmd_parent backref resolves to the owning Cmd which
+            // outlives the subprocess (freed only in `Cmd::deinit`).
+            unsafe { self.cmd_parent.cmd_mut() }.buffered_input_close();
+        }
+
         let exit_code: Option<u8> = 'brk: {
             if let Status::Exited(exited) = &status {
                 break 'brk Some(exited.code);
@@ -889,6 +966,10 @@
             let cmd = unsafe { handle.cmd_mut() };
             if cmd.exit_code.is_none() {
                 cmd.on_exit(code.into());
+            } else if cmd.has_finished() {
+                // Stdout/stderr closing set `exit_code` first; the stdin close
+                // above may have just satisfied `has_finished()`.
+                cmd.on_exit(cmd.exit_code.unwrap());
             }
         }
     }
@@ -902,6 +983,8 @@
 pub enum WritableInitError {
     #[error("UnexpectedCreatingStdin")]
     UnexpectedCreatingStdin,
+    #[error("{0}")]
+    Sys(bun_sys::Error),
 }
 
 pub enum Writable {
@@ -956,11 +1039,11 @@
                             (*pipe_ptr).writer.with_mut(|w| w.start_with_current_pipe())
                         } {
                             bun_sys::Result::Ok(()) => {}
-                            bun_sys::Result::Err(_err) => {
+                            bun_sys::Result::Err(e) => {
                                 // SAFETY: pipe_ptr is live with refcount 1;
                                 // deref frees it.
                                 unsafe { FileSink::deref(pipe_ptr) };
-                                return Err(WritableInitError::UnexpectedCreatingStdin);
+                                return Err(WritableInitError::Sys(e));
                             }
                         }
 
@@ -1079,8 +1162,36 @@
                 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");
+                    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_ptr = FileSink::create(event_loop, fd);
+                    // SAFETY: `create` returns a freshly-boxed non-null FileSink
+                    // with refcount 1; sole reference.
+                    match unsafe {
+                        (*pipe_ptr)
+                            .writer
+                            .with_mut(|w| w.start((*pipe_ptr).fd.get(), true))
+                    } {
+                        bun_sys::Result::Ok(()) => {}
+                        bun_sys::Result::Err(e) => {
+                            // SAFETY: `pipe_ptr` is live with refcount 1; deref frees it.
+                            unsafe { FileSink::deref(pipe_ptr) };
+                            return Err(WritableInitError::Sys(e));
+                        }
+                    }
+                    // SAFETY: sole reference; `w.handle` is the `PosixStreamingWriter`'s PollOrFd.
+                    unsafe {
+                        (*pipe_ptr).writer.with_mut(|w| {
+                            if let Some(poll) = w.handle.get_poll() {
+                                poll.set_flag(bun_io::FilePollFlag::Socket);
+                            }
+                        });
+                    }
+                    // SAFETY: `create` returns non-null with one owned ref; `adopt` takes it over.
+                    Ok(Writable::Pipe(unsafe { FileSinkPtr::adopt(pipe_ptr) }))
                 }
                 Stdio::SocketFd => {
                     // The shell never uses this; rejected at i < 3 anyway.