kernelfoundry.eval_pipeline.utils.subprocess

Functions

create_ForwardToStreamReaderProtocol(reader, ...)	Creates a ForwardToStreamReaderProtocol instance.
force_terminate(proc[, wait_after_terminate])	Helper function for forcefully terminating a subprocess.
monitor_output_stream(reader, proc, ...)	Monitors the output stream of a subprocess and applies monitor functions to each line.
robust_subprocess_run(cmd[, timeout, ...])	Runs a subprocess command that may not well behave and hang

Classes

EndMarkerMonitor(marker[, grace_period])
ForwardToStreamReaderProtocol(reader, limit, ...)
PytestEndMonitor()

async kernelfoundry.eval_pipeline.utils.subprocess.robust_subprocess_run(cmd: list[str] | str, timeout: float | None = None, output_inactivity_timeout: float | None = None, end_marker: str | None = None, **kwargs) → tuple[CompletedProcess, str | None]

Runs a subprocess command that may not well behave and hang

Parameters:

• cmd – List of command arguments to run

• timeout – Overall timeout for the subprocess

• output_inactivity_timeout – If set, terminate the process if no output is received for this many seconds

• end_marker – Controls end-of-process monitoring. "pytest" – terminate after the pytest summary line appears (10 s grace). Any other string – terminate when that literal string appears in output (2 s grace). None – no early termination monitoring.

• **kwargs – Additional keyword arguments to pass to subprocess.Popen

Returns:

A tuple of (CompletedProcess, termination_message)

async kernelfoundry.eval_pipeline.utils.subprocess.force_terminate(proc: Process, wait_after_terminate: float = 2.0)

Helper function for forcefully terminating a subprocess. :param proc: The subprocess to terminate. :param wait_after_terminate: Time to wait after sending terminate signal before killing the process.