running-process 3.0.0

A Rust-backed subprocess wrapper with split stdout/stderr streaming

running-process

running-process is a Rust-backed subprocess runtime with a thin Python API.

The pipe-backed API keeps stdout and stderr separate, preserves raw bytes until decode time, and defaults to UTF-8 with errors="replace" when you ask for text. The PTY API is separate because terminal sessions are chunk-oriented and should not be forced through line normalization.

Pipe-backed API

from running_process import RunningProcess

process = RunningProcess(
    ["python", "-c", "import sys; print('out'); print('err', file=sys.stderr)"]
)

process.wait()

print(process.stdout)          # stdout only
print(process.stderr)          # stderr only
print(process.combined_output) # combined compatibility view

Captured data values stay plain str | bytes. Live stream handles are exposed separately:

if process.stdout_stream.available():
    print(process.stdout_stream.drain())

Process priority is a first-class launch option:

from running_process import CpuPriority, RunningProcess

process = RunningProcess(
    ["python", "-c", "import time; time.sleep(1)"],
    nice=CpuPriority.LOW,
)

nice= behavior:

• accepts either a raw int niceness or a platform-neutral CpuPriority
• on Unix, it maps directly to process niceness
• on Windows, positive values map to below-normal or idle priority classes and negative values map to above-normal or high priority classes
• 0 leaves the default scheduler priority unchanged
• positive values are the portable default; negative values may require elevated privileges
• the enum intentionally stops at HIGH; there is no realtime tier

Available helpers:

• get_next_stdout_line(timeout)
• get_next_stderr_line(timeout)
• get_next_line(timeout) for combined compatibility reads
• drain_stdout()
• drain_stderr()
• drain_combined()
• stdout_stream.available()
• stderr_stream.available()
• combined_stream.available()

RunningProcess.run(...) supports common subprocess.run(...) style cases including:

• capture_output=True
• text=True
• encoding=...
• errors=...
• shell=True
• env=...
• nice=...
• stdin=subprocess.DEVNULL
• input=... in text or bytes form

Unsupported subprocess.run(...) kwargs now fail loudly instead of being silently ignored.

Expect API

expect(...) is available on both the pipe-backed and PTY-backed process APIs.

import re
import subprocess
from running_process import RunningProcess

process = RunningProcess(
    ["python", "-c", "print('prompt>'); import sys; print('echo:' + sys.stdin.readline().strip())"],
    stdin=subprocess.PIPE,
)

process.expect("prompt>", timeout=5, action="hello\n")
match = process.expect(re.compile(r"echo:(.+)"), timeout=5)
print(match.groups)

Supported action= forms:

• str or bytes: write to stdin
• "interrupt": send Ctrl-C style interrupt when supported
• "terminate"
• "kill"
• callable(match, process)

Pipe-backed expect(...) matches line-delimited output. If the child writes prompts without trailing newlines, use the PTY API instead.

PTY API

Use RunningProcess.pseudo_terminal(...) for interactive terminal sessions. It is chunk-oriented by design and preserves carriage returns and terminal control flow instead of normalizing it away.

from running_process import ExpectRule, RunningProcess

pty = RunningProcess.pseudo_terminal(
    ["python", "-c", "import sys; sys.stdout.write('name?'); sys.stdout.flush(); print('hello ' + sys.stdin.readline().strip())"],
    text=True,
    expect=[ExpectRule("name?", "world\n")],
    expect_timeout=5,
)

print(pty.output)

PTY behavior:

• accepts str and list[str] commands
• auto-splits simple string commands into argv when shell syntax is not present
• uses shell mode automatically when shell metacharacters are present
• keeps output chunk-buffered by default
• preserves \r for redraw-style terminal output
• supports write(...), read(...), drain(), available(), expect(...), resize(...), and send_interrupt()
• supports nice=... at launch
• supports restore_callback= and cleanup_callback= so terminal restoration and cleanup paths are explicit and testable
• supports interrupt_and_wait(...) for staged interrupt escalation
• supports wait_for_idle(...) with activity filtering
• exposes exit_reason, interrupt_count, interrupted_by_caller, and exit_status

There is also a compatibility alias: RunningProcess.psuedo_terminal(...).

You can also inspect the intended interactive launch semantics without launching a child:

from running_process import RunningProcess

spec = RunningProcess.interactive_launch_spec("console_isolated")
print(spec.ctrl_c_owner)
print(spec.creationflags)

Supported launch specs:

• pseudo_terminal
• console_shared
• console_isolated

For an actual launch, use RunningProcess.interactive(...):

process = RunningProcess.interactive(
    ["python", "-c", "print('hello from interactive mode')"],
    mode="console_shared",
    nice=5,
)
process.wait()

Abnormal Exits

By default, nonzero exits stay subprocess-like: you get a return code and can inspect exit_status.

process = RunningProcess(["python", "-c", "import sys; sys.exit(3)"])
process.wait()
print(process.exit_status)

If you want abnormal exits to raise, opt in:

from running_process import ProcessAbnormalExit, RunningProcess

try:
    RunningProcess.run(
        ["python", "-c", "import sys; sys.exit(3)"],
        capture_output=True,
        raise_on_abnormal_exit=True,
    )
except ProcessAbnormalExit as exc:
    print(exc.status.summary)

Notes:

• keyboard interrupts still raise KeyboardInterrupt
• kill -9 / SIGKILL is classified as an abnormal signal exit
• possible OOM conditions are exposed as a hint on exit_status.possible_oom
• OOM cannot be identified perfectly across platforms from exit status alone, so it is best-effort rather than guaranteed

Text and bytes

Pipe mode is byte-safe internally:

• invalid UTF-8 does not break capture
• text mode decodes with UTF-8 and errors="replace" by default
• binary mode returns bytes unchanged
• \r\n is normalized as a line break in pipe mode
• bare \r is preserved

PTY mode is intentionally more conservative:

• output is handled as chunks, not lines
• redraw-oriented \r is preserved
• no automatic terminal-output normalization is applied

Development

./install
./lint
./test

./install bootstraps the repo-local Rust toolchain into ./.cargo and ./.rustup.

./test runs the Rust tests, rebuilds the native extension with the unoptimized dev profile, runs the non-live Python tests, and then runs the @pytest.mark.live coverage that exercises real OS process and signal behavior.

If you want to invoke pytest directly, set RUNNING_PROCESS_LIVE_TESTS=1 and run uv run pytest -m live.

For direct Rust commands, prefer the repo-local trampolines:

./_cargo check --workspace
./_cargo fmt --all --check
./_cargo clippy --workspace --all-targets -- -D warnings

Notes

• stdout and stderr are no longer merged by default.
• combined_output exists for compatibility when you need the merged view.
• RunningProcess(..., use_pty=True) is no longer the preferred path; use RunningProcess.pseudo_terminal(...) for PTY sessions.
• The test suite checks that running_process.__version__, package metadata, and manifest versions stay in sync.