Input, environment, and deadlines

Input, environment, and directory

StdinInput(text=...) encodes text with the context's encoding; StdinInput(data=...) passes bytes. Give one, not both, and for input too large to hold in memory stream it instead. ExecutionContext.env overlays the live parent environment by default, so an empty mapping still inherits everything; cwd sets the child's working directory. Everything about how a command runs goes to run_sync() or run(), never to the builder, which rejects cwd=, env=, timeout=, and the other execution-option names with a TypeError rather than turning them into child flags.

import sys

from cuprum import Program, ExecutionContext, ProgramCatalogue, StdinInput, sh

catalogue = ProgramCatalogue.from_programs(sys.executable, name="input")
python = sh.make(Program(sys.executable), catalogue=catalogue)
command = python(
    "-c", "import os,sys; print(os.getenv('CUPRUM_EXAMPLE'), sys.stdin.read())"
)
result = command.run_sync(
    context=ExecutionContext(env={"CUPRUM_EXAMPLE": "ready"}),
    stdin=StdinInput(text="hello"),
    timeout=5,
)
assert result.stdout == "ready hello\n"

Stream input from a producer

StdinInput holds the whole payload in memory. For input that is large, or that is produced as the run goes, pass stdin=StdinStream(chunks=...) with an async iterable instead:

import sys
from collections.abc import AsyncIterator

from cuprum import Program, ProgramCatalogue, StdinStream, sh

catalogue = ProgramCatalogue.from_programs(sys.executable, name="stream-input")
python = sh.make(Program(sys.executable), catalogue=catalogue)


async def rows() -> AsyncIterator[str]:
    for number in range(1, 1001):
        yield f"row {number}\n"


count_lines = python("-c", "import sys; print(sum(1 for _ in sys.stdin))")
result = count_lines.run_sync(stdin=StdinStream(chunks=rows()))
assert result.stdout == "1000\n"

Cuprum pulls one chunk, writes it, and waits for the write to drain before pulling the next, so a slow reader slows the producer down rather than letting input pile up in memory. A drained write is not proof that the child has read the chunk: some bytes can still sit in the transport buffer or the OS pipe. Each chunk is held in full while it is written, so yield chunks of a bounded size when peak memory matters.

• Chunk types. A str chunk is encoded with the context's encoding and errors, and a character split across two chunks is still encoded correctly. A bytes chunk is written as it is.
• Producer failure. If the producer raises, or yields something that is neither str nor bytes, Cuprum terminates the child and raises StdinSourceError, with the original exception as __cause__. Cancellation propagates unchanged.
• A child that stops reading. Closing stdin early, as head does, is not an error. The run continues to the child's exit code, and the cut-short write is reported as a stdin_error event.
• Clean-up. On every exit path, Cuprum closes the child's stdin and then calls the producer's aclose(), when it has one. Both waits are bounded: a producer whose aclose() takes longer is cancelled, and one that ignores the cancellation may still be cleaning up after the run returns. A generator needs no closing by the caller.

Where the child's output should go to a file rather than the parent, see Send a stream to a file or the parent.

Compose the environment

EnvMode.OVERLAY is the default policy: values resolve against the live os.environ when each child is spawned, not against a snapshot, so a variable set after a scope starts is still seen. OVERLAY and INHERIT compose: each keeps the mode an outer scope chose and layers its values over the inherited overlay. REPLACE alone draws a boundary: it starts from an empty environment, applies only its mapping, and discards every outer overlay. The UNSET value removes a variable from a composed child environment, which an overlay could not do before.

import os
import sys

from cuprum import Program, ProgramCatalogue, sh
from cuprum.context import EnvMode, UNSET, env

catalogue = ProgramCatalogue.from_programs(sys.executable, name="env-modes")
python = sh.make(Program(sys.executable), catalogue=catalogue)
show = "-c", "import os; print(os.getenv('CUPRUM_DEMO'), os.getenv('CUPRUM_KEEP'))"

os.environ["CUPRUM_KEEP"] = "kept"
with env({"CUPRUM_DEMO": "overlaid"}, CUPRUM_DEMO="won"):
    assert python(*show).run_sync().stdout == "won kept\n"
with env({"CUPRUM_DEMO": UNSET, "CUPRUM_KEEP": UNSET}):
    assert python(*show).run_sync().stdout == "None None\n"
with env({"CUPRUM_DEMO": "only"}, mode=EnvMode.REPLACE):
    assert python(*show).run_sync().stdout == "only None\n"

Precedence, lowest first: the process's os.environ, read at spawn time; ambient scoped overlays, inner values winning, with a nested REPLACE discarding everything outside it; then the per-call ExecutionContext.env mapping, whose values win overall and whose env_mode can also be REPLACE. os.environ itself is never mutated. On POSIX the child's PATH is what resolves a bare program name, so a replacement policy that omits it can make the name fail to resolve; prefer an absolute program path, or an executable binding, with a deliberately minimal environment. The effective mode reaches every ExecEvent and labels the execution and failure counters, so a non-zero exit under a replacement policy is distinguishable from an ordinary one.

