""" ╔══════════════════════════════════════════════════════╗ ║ 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()