verify_live_smoke.py

"""R0-3 — post-deploy live smoke gate.

Verifies the ARTIFACT, not the pipeline: after a deploy, converts the kitchen
reference photo (extracted from the committed bundle, so no binary assets in
the repo) against the LIVE Space and asserts the bundle carries the metric
stack. Catches stale images, partial rsyncs, and dependency regressions that
deploy "successfully" while serving old behaviour — both deploy traps hit in
June 2026 (unpromoted Vercel build; rsync size+mtime skip) would have failed
this gate within minutes instead of costing a photo test round.

Pure stdlib (no pip installs in CI).

Usage:
    python verify_live_smoke.py [base_url]
    # default base_url: https://modventures-room-visualizer.hf.space

Exit 0 = live Space serves the expected stack; non-zero = block/alert.
"""

import base64
import io
import json
import os
import sys
import time
import urllib.error
import urllib.request
import uuid

HERE = os.path.dirname(os.path.abspath(__file__))
BUNDLE = os.path.join(HERE, "data", "ref_kitchen.vizbundle.json")
DEFAULT_BASE = "https://modventures-room-visualizer.hf.space"
SPACE_API = "https://huggingface.co/api/spaces/modventures/room-visualizer"

BUILD_WAIT_S = 20 * 60     # Space docker rebuild after a push
CONVERT_RETRY_S = 10 * 60  # wake-from-sleep + model loads on first request
JOB_WAIT_S = 6 * 60


def get_json(url, timeout=30):
    with urllib.request.urlopen(url, timeout=timeout) as r:
        return json.load(r)


def wait_for_space_ready():
    """Wait out a BUILDING stage; RUNNING/SLEEPING are both fine (a request
    wakes a sleeping Space)."""
    deadline = time.time() + BUILD_WAIT_S
    last = None
    while time.time() < deadline:
        try:
            stage = get_json(SPACE_API).get("runtime", {}).get("stage")
        except Exception as exc:
            stage = f"api-error: {exc}"
        if stage != last:
            print(f"  space stage: {stage}", flush=True)
            last = stage
        if stage in ("RUNNING", "SLEEPING", "PAUSED"):
            return stage
        time.sleep(20)
    raise SystemExit(f"FAIL: Space not ready within {BUILD_WAIT_S}s (stage={last})")


def post_convert(base, img_bytes):
    boundary = uuid.uuid4().hex
    body = (
        f"--{boundary}\r\nContent-Disposition: form-data; "
        f'name="file"; filename="smoke.jpg"\r\n'
        f"Content-Type: image/jpeg\r\n\r\n"
    ).encode() + img_bytes + f"\r\n--{boundary}--\r\n".encode()
    req = urllib.request.Request(
        base + "/viz2d/convert",
        data=body,
        method="POST",
        headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
    )
    deadline = time.time() + CONVERT_RETRY_S
    while True:
        try:
            with urllib.request.urlopen(req, timeout=120) as r:
                return json.load(r)
        except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError, OSError) as exc:
            if time.time() > deadline:
                raise SystemExit(f"FAIL: /viz2d/convert unreachable: {exc}")
            print(f"  convert not ready ({exc}), retrying...", flush=True)
            time.sleep(30)


def main():
    base = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_BASE
    print(f"smoke target: {base}", flush=True)

    img_bytes = base64.b64decode(json.load(open(BUNDLE))["pixels"])
    print(f"  reference photo: {len(img_bytes)} bytes (from committed kitchen bundle)")

    wait_for_space_ready()

    job = post_convert(base, img_bytes)
    jid = job.get("id") or job.get("jobId")
    if not jid:
        raise SystemExit(f"FAIL: convert returned no job id: {job}")
    print(f"  job: {jid}", flush=True)

    deadline = time.time() + JOB_WAIT_S
    status = None
    while time.time() < deadline:
        status = get_json(f"{base}/viz2d/jobs/{jid}").get("status")
        if status not in ("PENDING", "PROCESSING"):
            break
        time.sleep(5)
    if status != "COMPLETED":
        raise SystemExit(f"FAIL: job status {status}")

    bundle = get_json(f"{base}/viz2d/jobs/{jid}/file", timeout=120)
    segs = bundle.get("segments") or []
    if not segs:
        raise SystemExit("FAIL: no segments in live bundle")
    seg = max(segs, key=lambda s: s.get("metadata", {}).get("surfacePixels", 0))
    meta = seg.get("metadata", {})
    plane = seg.get("plane") or {}

    checks = [
        ("depthEnabled", meta.get("depthEnabled") is True),
        # R1-5 — either metric plane source is correct: moge-plane when the
        # point-map fit passes its gates, depth-plane otherwise.
        ("geometrySource in {depth-plane, moge-plane}",
         plane.get("geometrySource") in ("depth-plane", "moge-plane")),
        ("metersPerUnit in [0.5, 2.0]",
         isinstance(plane.get("metersPerUnit"), (int, float))
         and 0.5 <= plane["metersPerUnit"] <= 2.0),
        ("plane size sane (0.5..20 m)",
         0.5 <= plane.get("width", 0) <= 20 and 0.5 <= plane.get("height", 0) <= 20),
        ("shadeMap present", bool(seg.get("shadeMap"))),
        ("shadeRange present", bool(seg.get("shadeRange"))),
        # R4-1 — the intrinsic model must actually engage; a silent fall back
        # to the luminance heuristic (load failure, missing weights, trust
        # prompt) is exactly the class of regression this gate exists for.
        ("shadingSource=intrinsic", meta.get("shadingSource") == "intrinsic"),
    ]
    ok = True
    for label, passed in checks:
        print(f"  [{'PASS' if passed else 'FAIL'}] {label}")
        ok &= passed
    if not ok:
        print(f"\n  plane: { {k: plane.get(k) for k in ('geometrySource', 'metersPerUnit', 'width', 'height')} }")
        print(f"  metadata: { {k: meta.get(k) for k in ('depthEnabled', 'segmenter')} }")
        raise SystemExit("LIVE SMOKE FAILED — the deployed Space is not serving the expected stack")

    print("\nLIVE SMOKE PASSED")
    return 0


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