Set a deadline

Pass timeout to run() or run_sync() and catch TimeoutExpired, which subclasses the built-in TimeoutError. On expiry Cuprum asks the child to terminate, waits cancel_grace seconds, then kills it. Output captured before the deadline stays on the exception, as empty strings rather than None when capture is on.

import sys

from cuprum import ExecutionContext, Program, ProgramCatalogue, TimeoutExpired, sh

catalogue = ProgramCatalogue.from_programs(sys.executable, name="slow")
python = sh.make(Program(sys.executable), catalogue=catalogue)
slow = python("-c", "import time; time.sleep(30)")

try:
    slow.run_sync(timeout=0.5, context=ExecutionContext(cancel_grace=0.2))
except TimeoutExpired as exc:
    assert exc.timeout == 0.5
    assert exc.output == "" and exc.stderr == ""
else:
    raise AssertionError("the command should have timed out")

Cancelling an async run starts the same teardown. Catch the timeout separately from a non-zero exit: one means the deadline passed, the other that the child finished and failed.

Where a deadline comes from

Cuprum uses the first of these that is set:

1. The timeout argument to run() or run_sync().
2. ExecutionContext.timeout.
3. A ScopeConfig(timeout=...) default from the enclosing scoped() block, which applies to every command run inside it.

import sys

from cuprum import Program, ProgramCatalogue, ScopeConfig, TimeoutExpired, scoped, sh

catalogue = ProgramCatalogue.from_programs(sys.executable, name="timeouts")
python = sh.make(Program(sys.executable), catalogue=catalogue)
command = python("-c", "import time; print('started', flush=True); time.sleep(30)")

with scoped(ScopeConfig(timeout=1.0)):
    try:
        command.run_sync()
    except TimeoutExpired as exc:
        assert exc.timeout == 1.0
        # A slow start can expire before the child prints anything.
        assert exc.output in {"", "started\n"}
    else:
        raise AssertionError("the scope timeout should have expired")

Timeouts must be finite and non-negative; None means no limit. Zero or less is treated as a deadline already passed. A pipeline enforces one deadline for the whole run and reports its expiry once per stage. Every expiry also writes a structured WARNING to the cuprum.timeout logger, whether or not a hook is observing.

Reclaim the descendants a child leaves behind

A timeout terminates the process Cuprum spawned, but only that process. When the child has spawned descendants of its own, they are not signalled, and one that holds an inherited pipe open keeps the run waiting out its end-of-file window before TimeoutExpired is raised, while the work the deadline was meant to stop carries on. ExecutionContext.process_group takes a ProcessGroupPolicy. The default, INHERIT, tears down the direct child only; OWN_GROUP makes the child lead a new session and process group, so teardown signals the whole group and every descendant still in it:

import sys

from cuprum import ExecutionContext, ProcessGroupPolicy, Program, ProgramCatalogue, sh

catalogue = ProgramCatalogue.from_programs(sys.executable, name="process-group")
command = sh.make(Program(sys.executable), catalogue=catalogue)(
    "-c", "import os, sys; print(os.getpgrp() == os.getpid())"
)
owned = ExecutionContext(process_group=ProcessGroupPolicy.OWN_GROUP)
assert owned.process_group is ProcessGroupPolicy.OWN_GROUP
assert command.run_sync(context=owned).stdout == "True\n"
assert ExecutionContext().process_group is ProcessGroupPolicy.INHERIT

The last assertion is the compatibility promise: a context that says nothing keeps the child in the caller's group, so no existing run is affected. Under OWN_GROUP teardown signals exactly the group the spawn created, never the caller's, with the same two-phase grace: SIGTERM, cancel_grace, then SIGKILL. Settlement is then decided by the group rather than the direct child, so a descendant the child left behind is still reached after the child has exited.

Two trade-offs are worth knowing before opting in. Session leadership detaches the child from the controlling terminal: a terminal SIGINT or SIGHUP stops at the run rather than reaching the child, and a child that opens /dev/tty fails for want of one. And the guarantee is bounded by the kernel: a descendant that calls setsid() or setpgid() leaves the group and is out of reach, as are ancestors and siblings; containing a process that actively escapes needs a supervisor, a cgroup, or a container. OWN_GROUP raises ValueError at spawn on Windows, which has no POSIX process groups and whose Job Objects cannot be assigned atomically with process creation; INHERIT works everywhere. A teardown outcome Cuprum could not resolve is reported as a group_teardown event and a WARNING, so a teardown that left a process alive is never silent.