""" ╔══════════════════════════════════════════════════════════════╗ ║ WIZARD-VIBE CORE — Sandbox-First Architecture ║ ║ Single-file core: SSE server + orchestrator + self-heal ║ ║ Hot-reload watcher + iframe sandbox + A2A agent card ║ ╚══════════════════════════════════════════════════════════════╝ Architecture: core.py ← this file: everything in one module hot_reload.py ← watchdog-based file watcher for dev mode static/ ← Liquid Glass UI (minimalist HTML/CSS/JS) sandbox.sh ← one-command bootstrap script Dockerfile ← containerized deployment Protocol: GET / → Liquid Glass UI GET /api/health → Health check POST /api/stream → SSE streaming (AsyncIterator) POST /api/publish → GitHub + HF Spaces + agent.json GET /.well-known/agent.json → A2A agent card GET /api/preview?session_id=X → sandbox iframe content """ import asyncio import json import os import re import sys import time import uuid import base64 from pathlib import Path from typing import Dict, List, Optional, Tuple import aiohttp from aiohttp import web # ═══════════════════════════════════════════════════════════════ # CONFIGURATION # ═══════════════════════════════════════════════════════════════ PORT = int(os.environ.get("WIZARD_PORT", 8765)) STATIC_DIR = Path(__file__).parent / "static" GENERATED_DIR = Path(__file__).parent / "generated" GENERATED_DIR.mkdir(parents=True, exist_ok=True) HOT_RELOAD = os.environ.get("WIZARD_HOT_RELOAD", "0") == "1" # ═══════════════════════════════════════════════════════════════ # ORCHESTRATOR — Model Posing Engine # ═══════════════════════════════════════════════════════════════ class Orchestrator: """Routes tasks to optimal models by keyword matching.""" REGISTRY = { "vision": "microsoft/Phi-3-vision-128k-instruct", "logic": "deepseek-ai/DeepSeek-V3-0324", "code": "Qwen/Qwen3-Coder-30B-A3B-Instruct", "fallback": "mistralai/Mistral-7B-Instruct-v0.3", } KEYWORDS = { "vision": ["ui","design","css","html","layout","color","animation","svg","interface","visual","style","component","frontend","page","web"], "logic": ["algorithm","sort","search","optimize","compute","math","reasoning","backend","api","database","schema","query","logic"], "code": ["code","generate","build","create","app","website","script","function","class","module"], } def pose(self, prompt: str) -> Dict: p = prompt.lower() scores = {d: sum(1 for kw in self.KEYWORDS[d] if kw in p) for d in self.KEYWORDS} best = max(scores, key=scores.get) model = self.REGISTRY.get(best, self.REGISTRY["fallback"]) return {"model": model, "domain": best, "score": scores[best]} def generate(self, prompt: str): """Generator yielding code chunks — simulates model streaming.""" p = prompt.lower() if "game" in p: yield from self._game() elif "api" in p or "backend" in p or "server" in p: yield from self._api() elif "component" in p or "react" in p or "vue" in p: yield from self._component() else: yield from self._landing_page(prompt) def _landing_page(self, prompt: str): title = "Wizard-Vibe" for w in ["portfolio","startup","saas","agency","product"]: if w in prompt.lower(): title = f"{w.title()} — Wizard-Vibe"; break yield from ( '\n\n\n', '\n\n', f'{title}\n\n\n\n', '
\n', '\n', '\n', '\n', '\n', '
\n', '
\n', f'

{title}

\n', '

Built with Wizard-Vibe Core — sandbox-first, hot-reload, A2A-native.

\n', '
\n', '\n', '\n', '
\n', '
\n', '

⚡ Sandbox-First

Every line executes in an isolated iframe sandbox before publish.

\n', '

🔄 Hot Reload

File watcher restarts server on every change — zero latency dev loop.

\n', '

🚀 A2A Native

Every deploy auto-generates agent.json for ecosystem discovery.

\n', '
\n', '\n', '\n\n', ) def _game(self): yield from ( '\nWizard-Vibe Game', '

🧙‍♂️ Wizard-Vibe Game

