"""Alembic — AI Mini-App Factory. Gradio 6.x interface — Polished app builder with chat history: TOP: 60px header with logo + search + category pills + status LEFT (46%): "Build something new" textarea + chat history below RIGHT (54%): Live Preview browser mockup BOTTOM: Templates gallery (horizontal scroll) — shows built apps """ import asyncio import json import os import sys import uuid import html as html_module from pathlib import Path import gradio as gr sys.path.insert(0, str(Path(__file__).parent / "static")) from icons import ICONS, icon # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- MOCK_MODE = os.environ.get("ALEMBIC_MOCK", "0") == "1" API_URL = os.environ.get("ALEMBIC_API_URL", "") STATIC_DIR = Path(__file__).parent / "static" if not API_URL and not MOCK_MODE: MOCK_MODE = True def fetch_saved_apps() -> list: """Load globally saved apps from S3 manifest (called once at startup).""" if not API_URL: return [] try: import httpx with httpx.Client(timeout=8) as client: resp = client.get(f"{API_URL}/apps/list") if resp.status_code == 200: return resp.json().get("apps", []) except Exception as e: print(f"Could not load saved apps: {e}") return [] SAVED_APPS = fetch_saved_apps() # --------------------------------------------------------------------------- # Mock backend # --------------------------------------------------------------------------- MOCK_CALCULATOR_CODE = '''import gradio as gr def calculate(num1, num2, operation): if operation == "+": return num1 + num2 elif operation == "-": return num1 - num2 elif operation == "×": return num1 * num2 elif operation == "÷": return "Error: Division by zero" if num2 == 0 else num1 / num2 return "Invalid operation" with gr.Blocks(title="Calculator", css=""" .gradio-container { max-width: 400px; margin: auto; } button { font-size: 1.2rem !important; } """) as demo: gr.Markdown("## 🧮 Calculator") with gr.Row(): num1 = gr.Number(label="First number", value=0) num2 = gr.Number(label="Second number", value=0) with gr.Row(): op = gr.Radio(["+", "-", "×", "÷"], label="Operation", value="+") result = gr.Textbox(label="Result", interactive=False) btn = gr.Button("Calculate", variant="primary") btn.click(fn=calculate, inputs=[num1, num2, op], outputs=result) demo.launch() ''' def _detect_services_local(text: str) -> list[str]: """Detect required external services from text (no agent import needed).""" text_lower = text.lower() services = [] if any(w in text_lower for w in ["gmail", "email", "inbox", "messages from gmail"]): services.append("gmail") if any(w in text_lower for w in ["slack", "slack messages"]): services.append("slack") if any(w in text_lower for w in ["calendar", "events", "meetings", "google calendar"]): services.append("google_calendar") if any(w in text_lower for w in ["github", "pull requests", "git commits"]): services.append("github") if any(w in text_lower for w in ["google sheets", "spreadsheet"]): services.append("google_sheets") if any(w in text_lower for w in ["notion", "notion database"]): services.append("notion") return list(set(services)) async def mock_agent_response(user_message: str, state: dict) -> dict: """Simulate backend ChatResponse for mock mode (no agent import).""" await asyncio.sleep(0.6) msg_lower = user_message.lower() agent_state = state.get("agent_state") or {"turn_count": 0, "requirements_text": "", "services": []} # Track conversation turn_count = agent_state.get("turn_count", 0) + 1 requirements_text = agent_state.get("requirements_text", "") if requirements_text: requirements_text += " | " + user_message else: requirements_text = user_message # Detect required services on every turn detected_services = _detect_services_local(requirements_text) prev_services = agent_state.get("services", []) new_services = [s for s in detected_services if s not in prev_services] all_services = list(set(prev_services + detected_services)) # Simple confidence heuristic (no agent import) # High confidence: known simple apps OR user says "build"/"start"/"go" simple_apps = ["calculator", "timer", "counter", "todo", "notes"] has_simple_app = any(app in msg_lower for app in simple_apps) explicit_go = any(word in msg_lower for word in ["build", "start", "go", "create it", "make it"]) actions = [] # Show service connect prompt on first detection if new_services: actions.append({ "type": "show_connect", "required_services": new_services, }) if has_simple_app or explicit_go or turn_count >= 2: confidence = 0.95 plan = f"""✅ Purpose: {requirements_text[:50]} ✅ Input: User interaction ✅ Output: UI display ✅ Updates: Immediate ✅ Actions: Button clicks""" response_text = "Got it — I have a clear picture. Starting build now." actions.append({ "type": "start_build", "requirements": requirements_text, "design_prefs": state.get("design_prefs", {}), "plan": plan, "required_services": all_services, }) new_state = { "turn_count": turn_count, "requirements_text": requirements_text, "services": all_services, "phase": "building", } else: confidence = 0.5 plan = f"""❓ Purpose: {requirements_text[:50]} ❓ Data source: Not specified ❓ Output format: Not specified""" response_text = ( "Tell me more — what data will it work with, " "and how should results be displayed?" ) new_state = { "turn_count": turn_count, "requirements_text": requirements_text, "services": all_services, "phase": "intent", } return { "response": response_text, "new_state": new_state, "actions": actions, "confidence": confidence, "plan": plan, } async def mock_build(requirements: str) -> dict: """Simulate a build — returns code for known app types, or generic placeholder.""" await asyncio.sleep(2.0) req_lower = requirements.lower() # Return Space Shooter demo for testing if "space" in req_lower or "shooter" in req_lower: return { "status": "success", "code": "# Space Shooter game", "app_id": "space-shooter-demo", "url": "https://miniapps-alembic-hf-1781442002.s3.us-east-1.amazonaws.com/space-shooter-demo/" } if "calculator" in req_lower or "calc" in req_lower: return { "status": "success", "code": MOCK_CALCULATOR_CODE, "app_id": "calculator-001", "url": "https://miniapps-alembic-hf-1781442002.s3.us-east-1.amazonaws.com/test-calculator-app/" } return { "status": "success", "code": f"# Generated app for: {requirements}\nimport gradio as gr\nwith gr.Blocks() as demo:\n gr.Markdown('## Your App')\n gr.Textbox(label='Input')\ndemo.launch()\n", "app_id": f"app-{uuid.uuid4().hex[:8]}", "url": "https://miniapps-alembic-hf-1781442002.s3.us-east-1.amazonaws.com/test-gradio-lite-app/" } # --------------------------------------------------------------------------- # Backend communication # --------------------------------------------------------------------------- async def call_backend_agent(user_message: str, user_id: str, state: dict) -> dict: if MOCK_MODE: return await mock_agent_response(user_message, state) from model_runtime import _post messages = [{"role": "user", "content": user_message}] result = await _post("/chat", { "messages": messages, "user_id": user_id, "state": state.get("agent_state"), }) return result async def call_backend_generate(requirements: str, design_prefs: dict, user_id: str, connected_services: list = None) -> dict: if MOCK_MODE: return await mock_build(requirements) from model_runtime import _post payload = { "requirements": requirements, "design_prefs": design_prefs, "user_id": user_id, } if connected_services: payload["connected_services"] = connected_services result = await _post("/generate", payload) return result # --------------------------------------------------------------------------- # Gallery Data — default templates + user-built apps # --------------------------------------------------------------------------- DEFAULT_TEMPLATES = [ {"id": "gmail-dash", "name": "Gmail Dashboard", "desc": "Track email volume and team performance.", "tags": ["email", "productivity"], "icon": "mail"}, {"id": "pr-board", "name": "PR Board", "desc": "Monitor pull requests and merge status.", "tags": ["dev", "github"], "icon": "git-pr"}, {"id": "invoices", "name": "Invoice Gen", "desc": "Create professional invoices in seconds.", "tags": ["finance", "pdf"], "icon": "invoice"}, {"id": "standup", "name": "Standup Bot", "desc": "Collect and share daily standup summaries.", "tags": ["team", "ai"], "icon": "bot"}, {"id": "analytics", "name": "Analytics", "desc": "Visualize key metrics and track performance.", "tags": ["data", "charts"], "icon": "chart"}, ] CATEGORIES = ["All", "AI", "Productivity", "Data", "Dev", "Finance"] SUGGESTIONS = [ "Dashboard for spreadsheet data", "Form that notifies my team", "Tool to summarize meeting notes", ] # --------------------------------------------------------------------------- # UI Helpers — HTML builders # --------------------------------------------------------------------------- def make_header_html(categories: list, active: str = "All") -> str: pills = "" for cat in categories: active_cls = " active" if cat == active else "" pills += f'' search_icon = '' return f"""
Alembic
{search_icon} Search apps or ideas...
{pills}
Online
""" def make_build_header_html() -> str: sparkle = '' return f"""
{sparkle} Build something new

