scripts/verify_space.py

#!/usr/bin/env python3
"""Verify the deployed Docker Space and, optionally, a real conversion.

Default verification is read-only: ``hf spaces wait/info/logs`` followed by
GET probes of the public application. A conversion is submitted only when an
actual image is supplied with ``--e2e IMAGE``. The e2e path consumes the job's
SSE stream to a terminal event, downloads the real artifacts, and executes the
downloaded bundle with ``scripts/node_smoke.mjs``.

Examples:
    python3 scripts/verify_space.py
    python3 scripts/verify_space.py --e2e tests/fixtures/mug_photo.png
"""

from __future__ import annotations

import argparse
import json
import mimetypes
import shlex
import subprocess
import sys
from pathlib import Path
from typing import Any

import httpx

REPO_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_SPACE_ID = "Mike0021/img2threejs"
DEFAULT_SPACE_URL = "https://mike0021-img2threejs.hf.space"
EXPECTED_ARTIFACTS = (
    "reference.png",
    "probe.json",
    "spec.json",
    "compile-spec.json",
    "validation.json",
    "factory.ts",
    "model.bundle.js",
    "standalone.html",
)


def _command(argv: list[str], *, capture: bool = False) -> subprocess.CompletedProcess[str]:
    print("->", shlex.join(argv), flush=True)
    return subprocess.run(
        argv,
        check=False,
        text=True,
        capture_output=capture,
    )


def _show_failure_logs(space_id: str) -> None:
    for mode in (("--build", "--tail", "200"), ("--tail", "200")):
        result = _command(["hf", "spaces", "logs", space_id, *mode], capture=True)
        if result.stdout:
            print(result.stdout, end="" if result.stdout.endswith("\n") else "\n")
        if result.stderr:
            print(result.stderr, file=sys.stderr, end="" if result.stderr.endswith("\n") else "\n")


def _runtime_stage(payload: dict[str, Any]) -> tuple[str | None, Any, Any]:
    runtime = payload.get("runtime") if isinstance(payload.get("runtime"), dict) else payload
    raw = runtime.get("raw") if isinstance(runtime.get("raw"), dict) else {}
    return (
        runtime.get("stage") or raw.get("stage"),
        runtime.get("hardware") or raw.get("hardware"),
        runtime.get("requested_hardware") or raw.get("requested_hardware"),
    )


def _probe_get_surface(client: httpx.Client) -> bool:
    health_response = client.get("/health")
    health_response.raise_for_status()
    health = health_response.json()
    if health.get("status") != "ok":
        raise RuntimeError(f"unexpected /health payload: {health}")
    llm_configured = bool(health.get("llm_configured"))
    print(f"   GET /health ok (llm_configured={llm_configured})")

    ui = client.get("/")
    ui.raise_for_status()
    if "img2threejs" not in ui.text:
        raise RuntimeError("GET / did not return the img2threejs UI")
    print("   GET / ok")

    config_response = client.get("/api/config")
    config_response.raise_for_status()
    config = config_response.json()
    forbidden_keys = {"api_key", "llm_api_key", "anthropic_api_key", "token", "secret"}
    if forbidden_keys.intersection({str(key).lower() for key in config}):
        raise RuntimeError("GET /api/config exposed a credential-bearing key")
    serialized = json.dumps(config).lower()
    if "sk-ant-" in serialized or '"hf_' in serialized:
        raise RuntimeError("GET /api/config appears to contain credential material")
    print("   GET /api/config ok (credential-free)")

    gallery_response = client.get("/api/gallery", params={"offset": 0, "limit": 24})
    gallery_response.raise_for_status()
    gallery = gallery_response.json()
    if not isinstance(gallery.get("items"), list):
        raise RuntimeError("GET /api/gallery did not return an items array")
    if not isinstance(gallery.get("total"), int) or gallery["total"] < 0:
        raise RuntimeError("GET /api/gallery returned an invalid total")
    if gallery.get("offset") != 0 or gallery.get("limit") != 24:
        raise RuntimeError("GET /api/gallery did not honor pagination")
    for item in gallery["items"]:
        if not isinstance(item, dict) or not isinstance(item.get("id"), str):
            raise RuntimeError("GET /api/gallery returned a malformed item")
        if len(item["id"]) != 32 or any(c not in "0123456789abcdef" for c in item["id"]):
            raise RuntimeError("GET /api/gallery returned an invalid item id")
        if not str(item.get("thumbnailUrl", "")).startswith("/api/gallery/"):
            raise RuntimeError("GET /api/gallery returned an invalid thumbnail URL")
    print(f"   GET /api/gallery ok ({gallery['total']} persistent items)")
    return llm_configured