', '', '', '\n', ) def _api(self): yield from ( 'from flask import Flask,request,jsonify\nfrom flask_cors import CORS\n\napp=Flask(__name__)\nCORS(app)\n\n', '@app.route("/api/health")\ndef health():\n return jsonify({"status":"ok","engine":"Wizard-Vibe"})\n\n', 'if __name__=="__main__":\n app.run(host="0.0.0.0",port=8080)\n', ) def _component(self): yield from ( '\n', '\n', '\n', ) # ═══════════════════════════════════════════════════════════════ # REFLECT-SELECT — Self-Healing Engine # ═══════════════════════════════════════════════════════════════ class ReflectSelect: """Autonomous error detection and self-healing loop.""" VOID = {'br','hr','img','input','meta','link','area','base','col','embed','source','track','wbr'} PATTERNS = [ (r"console\.loge\(", "console.log("), (r"docment\.", "document."), (r"getElementbyId", "getElementById"), (r"innerHtml", "innerHTML"), (r"functon\s", "function "), (r"retrun", "return"), ] def heal(self, code: str, sandbox_errors: Optional[List[str]] = None) -> Tuple[str, int, int]: errors = self._detect(code) if sandbox_errors: errors.extend(sandbox_errors) found = len(errors) if not found: return code, 0, 0 fixed = 0 strategies = ["syntax", "logic", "structural"] for i in range(5): if not errors: break s = strategies[i % 3] if s == "syntax": code = self._fix_syntax(code) elif s == "logic": code = self._fix_logic(code) elif s == "structural": code = self._fix_structural(code) fixed += 1 errors = self._detect(code) return code, found, fixed def _detect(self, code: str) -> List[str]: e = [] if "', 'exec') except SyntaxError as se: e.append(f"Syntax: {se}") if code.count('(') != code.count(')'): e.append("Parenthesis mismatch") if code.count('{') != code.count('}'): e.append("Brace mismatch") return e def _html_errors(self, code: str) -> List[str]: e, stack = [], [] for closing, tag in re.findall(r'<(/?)(\w+)', code, re.IGNORECASE): t = tag.lower() if t in self.VOID: continue if closing: if stack and stack[-1] == t: stack.pop() else: e.append(f"Unmatched ") else: stack.append(t) for t in stack: e.append(f"Unclosed <{t}>") return e def _fix_syntax(self, code: str) -> str: for pat, rep in self.PATTERNS: code = re.sub(pat, rep, code) return code def _fix_logic(self, code: str) -> str: if 'fetch(' in code and '.catch' not in code: code = re.sub(r'(fetch\([^)]+\)[^;]*);', r'\1.catch(e => console.error(e));', code) return code def _fix_structural(self, code: str) -> str: c = code.strip() if (''): c = c.rstrip() + '\n' return c # ═══════════════════════════════════════════════════════════════ # SANDBOX VALIDATOR # ═══════════════════════════════════════════════════════════════ def sandbox_validate(code: str) -> Dict: """Validate code in a logical sandbox — tag balancing, syntax checks.""" rs = ReflectSelect() errors = rs._detect(code) return {"success": len(errors) == 0, "errors": errors} # ═══════════════════════════════════════════════════════════════ # GLOBAL STATE # ═══════════════════════════════════════════════════════════════ class State: def __init__(self): self.orchestrator = Orchestrator() self.reflect = ReflectSelect() self.sessions: Dict = {} self.codes: Dict[str, str] = {} self.sandbox: Dict[str, str] = {} self.publish_ready: Dict[str, bool] = {} state = State() # ═══════════════════════════════════════════════════════════════ # SSE STREAMING — AsyncIterator[StreamResponse] # ═══════════════════════════════════════════════════════════════ async def stream_gen(sid: str, prompt: str): """Core streaming generator — yields SSE events.""" state.sessions[sid]["status"] = "streaming" state.sandbox[sid] = "building" # Phase 1: Orchestrate yield _sse("phase", {"phase": "orchestrate"}) plan = state.orchestrator.pose(prompt) yield _sse("plan", plan) # Phase 2: Generate yield _sse("phase", {"phase": "generate"}) code = "" for chunk in state.orchestrator.generate(prompt): code += chunk state.codes[sid] = code yield _sse("code", {"chunk": chunk, "partial": code}) await asyncio.sleep(0.03) # Phase 3: Reflect-Select yield _sse("phase", {"phase": "heal"}) healed, found, fixed = state.reflect.heal(code) state.codes[sid] = healed yield _sse("heal", {"found": found, "fixed": fixed}) # Phase 4: Sandbox yield _sse("phase", {"phase": "sandbox"}) result = sandbox_validate(healed) if result["success"]: state.sandbox[sid] = "stable" state.publish_ready[sid] = True yield _sse("sandbox", {"status": "stable"}) else: state.sandbox[sid] = "error" yield _sse("sandbox", {"status": "error", "errors": result["errors"]}) for attempt in range(3): healed, _, _ = state.reflect.heal(healed, result["errors"]) state.codes[sid] = healed result = sandbox_validate(healed) if result["success"]: state.sandbox[sid] = "stable" state.publish_ready[sid] = True yield _sse("sandbox", {"status": "stable", "heal_attempts": attempt + 1}) break yield _sse("heal", {"attempt": attempt + 1}) state.sessions[sid]["status"] = "complete" yield _sse("done", {"status": state.sandbox[sid]}) def _sse(event: str, data: dict) -> str: return f"event: {event}\ndata: {json.dumps(data)}\n\n" # ═══════════════════════════════════════════════════════════════ # HTTP HANDLERS # ═══════════════════════════════════════════════════════════════ async def handle_stream(request: web.Request) -> web.StreamResponse: sid = str(uuid.uuid4())[:8] data = await request.json() prompt = data.get("prompt", "") state.sessions[sid] = {"id": sid, "prompt": prompt, "status": "init"} state.sandbox[sid] = "building" state.publish_ready[sid] = False resp = web.StreamResponse(status=200, headers={ "Content-Type": "text/event-stream", "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", }) await resp.prepare(request) try: async for event in stream_gen(sid, prompt): await resp.write(event.encode()) await resp.write(b"event: close\ndata: {}\n\n") except Exception as e: await resp.write(f"event: error\ndata: {json.dumps({'error': str(e)})}\n\n".encode()) return resp async def handle_publish(request: web.Request) -> web.Response: data = await request.json() sid = data.get("session_id") repo = data.get("repo_name", f"wizard-vibe-{sid}") if not state.publish_ready.get(sid): return web.json_response({"error": "Sandbox not stable"}, status=400) code = state.codes.get(sid, "") if not code: return web.json_response({"error": "No code generated"}, status=400) # Save locally out = GENERATED_DIR / repo out.mkdir(exist_ok=True) (out / "index.html").write_text(code) # Generate agent card agent = { "name": repo, "description": f"Wizard-Vibe Core generated app", "url": f"https://dryymatt-{repo}.hf.space", "version": "1.0.0", "a2aVersion": "1.0", "capabilities": {"streaming": True}, "skills": [{"id": "vibe-deploy", "name": "Vibe Deploy", "tags": ["wizard-vibe", "core", "a2a"]}], } (out / "agent.json").write_text(json.dumps(agent, indent=2)) # Try GitHub push token = os.environ.get("GITHUB_TOKEN") or os.environ.get("HF_TOKEN") github = {"success": False} if token: try: async with aiohttp.ClientSession() as s: headers = {"Authorization": f"Bearer {token}", "Accept": "application/vnd.github+json"} u = (await s.get("https://api.github.com/user", headers=headers)).json() owner = u["login"] await s.post("https://api.github.com/user/repos", headers=headers, json={"name": repo, "auto_init": True}) for path in ["index.html", ".well-known/agent.json"]: content = (out / path).read_text() if (out / path).exists() else "" await s.put(f"https://api.github.com/repos/{owner}/{repo}/contents/{path}", headers=headers, json={ "message": f"✨ Wizard-Vibe Core: {path}", "content": base64.b64encode(content.encode()).decode(), }) github = {"success": True, "url": f"https://github.com/{owner}/{repo}"} except Exception as e: github = {"success": False, "error": str(e)} state.sessions[sid]["published"] = True return web.json_response({"success": True, "github": github, "agent": agent}) async def handle_health(request: web.Request) -> web.Response: return web.json_response({"status": "alive", "engine": "Wizard-Vibe Core", "sandbox_first": True, "hot_reload": HOT_RELOAD}) async def handle_agent_card(request: web.Request) -> web.Response: return web.json_response({ "name": "wizard-vibe-core", "description": "Wizard-Vibe Core — Sandbox-first SSE streaming code generator with Reflect-Select self-healing and A2A native deploy.", "url": "https://dryymatt-wizard-vibe-core.hf.space", "version": "1.0.0", "a2aVersion": "1.0", "capabilities": {"streaming": True, "pushNotifications": False, "stateTransitionHistory": True}, "skills": [ {"id": "code-generation", "name": "Code Generation", "tags": ["wizard-vibe", "core", "sse", "streaming"]}, {"id": "reflect-select", "name": "Reflect-Select Heal", "tags": ["self-healing", "error-fix"]}, {"id": "github-publish", "name": "GitHub Publish", "tags": ["deploy", "git", "a2a"]}, ], }) async def handle_preview(request: web.Request) -> web.Response: sid = request.query.get("session_id", "") code = state.codes.get(sid, "") return web.Response(text=code, content_type="text/html") async def handle_status(request: web.Request) -> web.Response: sid = request.query.get("session_id", "") if sid in state.sessions: return web.json_response({ "status": state.sessions[sid]["status"], "sandbox": state.sandbox.get(sid), "publish_ready": state.publish_ready.get(sid), }) return web.json_response({"sessions": len(state.sessions)}) async def handle_static(request: web.Request) -> web.Response: path = request.match_info.get("path", "index.html") fp = STATIC_DIR / path if not fp.exists() or not fp.is_file(): return web.Response(text="Not found", status=404) ct = {".html": "text/html", ".css": "text/css", ".js": "application/javascript", ".svg": "image/svg+xml"} return web.Response(text=fp.read_text(), content_type=ct.get(fp.suffix, "text/plain")) # ═══════════════════════════════════════════════════════════════ # APP FACTORY # ═══════════════════════════════════════════════════════════════ def create_app() -> web.Application: app = web.Application() app.router.add_post("/api/stream", handle_stream) app.router.add_post("/api/publish", handle_publish) app.router.add_get("/api/status", handle_status) app.router.add_get("/api/preview", handle_preview) app.router.add_get("/api/health", handle_health) app.router.add_get("/.well-known/agent.json", handle_agent_card) app.router.add_get("/", handle_static) app.router.add_get("/{path:.*}", handle_static) return app def main(): print(f"🧙‍♂️ Wizard-Vibe Core [sandbox-first] :{PORT}") print(f" Hot-reload: {'ON' if HOT_RELOAD else 'OFF'}") print(f" Agent Card: /.well-known/agent.json") app = create_app() web.run_app(app, host="0.0.0.0", port=PORT) if __name__ == "__main__": main()