What would you like to build?

Describe your idea in natural language and let Alembic do the rest.

""" def make_textarea_footer_html() -> str: clip_icon = '' wand_icon = '' return f"""
{clip_icon} {wand_icon}
0 / 1000
""" def make_preview_header_html() -> str: phone_icon = '' desktop_icon = '' expand_icon = '' return f"""
Live Preview
{phone_icon} {desktop_icon} {expand_icon}
""" PREVIEW_PLACEHOLDER_HTML = """
✦ ✦ ✧ ✦
+

Start building your app

Describe what you need on the left. Alembic will generate the interface, logic, and data flow here.

Example: Create a finance dashboard
""" BUILDING_HTML_TEMPLATE = """
{status}

Building your app

{message}

{progress}% complete
""" IFRAME_PREVIEW_TEMPLATE = """
● Live Preview ↗
""" SERVICE_CONNECTION_TEMPLATE = """
🔌

Connect Services

This app needs access to the following services:

{service_cards}
""" SERVICE_CARD_TEMPLATE = """

{name}

{status_text}

""" def make_service_connection_html(required_services: list[str], service_status: dict) -> str: """Generate service connection UI.""" service_info = { "gmail": {"name": "Gmail", "logo": "📧"}, "slack": {"name": "Slack", "logo": "💬"}, "google_calendar": {"name": "Google Calendar", "logo": "📅"}, "notion": {"name": "Notion", "logo": "📝"}, "twitter": {"name": "Twitter", "logo": "🐦"}, "github": {"name": "GitHub", "logo": "🐙"}, "google_sheets": {"name": "Google Sheets", "logo": "📊"}, } cards = [] all_connected = True for slug in required_services: info = service_info.get(slug, {"name": slug.title(), "logo": "🔗"}) is_connected = service_status.get("connected", {}).get(slug, False) connect_url = service_status.get("connect_urls", {}).get(slug, "") if is_connected: status_class = "connected" status_text = "✓ Connected" button_text = "✓ Connected" disabled = "disabled" else: status_class = "not-connected" status_text = "Not connected" button_text = "Connect" disabled = "" all_connected = False cards.append(SERVICE_CARD_TEMPLATE.format( status_class=status_class, logo=info["logo"], name=info["name"], status_text=status_text, slug=slug, url=connect_url, button_text=button_text, disabled=disabled )) continue_disabled = "" if all_connected else "disabled" return SERVICE_CONNECTION_TEMPLATE.format( service_cards="\n".join(cards), disabled=continue_disabled ) def make_code_preview_html(code: str, app_name: str = "Your App") -> str: """Render generated code in a nice preview panel.""" escaped = html_module.escape(code) return f"""
{app_name} 
{escaped}
""" def make_gallery_html(templates: list, user_apps: list = None) -> str: grid_icon = '' cards = "" # User-built apps first (highlighted) — ONLY show user apps if they exist if user_apps and len(user_apps) > 0: for app_data in user_apps: tags_html = "".join(f'{t}' for t in app_data.get("tags", ["custom"])) app_url = app_data.get("url", "") onclick_attr = f'onclick="window.viewAppInPreview(\'{app_url}\')"' if app_url else "" cards += f"""
✦
{app_data["name"]}
{app_data.get("desc", "Built with Alembic")}
{tags_html}
View App →
""" # Default templates — only show if no user apps if not user_apps or len(user_apps) == 0: for app_data in templates: tags_html = "".join(f'{t}' for t in app_data["tags"]) app_icon = icon(app_data["icon"], "gallery-card-icon") template_desc = app_data["desc"] cards += f"""
{app_icon}
{app_data["name"]}
{app_data["desc"]}
{tags_html}
Use Template →
""" cards += f"""
{icon("plus", "gallery-card-icon")}
New App
Start from scratch.
Create New App →
""" # Change header based on content if user_apps and len(user_apps) > 0: header_title = "Your Apps" header_subtitle = f"{len(user_apps)} app{'s' if len(user_apps) > 1 else ''} built with Alembic." else: header_title = "Templates" header_subtitle = "Kickstart your next project with a template." return f"""
{grid_icon}

