scripts/deploy_preflight.py

"""Deploy / record preflight — is the app ready to ship to the Space and record?

Distinct from `scripts/preflight.py` (which gates the live *model* on the real
stack). This one gates *deployment + recording readiness* and runs fully offline:

  • local build is sound (imports, builds the UI, core tests pass, no errors)
  • every file the Space needs is present + the README frontmatter is valid
  • the prompt won't regress (reference block is lean, ledger is the clean baseline)
  • the credentials/tooling to actually push exist — and if not, says exactly what
    to set and where (this remote session does NOT carry HF_TOKEN by default)
  • (only if a token is present) the live Space exists and what's deployed

GO  → safe to `hf upload` + reboot + record.   Run:  make deploy-check
"""

from __future__ import annotations

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

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
SPACE = "build-small-hackathon/microfactory-lab"
FIELD_LOG_DATASET = "build-small-hackathon/chief-engineer-field-log"
# Uploaded to the Space = everything EXCEPT these (keeps learn/ + assets/ + data/*.jsonl,
# which the app imports/needs; drops docs, spikes, secrets, caches, runtime/transient files).
# Public-facing learn/finetune docs that stay on the Space: README.md, MODEL_CARD*.md,
# SERVING.md, OLLAMA_PUBLISHING.md. Internal session/budget/iteration logs stay in the
# GitHub repo but are kept out of the Space.
SPACE_IGNORE = [
    # Docs are internal by default, but two reference docs are public-facing.
    "docs/**",
    "!docs/reference/DEPLOYMENT.md",
    "!docs/reference/SIMULATION.md",
    "spike/**", "field_logs/**", "deliberation_logs/**", ".venv/**", "node_modules/**",
    "recordings/**", "**/__pycache__/**", "**/*.pyc", ".git/**", ".gitignore", ".env", ".agents/**",
    ".codeboarding/**", ".codeboardingignore", "data/policy.json", "data/_generated.glb",
    "data/_vprint.gif", "uv.lock", ".pytest_cache/**", "*.cap/**",
    # Internal finetune workpapers kept internal.
    "learn/finetune/REPORT.md",
    "learn/finetune/REPORT_v1.md",
]
_fail: list[str] = []
_warn: list[str] = []


def ok(gate: str, detail: str = "") -> None:
    print(f"🟢 {gate}{' — ' + detail if detail else ''}")


def warn(gate: str, detail: str) -> None:
    _warn.append(gate)
    print(f"🟡 {gate} — {detail}")


def fail(gate: str, detail: str) -> None:
    _fail.append(gate)
    print(f"🔴 {gate} — {detail}")


# ── D1 · local build is sound ────────────────────────────────────────────────
def d1_build() -> None:
    try:
        import app  # noqa: F401  (also exercises the whole import graph)
        if type(getattr(app, "demo", None)).__name__ != "Blocks":
            fail("D1 build", "app.demo is not a Gradio Blocks — build() did not run")
            return
        ok("D1 build", "app imports + builds the UI (Studio→Build→Print→Review)")
    except Exception as e:  # noqa: BLE001
        fail("D1 build", f"import/build error: {e!r}")


def d1b_tests() -> None:
    r = subprocess.run([sys.executable, str(ROOT / "test_core.py")],
                       cwd=ROOT, capture_output=True, text=True)
    if r.returncode == 0 and "ALL CORE TESTS PASSED" in (r.stdout + r.stderr):
        ok("D1 tests", "core tests pass (offline)")
    else:
        fail("D1 tests", f"test_core.py failed (rc={r.returncode}); run `make test`")


# ── D2 · everything the Space needs is present ───────────────────────────────
def d2_files() -> None:
    required = ["app.py", "README.md", "requirements.txt", "core", "ingest", "sim",
                "scripts", "learn", "data/seed_lessons.jsonl", "data/references.jsonl",
                "data/lessons.jsonl"]
    missing = [p for p in required if not (ROOT / p).exists()]
    if missing:
        fail("D2 files", f"missing for the Space: {', '.join(missing)}")
    else:
        ok("D2 files", "all app + data files present")
    if not (ROOT / "assets" / "benchy.glb").exists():
        warn("D2 assets", "assets/benchy.glb missing — the hero quick-load won't render")


