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,17 @@
 
         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);
+                }
+                return Err(ShellErr::Sys(e.to_shell_system_error()));
+            }
         };
         let stdout = Readable::init(
             OutKind::Stdout,
@@ -772,9 +789,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 +816,40 @@
             }
         }
 
+        // 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 assign_err: Option<Box<[u8]>> = 'assign: {
+                let Some(global) = cmd_parent.interp.get().global_this_ref() else {
+                    break 'assign None;
+                };
+                // SAFETY: borrow of the stdin slot scoped to this block; single-threaded.
+                let Writable::Pipe(pipe) = (unsafe { &mut (*subprocess).stdin }) else {
+                    break 'assign None;
+                };
+                match pipe.assign_to_stream(&mut stream, global).to_error() {
+                    None => None,
+                    Some(err) => {
+                        use std::io::Write;
+                        let mut msg = Vec::<u8>::new();
+                        let _ = write!(
+                            &mut msg,
+                            "Failed to pipe ReadableStream to stdin: {}",
+                            err.fmt_string(global)
+                        );
+                        Some(msg.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 +914,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 +959,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 +976,8 @@
 pub enum WritableInitError {
     #[error("UnexpectedCreatingStdin")]
     UnexpectedCreatingStdin,
+    #[error("{0}")]
+    Sys(bun_sys::Error),
 }
 
 pub enum Writable {
@@ -1079,8 +1155,28 @@
                 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: `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.

test/js/bun/shell/bunshell.test.ts

@@ -3085,6 +3085,125 @@ test("stdin redirect from a Uint8Array sends the bytes captured when the command
   expect(result.exitCode).toBe(0);
 }, 60_000);
 
+describe("redirect stdin from ReadableStream", () => {
+  // External command: on POSIX `cat` is external, on Windows it's a builtin,
+  // so use a spawned Bun child everywhere.
+  const childPump = `process.stdout.write(await Bun.stdin.text())`;
+
+  test.concurrent("native source (subprocess stdout)", async () => {
+    // #18262 repro: a Bun.spawn stdout ReadableStream (backed by a native
+    // ByteStream) piped into a shell command's stdin via the SinkHandle path.
+    await using proc = Bun.spawn({
+      cmd: [BUN, "-e", "process.stdout.write('hello from stream')"],
+      env: bunEnv,
+      stdout: "pipe",
+    });
+    const out = await $`${BUN} -e ${childPump} < ${proc.stdout}`.env(bunEnv).nothrow().quiet();
+    expect({ stdout: out.stdout.toString(), exitCode: out.exitCode }).toEqual({
+      stdout: "hello from stream",
+      exitCode: 0,
+    });
+  });
+
+  test.concurrent("native source, large payload with backpressure", async () => {
+    const size = 256 * 1024;
+    await using proc = Bun.spawn({
+      cmd: [BUN, "-e", `process.stdout.write(Buffer.alloc(${size}, 'x'))`],
+      env: bunEnv,
+      stdout: "pipe",
+    });
+    const out = await $`${BUN} -e ${childPump} < ${proc.stdout}`.env(bunEnv).nothrow().quiet();
+    expect(out.stdout.equals(Buffer.alloc(size, "x"))).toBe(true);
+    expect(out.exitCode).toBe(0);
+  });
+
+  test.concurrent("native file source (Bun.file().stream())", async () => {
+    using dir = tempDir("shell-stream-file", { "input.txt": "file-content" });
+    const stream = Bun.file(join(String(dir), "input.txt")).stream();
+    const out = await $`${BUN} -e ${childPump} < ${stream}`.env(bunEnv).nothrow().quiet();
+    expect({ stdout: out.stdout.toString(), exitCode: out.exitCode }).toEqual({
+      stdout: "file-content",
+      exitCode: 0,
+    });
+  });
+
+  test.concurrent("JS-backed multi-chunk stream", async () => {
+    const stream = new ReadableStream({
+      async pull(c) {
+        c.enqueue(new TextEncoder().encode("chunk1 "));
+        await Bun.sleep(0);
+        c.enqueue(new TextEncoder().encode("chunk2"));
+        c.close();
+      },
+    });
+    const out = await $`${BUN} -e ${childPump} < ${stream}`.env(bunEnv).nothrow().quiet();
+    expect({ stdout: out.stdout.toString(), exitCode: out.exitCode }).toEqual({
+      stdout: "chunk1 chunk2",
+      exitCode: 0,
+    });
+  });
+
+  test.concurrent("child exits while stream is still producing", async () => {
+    const { promise: cancelled, resolve: onCancel } = Promise.withResolvers<true>();
+    const stream = new ReadableStream({
+      async pull(c) {
+        c.enqueue(new TextEncoder().encode("x"));
+        await Bun.sleep(0);
+      },
+      cancel() {
+        onCancel(true);
+      },
+    });
+    // Child reads 3 bytes then exits; the shell must cancel the stream and
+    // finish the command with the child's exit status.
+    const child = `
+      const b = Buffer.alloc(3);
+      let n = 0;
+      while (n < 3) { const r = require('fs').readSync(0, b, n, 3 - n); if (r <= 0) break; n += r; }
+      process.stdout.write(b.subarray(0, n));
+    `;
+    const out = await $`${BUN} -e ${child} < ${stream}`.env(bunEnv).nothrow().quiet();
+    expect({ stdout: out.stdout.toString(), exitCode: out.exitCode }).toEqual({
+      stdout: "xxx",
+      exitCode: 0,
+    });
+    expect(await cancelled).toBe(true);
+  });
+
+  test("stdout/stderr redirect throws", async () => {
+    const s1 = new ReadableStream({ pull: c => c.close() });
+    await expect(runWithErrorPromise(() => $`${BUN} -e 0 > ${s1}`)).resolves.toThrow(
+      /ReadableStream cannot be used for stdout or stderr/,
+    );
+    const s2 = new ReadableStream({ pull: c => c.close() });
+    await expect(runWithErrorPromise(() => $`${BUN} -e 0 2> ${s2}`)).resolves.toThrow(
+      /ReadableStream cannot be used for stdout or stderr/,
+    );
+  });
+
+  test("locked stream throws", async () => {
+    const s = new ReadableStream({ pull: c => (c.enqueue(new Uint8Array([65])), c.close()) });
+    const r = s.getReader();
+    await expect(runWithErrorPromise(() => $`${BUN} -e 0 < ${s}`)).resolves.toThrow(/already been used/);
+    r.releaseLock();
+  });
+
+  test("disturbed stream throws", async () => {
+    const s = new ReadableStream({ pull: c => (c.enqueue(new Uint8Array([65])), c.close()) });
+    const r = s.getReader();
+    await r.read();
+    r.releaseLock();
+    await expect(runWithErrorPromise(() => $`${BUN} -e 0 < ${s}`)).resolves.toThrow(/already been used/);
+  });
+
+  test("redirect to a builtin throws", async () => {
+    const s = new ReadableStream({ pull: c => (c.enqueue(new Uint8Array([65])), c.close()) });
+    await expect(runWithErrorPromise(() => $`echo < ${s}`)).resolves.toThrow(
+      /ReadableStream cannot be redirected to a builtin command/,
+    );
+  });
+});
+
 describe("stdin redirect from a zero-length buffer delivers EOF to the spawned command", () => {
   // A spawned command reading stdin (cat) must see EOF when the redirect
   // source is an empty ArrayBuffer/TypedArray, same as an empty Blob.