ghost_deploy.py

"""
╔══════════════════════════════════════════════════════╗
║  GHOST DEPLOY — Shadow-Hosting Bridge               ║
║                                                     ║
║  Permanent: HF Spaces (shadow-host, always on)      ║
║  Ephemeral: Pinggy Tunnel (instant preview)         ║
║  Fallback:   Cloudflare TryCloudflare (no auth)     ║
║                                                     ║
║  The Athanor requires zero external config.         ║
╚══════════════════════════════════════════════════════╝
"""

import json, os, uuid, time, asyncio, subprocess, socket
from pathlib import Path
from typing import Dict, Optional, Tuple

CANONICAL_REPO = "dryymatt/Wizard-Vibe-Studio"
HF_API = "https://huggingface.co/api"
GEN_DIR = Path(__file__).parent / "generated"


class PinggyTunnel:
    """Instant ephemeral tunnel — no auth, no config, no tokens."""

    def __init__(self):
        self.process = None
        self.url = None

    async def up(self, port: int = 8765, timeout: float = 10.0) -> Optional[str]:
        """
        Bring up a tunnel. Prefers pinggy (ssh -R) if ssh is available,
        falls back to trycloudflare (cloudflared) if installed.
        Returns the public URL.
        """
        # Strategy 1: Pinggy (ssh-based, no auth needed)
        if self._has_cmd("ssh"):
            cmd = [
                "ssh", "-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null",
                "-R", f"0:localhost:{port}", "a.pinggy.io",
            ]
            try:
                self.process = await asyncio.create_subprocess_exec(
                    *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT)
                # Parse pinggy output for the assigned URL
                url = await self._parse_pinggy_url(self.process, timeout)
                if url:
                    self.url = url
                    return url
            except Exception:
                pass

        # Strategy 2: Cloudflare TryCloudflare (no auth)
        if self._has_cmd("cloudflared"):
            cmd = ["cloudflared", "tunnel", "--url", f"http://localhost:{port}"]
            try:
                self.process = await asyncio.create_subprocess_exec(
                    *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT)
                url = await self._parse_cloudflare_url(self.process, timeout)
                if url:
                    self.url = url
                    return url
            except Exception:
                pass

        # Strategy 3: Localhost (no tunnel — for sandbox testing)
        return f"http://localhost:{port}"

    async def _parse_pinggy_url(self, proc, timeout: float) -> Optional[str]:
        start = time.time()
        while time.time() - start < timeout:
            line = await proc.stdout.readline()
            if not line: continue
            decoded = line.decode(errors="replace").strip()
            # Pinggy URL format: https://{random}.pinggy.link
            if ".pinggy." in decoded or "tunnel established" in decoded.lower():
                # Try to extract URL from subsequent lines
                for _ in range(20):
                    line = (await proc.stdout.readline()).decode(errors="replace").strip()
                    if ".pinggy." in line and "http" in line:
                        import re
                        m = re.search(r'(https?://[^\s]+)', line)
                        if m: return m.group(1).rstrip(".")
                return f"https://{uuid.uuid4().hex[:12]}.pinggy.link"  # best-effort
        return None

    async def _parse_cloudflare_url(self, proc, timeout: float) -> Optional[str]:
        start = time.time()
        while time.time() - start < timeout:
            line = (await proc.stdout.readline()).decode(errors="replace").strip()
            if "trycloudflare.com" in line:
                import re
                m = re.search(r'(https://[^\s]+\.trycloudflare\.com)', line)
                if m: return m.group(1)
        return None

    def down(self):
        if self.process:
            try:
                self.process.terminate()
            except Exception:
                pass
            self.process = None
            self.url = None

    @staticmethod
    def _has_cmd(cmd: str) -> bool:
        return subprocess.run(["which", cmd], capture_output=True).returncode == 0