# ── D3 · README frontmatter the Space build reads ────────────────────────────
def d3_frontmatter() -> None:
    txt = (ROOT / "README.md").read_text(encoding="utf-8")
    m = re.match(r"^---\n(.*?)\n---", txt, re.S)
    if not m:
        fail("D3 README", "no YAML frontmatter block")
        return
    fm = m.group(1)
    need = {"sdk": "gradio", "app_file": "app.py"}
    for k, v in need.items():
        if not re.search(rf"^{k}:\s*{re.escape(v)}\s*$", fm, re.M):
            fail("D3 README", f"frontmatter `{k}: {v}` missing/wrong")
            return
    if not re.search(r"^sdk_version:\s*\S+", fm, re.M):
        fail("D3 README", "sdk_version missing")
        return
    sd = re.search(r"^short_description:\s*(.+)$", fm, re.M)
    if not sd:
        warn("D3 README", "no short_description")
    elif len(sd.group(1).strip().strip('"')) > 60:
        fail("D3 README", f"short_description >60 chars ({len(sd.group(1).strip())}) — upload will reject")
    else:
        ok("D3 README", "frontmatter valid (sdk/app_file/sdk_version/short_description)")


# ── D4 · requirements carry the Space (zerogpu) deps ─────────────────────────
def d4_requirements() -> None:
    req = (ROOT / "requirements.txt").read_text(encoding="utf-8").lower()
    needed = ["gradio", "spaces", "torch", "transformers", "trimesh", "shapely", "pydantic"]
    missing = [p for p in needed if p not in req]
    if missing:
        fail("D4 requirements", f"missing pins: {', '.join(missing)} (Space build/zerogpu needs them)")
    else:
        ok("D4 requirements", "core + zerogpu deps inlined")


# ── D5 · the prompt won't regress (lean reference block) ─────────────────────
def d5_reference_block() -> None:
    try:
        from ingest.distill import reference_block
        worst = max((len(reference_block(m)) for m in ("PLA", "PETG", "ABS", "TPU")), default=0)
        if worst == 0:
            warn("D5 reference", "reference_block returned nothing — references not loaded?")
        elif worst > 12:
            fail("D5 reference", f"reference block is {worst} lines/material — prompt flood regression")
        else:
            ok("D5 reference", f"lean ({worst} lines/material max)")
    except Exception as e:  # noqa: BLE001
        fail("D5 reference", f"reference_block error: {e!r}")


# ── D6 · the deployed ledger is the clean baseline (no demo junk) ────────────
def d6_ledger() -> None:
    srcs: dict[str, int] = {}
    for line in (ROOT / "data" / "lessons.jsonl").read_text(encoding="utf-8").splitlines():
        if line.strip():
            try:
                srcs[json.loads(line).get("source", "?")] = srcs.get(json.loads(line).get("source", "?"), 0) + 1
            except Exception:
                continue
    runtime = srcs.get("earned", 0) + srcs.get("sim", 0)
    summary = ", ".join(f"{k}:{v}" for k, v in sorted(srcs.items()))
    if runtime:
        warn("D6 ledger", f"{runtime} runtime lesson(s) in the ledger ({summary}) — "
                          "reset before upload: git checkout -- data/lessons.jsonl")
    else:
        ok("D6 ledger", f"clean baseline ({summary})")


# ── D7 · data integrity (valid JSONL, enums) ─────────────────────────────────
def d7_data() -> None:
    try:
        from core.models import MATERIALS, OUTCOMES
        bad = 0
        for line in (ROOT / "data" / "references.jsonl").read_text(encoding="utf-8").splitlines():
            if line.strip():
                r = json.loads(line)
                if not {"material", "param", "value", "source"} <= set(r):
                    bad += 1
        obs = ROOT / "sim" / "calibration" / "observations.modal.jsonl"
        obad = sum(1 for l in obs.read_text().splitlines()
                   if l.strip() and json.loads(l).get("outcome") not in OUTCOMES) if obs.exists() else 0
        if bad or obad:
            warn("D7 data", f"{bad} malformed ref rows, {obad} bad-outcome obs rows")
        else:
            ok("D7 data", "references + calibration obs well-formed")
    except Exception as e:  # noqa: BLE001
        warn("D7 data", f"could not validate: {e!r}")


# ── D8 · credentials + tooling to actually deploy ────────────────────────────
def d8_credentials() -> None:
    has_hf_cli = shutil.which("hf") is not None
    token = (os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN")
             or os.environ.get("HUGGINGFACE_TOKEN"))
    whoami = None
    if has_hf_cli:
        r = subprocess.run(["hf", "auth", "whoami"], capture_output=True, text=True)
        if r.returncode == 0 and "Not logged in" not in r.stdout:
            whoami = r.stdout.strip().splitlines()[0] if r.stdout.strip() else "ok"
    if not has_hf_cli:
        fail("D8 hf-cli", "`hf` not installed — `uv pip install -U huggingface_hub`")
    if token or whoami:
        ok("D8 hf-auth", f"authenticated ({'token env' if token else 'hf login: ' + str(whoami)})")
    else:
        warn("D8 hf-auth", "NO HF credentials in this session. To deploy: run the `hf upload` "
                           "from a machine where you've run `hf auth login`, OR set HF_TOKEN "
                           "in the environment (HF write token, member of build-small-hackathon).")
    return bool(token or whoami)


