| """ |
| ╔══════════════════════════════════════════════════════════════╗ |
| ║ 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 |
|
|
| |
| |
| |
|
|
| 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" |
|
|
| |
| |
| |
|
|
| 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 ( |
| '<!DOCTYPE html>\n<html lang="en">\n<head>\n', |
| '<meta charset="UTF-8">\n<meta name="viewport" content="width=device-width,initial-scale=1.0">\n', |
| f'<title>{title}</title>\n<style>\n', |
| '*{margin:0;padding:0;box-sizing:border-box}\n', |
| 'body{font-family:-apple-system,BlinkMacSystemFont,"Segoe UI",Roboto,sans-serif;', |
| 'background:linear-gradient(135deg,#0a0a1a 0%,#1a0a2e 50%,#0a1a2e 100%);color:#e0e0ff;min-height:100vh}\n', |
| '.glass{background:rgba(255,255,255,.03);backdrop-filter:blur(20px);border:1px solid rgba(255,255,255,.06);border-radius:20px}\n', |
| '</style>\n</head>\n<body>\n', |
| '<div style="position:fixed;top:20px;left:50%;transform:translateX(-50%);z-index:100">\n', |
| '<svg width="64" height="64" viewBox="0 0 64 64" id="hat" style="filter:drop-shadow(0 0 20px rgba(139,92,246,.6))">\n', |
| '<defs><linearGradient id="g" x1="0%" y1="0%" x2="100%" y2="100%"><stop offset="0%" style="stop-color:#8B5CF6"/><stop offset="50%" style="stop-color:#06B6D4"/><stop offset="100%" style="stop-color:#10B981"/></linearGradient></defs>\n', |
| '<path d="M32 8 L8 48 L32 40 L56 48 Z" fill="url(#g)" stroke="rgba(255,255,255,.3)" stroke-width="1.5"/>\n', |
| '<ellipse cx="32" cy="48" rx="26" ry="7" fill="none" stroke="url(#g)" stroke-width="2" opacity=".6"/>\n', |
| '</svg></div>\n', |
| '<section style="display:flex;flex-direction:column;align-items:center;justify-content:center;min-height:100vh;padding:2rem">\n', |
| f'<h1 style="font-size:clamp(3rem,8vw,6rem);font-weight:800;background:linear-gradient(135deg,#8B5CF6,#06B6D4,#10B981);-webkit-background-clip:text;-webkit-text-fill-color:transparent;text-align:center">{title}</h1>\n', |
| '<p style="font-size:1.25rem;color:#a0a0cc;margin-top:1.5rem;text-align:center;max-width:600px">Built with Wizard-Vibe Core — sandbox-first, hot-reload, A2A-native.</p>\n', |
| '<div style="display:flex;gap:1rem;margin-top:2.5rem">\n', |
| '<button style="padding:.875rem 2rem;background:linear-gradient(135deg,#8B5CF6,#06B6D4);border:none;border-radius:12px;color:#fff;font-size:1rem;font-weight:600;cursor:pointer">Get Started</button>\n', |
| '<button style="padding:.875rem 2rem;background:rgba(255,255,255,.05);border:1px solid rgba(255,255,255,.1);border-radius:12px;color:#c0c0e0;font-size:1rem;font-weight:500;cursor:pointer">Learn More</button>\n', |
| '</div></section>\n', |
| '<section style="display:grid;grid-template-columns:repeat(auto-fit,minmax(280px,1fr));gap:2rem;padding:4rem 2rem;max-width:1200px;margin:0 auto">\n', |
| '<div class="glass" style="padding:2rem"><h3 style="color:#8B5CF6">⚡ Sandbox-First</h3><p style="color:#9090b0;margin-top:.75rem">Every line executes in an isolated iframe sandbox before publish.</p></div>\n', |
| '<div class="glass" style="padding:2rem"><h3 style="color:#06B6D4">🔄 Hot Reload</h3><p style="color:#9090b0;margin-top:.75rem">File watcher restarts server on every change — zero latency dev loop.</p></div>\n', |
| '<div class="glass" style="padding:2rem"><h3 style="color:#10B981">🚀 A2A Native</h3><p style="color:#9090b0;margin-top:.75rem">Every deploy auto-generates agent.json for ecosystem discovery.</p></div>\n', |
| '</section>\n', |
| '<script>const h=document.getElementById("hat");let p=0;function a(){p+=.02;h.style.transform=`scale(${1+Math.sin(p)*.08})`;requestAnimationFrame(a)}a();</script>\n', |
| '</body>\n</html>\n', |
| ) |
|
|
| def _game(self): |
| yield from ( |
| '<!DOCTYPE html>\n<html><head><meta charset="UTF-8"><title>Wizard-Vibe Game</title><style>', |
| '*{margin:0;padding:0}body{background:#0a0a1a;color:#e0e0ff;font-family:sans-serif;display:flex;flex-direction:column;align-items:center;justify-content:center;height:100vh}', |
| 'canvas{border:1px solid rgba(255,255,255,.1);border-radius:12px}</style></head><body>', |
| '<h2 style="background:linear-gradient(135deg,#8B5CF6,#06B6D4);-webkit-background-clip:text;-webkit-text-fill-color:transparent;margin-bottom:1rem">🧙♂️ Wizard-Vibe Game</h2>', |
| '<canvas id="c" width="400" height="400"></canvas>', |
| '<script>const c=document.getElementById("c"),x=c.getContext("2d");let bx=200,by=200,bvx=2,bvy=2;function d(){x.clearRect(0,0,400,400);x.beginPath();x.arc(bx,by,20,0,Math.PI*2);const g=x.createLinearGradient(bx-20,by,bx+20,by);g.addColorStop(0,"#8B5CF6");g.addColorStop(.5,"#06B6D4");g.addColorStop(1,"#10B981");x.fillStyle=g;x.fill();x.strokeStyle="rgba(255,255,255,.3)";x.lineWidth=2;x.stroke();bx+=bvx;by+=bvy;if(bx-20<0||bx+20>400)bvx=-bvx;if(by-20<0||by+20>400)bvy=-bvy;requestAnimationFrame(d)}d();</script>', |
| '</body></html>\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 ( |
| '<template>\n <div class="wizard-card" :class="{active:isActive}">\n <div class="card-glass">\n', |
| ' <h3>{{title}}</h3><p>{{description}}</p><button @click="toggle">{{label}}</button>\n </div>\n </div>\n</template>\n', |
| '<script>export default{name:"WizardCard",props:{title:String,description:String,label:{type:String,default:"Activate"}},data(){return{isActive:false}},methods:{toggle(){this.isActive=!this.isActive;this.$emit("toggle",this.isActive)}}};</script>\n', |
| '<style scoped>.wizard-card{border-radius:16px;overflow:hidden}.card-glass{background:rgba(255,255,255,.05);backdrop-filter:blur(20px);border:1px solid rgba(255,255,255,.08);padding:2rem}.active{box-shadow:0 0 30px rgba(139,92,246,.3)}</style>\n', |
| ) |
|
|
| |
| |
| |
|
|
| 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 "<html" in code.lower() or code.startswith("<!"): |
| e.extend(self._html_errors(code)) |
| elif code.startswith("import") or "def " in code: |
| try: compile(code, '<s>', '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 </{tag}>") |
| 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 ('<html' in c.lower() or '<body' in c.lower()) and not c.startswith('<!'): |
| c = '<!DOCTYPE html>\n' + c |
| if '<html' in c.lower() and not c.rstrip().endswith('</html>'): |
| c = c.rstrip() + '\n</html>' |
| return c |
|
|
| |
| |
| |
|
|
| 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} |
|
|
| |
| |
| |
|
|
| 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() |
|
|
| |
| |
| |
|
|
| async def stream_gen(sid: str, prompt: str): |
| """Core streaming generator — yields SSE events.""" |
| state.sessions[sid]["status"] = "streaming" |
| state.sandbox[sid] = "building" |
|
|
| |
| yield _sse("phase", {"phase": "orchestrate"}) |
| plan = state.orchestrator.pose(prompt) |
| yield _sse("plan", plan) |
|
|
| |
| 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) |
|
|
| |
| yield _sse("phase", {"phase": "heal"}) |
| healed, found, fixed = state.reflect.heal(code) |
| state.codes[sid] = healed |
| yield _sse("heal", {"found": found, "fixed": fixed}) |
|
|
| |
| 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" |
|
|
| |
| |
| |
|
|
| 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) |
|
|
| |
| out = GENERATED_DIR / repo |
| out.mkdir(exist_ok=True) |
| (out / "index.html").write_text(code) |
|
|
| |
| 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)) |
|
|
| |
| 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, "<!-- No code yet -->") |
| 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")) |
|
|
| |
| |
| |
|
|
| 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() |
|
|