{header_title}

{header_subtitle}

See all →
{cards}
""" # --------------------------------------------------------------------------- # Chat history rendering # --------------------------------------------------------------------------- def render_chat_html(chat_history: list) -> str: if not chat_history: return "" html_parts = [] for msg in chat_history: role = msg["role"] content = msg["content"] if role == "user": html_parts.append( f'
' f'
{content}
' f'
' ) else: html_parts.append( f'
' f'
{content}
' f'
' ) return f'
{"".join(html_parts)}
' # --------------------------------------------------------------------------- # Main handler — chat-driven with history # --------------------------------------------------------------------------- async def handle_generate( user_message: str, js_user_id: str, chat_html: str, preview_frame: str, gallery_html: str, session_state: dict, ): """Process user message: show in chat, call agent, stream response, build if triggered.""" HIDE_SAVE = gr.update(visible=False) SHOW_SAVE = gr.update(visible=True) if not user_message.strip(): yield chat_html, preview_frame, gallery_html, session_state, "", HIDE_SAVE return # Use persistent user_id from localStorage (via JS hidden field) if js_user_id and js_user_id.strip(): session_state["user_id"] = js_user_id.strip() elif not session_state.get("user_id"): session_state["user_id"] = str(uuid.uuid4()) if "agent_state" not in session_state: session_state["agent_state"] = None if "chat_history" not in session_state: session_state["chat_history"] = [] if "built_apps" not in session_state: session_state["built_apps"] = [] # Add user message to history session_state["chat_history"].append({"role": "user", "content": user_message}) # Show user message + thinking indicator thinking_history = session_state["chat_history"] + [ {"role": "assistant", "content": ' Analyzing...'} ] yield render_chat_html(thinking_history), preview_frame, gallery_html, session_state, "", HIDE_SAVE await asyncio.sleep(0.3) # Call agent backend try: response = await call_backend_agent(user_message, session_state["user_id"], session_state) except Exception as e: print(f"Backend error: {e}, falling back to mock") try: response = await mock_agent_response(user_message, session_state) except Exception as e2: error_msg = f"Connection issue — retrying... ({str(e)[:60]})" session_state["chat_history"].append({"role": "assistant", "content": error_msg}) yield render_chat_html(session_state["chat_history"]), preview_frame, gallery_html, session_state, "", HIDE_SAVE return response_text = response.get("response", "") new_agent_state = response.get("new_state", {}) actions = response.get("actions", []) resp_confidence = response.get("confidence", 0.5) plan = response.get("plan") session_state["agent_state"] = new_agent_state assistant_content = response_text resp_pct = int(resp_confidence * 100) assistant_content += f' Confidence: {resp_pct}%' session_state["chat_history"].append({"role": "assistant", "content": assistant_content}) yield render_chat_html(session_state["chat_history"]), preview_frame, gallery_html, session_state, "", HIDE_SAVE # --- Service connection: check on every turn --- all_required_services = [] for action in actions: if action.get("type") == "show_connect": all_required_services.extend(action.get("required_services", [])) if action.get("type") == "start_build": all_required_services.extend(action.get("required_services", [])) all_required_services = list(set(all_required_services)) if all_required_services and not session_state.get("services_connected"): try: from model_runtime import _post service_status = await _post("/check-services", { "user_id": session_state["user_id"], "services": all_required_services }) all_connected = all(service_status.get("connected", {}).values()) if not all_connected: service_panel = make_service_connection_html(all_required_services, service_status) session_state["pending_build"] = { "requirements": user_message, "required_services": all_required_services } service_names = ", ".join(s.replace("_", " ").title() for s in all_required_services) session_state["chat_history"].append({ "role": "assistant", "content": f"🔌 This app needs {service_names} access. Please connect the required services to continue." }) yield render_chat_html(session_state["chat_history"]), service_panel, gallery_html, session_state, "", HIDE_SAVE return else: session_state["services_connected"] = True except Exception as e: print(f"Service check error: {e}") # Check if build was triggered build_triggered = any(a.get("type") == "start_build" for a in actions) if build_triggered: await asyncio.sleep(0.3) new_preview = BUILDING_HTML_TEMPLATE.format( status="Generating code...", message="Writing your application...", progress=20 ) yield render_chat_html(session_state["chat_history"]), new_preview, gallery_html, session_state, "", HIDE_SAVE requirements = user_message design_prefs = {} for action in actions: if action.get("type") == "start_build": requirements = action.get("requirements", user_message) design_prefs = action.get("design_prefs", {}) break new_preview = BUILDING_HTML_TEMPLATE.format( status="Generating code...", message="Building application logic...", progress=50 ) yield render_chat_html(session_state["chat_history"]), new_preview, gallery_html, session_state, "", HIDE_SAVE try: # Pass connected services so codegen injects the API helper services = (session_state.get("pending_build", {}).get("required_services") or all_required_services or []) result = await call_backend_generate(requirements, design_prefs, session_state["user_id"], services) new_preview = BUILDING_HTML_TEMPLATE.format( status="Deploying...", message="Uploading to CloudFront...", progress=85 ) yield render_chat_html(session_state["chat_history"]), new_preview, gallery_html, session_state, "", HIDE_SAVE await asyncio.sleep(0.5) code = result.get("code", "") app_id = result.get("app_id", f"app-{uuid.uuid4().hex[:8]}") cloudfront_url = result.get("url") if code and cloudfront_url: app_name = requirements.strip().split("\n")[0][:40] if len(app_name) > 30: app_name = app_name[:27] + "..." new_preview = IFRAME_PREVIEW_TEMPLATE.format(url=cloudfront_url) new_app = { "id": app_id, "name": app_name, "desc": "Built with Alembic", "tags": ["custom", "ai-generated"], "code": code, "url": cloudfront_url, } session_state["built_apps"].append(new_app) new_gallery = make_gallery_html(DEFAULT_TEMPLATES, session_state["built_apps"]) session_state["chat_history"].append({ "role": "assistant", "content": f'✅ Your app {app_name} is live! Deployed' }) # SUCCESS — show Save button yield render_chat_html(session_state["chat_history"]), new_preview, new_gallery, session_state, "", SHOW_SAVE elif code: new_preview = make_code_preview_html(code, "Generated Code (Deploy Failed)") session_state["chat_history"].append({ "role": "assistant", "content": "⚠️ App generated but deploy failed. Code shown in preview." }) yield render_chat_html(session_state["chat_history"]), new_preview, gallery_html, session_state, "", HIDE_SAVE else: session_state["chat_history"].append({ "role": "assistant", "content": "Build didn't produce code. Try being more specific." }) yield render_chat_html(session_state["chat_history"]), PREVIEW_PLACEHOLDER_HTML, gallery_html, session_state, "", HIDE_SAVE except Exception as e: session_state["chat_history"].append({ "role": "assistant", "content": f"Build error: {str(e)[:80]}. Try again." }) yield render_chat_html(session_state["chat_history"]), PREVIEW_PLACEHOLDER_HTML, gallery_html, session_state, "", HIDE_SAVE # --------------------------------------------------------------------------- # Suggestion handler — fills textarea # --------------------------------------------------------------------------- def fill_suggestion(suggestion_text: str) -> str: """Return the suggestion text to fill into the textarea.""" return suggestion_text # --------------------------------------------------------------------------- # Service recheck — called by JS after OAuth popup closes # --------------------------------------------------------------------------- async def handle_recheck_services(chat_html, preview_html, gallery_html, session_state): """Re-check service connection after OAuth popup and auto-build if ready.""" HIDE_SAVE = gr.update(visible=False) SHOW_SAVE = gr.update(visible=True) pending = session_state.get("pending_build", {}) required_services = pending.get("required_services", []) if not required_services: yield chat_html, preview_html, gallery_html, session_state, "", HIDE_SAVE return try: from model_runtime import _post service_status = await _post("/check-services", { "user_id": session_state["user_id"], "services": required_services }) all_connected = all(service_status.get("connected", {}).values()) if not all_connected: service_panel = make_service_connection_html(required_services, service_status) yield chat_html, service_panel, gallery_html, session_state, "", HIDE_SAVE return session_state["services_connected"] = True session_state["chat_history"].append({ "role": "assistant", "content": "✅ All services connected! Starting build..." }) new_preview = BUILDING_HTML_TEMPLATE.format( status="Generating code...", message="Writing your application...", progress=20 ) yield render_chat_html(session_state["chat_history"]), new_preview, gallery_html, session_state, "", HIDE_SAVE requirements = pending.get("requirements", "Build the app") new_preview = BUILDING_HTML_TEMPLATE.format( status="Generating code...", message="Building application logic...", progress=50 ) yield render_chat_html(session_state["chat_history"]), new_preview, gallery_html, session_state, "", HIDE_SAVE try: result = await call_backend_generate(requirements, {}, session_state["user_id"], required_services) new_preview = BUILDING_HTML_TEMPLATE.format( status="Deploying...", message="Uploading to CloudFront...", progress=85 ) yield render_chat_html(session_state["chat_history"]), new_preview, gallery_html, session_state, "", HIDE_SAVE await asyncio.sleep(0.5) code = result.get("code", "") app_id = result.get("app_id", f"app-{uuid.uuid4().hex[:8]}") cloudfront_url = result.get("url") if code and cloudfront_url: app_name = requirements.strip().split("\n")[0][:40] if len(app_name) > 30: app_name = app_name[:27] + "..." new_preview = IFRAME_PREVIEW_TEMPLATE.format(url=cloudfront_url) new_app = { "id": app_id, "name": app_name, "desc": "Built with Alembic", "tags": ["custom", "ai-generated"], "code": code, "url": cloudfront_url, } session_state["built_apps"].append(new_app) new_gallery = make_gallery_html(DEFAULT_TEMPLATES, session_state["built_apps"]) session_state["chat_history"].append({ "role": "assistant", "content": f'✅ Your app {app_name} is live! Deployed' }) yield render_chat_html(session_state["chat_history"]), new_preview, new_gallery, session_state, "", SHOW_SAVE else: session_state["chat_history"].append({ "role": "assistant", "content": "Build completed but no output. Try again." }) yield render_chat_html(session_state["chat_history"]), PREVIEW_PLACEHOLDER_HTML, gallery_html, session_state, "", HIDE_SAVE except Exception as e: session_state["chat_history"].append({ "role": "assistant", "content": f"Build error: {str(e)[:80]}. Try again." }) yield render_chat_html(session_state["chat_history"]), PREVIEW_PLACEHOLDER_HTML, gallery_html, session_state, "", HIDE_SAVE except Exception as e: session_state["chat_history"].append({ "role": "assistant", "content": f"Service check error: {str(e)[:80]}. Try connecting again." }) yield render_chat_html(session_state["chat_history"]), preview_html, gallery_html, session_state, "", HIDE_SAVE # --------------------------------------------------------------------------- # CSS & Head # --------------------------------------------------------------------------- def load_css() -> str: css_path = STATIC_DIR / "style.css" if css_path.exists(): return css_path.read_text() return "" HEAD_HTML = """ """ JS_INIT = """ () => { document.body.classList.add('alembic-light'); document.documentElement.classList.remove('dark'); const observer = new MutationObserver((mutations) => { for (const m of mutations) { if (m.type === 'attributes' && m.attributeName === 'class') { if (m.target.classList && m.target.classList.contains('dark')) { m.target.classList.remove('dark'); } } } }); observer.observe(document.documentElement, { attributes: true }); // Auto-scroll chat area const chatObserver = new MutationObserver(() => { const chatEl = document.querySelector('.chat-history'); if (chatEl) chatEl.scrollTop = chatEl.scrollHeight; }); setTimeout(() => { const target = document.querySelector('.chat-area'); if (target) chatObserver.observe(target, {childList: true, subtree: true}); }, 1000); // Persist user_id: cookie (primary) + localStorage (fallback for iframe) // Cookie with SameSite=Lax; Secure — blocked as 3rd-party in some browsers // when inside HF Space iframe, so localStorage is the fallback. function getCookie(name) { var m = document.cookie.match('(?:^|; )' + name + '=([^;]*)'); return m ? m[1] : null; } function setCookie(name, val) { var exp = new Date(Date.now() + 365*24*60*60*1000).toUTCString(); document.cookie = name + '=' + val + '; expires=' + exp + '; path=/; SameSite=Lax; Secure'; } var userId = getCookie('alembic_uid'); if (!userId) { try { userId = localStorage.getItem('alembic_user_id'); } catch(e) {} } if (!userId) { userId = 'user-' + crypto.randomUUID(); } // Write to both stores setCookie('alembic_uid', userId); try { localStorage.setItem('alembic_user_id', userId); } catch(e) {} window.ALEMBIC_USER_ID = userId; // Fill the hidden user-id field so Python can read it setTimeout(function() { var field = document.querySelector('#user-id-store textarea') || document.querySelector('#user-id-store input'); if (field) { field.value = userId; field.dispatchEvent(new Event('input', { bubbles: true })); } }, 800); } """ # --------------------------------------------------------------------------- # Save app handler # --------------------------------------------------------------------------- async def handle_save_app(chat_html, preview_html, gallery_html, session_state): """Save the last built app to the global manifest.""" HIDE_SAVE = gr.update(visible=False) built_apps = session_state.get("built_apps", []) if not built_apps: yield chat_html, preview_html, gallery_html, session_state, "", HIDE_SAVE return last_app = built_apps[-1] if last_app.get("saved"): session_state["chat_history"].append({ "role": "assistant", "content": "This app is already saved." }) yield render_chat_html(session_state["chat_history"]), preview_html, gallery_html, session_state, "", HIDE_SAVE return try: from model_runtime import _post await _post("/apps/save", { "app_id": last_app["id"], "name": last_app["name"], "description": last_app.get("desc", ""), "tags": last_app.get("tags", []), "url": last_app.get("url", ""), "user_id": session_state.get("user_id", ""), }) last_app["saved"] = True session_state["chat_history"].append({ "role": "assistant", "content": f'💾 {last_app["name"]} saved to global gallery! Anyone can find it now.' }) # Hide save button after successful save yield render_chat_html(session_state["chat_history"]), preview_html, gallery_html, session_state, "", HIDE_SAVE except Exception as e: session_state["chat_history"].append({ "role": "assistant", "content": f"Save failed: {str(e)[:80]}. Try again." }) # Keep save button visible on failure so user can retry yield render_chat_html(session_state["chat_history"]), preview_html, gallery_html, session_state, "", gr.update(visible=True) # --------------------------------------------------------------------------- # Build the Gradio app # --------------------------------------------------------------------------- # Pre-load globally saved apps for the gallery initial_gallery_apps = SAVED_APPS if SAVED_APPS else None with gr.Blocks(title="Alembic") as demo: # --- Header Bar --- gr.HTML(make_header_html(CATEGORIES, "All"), elem_classes="header-wrap") # --- Session state --- session_state = gr.State({ "user_id": None, "agent_state": None, "chat_history": [], "built_apps": [], }) # Hidden user-id field — JS fills from localStorage on page load user_id_field = gr.Textbox( value="", elem_id="user-id-store", elem_classes="sr-only-btn", show_label=False, ) # --- Main split layout --- with gr.Row(elem_classes="main-row"): # LEFT: Chat panel (history + input at bottom) with gr.Column(scale=46, elem_classes="build-col"): gr.HTML(make_build_header_html(), elem_classes="build-section") # Chat history area (scrollable, grows with messages) chat_area = gr.HTML(value="", elem_classes="chat-area") # Input area at BOTTOM user_input = gr.Textbox( show_label=False, placeholder="Describe your app...", lines=3, max_lines=6, elem_classes="prompt-area", ) gr.HTML(make_textarea_footer_html(), elem_classes="textarea-footer-wrap") with gr.Row(elem_classes="action-buttons-row"): generate_btn = gr.Button( "✦ Generate App ⌘↵", variant="primary", elem_classes="generate-btn-wrap", ) save_btn = gr.Button( "💾 Save App", variant="secondary", elem_classes="save-btn-wrap", visible=False, ) # Suggestion buttons — clicking fills the textarea with gr.Row(elem_classes="suggestions-row"): sug1 = gr.Button("💡 Dashboard for spreadsheet data", elem_classes="suggestion-btn") sug2 = gr.Button("💡 Form that notifies my team", elem_classes="suggestion-btn") sug3 = gr.Button("💡 Tool to summarize meeting notes", elem_classes="suggestion-btn") # RIGHT: Preview panel with gr.Column(scale=54, elem_classes="preview-col"): gr.HTML(make_preview_header_html(), elem_classes="preview-label-wrap") preview_frame = gr.HTML( value=PREVIEW_PLACEHOLDER_HTML, elem_classes="preview-frame", ) # --- Templates / Saved Apps Gallery --- gallery_section = gr.HTML( make_gallery_html(DEFAULT_TEMPLATES, initial_gallery_apps), elem_classes="gallery-section", ) # Recheck button — CSS-hidden (not visible=False, which removes from DOM). recheck_btn = gr.Button("Recheck Services", elem_id="recheck-services-btn", elem_classes="sr-only-btn") # --- Event handlers --- outputs = [chat_area, preview_frame, gallery_section, session_state, user_input, save_btn] gen_inputs = [user_input, user_id_field, chat_area, preview_frame, gallery_section, session_state] generate_btn.click( fn=handle_generate, inputs=gen_inputs, outputs=outputs, ) user_input.submit( fn=handle_generate, inputs=gen_inputs, outputs=outputs, ) recheck_btn.click( fn=handle_recheck_services, inputs=[chat_area, preview_frame, gallery_section, session_state], outputs=outputs, ) save_btn.click( fn=handle_save_app, inputs=[chat_area, preview_frame, gallery_section, session_state], outputs=outputs, ) # Suggestion buttons fill textarea sug1.click(fn=lambda: "Dashboard for spreadsheet data", outputs=user_input) sug2.click(fn=lambda: "Form that notifies my team", outputs=user_input) sug3.click(fn=lambda: "Tool to summarize meeting notes", outputs=user_input) # --------------------------------------------------------------------------- # Launch # --------------------------------------------------------------------------- if __name__ == "__main__": demo.launch( server_name="0.0.0.0", server_port=7860, css=load_css(), head=HEAD_HTML, js=JS_INIT, )