promptkit/scripts/check-examples.py

#!/usr/bin/env python3
"""Run the maintained consumer examples and check their semantic output."""

import json
import os
import subprocess
import sys
from pathlib import Path


REPOSITORY_ROOT = Path(__file__).resolve().parent.parent


def example_environment():
    environment = os.environ.copy()
    environment["GOTOOLCHAIN"] = "local"
    try:
        host = subprocess.run(
            ["go", "env", "GOHOSTOS", "GOHOSTARCH"],
            cwd=REPOSITORY_ROOT,
            env=environment,
            text=True,
            capture_output=True,
            check=False,
        )
    except OSError as error:
        raise RuntimeError(f"could not run go env: {error}") from error

    if host.returncode != 0:
        raise RuntimeError(
            "go env GOHOSTOS GOHOSTARCH failed "
            f"with exit status {host.returncode}: {host.stderr.strip()}"
        )

    values = host.stdout.splitlines()
    if len(values) != 2 or not all(values):
        raise RuntimeError(
            "go env GOHOSTOS GOHOSTARCH returned unexpected output: "
            f"{host.stdout!r}"
        )

    environment["GOOS"], environment["GOARCH"] = values
    return environment


def run_example(name, environment, failures):
    command = ["go", "run", f"./examples/go-library/{name}"]
    try:
        result = subprocess.run(
            command,
            cwd=REPOSITORY_ROOT,
            env=environment,
            text=True,
            capture_output=True,
            check=False,
        )
    except OSError as error:
        failures.append(f"{name}: could not start {command[0]}: {error}")
        return None

    if result.returncode != 0:
        failures.append(f"{name}: command exited with status {result.returncode}")
        if result.stderr:
            failures.append(f"{name}: stderr:\n{result.stderr.rstrip()}")
        if result.stdout:
            failures.append(f"{name}: stdout:\n{result.stdout.rstrip()}")
        return None

    try:
        value = json.loads(result.stdout)
    except json.JSONDecodeError as error:
        failures.append(f"{name}: output is not valid JSON: {error}")
        if result.stdout:
            failures.append(f"{name}: stdout:\n{result.stdout.rstrip()}")
        return None

    if not isinstance(value, dict):
        failures.append(
            f"{name}: expected a JSON object, got {type(value).__name__}"
        )
        return None

    return value


def expect_equal(example_name, value, key, expected, failures):
    actual = value.get(key)
    matches = actual is expected if isinstance(expected, bool) else actual == expected
    if not matches:
        failures.append(
            f"{example_name}: expected {key}={expected!r}, got {actual!r}"
        )


def expect_positive(example_name, value, key, failures):
    actual = value.get(key)
    try:
        matches = actual > 0
    except TypeError:
        matches = False
    if not matches:
        failures.append(f"{example_name}: expected {key} to be positive, got {actual!r}")


def main():
    try:
        environment = example_environment()
    except RuntimeError as error:
        print(error, file=sys.stderr)
        return 1

    failures = []
    prepared = run_example("prepare", environment, failures)
    result = run_example("run", environment, failures)
    observed = run_example("observe", environment, failures)

    if prepared is not None:
        expect_equal("prepare", prepared, "prompt_id", "example.prepare", failures)
        expect_equal(
            "prepare", prepared, "selected_profile", "offline-example", failures
        )
        expect_equal("prepare", prepared, "model", "offline-model", failures)
        expect_positive("prepare", prepared, "message_count", failures)

    if result is not None:
        expect_equal(
            "run", result, "output", "Ada finished the migration review.", failures
        )
        expect_equal("run", result, "is_valid", True, failures)
        expect_equal("run", result, "validation_status", "passed", failures)
        expect_equal("run", result, "model", "offline-model", failures)
        expect_equal("run", result, "total_tokens", 18, failures)

    if observed is not None:
        expect_equal("observe", observed, "proposal_attempts", 2, failures)
        expect_equal(
            "observe", observed, "proposal_transient_retries", 1, failures
        )
        expect_equal(
            "observe", observed, "proposal_usage_complete", False, failures
        )
        expect_equal("observe", observed, "assessment_attempts", 1, failures)
        expect_equal(
            "observe", observed, "assessment_transient_retries", 0, failures
        )
        expect_equal(
            "observe", observed, "assessment_outcome", "accepted", failures
        )

    if failures:
        print("Consumer example checks failed:", file=sys.stderr)
        for failure in failures:
            print(f"- {failure}", file=sys.stderr)
        return 1

    print("Consumer example checks passed: prepare, run, observe")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())