swarmauri_evaluator_subprocess 0.3.0.dev33

A subprocess-based evaluator for executing and measuring program performance in isolated subprocesses

---

Swarmauri Evaluator Subprocess

SubprocessEvaluator executes programs inside sandboxed subprocesses while enforcing CPU, memory, and file size quotas. It captures stdout/stderr, tracks exit codes, and returns a normalized score plus structured metadata describing each run.

Highlights

• Apply CPU timeouts, memory ceilings, file size limits, and process count caps before user code starts (resource.setrlimit).
• Automatically choose the appropriate command: launch executables directly or wrap Python and shell scripts with the correct interpreter.
• Compare stdout against an expected_output string and annotate mismatches in the result metadata.
• Record execution context (command, args, working_dir) alongside collected streams for easy debugging.
• Aggregate multiple runs with reason counts, timeout rates, and success rates via aggregate_scores.

Installation

Pick the tool that matches your workflow:

# pip
pip install swarmauri_evaluator_subprocess

# Poetry
poetry add swarmauri_evaluator_subprocess

# uv
uv add swarmauri_evaluator_subprocess

Quickstart

The example below writes a temporary Python script to disk, wraps it in a small IProgram implementation, and evaluates it inside a subprocess. The evaluator returns 1.0 when the exit code is in success_exit_codes and, when provided, stdout matches expected_output.

from pathlib import Path
import tempfile

from swarmauri_evaluator_subprocess import SubprocessEvaluator
from swarmauri_core.programs.IProgram import DiffType, IProgram


class ScriptProgram(IProgram):
    """Minimal IProgram wrapper for a script stored on disk."""

    def __init__(self, path: Path):
        self._path = Path(path)

    # Required IProgram interface methods -------------------------------
    def diff(self, other: IProgram) -> DiffType:  # pragma: no cover - example
        return {}

    def apply_diff(self, diff: DiffType) -> "ScriptProgram":  # pragma: no cover
        return ScriptProgram(self._path)

    def validate(self) -> bool:  # pragma: no cover
        return self._path.exists()

    def clone(self) -> "ScriptProgram":  # pragma: no cover
        return ScriptProgram(self._path)

    # Methods consumed by SubprocessEvaluator ---------------------------
    def get_path(self) -> str:
        return str(self._path)

    def is_executable(self) -> bool:
        return False


def run_example(expected_output: str = "hello from subprocess\n"):
    evaluator = SubprocessEvaluator(timeout=5)

    with tempfile.TemporaryDirectory() as tmpdir:
        script_path = Path(tmpdir) / "echo.py"
        script_path.write_text("print('hello from subprocess')\n", encoding="utf-8")

        program = ScriptProgram(script_path)

        score, metadata = evaluator.evaluate(
            program,
            expected_output=expected_output,
        )

    return score, metadata


def main():
    score, metadata = run_example()
    print("Score:", score)
    print("Stdout:", metadata["stdout"].strip())
    print("Reason:", metadata["reason"])


if __name__ == "__main__":
    main()

Evaluation options

SubprocessEvaluator.evaluate(program, **kwargs) accepts runtime controls in addition to the evaluator's model fields:

Argument	Description
args	List of command-line arguments appended to the prepared command.
input_data	String provided on stdin; useful for feeding sample input.
expected_output	Optional stdout string; mismatches lower the score to 0.7.
timeout	Overrides the evaluator's timeout for a single run.

Returned metadata

Each evaluation returns (score, metadata) where metadata always contains:

• stdout, stderr, and exit_code from the subprocess.
• timed_out flag plus a human-readable reason such as success, timeout, or exit_code_<value>.
• command, args, and working_dir to show how the program was launched.
• execution_time (seconds) measured by the evaluator wrapper.

When aggregating multiple runs, aggregate_scores adds reason_counts, timeout_rate, success_rate, and total_executions to the combined metadata so callers can evaluate fleet-wide behavior.

Want to help?

If you want to contribute to swarmauri-sdk, read up on our guidelines for contributing that will help you get started.