def _terminal_event(client: httpx.Client, events_url: str) -> dict[str, Any]:
    terminal: dict[str, Any] | None = None
    print(f"   streaming {events_url}")
    with client.stream("GET", events_url) as response:
        response.raise_for_status()
        for line in response.iter_lines():
            if not line.startswith("data:"):
                continue
            payload = json.loads(line[5:].strip())
            stage = payload.get("stage", "?")
            status = payload.get("status", "?")
            message = payload.get("message", "")
            print(f"   SSE {stage}/{status}: {message}", flush=True)
            if status == "error" or (stage == "done" and status == "done"):
                terminal = payload
                break
    if terminal is None:
        raise RuntimeError("SSE stream ended without a terminal event")
    return terminal


def _download_artifacts(
    client: httpx.Client,
    job_id: str,
    result: dict[str, Any],
    output_root: Path,
) -> Path:
    job_dir = output_root.resolve() / job_id
    job_dir.mkdir(parents=True, exist_ok=False)
    advertised = result.get("artifacts") if isinstance(result.get("artifacts"), dict) else {}

    for name in EXPECTED_ARTIFACTS:
        url = advertised.get(name) or f"/api/jobs/{job_id}/artifacts/{name}"
        response = client.get(str(url))
        response.raise_for_status()
        if not response.content:
            raise RuntimeError(f"downloaded artifact is empty: {name}")
        (job_dir / name).write_bytes(response.content)
        print(f"   artifact {name}: {len(response.content)} bytes")

    spec = json.loads((job_dir / "spec.json").read_text(encoding="utf-8"))
    if spec.get("reviewHistory"):
        raise RuntimeError("downloaded spec claims review history for an unreviewed hosted preview")
    compile_spec = json.loads(
        (job_dir / "compile-spec.json").read_text(encoding="utf-8"))
    if compile_spec.get("reviewHistory"):
        raise RuntimeError("compile-spec.json contains fabricated review history")
    pipeline = compile_spec.get("sculptPipeline") or {}
    if pipeline.get("passGateMode") != "hosted-unreviewed-preview":
        raise RuntimeError("compile-spec.json is not labeled as an unreviewed hosted preview")
    validation = json.loads(
        (job_dir / "validation.json").read_text(encoding="utf-8"))
    if validation.get("ok") is not True:
        raise RuntimeError("live validation.json does not record a passing deterministic gate")
    factory = (job_dir / "factory.ts").read_text(encoding="utf-8")
    if "TODO:" in factory:
        raise RuntimeError("factory.ts contains an unsupported-primitive TODO fallback")
    standalone = (job_dir / "standalone.html").read_text(encoding="utf-8")
    if "await import(url)" not in standalone:
        raise RuntimeError("standalone.html does not import its embedded ESM bundle")
    return job_dir


def _node_smoke(bundle_path: Path, spec_path: Path) -> None:
    harness = REPO_ROOT / "scripts" / "node_smoke.mjs"
    if not (REPO_ROOT / "node_modules" / "three").is_dir():
        raise RuntimeError("node_modules/three is missing; run `npm ci` before live e2e verification")
    result = _command(["node", str(harness), str(bundle_path), str(spec_path)])
    if result.returncode != 0:
        raise RuntimeError(f"node smoke failed with exit code {result.returncode}")


def _verify_gallery_item(
    client: httpx.Client,
    *,
    job_id: str,
    result: dict[str, Any],
    job_dir: Path,
) -> dict[str, Any]:
    if result.get("shareRequested") is not True:
        raise RuntimeError("default submission did not record shareRequested=true")
    if result.get("shared") is not True:
        raise RuntimeError("default submission was not committed to the gallery")
    summary = result.get("galleryItem")
    if not isinstance(summary, dict) or summary.get("id") != job_id:
        raise RuntimeError("terminal result has no matching galleryItem")

    response = client.get(f"/api/gallery/{job_id}")
    response.raise_for_status()
    item = response.json()
    if item.get("id") != job_id or item.get("targetName") != result.get("targetName"):
        raise RuntimeError("gallery detail does not match the completed job")
    if item.get("generation", {}).get("reviewStatus") != "unreviewed":
        raise RuntimeError("gallery item lost the hosted-preview review status")

    advertised = item.get("artifacts")
    if not isinstance(advertised, dict):
        raise RuntimeError("gallery detail has no artifact map")
    for name in EXPECTED_ARTIFACTS:
        url = advertised.get(name)
        if not isinstance(url, str):
            raise RuntimeError(f"gallery item does not advertise {name}")
        gallery_response = client.get(url)
        gallery_response.raise_for_status()
        local_bytes = (job_dir / name).read_bytes()
        if gallery_response.content != local_bytes:
            raise RuntimeError(f"gallery artifact differs from job artifact: {name}")
        if "immutable" not in gallery_response.headers.get("cache-control", ""):
            raise RuntimeError(f"gallery artifact is not immutable: {name}")
        print(f"   gallery artifact {name}: stable and byte-identical")

    listing = client.get("/api/gallery", params={"offset": 0, "limit": 24})
    listing.raise_for_status()
    if job_id not in {entry.get("id") for entry in listing.json().get("items", [])}:
        raise RuntimeError("newly published item is absent from the gallery index")
    friendly = client.get(f"/gallery/{job_id}")
    friendly.raise_for_status()
    if "img2threejs" not in friendly.text:
        raise RuntimeError("friendly gallery route did not return the SPA")
    print(f"   gallery item {job_id}: detail, index, artifacts and SPA route ok")
    return item


