ai-agent/sidecar/scaffold.py

"""
``POST /scaffold`` — create a project on this host from one of the owner's
templates: the host half of the hub's *New project* flow (backend
``newproject/``).

The hub can't scaffold by itself. The generator needs the host's node/npm, the
private npm registry bound to the host's loopback, the Gitea credentials, and
SSH to the zipgo host (a password lock is set before the first deploy). So the
sidecar runs the configured generator (``SIDECAR_SCAFFOLD_CMD``, default
``new-project`` — the homelab's new-frontend-project skill) with a vars JSON and
reports what it printed. The generator's contract:

* ``<cmd> --template <name> --vars <file> [flags]`` scaffolds
  ``~/projects/<NAME>`` and, last, prints one ``SCAFFOLD_RESULT={json}`` line
  (``dir``, ``url``, ``host`` and, with ``--password``, ``password``);
* with ``--check`` it creates nothing and prints ``SCAFFOLD_CHECK={json}``:
  ``{"ok": bool, "problems": [{"field", "message"}]}`` — the name already used
  by a project, the address already served, …

Only the flags below go onto the command line; every other choice travels in
the vars JSON (``NAME``, ``LABEL``, ``SUBDOMAIN``, ``GITEA_OWNER``, …), which
the generator validates itself.
"""

from __future__ import annotations

import json
import os
import re
import shlex
import subprocess
import tempfile

from fastapi import APIRouter, Header, HTTPException
from pydantic import BaseModel

from config import RUN_ENV_STRIP
from validate import _auth

SCAFFOLD_CMD = shlex.split(os.environ.get("SIDECAR_SCAFFOLD_CMD", "new-project"))
# npm install + the first Gitea push; a cold registry cache takes a few minutes.
SCAFFOLD_TIMEOUT_S = float(os.environ.get("SIDECAR_SCAFFOLD_TIMEOUT_S", "900"))
CHECK_TIMEOUT_S = 60.0

_TEMPLATE_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,63}$")
_KEY_RE = re.compile(r"^[A-Z][A-Z0-9_]{0,31}$")
_LOG_TAIL = 6000

router = APIRouter()


class ScaffoldBody(BaseModel):
    template: str
    vars: dict
    noGit: bool = False
    password: bool = False          # lock the site (zipgo password, generated)


def _args(body: ScaffoldBody, vars_path: str, check: bool) -> list[str]:
    template = (body.template or "").strip()
    if not _TEMPLATE_RE.match(template):
        raise HTTPException(400, f"invalid template: {template!r}")
    args = [*SCAFFOLD_CMD, "--template", template, "--vars", vars_path]
    if check:
        return args + ["--check"]
    if body.noGit:
        args.append("--no-git")
    if body.password:
        args.append("--password")
    return args


def _vars(raw: dict) -> dict[str, str]:
    out: dict[str, str] = {}
    for k, v in (raw or {}).items():
        if not isinstance(k, str) or not _KEY_RE.match(k):
            raise HTTPException(400, f"invalid vars key: {k!r}")
        if not isinstance(v, (str, int, float)):
            raise HTTPException(400, f"vars[{k}] must be a string")
        out[k] = str(v)[:2000]
    if not out.get("NAME"):
        raise HTTPException(400, "vars.NAME is required")
    return out


def _marker(out: str, name: str) -> dict | None:
    """The JSON after the last ``<name>=`` line the generator printed."""
    for line in reversed(out.splitlines()):
        if line.startswith(f"{name}="):
            try:
                return json.loads(line[len(name) + 1:])
            except ValueError:
                return None
    return None


def _run(body: ScaffoldBody, check: bool) -> dict:
    env = {k: v for k, v in os.environ.items() if k not in RUN_ENV_STRIP}
    with tempfile.NamedTemporaryFile("w", suffix=".json", prefix="scaffold-",
                                     delete=False) as f:
        json.dump(_vars(body.vars), f)
        vars_path = f.name
    try:
        args = _args(body, vars_path, check)
        try:
            p = subprocess.run(args, cwd=os.path.expanduser("~"), env=env,
                               stdin=subprocess.DEVNULL, capture_output=True,
                               text=True,
                               timeout=CHECK_TIMEOUT_S if check else SCAFFOLD_TIMEOUT_S)
        except FileNotFoundError:
            raise HTTPException(501, f"scaffold command not found: {SCAFFOLD_CMD[0]}")
        except subprocess.TimeoutExpired as e:
            out = (e.stdout or b"").decode(errors="ignore") if isinstance(e.stdout, bytes) else (e.stdout or "")
            return {"ok": False, "exitCode": None, "error": "timed out",
                    "log": out[-_LOG_TAIL:]}
    finally:
        os.unlink(vars_path)
    out = (p.stdout or "") + (("\n" + p.stderr) if p.stderr else "")
    marker = _marker(p.stdout or "", "SCAFFOLD_CHECK" if check else "SCAFFOLD_RESULT")
    ok = p.returncode == 0 and marker is not None
    if check and marker is not None:
        ok = bool(marker.get("ok"))
    # A finished scaffold's password is the guest's to read, once — the log
    # only goes to the hub's server log, so keep it out of there.
    log = out[-_LOG_TAIL:]
    if not check and marker and marker.get("password"):
        log = log.replace(marker["password"], "••••••")
    return {"ok": ok, "exitCode": p.returncode, "result": marker, "log": log}


@router.post("/scaffold")
def scaffold(body: ScaffoldBody, authorization: str | None = Header(default=None)) -> dict:
    """Scaffold ``~/projects/<vars.NAME>`` from ``template`` (blocking: npm
    install and the first Gitea push run before it answers)."""
    _auth(authorization)
    return _run(body, check=False)


@router.post("/scaffold/check")
def scaffold_check(body: ScaffoldBody, authorization: str | None = Header(default=None)) -> dict:
    """Would that scaffold work? The name, the address, the template — nothing
    is created."""
    _auth(authorization)
    return _run(body, check=True)