# ── D9 · (token only) what's live on the Space right now ─────────────────────
def d9_space(authed: bool) -> None:
    if not authed:
        warn("D9 space", "skipped — no credentials to query the live Space")
        return
    try:
        from huggingface_hub import HfApi
        api = HfApi()
        rt = api.get_space_runtime(SPACE)
        files = api.list_repo_files(SPACE, repo_type="space")
        has_core = "app.py" in files and any(f.startswith("core/") for f in files)
        stage = getattr(rt, "stage", "?")
        ok("D9 space", f"reachable · stage={stage} · app.py+core present={has_core} · {len(files)} files")
        if not has_core:
            warn("D9 space", "Space is missing app.py/core — likely the pre-restructure build; push the update")
    except Exception as e:  # noqa: BLE001
        warn("D9 space", f"could not query Space: {e!r}")


# ── D10 · the field-log dataset is set + reachable (Sharing-is-Caring / all-runs) ──
def d10_dataset(authed: bool) -> None:
    if not authed:
        warn("D10 dataset", "skipped — no credentials to verify the field-log dataset")
        return
    try:
        from huggingface_hub import HfApi
        api = HfApi()
        if api.repo_exists(FIELD_LOG_DATASET, repo_type="dataset"):
            files = api.list_repo_files(FIELD_LOG_DATASET, repo_type="dataset")
            logged = any(f.endswith("interactions.jsonl") for f in files)
            ok("D10 dataset", f"{FIELD_LOG_DATASET} exists"
               + (" · interactions.jsonl present (runs are logging)" if logged
                  else " · no interactions.jsonl yet (do one BUILD on the Space to confirm)"))
        else:
            warn("D10 dataset", f"{FIELD_LOG_DATASET} not found — create it, or let CommitScheduler "
                                "make it on first run (needs HF_TOKEN as a Space secret).")
    except Exception as e:  # noqa: BLE001
        warn("D10 dataset", f"could not verify dataset: {e!r}")


# ── push: actually update the Space files (gated on green + auth) ─────────────
def push_space(factory_reboot: bool = True) -> None:
    """Upload the app to the Space (everything except SPACE_IGNORE) and reboot.
    Only runs after the gates pass + credentials are present."""
    try:
        from huggingface_hub import HfApi
    except Exception as e:  # noqa: BLE001
        fail("PUSH", f"huggingface_hub unavailable: {e!r}")
        return
    api = HfApi()
    print(f"\n⏫ uploading {ROOT.name}/ → {SPACE} (excluding docs, spike, caches, secrets)…")
    try:
        api.upload_folder(repo_id=SPACE, repo_type="space", folder_path=str(ROOT),
                          ignore_patterns=SPACE_IGNORE,
                          commit_message="deploy: update Space from deploy_preflight --push")
        ok("PUSH", "files uploaded")
        if factory_reboot:
            api.restart_space(SPACE, factory_reboot=True)
            ok("PUSH", "factory reboot requested — Space rebuilding (~1-2 min)")
        print("   Next: wait for build, then smoke-test (BUILD shows reasoning not Error; "
              "O'Brien/La Forge; reset button; wide UI).")
    except Exception as e:  # noqa: BLE001
        fail("PUSH", f"upload/restart failed: {e!r}")


def main() -> None:
    ap = argparse.ArgumentParser(description="Deploy/record readiness gate (+ optional Space push).")
    ap.add_argument("--push", action="store_true",
                    help="after the gates pass, UPDATE the Space files (hf upload) + factory reboot")
    ap.add_argument("--no-reboot", action="store_true", help="with --push, skip the factory reboot")
    args = ap.parse_args()

    print("Deploy / record preflight — " + SPACE)
    print("=" * 70)
    d1_build()
    d1b_tests()
    d2_files()
    d3_frontmatter()
    d4_requirements()
    d5_reference_block()
    d6_ledger()
    d7_data()
    authed = d8_credentials()
    d9_space(authed)
    d10_dataset(authed)
    print("=" * 70)
    if _fail:
        print(f"🔴 NO-GO: fix {len(_fail)} blocker(s) — {', '.join(_fail)}")
        sys.exit(1)
    if args.push:
        if not authed:
            print("🔴 --push needs HF credentials (HF_TOKEN or `hf auth login`). Nothing pushed.")
            sys.exit(1)
        push_space(factory_reboot=not args.no_reboot)
        sys.exit(1 if _fail else 0)
    if _warn:
        print(f"🟡 GO with warnings ({', '.join(_warn)}) — read them; credentials/Space "
              "warnings just mean 'deploy from an authenticated machine'. "
              "Run with --push (authenticated) to update the Space.")
        sys.exit(0)
    print("🟢 GO — local build clean, files + frontmatter ready, authenticated. "
          "Re-run with --push to update the Space + reboot, then smoke-test → record.")


if __name__ == "__main__":
    main()