feat(rust): add exact-argv streaming (#190)

rust/README.md

@@ -80,6 +80,30 @@ Parity guarantees with the JavaScript implementation:
   signal) or `stream.kill_with(signal)` (explicit override); dropping the stream
   (e.g. `break`) stops the process too.
 
+`StreamingRunner::new(command)` interprets a completed command string with the
+platform shell. When argument boundaries must be preserved exactly, pass the
+executable and arguments separately with `from_argv`:
+
+```rust,no_run
+use command_stream::StreamingRunner;
+
+#[tokio::main]
+async fn main() -> command_stream::Result<()> {
+    let runner = StreamingRunner::from_argv(
+        "my-program",
+        ["argument with spaces", "literal&metacharacters"],
+    );
+    let result = runner.collect().await?;
+
+    assert!(result.is_success());
+    Ok(())
+}
+```
+
+The exact-argv form bypasses `/bin/sh -c` and `cmd.exe /c`, so it does not
+require shell-specific quoting. It also accepts OS-native executable and
+argument values such as `PathBuf` and `OsString`.
+
 ## Command Line
 
 The crate also builds a `command-stream` binary:

rust/changelog.d/20260811_110000_exact_streaming_argv.md

@@ -0,0 +1,7 @@
+---
+bump: minor
+---
+
+### Added
+
+- Add `StreamingRunner::from_argv` for shell-free streaming execution with exact executable and argument boundaries on every platform.

rust/src/stream.rs

@@ -55,6 +55,7 @@
 //! ```
 
 use std::collections::HashMap;
+use std::ffi::OsString;
 use std::path::PathBuf;
 use std::process::Stdio;
 use std::time::Duration;
@@ -86,19 +87,50 @@ pub enum OutputChunk {
 
 /// A streaming process runner that allows async iteration over output
 pub struct StreamingRunner {
-    command: String,
+    command: StreamingCommand,
     cwd: Option<PathBuf>,
     env: Option<HashMap<String, String>>,
     stdin_content: Option<String>,
     kill_signal: String,
     exit_pump_grace_ms: u64,
 }
 
+#[derive(Clone)]
+enum StreamingCommand {
+    Shell(String),
+    Argv {
+        program: OsString,
+        args: Vec<OsString>,
+    },
+}
+
 impl StreamingRunner {
-    /// Create a new streaming runner
+    /// Create a streaming runner for a command string interpreted by the
+    /// platform shell.
     pub fn new(command: impl Into<String>) -> Self {
+        Self::with_command(StreamingCommand::Shell(command.into()))
+    }
+
+    /// Create a streaming runner for an executable and exact argument vector.
+    ///
+    /// Unlike [`StreamingRunner::new`], this constructor bypasses the platform
+    /// shell. Argument boundaries are therefore preserved on every platform,
+    /// including Windows, without requiring shell-specific quoting.
+    pub fn from_argv<P, I, S>(program: P, args: I) -> Self
+    where
+        P: Into<OsString>,
+        I: IntoIterator<Item = S>,
+        S: Into<OsString>,
+    {
+        Self::with_command(StreamingCommand::Argv {
+            program: program.into(),
+            args: args.into_iter().map(Into::into).collect(),
+        })
+    }
+
+    fn with_command(command: StreamingCommand) -> Self {
         StreamingRunner {
-            command: command.into(),
+            command,
             cwd: None,
             env: None,
             stdin_content: None,
@@ -279,22 +311,34 @@ impl Drop for OutputStream {
 
 /// Run a streaming process and send output to the channel
 async fn run_streaming_process(
-    command: String,
+    command: StreamingCommand,
     cwd: Option<PathBuf>,
     env: Option<HashMap<String, String>>,
     stdin_content: Option<String>,
     exit_pump_grace_ms: u64,
     tx: mpsc::Sender<OutputChunk>,
     mut kill_rx: mpsc::UnboundedReceiver<String>,
 ) -> Result<()> {
-    trace_lazy("StreamingRunner", || format!("Starting: {}", command));
+    trace_lazy("StreamingRunner", || match &command {
+        StreamingCommand::Shell(command) => format!("Starting: {command}"),
+        StreamingCommand::Argv { program, args } => {
+            format!("Starting argv command: {program:?} {args:?}")
+        }
+    });
 
-    let shell = find_available_shell();
-    let mut cmd = Command::new(&shell.cmd);
-    for arg in &shell.args {
-        cmd.arg(arg);
-    }
-    cmd.arg(&command);
+    let mut cmd = match command {
+        StreamingCommand::Shell(command) => {
+            let shell = find_available_shell();
+            let mut cmd = Command::new(&shell.cmd);
+            cmd.args(&shell.args).arg(command);
+            cmd
+        }
+        StreamingCommand::Argv { program, args } => {
+            let mut cmd = Command::new(program);
+            cmd.args(args);
+            cmd
+        }
+    };
 
     // Configure stdio
     if stdin_content.is_some() {

rust/tests/stream.rs

@@ -20,6 +20,29 @@ async fn test_streaming_runner_with_stdin() {
     assert!(result.stdout.contains("test input"));
 }
 
+/// An executable and its arguments must reach the OS as separate values rather
+/// than being reconstructed as a command string for the platform shell.
+#[tokio::test]
+async fn test_streaming_runner_preserves_exact_argv() {
+    let test_executable = std::env::current_exe().unwrap();
+    let shell_metacharacters = "argument with spaces & | < > ^ % ! ' \"";
+    let runner =
+        StreamingRunner::from_argv(test_executable, ["--list", "--skip", shell_metacharacters]);
+
+    let result = runner.collect().await.unwrap();
+
+    assert!(
+        result.is_success(),
+        "exact-argv child failed: {}",
+        result.stderr
+    );
+    assert!(
+        result.stdout.contains("test_streaming_runner_basic"),
+        "exact-argv child did not produce its test list: {}",
+        result.stdout
+    );
+}
+
 #[tokio::test]
 async fn test_output_stream_chunks() {
     let runner = StreamingRunner::new("echo chunk1 && echo chunk2");