STABLE VIBE: ghost_deploy.py — port 8765, relative paths

ghost_deploy.py

@@ -0,0 +1,261 @@
+"""
+╔══════════════════════════════════════════════════════╗
+║  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()