Agentic-AI-Template/Codex/Coding/harness framework cs/scripts/dotnet_harness/process.py

import json
import locale
import re
import xml.etree.ElementTree as ET
import subprocess
import time

from .models import CommandResult, ResultCheckKind

DIAGNOSTIC_LIMIT = 8000


class ValidationFailure(RuntimeError):
    pass


def _decode(value):
    if value is None:
        return ""
    if isinstance(value, str):
        return value
    for encoding in ("utf-8", locale.getpreferredencoding(False)):
        try:
            return value.decode(encoding)
        except (UnicodeDecodeError, LookupError):
            continue
    return value.decode("utf-8", errors="replace")


def _check_result(check, result):
    if check.kind is ResultCheckKind.PROJECT_IS_NET8:
        try:
            properties = json.loads(result.stdout)["Properties"]
            target = properties.get("TargetFrameworks") or properties["TargetFramework"]
            frameworks = target.split(";")
            if not all(re.fullmatch(r"net8\.0(?:-[A-Za-z0-9.]+)?", item.strip()) for item in frameworks):
                raise ValueError(f"expected net8.0 targets, got {target!r}")
        except (ValueError, KeyError, TypeError, AttributeError) as exc:
            raise ValidationFailure(f"cannot verify evaluated net8.0 target: {exc}") from exc
        return
    if check.kind is not ResultCheckKind.TRX_HAS_TESTS:
        raise ValidationFailure("unsupported result check")
    files = sorted(check.path.glob("*.trx"))
    if not files:
        raise ValidationFailure("test run produced no TRX results")
    for path in files:
        try:
            if path.resolve().parent != check.path.resolve():
                raise ValueError("TRX result is outside its result directory")
            tree = ET.parse(path)
            summary = tree.getroot().find("{*}ResultSummary")
            if summary is None or summary.get("outcome") not in {"Completed", "Passed"}:
                raise ValueError("test run did not complete successfully")
            counters = summary.find("{*}Counters")
            if counters is None:
                raise ValueError("missing test counters")
            counts = {key: int(value) for key, value in counters.attrib.items()}
            if not (0 < counts["passed"] <= counts["executed"] <= counts["total"]):
                raise ValueError("no executed passing tests or inconsistent counters")
            if counts["failed"] or any(counts.get(key, 0) for key in
                    ("error", "timeout", "aborted", "inconclusive", "passedButRunAborted", "notRunnable", "disconnected")):
                raise ValueError("unsuccessful tests in TRX")
            if any(value < 0 for value in counts.values()):
                raise ValueError("negative test counters")
        except (ET.ParseError, OSError, ValueError, KeyError) as exc:
            raise ValidationFailure(f"invalid TRX {path.name}: {exc}") from exc


def _diagnostic(stdout, stderr):
    diagnostic = (stdout + "\n" + stderr).rstrip()[-DIAGNOSTIC_LIMIT:]
    return diagnostic.replace("\r\n", " | ").replace("\n", " | ")


def _command_context(argv, cwd):
    return f"argv={list(argv)!r}; cwd={str(cwd)!r}"


def execute_plan(
    plan,
    root,
    *,
    env=None,
    total_timeout_seconds=1800,
    deadline=None,
    run=subprocess.run,
    clock=time.monotonic,
):
    root = root.resolve()
    if deadline is None:
        deadline = clock() + total_timeout_seconds
    results = []

    for step in plan.steps:
        cwd = step.command.cwd.resolve()
        context = _command_context(step.command.argv, cwd)
        try:
            cwd.relative_to(root)
        except ValueError as exc:
            raise ValidationFailure(
                f"{step.command.stage} working directory is outside the repository; "
                f"{context}"
            ) from exc

        for check in step.checks:
            if check.kind is not ResultCheckKind.TRX_HAS_TESTS:
                continue
            if check.path is None:
                raise ValidationFailure(f"{step.command.stage} missing results directory; {context}")
            result_path = check.path.absolute()
            if not result_path.resolve().is_relative_to(root) or result_path.resolve() != result_path:
                raise ValidationFailure(f"{step.command.stage} results directory is outside repository or linked; {context}")
            try:
                result_path.mkdir(parents=True, exist_ok=False)
            except OSError as exc:
                raise ValidationFailure(f"{step.command.stage} requires a fresh results directory: {exc}; {context}") from exc

        remaining = deadline - clock()
        if remaining <= 0:
            raise ValidationFailure(
                f"{step.command.stage} timed out before it could start; {context}"
            )

        timeout = min(step.command.timeout_seconds, remaining)
        try:
            completed = run(
                list(step.command.argv),
                cwd=cwd,
                env=env,
                capture_output=True,
                shell=False,
                timeout=timeout,
            )
        except subprocess.TimeoutExpired as exc:
            stdout = _decode(exc.output)
            stderr = _decode(exc.stderr)
            diagnostic = _diagnostic(stdout, stderr)
            message = (
                f"{step.command.stage} timed out after {timeout} seconds; {context}"
            )
            if diagnostic:
                message = f"{message}; output tail: {diagnostic}"
            raise ValidationFailure(message) from exc

        stdout = _decode(completed.stdout)
        stderr = _decode(completed.stderr)
        result = CommandResult(step.command, completed.returncode, stdout, stderr)
        if completed.returncode != 0:
            diagnostic = _diagnostic(stdout, stderr)
            message = (
                f"{step.command.stage} failed with exit code {completed.returncode}; "
                f"{context}"
            )
            if diagnostic:
                message = f"{message}; output tail: {diagnostic}"
            raise ValidationFailure(message)

        try:
            for check in step.checks:
                _check_result(check, result)
        except ValidationFailure as exc:
            diagnostic = _diagnostic(stdout, stderr)
            message = (
                f"{step.command.stage} result check failed: {exc}; {context}"
            )
            if diagnostic:
                message = f"{message}; output tail: {diagnostic}"
            raise ValidationFailure(message) from exc
        results.append(result)

    return tuple(results)