tasks.py

#!/usr/bin/env python3
"""Cross-platform task runner — stdlib only.

Every Makefile target aliases here. Works on Linux, macOS, and Windows.
The runner sets the WindowsSelectorEventLoopPolicy before importing any
asyncio module that needs it (Phase-3.5 must-fix #8).
"""
from __future__ import annotations

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

# ── Windows event-loop policy guard ────────────────────────────────────────────
if os.name == "nt":  # pragma: no cover
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

ROOT = Path(__file__).resolve().parent
os.environ.setdefault("TRITON_CACHE_DIR", str(ROOT / ".triton_cache"))
os.environ.setdefault("PYTHONPATH", str(ROOT) + os.pathsep + os.environ.get("PYTHONPATH", ""))


# ── helpers ────────────────────────────────────────────────────────────────────


def _run(cmd: list[str], **kwargs) -> int:
    """Run a subprocess, streaming output. Return its exit code."""
    print(f"$ {' '.join(cmd)}")
    try:
        return subprocess.call(cmd, cwd=ROOT, **kwargs)
    except FileNotFoundError as exc:
        print(f"  error: {exc}", file=sys.stderr)
        return 127


def _have(tool: str) -> bool:
    return shutil.which(tool) is not None


def _python() -> str:
    return sys.executable or "python"


# ── tasks ──────────────────────────────────────────────────────────────────────


def task_install(args: argparse.Namespace) -> int:
    """Install runtime deps in editable mode + dev group."""
    rc = _run([_python(), "-m", "pip", "install", "--upgrade", "pip"])
    if rc != 0:
        return rc
    rc = _run([_python(), "-m", "pip", "install", "-e", "."])
    if rc != 0:
        return rc
    # Best-effort dev group
    rc = _run([_python(), "-m", "pip", "install", "pytest", "pytest-asyncio", "pytest-httpx",
               "hypothesis", "ruff", "mypy", "coverage[toml]", "pip-audit", "import-linter"])
    return rc


def task_test(args: argparse.Namespace) -> int:
    """Run the full test suite."""
    return _run([_python(), "-m", "pytest", "-q"])


def task_smoke(args: argparse.Namespace) -> int:
    """Run only the smoke value-path test (AC-1)."""
    return _run([_python(), "-m", "pytest", "-q", "-m", "smoke", "tests/smoke/"])


def task_lint(args: argparse.Namespace) -> int:
    """Ruff lint + format check."""
    if not _have("ruff"):
        return _run([_python(), "-m", "ruff", "check", "."])
    return _run(["ruff", "check", "."])


def task_format(args: argparse.Namespace) -> int:
    """Ruff format (auto-fix)."""
    if not _have("ruff"):
        return _run([_python(), "-m", "ruff", "format", "."])
    return _run(["ruff", "format", "."])


def task_typecheck(args: argparse.Namespace) -> int:
    """mypy strict on selected modules."""
    return _run([_python(), "-m", "mypy", "echoform", "ghost_memory"])


def task_licenses(args: argparse.Namespace) -> int:
    """pip-licenses + deny.toml policy."""
    return _run([_python(), "-m", "piplicenses",
                 "--fail-on=GPL;LGPL;AGPL;SSPL;BUSL;Commons-Clause"])


def task_audit(args: argparse.Namespace) -> int:
    """pip-audit for vulnerabilities."""
    return _run([_python(), "-m", "pip_audit"])


def task_clean(args: argparse.Namespace) -> int:
    """Remove caches and build artifacts."""
    for pat in [".pytest_cache", ".ruff_cache", ".mypy_cache", "htmlcov", "build", "dist",
                ".coverage", ".coverage.*", ".triton_cache", "local-archive"]:
        for p in ROOT.glob(pat):
            try:
                if p.is_dir():
                    shutil.rmtree(p, ignore_errors=True)
                else:
                    p.unlink(missing_ok=True)
            except Exception as exc:
                print(f"  skip {p}: {exc}")
    return 0


def task_docs(args: argparse.Namespace) -> int:
    """Build mkdocs site."""
    if not _have("mkdocs"):
        print("install with: pip install mkdocs mkdocs-material mkdocstrings[python]")
        return 1
    return _run(["mkdocs", "build", "--strict"])


def task_docker_build(args: argparse.Namespace) -> int:
    """Build the api Docker image."""
    if not _have("docker"):
        print("docker not on PATH")
        return 1
    return _run(["docker", "build", "-t", "echoform/api:latest", "."])


def task_compose_up(args: argparse.Namespace) -> int:
    if not _have("docker"):
        print("docker not on PATH")
        return 1
    return _run(["docker", "compose", "up", "-d"])


def task_compose_down(args: argparse.Namespace) -> int:
    if not _have("docker"):
        return 0
    return _run(["docker", "compose", "down", "-v"])


def task_migrate(args: argparse.Namespace) -> int:
    """Run Alembic migrations to head."""
    if not _have("alembic"):
        return _run([_python(), "-m", "alembic", "upgrade", "head"])
    return _run(["alembic", "upgrade", "head"])


def task_serve(args: argparse.Namespace) -> int:
    """Run the dev server."""
    return _run([_python(), "-m", "echoform", "serve", "--host", args.host, "--port", str(args.port)])


def task_built_from(args: argparse.Namespace) -> int:
    """Print BUILT_FROM.md content. Useful in CI/release."""
    path = ROOT / "BUILT_FROM.md"
    if not path.exists():
        print("BUILT_FROM.md missing", file=sys.stderr)
        return 2
    sys.stdout.write(path.read_text(encoding="utf-8"))
    return 0


# ── dispatch ───────────────────────────────────────────────────────────────────


TASKS = {
    "install":       task_install,
    "test":          task_test,
    "smoke":         task_smoke,
    "lint":          task_lint,
    "format":        task_format,
    "typecheck":     task_typecheck,
    "licenses":      task_licenses,
    "audit":         task_audit,
    "clean":         task_clean,
    "docs":          task_docs,
    "docker-build":  task_docker_build,
    "compose-up":    task_compose_up,
    "compose-down":  task_compose_down,
    "migrate":       task_migrate,
    "serve":         task_serve,
    "built-from":    task_built_from,
}


def main() -> int:
    parser = argparse.ArgumentParser(description="ECHOFORM task runner")
    sub = parser.add_subparsers(dest="task", required=True)
    for name in TASKS:
        sp = sub.add_parser(name)
        if name == "serve":
            sp.add_argument("--host", default="0.0.0.0")
            sp.add_argument("--port", type=int, default=8080)
    args = parser.parse_args()
    return TASKS[args.task](args)


if __name__ == "__main__":  # pragma: no cover
    sys.exit(main())