Greyhorse process library

Running processes -- on this machine or on another one over SSH -- through a single contract, so the same code drives both.

Three levels, and only the first is expensive:

• a connection carries the lifecycle (start / stop / is_alive / active). For SSH that is a real TCP + crypto channel, opened once and reused; for the local machine it opens nothing and says so;
• a session is borrowed from a connection, cheaply, one per unit of work. This is the level to type-hint;
• a run comes back as Result[CompletedProcess, ProcessError] (exit code, stdout, stderr, or the reason there is no CompletedProcess at all), or as an interactive handle when the process must stay alive and be talked to.

A non-zero exit is a result, not an exception -- run()/sudo() return Ok(CompletedProcess) for that. Err(ProcessError...) is reserved for the cases where there is no CompletedProcess to report: the process never started, or timeout=/max_output= cut it off before it finished.

Installation

uv add greyhorse-process

The local road is the base install. Add the ssh extra for the SSH one -- asyncssh is a real TCP+crypto client, not carried by a service that only ever touches the local machine:

uv add "greyhorse-process[ssh]"

Without the extra the package still imports and the local road still works; greyhorse_process.ssh raises asyncssh's own ImportError if reached for directly.

Usage

Every snippet below matches the real API. Full, runnable programs live in examples/ and are executed by the test suite, so they cannot rot silently.

Local

from greyhorse_process.local import AsyncLocalConnection

async with AsyncLocalConnection().session() as session:
    result = (await session.run('echo hello | tr a-z A-Z', shell=True)).unwrap()
    print(result.stdout, result.returncode)

shell=False (the default) splits the command into an argv and starts the program directly, with no shell in between. shell=True sends the whole line to a shell, so pipes, redirections and quoting mean what they mean in a terminal. SyncLocalConnection is the blocking twin, same arguments.

SSH

There is no argv road here -- every entry point on this road takes a command STRING that the remote login shell interprets, unlike the local road's shell=False, which splits into an argv and starts the program directly with no shell in between at all. shell=False over SSH means "quote it so the remote shell cannot reinterpret it", not "no shell" -- a server hands whatever it receives to the user's login shell either way.

The practical hazard that follows is INTERPOLATION, not "shell versus exec": building a command string with an f-string, .format(), or concatenation that mixes in an untrusted value hands that value to the remote shell as syntax, not as one opaque piece of data. For example:

name = 'a b --force'
await session.run(f'echo {name}')  # BROKEN: three shell tokens, not one value

{name} lands unquoted, so the remote shell sees echo a b --force -- three separate words, one of them a flag echo never asked for -- instead of one value. This package does not escape interpolated values for the caller; there is no argv-based escape hatch on the SSH road the way shell=False gives one locally. shlex.quote is what a caller must reach for themselves before interpolating anything untrusted into a command string:

import shlex

await session.run(f'echo {shlex.quote(name)}')  # 'echo' 'a b --force' -- one value

SshConf.from_uri reads ssh://user:password@host:port so a deployment does not need four separate settings:

from greyhorse_process.config import SshConf
from greyhorse_process.ssh import AsyncSshConnection

conf = SshConf.from_uri('ssh://deploy@10.0.0.5:22')

async with AsyncSshConnection(conf).session() as session:
    result = (await session.run('systemctl status myapp')).unwrap()
    print(result.stdout)

The connection is the expensive part -- open it once and borrow a session per command, rather than opening one per run. verify_host_key defaults to True (asyncssh's own known_hosts checking); turning it off is a decision the caller writes down explicitly (SshConf(..., verify_host_key=False)), not a silent default.

sudo

result = (await session.sudo('systemctl restart myapp')).unwrap()

is run(..., sudo=True). With LocalConf(sudo_password=...) / SshConf(sudo_password=...) set, the password reaches sudo through a private pipe file descriptor (local) or sudo's own stdin (SSH) -- never through the child's environment or command line. With no password configured, sudo -n runs: a command that actually needs elevation fails immediately instead of hanging on a prompt nobody can answer.

Timeouts and output limits

result = await session.run('sleep 30', timeout=5.0)
# result.is_err() and isinstance(result.unwrap_err(), ProcessError.Timeout)

result = await session.run('some-noisy-command', max_output=1_000_000)
# result.is_err() and isinstance(result.unwrap_err(), ProcessError.OutputTooLarge)

Both default to None (no limit) and are purely additive over a plain run() call. timeout= bounds how long the process may run before it is killed and Err(ProcessError.Timeout(seconds=...)) comes back. max_output= bounds combined stdout+stderr, in bytes, before the same happens with Err(ProcessError.OutputTooLarge(limit=...)) -- read incrementally under the hood, specifically so that a command with no output limit of its own cannot exhaust memory before this library even gets a chance to stop it. Neither changes what a NORMAL run reports: a process that finishes within both limits returns Ok(CompletedProcess), identical to a call made without them.

timeout= is a bound on how long run() takes to RETURN, not a promise that it returns at the exact instant the deadline passes. On the local async road, the one shape that can overshoot is a still-alive tracked process whose output cannot be drained the moment it is killed -- most commonly a shell=True line with a surviving descendant still holding stdout/stderr open (sleep 20 & sleep 30, where killing the tracked shell does not touch the backgrounded sleep). There, cleanup falls back to a bounded drain-and-wait (local.py's _DRAIN_GRACE_SECONDS, five seconds) before giving up on that descendant and returning anyway -- measured directly: run('sleep 20 & sleep 30', shell=True, timeout=1.0) returns Err(ProcessError.Timeout(...)) at 1.0 + 5.0 seconds, not at 1.0. This is NOT true of every timeout, only that one shape (a still-alive process whose output genuinely cannot be drained before the process itself would be); an ordinary sleep 30 with no output at all, or a process that has already exited by the time the deadline fires, returns right at timeout= with no added delay. The local SYNC road does not have this overshoot for the equivalent case -- measured the same way, the identical command returns at 1.0 seconds flat on the sync road, because subprocess.Popen.wait() does not gate on the pipes being drained the way asyncio.subprocess. Process.wait() does, so killing the tracked process alone is already enough to unblock it; see local.py's _kill_and_drain and _kill_and_close_sync docstrings for the full reasoning behind why the two roads genuinely differ here rather than one of them being a bug. This is a documented, accepted asymmetry between the two roads, not something a caller needs to work around -- a caller relying on a hard deadline should still treat timeout= as "at least this long" on the async road, exactly as most operating-system-level timeouts already are.

Interactive processes

run()/sudo() wait for the process to finish. create_process is the other half -- the process stays alive inside the block, and lines go back and forth while it does:

async with session.create_process('cat') as proc:
    await proc.write_line('hello')
    print(await proc.read_line())

This is what a sudo prompt, an interactive installer, or a long-running job's progress output needs. The process is killed on the way out of the block if it is still running -- whether the caller read everything, read nothing, or stopped reading partway through -- and that cleanup is BOUNDED, not an unbounded wait: on the async road within _DRAIN_GRACE_SECONDS (local.py) of the kill, on the sync road immediately (Popen.wait() does not wait on unread pipe data the way the async road's process does). The one case this cannot fully close -- a shell=True command running more than one program on one line, where a descendant of the killed process inherited the same stdout/stderr descriptors and is still alive -- can leave that descendant running past the block's exit; see local.py's _kill_and_drain/_kill_and_close_sync docstrings for the full reasoning.

Error handling

run()/sudo() return Result[CompletedProcess, ProcessError] (greyhorse.result/greyhorse.error). Three outcomes, not two:

• Ok(CompletedProcess) with .ok is True -- exited zero;
• Ok(CompletedProcess) with .ok is False -- exited NON-zero. STILL an ANSWER, not a failure -- grep saying "no match" is not a malfunction, and this library never turns a non-zero exit into an Err;
• Err(ProcessError...) -- there is no CompletedProcess to report at all. ProcessError (greyhorse_process.abc) has five cases:
  • NotFound(command=...) -- local shell=False only: the argv never reached a shell, so there was no process to report a result for and the spawning layer itself could not find the program. Local shell=True and the ENTIRE SSH road (shell=True or shell=False) do NOT produce this: the command reaches an actual shell (local /bin/sh -c, or the remote login shell -- see "SSH" above for why shell=False still means a shell is involved there), which reports "not found" as its own exit status -- a normal Ok(CompletedProcess) with returncode == 127, not this case;
  • ConnectionFailed(details=...) -- SSH only: the connection dropped or the transport otherwise failed while THIS run's own create_process() was in flight (a sudo probe/auth round trip, or the command's channel itself);
  • Timeout(seconds=...) -- timeout= fired before the process finished. The process is always killed first;
  • OutputTooLarge(limit=...) -- max_output= fired: combined stdout+stderr crossed the cap before the process finished. Also always killed first;
  • Unexpected(details=...) -- anything else (a pipe-setup failure in the sudo/askpass path, say).

Matching:

from greyhorse_process.abc import ProcessError

match await session.run('some-command'):
    case Ok(completed) if completed.ok:
        ...
    case Ok(completed):
        ...  # non-zero exit; still an answer
    case Err(ProcessError.NotFound()):
        ...
    case Err(err):
        ...  # Timeout / OutputTooLarge / ConnectionFailed / Unexpected

What this does NOT cover -- and still raises exactly as before -- is connection LIFECYCLE, which this redesign does not touch: opening a connection (AsyncSshConnection.start()/.session(), or the implicit start() an async with connection: block performs) can raise asyncssh.Error on a bad address, a refused connection, or failed authentication. asyncssh.Error's own MRO is Exception, BaseException, object -- it is NOT an OSError subclass, so except OSError around connection-opening code does not catch these; a caller who wants to handle SSH-specific connection failures needs a separate except asyncssh.Error clause around session()/start(), distinct from the Err(ProcessError. ConnectionFailed(...)) a run() call returns for a connection that drops MID-command.

Runnable examples live in examples/ and are executed by the test suite, so they cannot rot:

uv run python examples/01_local_run.py

Development

uv sync
uv run pytest tests -q
uv run mypy greyhorse_process

Linting and formatting run from the REPOSITORY ROOT, where the shared ruff configuration lives:

ruff check exec/process
ruff format exec/process

Tests

Everything on the local road needs nothing but this machine. The SSH tests need a real server and are skipped -- with a reason, not an error -- unless one is pointed at:

export PROCESS_TEST_SSH_URI='ssh://user:password@host:22'
uv run pytest tests -q

That URI is the only knob: host, port, user and password all come out of it (SshConf.from_uri). Host-key verification is ON by default in the library and switched off only by the test helper, which talks to throwaway hosts.