Execution runtime

SafeCmd.run executes curated commands asynchronously with predictable capture and echo semantics and returns a structured CommandResult:

• stdout and stderr are captured by default. Set capture=False to stream only; the result will carry None for output fields.
• echo=True tees stdout/stderr to the parent process while still capturing them when capture=True.
• Pass an ExecutionContext via the context parameter to override execution details:
• env overlays key/value pairs on top of the current environment without mutating os.environ; use it to pass per-command settings.
• cwd sets the working directory for the subprocess when provided.
• cancel_grace controls how long Cuprum waits after SIGTERM (termination signal) before escalating to SIGKILL (kill signal).
• stdout_sink and stderr_sink route echoed output to alternative text streams when echo=True.
• encoding and errors configure how captured output is decoded; defaults are "utf-8" with "replace".
• exit_code, pid, and ok on the CommandResult make it easy to branch on success.

from cuprum import ECHO, ExecutionContext, sh


async def greet() -> None:
    cmd = sh.make(ECHO)("-n", "hello runtime")
    ctx = ExecutionContext(env={"GREETING": "1"})
    result = await cmd.run(echo=True, context=ctx)
    if not result.ok:
        raise RuntimeError(f"echo failed: {result.exit_code}")
    print(result.stdout)

If the awaiting task is cancelled while a command is running, Cuprum sends SIGTERM to the subprocess, waits for a short grace period, and then escalates to SIGKILL to ensure the child process is cleaned up.

For scripts or contexts where async/await is not available, use run_sync():

from cuprum import ECHO, ExecutionContext, sh


def greet() -> None:
    cmd = sh.make(ECHO)("-n", "hello sync")
    ctx = ExecutionContext(env={"GREETING": "1"})
    result = cmd.run_sync(echo=True, context=ctx)
    if not result.ok:
        raise RuntimeError(f"echo failed: {result.exit_code}")
    print(result.stdout)

run_sync() accepts the same parameters as run() and returns an identical CommandResult. It drives the event loop internally via asyncio.run().