alembic / app.py
RafalB
Pass connected_services to /generate so ALEMBIC_API helper gets injected
379dfef
Raw
History Blame Contribute Delete
54 kB
"""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'<button class="cat-pill{active_cls}">{cat}</button>'
search_icon = '<svg width="16" height="16" viewBox="0 0 24 24" fill="none"><circle cx="11" cy="11" r="7" stroke="currentColor" stroke-width="1.5"/><path d="M16.5 16.5L21 21" stroke="currentColor" stroke-width="1.5" stroke-linecap="round"/></svg>'
return f"""
<div class="alembic-header">
<div class="header-brand">
<span class="header-logo">{icon("alembic")}</span>
<span class="header-title">Alembic</span>
</div>
<div class="header-search">
{search_icon}
<span class="header-search-text">Search apps or ideas...</span>
</div>
<div class="header-pills">{pills}</div>
<div class="header-right">
<span class="header-status"><span class="status-dot"></span>Online</span>
<div class="header-avatar"></div>
</div>
</div>
"""
def make_build_header_html() -> str:
sparkle = '<svg width="16" height="16" viewBox="0 0 24 24" fill="none"><path d="M12 2l2.4 7.2L22 12l-7.6 2.8L12 22l-2.4-7.2L2 12l7.6-2.8L12 2z" stroke="currentColor" stroke-width="1.5" stroke-linejoin="round"/></svg>'
return f"""
<div class="build-header">
<div class="build-subhead">{sparkle} Build something new</div>
<h2 class="build-title">What would you like to build?</h2>
<p class="build-subtitle">Describe your idea in natural language and let Alembic do the rest.</p>
</div>
"""
def make_textarea_footer_html() -> str:
clip_icon = '<svg width="16" height="16" viewBox="0 0 24 24" fill="none"><path d="M21.44 11.05l-9.19 9.19a6 6 0 01-8.49-8.49l9.19-9.19a4 4 0 015.66 5.66l-9.2 9.19a2 2 0 01-2.83-2.83l8.49-8.48" stroke="currentColor" stroke-width="1.5" stroke-linecap="round" stroke-linejoin="round"/></svg>'
wand_icon = '<svg width="16" height="16" viewBox="0 0 24 24" fill="none"><path d="M15 4V2M15 16v-2M8 9h2M20 9h2M17.8 11.8L19 13M17.8 6.2L19 5M12.2 11.8L11 13M12.2 6.2L11 5M15 9a2 2 0 11-4 0 2 2 0 014 0z" stroke="currentColor" stroke-width="1.5" stroke-linecap="round"/><path d="M2 22l10-10" stroke="currentColor" stroke-width="1.5" stroke-linecap="round"/></svg>'
return f"""
<div class="textarea-footer">
<div class="textarea-icons">
<span class="textarea-icon-btn">{clip_icon}</span>
<span class="textarea-icon-btn">{wand_icon}</span>
</div>
<span class="char-counter">0 / 1000</span>
</div>
"""
def make_preview_header_html() -> str:
phone_icon = '<svg width="16" height="16" viewBox="0 0 24 24" fill="none"><rect x="6" y="2" width="12" height="20" rx="2" stroke="currentColor" stroke-width="1.5"/><path d="M10 18h4" stroke="currentColor" stroke-width="1.5" stroke-linecap="round"/></svg>'
desktop_icon = '<svg width="16" height="16" viewBox="0 0 24 24" fill="none"><rect x="3" y="4" width="18" height="13" rx="2" stroke="currentColor" stroke-width="1.5"/><path d="M8 21h8M12 17v4" stroke="currentColor" stroke-width="1.5" stroke-linecap="round"/></svg>'
expand_icon = '<svg width="16" height="16" viewBox="0 0 24 24" fill="none"><path d="M15 3h6v6M9 21H3v-6M21 3l-7 7M3 21l7-7" stroke="currentColor" stroke-width="1.5" stroke-linecap="round" stroke-linejoin="round"/></svg>'
return f"""
<div class="preview-header">
<div class="preview-label-left">
<span class="preview-label-text">Live Preview</span>
<span class="preview-dot"></span>
</div>
<div class="preview-controls">
<span class="preview-ctrl-btn">{phone_icon}</span>
<span class="preview-ctrl-btn">{desktop_icon}</span>
<span class="preview-ctrl-btn">{expand_icon}</span>
</div>
</div>
"""
PREVIEW_PLACEHOLDER_HTML = """
<div class="preview-mockup">
<div class="browser-bar">
<span class="browser-dot green"></span>
<span class="browser-dot green"></span>
<span class="browser-dot green"></span>
</div>
<div class="preview-body">
<span class="sparkle sparkle-1">✦</span>
<span class="sparkle sparkle-2">✦</span>
<span class="sparkle sparkle-3">✧</span>
<span class="sparkle sparkle-4">✦</span>
<div class="mockup-window">
<div class="mockup-titlebar">
<span class="mockup-titlebar-dot"></span>
<span class="mockup-titlebar-dot"></span>
<span class="mockup-titlebar-dot"></span>
</div>
<div class="mockup-content">
<div class="mockup-sidebar">
<div class="mockup-sidebar-line" style="width:80%"></div>
<div class="mockup-sidebar-line" style="width:60%"></div>
<div class="mockup-sidebar-line" style="width:90%"></div>
<div class="mockup-sidebar-line" style="width:50%"></div>
</div>
<div class="mockup-main">
<span class="mockup-plus">+</span>
</div>
</div>
</div>
<h3>Start building your app</h3>
<p>Describe what you need on the left. Alembic will generate the interface, logic, and data flow here.</p>
<div class="example-btn">
<svg width="14" height="14" viewBox="0 0 24 24" fill="none"><path d="M12 2l2.4 7.2L22 12l-7.6 2.8L12 22l-2.4-7.2L2 12l7.6-2.8L12 2z" stroke="currentColor" stroke-width="1.5" stroke-linejoin="round"/></svg>
Example: Create a finance dashboard
</div>
</div>
</div>
"""
BUILDING_HTML_TEMPLATE = """
<div class="preview-mockup">
<div class="browser-bar">
<span class="browser-dot green"></span>
<span class="browser-dot green"></span>
<span class="browser-dot green"></span>
</div>
<div class="preview-body">
<div class="build-indicator">
<div class="spinner"></div>
<span class="build-status">{status}</span>
</div>
<h3>Building your app</h3>
<p>{message}</p>
<div class="progress-bar-wrap">
<div class="progress-bar" style="width: {progress}%"></div>
</div>
<div class="progress-text">{progress}% complete</div>
</div>
</div>
"""
IFRAME_PREVIEW_TEMPLATE = """
<div class="preview-container" style="width:100%;height:600px;position:relative;border-radius:12px;overflow:hidden;border:1px solid #e2e5ef;">
<div style="padding:10px 14px;border-bottom:1px solid #e2e5ef;background:#fafafa;display:flex;align-items:center;gap:8px;">
<span style="color:#10b981;font-size:10px;">&#9679;</span>
<span style="font-family:var(--font-display);font-size:0.85rem;font-weight:600;color:#0f1729;">Live Preview</span>
<a href="{url}" target="_blank" style="margin-left:auto;font-size:0.75rem;color:#94a3b8;text-decoration:none;opacity:0.7;">&#x2197;</a>
</div>
<iframe
id="mini-app-iframe"
src="{url}"
style="width:100%;height:calc(100% - 42px);border:none;background:#ffffff;"
title="Mini-App Preview"
sandbox="allow-scripts allow-same-origin allow-forms allow-popups"
loading="lazy">
</iframe>
</div>
"""
SERVICE_CONNECTION_TEMPLATE = """
<div class="service-connect-panel">
<div class="service-header">
<span class="service-icon">🔌</span>
<h3>Connect Services</h3>
<p>This app needs access to the following services:</p>
</div>
<div class="service-list">
{service_cards}
</div>
<div style="display:flex;gap:10px;margin-top:12px;">
<button class="continue-build-btn" onclick="window.continueBuild()" {disabled}>
✓ Continue Building
</button>
<button class="refresh-status-btn" onclick="window.recheckServices()" style="padding:10px 20px;border-radius:10px;border:1.5px solid #c7d2fe;background:#f8f9fc;color:#4f46e5;font-weight:600;cursor:pointer;font-size:0.9rem;">
&#x21bb; Refresh Status
</button>
</div>
</div>
"""
SERVICE_CARD_TEMPLATE = """
<div class="service-card {status_class}">
<div class="service-info">
<span class="service-logo">{logo}</span>
<div>
<h4>{name}</h4>
<p class="service-status">{status_text}</p>
</div>
</div>
<button class="service-connect-btn" onclick="window.connectService('{slug}', '{url}')" {disabled}>
{button_text}
</button>
</div>
"""
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"""
<div class="preview-mockup">
<div class="browser-bar">
<span class="browser-dot green"></span>
<span class="browser-dot green"></span>
<span class="browser-dot green"></span>
<span class="browser-bar-title">{app_name}</span>
</div>
<div class="code-preview-body">
<pre class="code-block"><code>{escaped}</code></pre>
</div>
</div>
"""
def make_gallery_html(templates: list, user_apps: list = None) -> str:
grid_icon = '<svg width="18" height="18" viewBox="0 0 24 24" fill="none"><rect x="3" y="3" width="7" height="7" rx="1.5" stroke="currentColor" stroke-width="1.5"/><rect x="14" y="3" width="7" height="7" rx="1.5" stroke="currentColor" stroke-width="1.5"/><rect x="3" y="14" width="7" height="7" rx="1.5" stroke="currentColor" stroke-width="1.5"/><rect x="14" y="14" width="7" height="7" rx="1.5" stroke="currentColor" stroke-width="1.5"/></svg>'
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'<span class="gallery-tag">{t}</span>' 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"""
<div class="gallery-card gallery-card-built" {onclick_attr} style="cursor: pointer;">
<div class="gallery-card-icon built-icon">✦</div>
<div class="gallery-card-name">{app_data["name"]}</div>
<div class="gallery-card-desc">{app_data.get("desc", "Built with Alembic")}</div>
<div class="gallery-card-tags">{tags_html}</div>
<div class="gallery-card-action">View App →</div>
</div>
"""
# 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'<span class="gallery-tag">{t}</span>' for t in app_data["tags"])
app_icon = icon(app_data["icon"], "gallery-card-icon")
template_desc = app_data["desc"]
cards += f"""
<div class="gallery-card" onclick="window.useTemplate('{template_desc}')">
{app_icon}
<div class="gallery-card-name">{app_data["name"]}</div>
<div class="gallery-card-desc">{app_data["desc"]}</div>
<div class="gallery-card-tags">{tags_html}</div>
<div class="gallery-card-action">Use Template →</div>
</div>
"""
cards += f"""
<div class="gallery-card gallery-card-new" onclick="window.focusInput()">
{icon("plus", "gallery-card-icon")}
<div class="gallery-card-name">New App</div>
<div class="gallery-card-desc">Start from scratch.</div>
<div class="gallery-card-action">Create New App →</div>
</div>
"""
# 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"""
<div class="gallery-header">
<div class="gallery-header-left">
<span class="gallery-header-icon">{grid_icon}</span>
<div class="gallery-header-text">
<h4>{header_title}</h4>
<p>{header_subtitle}</p>
</div>
</div>
<a class="gallery-see-all" href="#">See all →</a>
</div>
<div class="gallery-scroll">{cards}</div>
"""
# ---------------------------------------------------------------------------
# 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'<div class="chat-msg chat-msg-user">'
f'<div class="chat-msg-bubble user-bubble">{content}</div>'
f'</div>'
)
else:
html_parts.append(
f'<div class="chat-msg chat-msg-assistant">'
f'<div class="chat-msg-bubble assistant-bubble">{content}</div>'
f'</div>'
)
return f'<div class="chat-history">{"".join(html_parts)}</div>'
# ---------------------------------------------------------------------------
# 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": '<span class="thinking-dot"></span> 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' <span class="confidence-badge">Confidence: {resp_pct}%</span>'
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 <strong>{service_names}</strong> 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 <strong>{app_name}</strong> is live! <span class="confidence-badge">Deployed</span>'
})
# 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 <strong>{app_name}</strong> is live! <span class="confidence-badge">Deployed</span>'
})
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 = """
<link rel="preconnect" href="https://api.fontshare.com" crossorigin>
<link href="https://api.fontshare.com/v2/css?f[]=general-sans@400,500,600,700&display=swap" rel="stylesheet">
<link href="https://fonts.googleapis.com/css2?family=JetBrains+Mono:wght@400;500;600&display=swap" rel="stylesheet">
<meta name="theme-color" content="#f8f9fc">
<script>
// Dynamic iframe loader for mini-apps
function loadMiniApp(url) {
const iframe = document.querySelector('#mini-app-iframe');
if (!iframe) {
console.warn('iframe #mini-app-iframe not found');
return;
}
iframe.style.opacity = '0.5';
iframe.style.filter = 'blur(2px)';
iframe.src = url;
iframe.onload = function() {
iframe.style.opacity = '1';
iframe.style.filter = 'none';
};
setTimeout(function() {
if (iframe.style.opacity !== '1') {
iframe.style.opacity = '1';
iframe.style.filter = 'none';
}
}, 10000);
}
window.addEventListener('message', function(event) {
if (event.data.type === 'alembic_app_ready') {
loadMiniApp(event.data.url);
}
});
// Click the hidden Gradio recheck button
window.recheckServices = function() {
// Try multiple selectors — Gradio wraps buttons in divs
const wrapper = document.getElementById('recheck-services-btn');
if (wrapper) {
const btn = wrapper.querySelector('button');
if (btn) { console.log('Recheck: clicking button inside wrapper'); btn.click(); return; }
console.log('Recheck: clicking wrapper directly'); wrapper.click(); return;
}
// Fallback: find by class
const byClass = document.querySelector('.sr-only-btn button');
if (byClass) { console.log('Recheck: clicking by class'); byClass.click(); return; }
console.warn('Recheck: could not find recheck button');
};
// Service connection handlers
// NOTE: Runs inside HF Space iframe — popup.closed can fail cross-origin,
// so we have multiple fallback strategies.
window.connectService = function(slug, url) {
if (!url) return;
var popup = null;
try {
popup = window.open(url, 'Connect_' + slug, 'width=600,height=700,scrollbars=yes');
} catch (e) {
// If popup blocked in iframe, open in parent or fallback
console.warn('Popup blocked, trying parent window');
try { popup = window.parent.open(url, '_blank'); } catch (e2) {}
}
if (!popup) {
// Popup fully blocked — open as link in new tab
window.open(url, '_blank');
console.log('Popup blocked, opened in new tab. User should click Refresh Status after connecting.');
return;
}
// Poll for popup close
var recheckDone = false;
var pollInterval = setInterval(function() {
try {
if (popup.closed) {
clearInterval(pollInterval);
if (!recheckDone) {
recheckDone = true;
console.log('OAuth popup closed, rechecking services in 2s...');
setTimeout(function() { window.recheckServices(); }, 2000);
}
}
} catch (e) {
// Cross-origin — can't check popup.closed, rely on fallback
}
}, 800);
// Fallback: recheck after 20s regardless (handles iframe cross-origin issues)
setTimeout(function() {
clearInterval(pollInterval);
if (!recheckDone) {
recheckDone = true;
console.log('Fallback recheck after 20s');
window.recheckServices();
}
}, 20000);
};
window.continueBuild = function() {
window.recheckServices();
};
// Load a mini-app URL into the Alembic Live Preview panel (no new tab)
window.viewAppInPreview = function(url) {
const container = document.querySelector('.preview-frame .html-container') ||
document.querySelector('.preview-frame');
if (!container) return;
container.innerHTML =
'<div style="width:100%;height:600px;position:relative;border-radius:12px;overflow:hidden;border:1px solid #e2e5ef;">' +
'<div style="padding:10px 14px;border-bottom:1px solid #e2e5ef;background:#fafafa;display:flex;align-items:center;gap:8px;">' +
'<span style="color:#10b981;font-size:10px;">\\u25CF</span>' +
'<span style="font-family:var(--font-display);font-size:0.85rem;font-weight:600;color:#0f1729;">Live Preview</span>' +
'</div>' +
'<iframe src="' + url + '" style="width:100%;height:calc(100% - 42px);border:none;background:#fff;" ' +
'sandbox="allow-scripts allow-same-origin allow-forms allow-popups" loading="lazy"></iframe></div>';
container.scrollIntoView({behavior: 'smooth', block: 'start'});
};
// Template gallery helpers
window.useTemplate = function(description) {
const textarea = document.querySelector('.prompt-area textarea');
if (textarea) {
textarea.value = description;
textarea.dispatchEvent(new Event('input', { bubbles: true }));
textarea.focus();
textarea.scrollIntoView({ behavior: 'smooth', block: 'center' });
}
};
window.focusInput = function() {
const textarea = document.querySelector('.prompt-area textarea');
if (textarea) {
textarea.focus();
textarea.scrollIntoView({ behavior: 'smooth', block: 'center' });
}
};
</script>
"""
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'💾 <strong>{last_app["name"]}</strong> 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,
)