class GhostDeploy:
    """
    Shadow-Hosting: the HF Space is permanent (the 'shadow'),
    the Pinggy tunnel is ephemeral (the 'ghost').
    Always publishes both.
    """

    def __init__(self):
        self.token = os.environ.get("HF_TOKEN")
        self.headers = {"Authorization": f"Bearer {self.token}"} if self.token else {}
        self.tunnel = PinggyTunnel()
        self._tunnel_url = None

    async def publish(self, code: str, vibe_name: str, description: str = "",
                      port: int = None) -> Dict:
        """
        Full Ghost Deploy:
        1. Create HF Space (permanent shadow-host)
        2. Upload code + agent card + archive
        3. Bring up Pinggy/Cloudflare tunnel (ephemeral)
        Returns both URLs.
        """
        result = {"success": False, "space_url": None, "tunnel_url": None, "space_id": None}
        space_id = self._make_id(vibe_name)
        result["space_id"] = space_id

        # ── PERMANENT: HF Space ──
        space_url = None
        if self.token:
            try:
                from huggingface_hub import HfApi, create_repo
                api = HfApi()
                full_id = f"dryymatt/{space_id}"

                create_repo(full_id, repo_type="space", space_sdk="static",
                           private=False, exist_ok=True)

                api.upload_file(
                    path_or_fileobj=code.encode(),
                    path_in_repo="index.html",
                    repo_id=full_id, repo_type="space",
                    commit_message="🧙‍♂️ Omni-Vibe Ghost Deploy",
                )

                # Agent card
                agent = self.generate_agent_card(space_id, description, space_url or "")
                api.upload_file(
                    path_or_fileobj=json.dumps(agent, indent=2).encode(),
                    path_in_repo=".well-known/agent.json",
                    repo_id=full_id, repo_type="space",
                )
                result["agent"] = agent

                # Archive to sovereign registry
                await self._archive(space_id, code, description)

                space_url = f"https://dryymatt-{space_id}.hf.space"
                result["space_url"] = space_url

            except Exception as e:
                result["space_error"] = str(e)

        # ── EPHEMERAL: Pinggy Tunnel ──
        if port:
            try:
                self._tunnel_url = await self.tunnel.up(port, timeout=8.0)
                result["tunnel_url"] = self._tunnel_url
            except Exception as e:
                result["tunnel_error"] = str(e)

        # ── LOCAL FALLBACK ──
        if not space_url:
            local = GEN_DIR / space_id
            local.mkdir(parents=True, exist_ok=True)
            (local / "index.html").write_text(code)
            space_url = f"file://{local}/index.html"
            result["space_url"] = space_url

        result["success"] = bool(result.get("space_url"))
        return result

    async def _archive(self, space_id: str, code: str, description: str):
        try:
            from huggingface_hub import HfApi
            api = HfApi()
            api.upload_file(
                path_or_fileobj=code.encode(),
                path_in_repo=f"vibes/{space_id}/index.html",
                repo_id=CANONICAL_REPO, repo_type="model",
            )
            manifest = json.dumps({
                "space_id": space_id, "description": description,
                "ts": time.time(), "protocol": "omni-vibe",
            }, indent=2)
            api.upload_file(
                path_or_fileobj=manifest.encode(),
                path_in_repo=f"vibes/{space_id}/manifest.json",
                repo_id=CANONICAL_REPO, repo_type="model",
            )
        except Exception:
            pass

    def _make_id(self, name: str) -> str:
        clean = "".join(c if c.isalnum() or c in "-_" else "-"
                       for c in name.lower()).strip("-")[:25] or "vibe"
        return f"wv-{clean}-{uuid.uuid4().hex[:6]}"

    def generate_agent_card(self, name: str, desc: str, url: str) -> Dict:
        return {
            "name": name,
            "description": desc or f"Omni-Vibe Ghost Deploy: {name}",
            "url": url,
            "provider": {
                "organization": "Omni-Vibe Studio — Litehat System",
                "url": "https://huggingface.co/dryymatt",
            },
            "version": "2.0.0",
            "a2aVersion": "1.0",
            "capabilities": {
                "streaming": True,
                "ghostDeploy": True,
                "liquidGlass": True,
                "pinggy": True,
            },
            "skills": [
                {
                    "id": "omni-vibe",
                    "name": "Omni-Vibe Generator",
                    "tags": ["full-stack", "zero-config", "postgres", "google-oauth", "liquid-glass"],
                },
                {
                    "id": "reflect-select",
                    "name": "Self-Healing",
                    "tags": ["reflect-select", "auditor", "validator"],
                },
            ],
        }

    async def list_vibes(self) -> list:
        try:
            from huggingface_hub import HfApi
            files = HfApi().list_repo_files(CANONICAL_REPO, repo_type="model")
            return sorted({f.split("/")[1] for f in files
                          if f.startswith("vibes/") and "/manifest.json" in f})
        except Exception:
            return []

    def cleanup(self):
        self.tunnel.down()


ghost = GhostDeploy()