def _run_e2e(
    client: httpx.Client,
    image_path: Path,
    output_root: Path,
    llm_configured: bool,
) -> Path:
    if not llm_configured:
        raise RuntimeError("--e2e requires configured live LLM secrets; /health reports false")
    image_path = image_path.resolve()
    if not image_path.is_file():
        raise RuntimeError(f"e2e image does not exist: {image_path}")
    media_type = mimetypes.guess_type(image_path.name)[0] or "application/octet-stream"
    with image_path.open("rb") as stream:
        response = client.post(
            "/api/jobs",
            files={"file": (image_path.name, stream, media_type)},
        )
    if response.status_code != 202:
        raise RuntimeError(
            f"POST /api/jobs returned {response.status_code}: {response.text[:500]}"
        )
    accepted = response.json()
    job_id = str(accepted["job_id"])
    events_url = str(accepted.get("events_url") or f"/api/jobs/{job_id}/events")
    print(f"   accepted real image as job {job_id}")

    terminal = _terminal_event(client, events_url)
    if terminal.get("status") != "done":
        raise RuntimeError("live job failed: " + json.dumps(terminal, ensure_ascii=False)[:1000])
    data = terminal.get("data") if isinstance(terminal.get("data"), dict) else {}
    result = data.get("result") if isinstance(data.get("result"), dict) else {}
    if not result:
        raise RuntimeError("terminal SSE event has no result payload")
    if result.get("generationMode") != "hosted-unreviewed-preview":
        raise RuntimeError(f"unexpected generation mode: {result.get('generationMode')}")
    if result.get("reviewStatus") != "unreviewed":
        raise RuntimeError(f"unexpected hosted review status: {result.get('reviewStatus')}")

    job_dir = _download_artifacts(client, job_id, result, output_root)
    _node_smoke(job_dir / "model.bundle.js", job_dir / "compile-spec.json")
    _verify_gallery_item(
        client,
        job_id=job_id,
        result=result,
        job_dir=job_dir,
    )
    return job_dir


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--space-id", default=DEFAULT_SPACE_ID)
    parser.add_argument("--space-url", default=DEFAULT_SPACE_URL)
    parser.add_argument("--timeout", default="40m",
                        help="timeout passed to `hf spaces wait` (default: 40m)")
    parser.add_argument("--logs-tail", type=int, default=80)
    parser.add_argument("--e2e", metavar="IMAGE", type=Path,
                        help="submit this real image and require terminal artifacts")
    parser.add_argument("--output-dir", type=Path,
                        default=REPO_ROOT / "verification" / "live-artifacts")
    args = parser.parse_args(argv)

    try:
        wait = _command([
            "hf", "spaces", "wait", args.space_id,
            "--timeout", args.timeout, "--format", "json",
        ])
        if wait.returncode != 0:
            _show_failure_logs(args.space_id)
            return wait.returncode or 1

        info = _command([
            "hf", "spaces", "info", args.space_id,
            "--expand", "runtime", "--format", "json",
        ], capture=True)
        if info.returncode != 0:
            if info.stderr:
                print(info.stderr, file=sys.stderr)
            _show_failure_logs(args.space_id)
            return info.returncode or 1
        print(info.stdout, end="" if info.stdout.endswith("\n") else "\n")
        payload = json.loads(info.stdout)
        stage, hardware, requested = _runtime_stage(payload)
        print(f"-> runtime stage={stage} hardware={hardware} requested={requested}")
        if stage != "RUNNING":
            _show_failure_logs(args.space_id)
            raise RuntimeError(f"Space is not RUNNING (stage={stage})")

        logs = _command([
            "hf", "spaces", "logs", args.space_id,
            "--tail", str(max(1, args.logs_tail)),
        ])
        if logs.returncode != 0:
            raise RuntimeError(f"runtime log retrieval failed with exit code {logs.returncode}")

        timeout = httpx.Timeout(connect=30.0, read=30.0, write=60.0, pool=30.0)
        with httpx.Client(
            base_url=args.space_url.rstrip("/"),
            follow_redirects=True,
            timeout=timeout,
        ) as client:
            llm_configured = _probe_get_surface(client)
            artifact_dir: Path | None = None
            if args.e2e is not None:
                artifact_dir = _run_e2e(
                    client, args.e2e, args.output_dir, llm_configured
                )

        print("-> LIVE VERIFICATION PASSED")
        print(f"-> {args.space_url.rstrip('/')}")
        if artifact_dir is not None:
            print(f"-> downloaded artifacts: {artifact_dir}")
        return 0
    except (OSError, ValueError, KeyError, RuntimeError, httpx.HTTPError) as exc:
        print(f"verification failed: {exc}", file=sys.stderr)
        return 1


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