Add Command.prefer_local usage example to docs/commands.md

Demonstrates preferring a local project directory over PATH for bare-name
resolution, including stacking multiple directories in priority order and
the bare-name-only boundary, consistent with the prefer_local docs from T-072.

Author: ZelAnton

docs/streaming.md

@@ -291,9 +291,21 @@ assert finished.exited_zero
 ```
 
 `ProcessStdin` is fully awaitable: `await write(bytes)`, `write_line(str)`
-(newline + flush), `flush()`, and `close()` (EOF). `take_stdin()` **raises**
-`ProcessError` if the `Command` didn't `keep_stdin_open()` or the writer was
-already taken — so a missing setup fails right here, not later on a `None`.
+(newline + flush), `send_control(str)`, `flush()`, and `close()` (EOF).
+`send_control()` accepts exactly one recognized control character and writes
+the mapped control byte to the child's stdin pipe: for example,
+`await stdin.send_control("c")` writes Ctrl-C (`\x03`) and
+`await stdin.send_control("d")` writes Ctrl-D (`\x04`). Invalid input raises
+`ValueError`.
+
+This is a byte in a normal pipe, not a terminal signal. It only affects
+children that read and interpret that byte from stdin; real terminal semantics
+such as SIGINT/SIGTSTP delivery require a pseudoterminal, which `processkit`
+does not provide yet.
+
+`take_stdin()` **raises** `ProcessError` if the `Command` didn't
+`keep_stdin_open()` or the writer was already taken — so a missing setup fails
+right here, not later on a `None`.
 
 **Avoid the full-duplex deadlock.** A child's stdout pipe has a finite OS
 buffer; once it fills, the child blocks *writing* stdout until something reads

src/processkit/_processkit.pyi

@@ -456,6 +456,13 @@ class ProcessStdin:
 
     async def write(self, data: ReadableBuffer) -> None: ...
     async def write_line(self, line: str) -> None: ...
+    async def send_control(self, control: str) -> None:
+        """Write one mapped control byte, e.g. ``"c"`` -> Ctrl-C (``\\x03``).
+
+        This writes a byte to the child's stdin pipe, not a terminal signal;
+        real SIGINT/SIGTSTP delivery requires a pseudoterminal.
+        """
+
     async def flush(self) -> None: ...
     async def close(self) -> None: ...
 

src/running.rs

@@ -8,7 +8,7 @@ use processkit::OutputEvents as PkOutputEvents;
 use processkit::ProcessStdin as PkProcessStdin;
 use processkit::RunningProcess as PkRunningProcess;
 use processkit::StdoutLines as PkStdoutLines;
-use pyo3::exceptions::{PyOSError, PyStopAsyncIteration};
+use pyo3::exceptions::{PyOSError, PyStopAsyncIteration, PyValueError};
 use pyo3::prelude::*;
 use tokio::sync::Mutex;
 
@@ -53,6 +53,38 @@ impl PyProcessStdin {
         })
     }
 
+    /// Send a single control byte to the child's stdin.
+    fn send_control<'py>(
+        &self,
+        py: Python<'py>,
+        control: String,
+    ) -> PyResult<Bound<'py, PyAny>> {
+        let mut chars = control.chars();
+        let c = chars.next().ok_or_else(|| {
+            PyValueError::new_err("send_control() requires exactly one control character")
+        })?;
+        if chars.next().is_some() {
+            return Err(PyValueError::new_err(
+                "send_control() requires exactly one control character",
+            ));
+        }
+
+        let stdin = self.inner.clone();
+        pyo3_async_runtimes::tokio::future_into_py(py, async move {
+            let mut guard = stdin.lock().await;
+            let writer = guard
+                .as_mut()
+                .ok_or_else(|| PyOSError::new_err("stdin is closed"))?;
+            writer.send_control(c).await.map_err(|err| {
+                if err.kind() == std::io::ErrorKind::InvalidInput {
+                    PyValueError::new_err(err.to_string())
+                } else {
+                    PyErr::from(err)
+                }
+            })
+        })
+    }
+
     /// Flush buffered writes to the child.
     fn flush<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
         let stdin = self.inner.clone();

tests/test_pipelines.py

@@ -136,7 +136,10 @@ def test_pipeline_stage_timeout_kills_its_whole_subtree(pid_file: pathlib.Path)
     # stage's whole subtree, including a grandchild it forks off -- previously
     # the stage's own kill reached only its direct child, so a forking stage's
     # grandchild survived, kept the pipe open, and stalled the downstream stage.
-    spawner = spawn_grandchild_command(pid_file).timeout(0.3)
+    # Keep this comfortably above Windows process-start latency under xdist:
+    # the timeout must fire after the grandchild PID is observable, otherwise
+    # the probe races the very timeout whose teardown behavior it is testing.
+    spawner = spawn_grandchild_command(pid_file).timeout(2.0)
     downstream = Command(PY, ["-c", "import sys; sys.stdin.read()"])
     pipe = spawner | downstream
 

tests/test_streaming.py

@@ -43,6 +43,14 @@
     "import sys; [(sys.stdout.write(line.upper()), sys.stdout.flush()) for line in sys.stdin]"
 )
 
+# Echoes the first raw stdin byte to stdout.
+_ECHO_ONE_STDIN_BYTE = (
+    "import sys; "
+    "data = sys.stdin.buffer.read(1); "
+    "sys.stdout.buffer.write(data); "
+    "sys.stdout.buffer.flush()"
+)
+
 # stdout + stderr on both streams.
 _BOTH_STREAMS = (
     "import sys; "
@@ -223,6 +231,32 @@ async def scenario() -> list[str]:
     assert asyncio.run(scenario()) == ["FROM-BYTEARRAY", "FROM-MEMORYVIEW"]
 
 
+def test_interactive_stdin_send_control_writes_control_byte() -> None:
+    async def scenario() -> bytes:
+        proc = await Command(PY, ["-c", _ECHO_ONE_STDIN_BYTE]).keep_stdin_open().astart()
+        stdin = proc.take_stdin()
+        await stdin.send_control("d")
+        await stdin.close()
+        result = await proc.aoutput_bytes()
+        return result.stdout
+
+    assert asyncio.run(scenario()) == b"\x04"
+
+
+def test_interactive_stdin_send_control_rejects_invalid_argument() -> None:
+    async def scenario() -> None:
+        proc = await Command(PY, ["-c", _ECHO_ONE_STDIN_BYTE]).keep_stdin_open().astart()
+        stdin = proc.take_stdin()
+        with pytest.raises(ValueError):
+            await stdin.send_control("0")
+        with pytest.raises(ValueError):
+            await stdin.send_control("cc")
+        await stdin.close()
+        await proc.aoutcome()
+
+    asyncio.run(scenario())
+
+
 def test_take_stdin_is_once() -> None:
     # The first take hands over the handle; a second take raises (consumed).
     async def scenario() -> None: