scripts/task.py

#!/usr/bin/env python3
"""Project task runner.

This is an optional local wrapper around direct uv commands. The demo server
still runs directly through app.py.
"""

from __future__ import annotations

import argparse
import os
import shutil
import subprocess
import sys
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_ENV = {
    "UV_PROJECT_ENVIRONMENT": ".venv",
    "UV_CACHE_DIR": ".uv-cache",
    "UV_LINK_MODE": "copy",
}
CLEAN_PATHS = [
    ".pytest_cache",
    "arena/__pycache__",
    "tests_python/__pycache__",
]


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Run project tasks")
    parser.add_argument("task", choices=sorted(_tasks()))
    parser.add_argument("args", nargs=argparse.REMAINDER)
    parsed = parser.parse_args(argv)

    env = task_env()
    task_args = _strip_separator(parsed.args)
    if parsed.task == "clean":
        clean()
        return 0

    if parsed.task == "verify":
        return verify(env, task_args)

    command, overrides = command_for_invocation(parsed.task, task_args)
    run_env = env | overrides
    return subprocess.run(command + task_args, cwd=ROOT, env=run_env, check=False).returncode


def task_env() -> dict[str, str]:
    env = dict(os.environ)
    _load_dotenv(ROOT / ".env", env)
    for key, value in DEFAULT_ENV.items():
        env.setdefault(key, value)
    return env


def command_for(task: str) -> tuple[list[str], dict[str, str]]:
    command = _tasks().get(task)
    if command is None:
        raise ValueError(f"unknown task: {task}")
    return command


def command_for_invocation(task: str, args: list[str]) -> tuple[list[str], dict[str, str]]:
    command, overrides = command_for(task)
    if task == "test" and args:
        return ["uv", "run", "pytest"], overrides
    return command, overrides


def clean() -> None:
    for relative_path in CLEAN_PATHS:
        shutil.rmtree(ROOT / relative_path, ignore_errors=True)


def verify(env: dict[str, str], args: list[str]) -> int:
    if args:
        raise ValueError("verify does not accept extra arguments")
    for command, overrides in _verify_commands():
        result = subprocess.run(command, cwd=ROOT, env=env | overrides, check=False)
        if result.returncode != 0:
            return result.returncode
    return 0


def _verify_commands() -> list[tuple[list[str], dict[str, str]]]:
    return [
        (["uv", "run", "pytest", "tests_python"], {}),
        (
            [
                "uv",
                "run",
                "python",
                "-m",
                "arena.harness",
                "--prompt",
                "Build a tiny CLI",
                "--test",
                "test -f README.md",
            ],
            {},
        ),
    ]


def _tasks() -> dict[str, tuple[list[str], dict[str, str]]]:
    return {
        "dev": (
            ["uv", "run", "uvicorn", "app:app", "--host", "127.0.0.1", "--port", "8790"],
            {},
        ),
        "gradio": (["uv", "run", "python", "app.py"], {}),
        "harness": (["uv", "run", "python", "-m", "arena.harness"], {}),
        "test": (["uv", "run", "pytest", "tests_python"], {}),
        "verify": ([], {}),
        "smoke": (["uv", "run", "python", "-m", "arena.smoke", "agent"], {}),
        "validator-smoke": (["uv", "run", "python", "-m", "arena.smoke", "validator"], {}),
        "daytona-validator-smoke": (
            ["uv", "run", "python", "-m", "arena.smoke", "validator"],
            {"VALIDATOR_SANDBOX_PROVIDER": "daytona"},
        ),
        "clean": ([], {}),
    }


def _load_dotenv(path: Path, env: dict[str, str]) -> None:
    if not path.exists():
        return
    for raw_line in path.read_text().splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        key = key.strip()
        value = value.strip().strip("\"'")
        if key and key not in env:
            env[key] = value


def _strip_separator(args: list[str]) -> list[str]:
    if args and args[0] == "--":
        return args[1:]
    return args


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))