🔥 Connect to real CAPT backend

app.py

@@ -1,15 +1,56 @@
 #!/usr/bin/env python3
-"""GIT HOLOGRAM — Visualize any GitHub repo as causal threads."""
+"""
+GIT HOLOGRAM — Visualize any GitHub repo as causal threads.
+Now with real bioCAPT integration: analysis, memory recall, and security audit.
+"""
 
 import subprocess
 import tempfile
 import shutil
 import os
+import hashlib
+import json
 from datetime import datetime
 from typing import List, Dict
 
 import gradio as gr
 
+from capt_backend import CAPTBackend, health_banner_html, BRAINS
+
+backend = CAPTBackend()
+
+# ── Persistent cache dir ───────────────────────────────────────────────────
+CACHE_DIR = "/tmp/git-hologram-cache"
+os.makedirs(CACHE_DIR, exist_ok=True)
+
+def _repo_key(repo_url: str) -> str:
+    return hashlib.sha256(repo_url.encode()).hexdigest()[:16]
+
+def _get_cached(repo_url: str):
+    key = _repo_key(repo_url)
+    meta_path = os.path.join(CACHE_DIR, f"{key}.json")
+    if os.path.exists(meta_path):
+        try:
+            with open(meta_path) as f:
+                meta = json.load(f)
+            repo_path = os.path.join(CACHE_DIR, key)
+            if os.path.isdir(repo_path):
+                return repo_path, meta
+        except Exception:
+            pass
+    return None, None
+
+def _set_cached(repo_url: str, repo_path: str, meta: dict):
+    key = _repo_key(repo_url)
+    dest = os.path.join(CACHE_DIR, key)
+    if dest != repo_path and os.path.exists(repo_path):
+        if os.path.exists(dest):
+            shutil.rmtree(dest, ignore_errors=True)
+        shutil.move(repo_path, dest)
+    with open(os.path.join(CACHE_DIR, f"{key}.json"), "w") as f:
+        json.dump(meta, f)
+    return dest
+
 # ── Git logic ──────────────────────────────────────────────────────────
 
 def get_git_commits(repo_path: str, max_commits: int = 50) -> List[Dict]:
@@ -156,24 +197,33 @@ def render_summary(commits: List[Dict]) -> str:
 
 # ── Repo fetch ─────────────────────────────────────────────────────────
 
-def fetch_repo(repo_url: str, max_commits: int) -> str:
-    """Clone shallowly, run hologram, clean up."""
+def fetch_repo(repo_url: str, max_commits: int) -> tuple:
+    """Clone shallowly (with cache), run hologram, return (html, repo_url, status)."""
     repo_url = repo_url.strip()
     if not repo_url:
-        return "<p style='color:#f66'>Please enter a repo URL.</p>"
+        return "<p style='color:#f66'>Please enter a repo URL.</p>", "", "empty"
 
-    # Normalize URL to git clone format
+    # Normalize URL
     if repo_url.startswith("http"):
         if not repo_url.endswith(".git"):
             repo_url += ".git"
     elif repo_url.startswith("git@"):
-        pass  # already SSH format
+        pass
     else:
-        # Assume GitHub username/repo shorthand
         if "/" not in repo_url:
             repo_url = f"knowurknot/{repo_url}"
         repo_url = f"https://github.com/{repo_url}.git"
 
+    # Check cache
+    cached_path, meta = _get_cached(repo_url)
+    if cached_path:
+        commits = get_git_commits(cached_path, max_commits)
+        if commits:
+            hologram_html = render_hologram(commits)
+            summary_html = render_summary(commits)
+            full_html = f'<div class="summary-card">{summary_html}</div>{hologram_html}'
+            return full_html, repo_url, "cached"
+
     tmpdir = tempfile.mkdtemp(prefix="hologram_")
     try:
         depth = max(max_commits + 10, 50)
@@ -182,25 +232,61 @@ def fetch_repo(repo_url: str, max_commits: int) -> str:
             capture_output=True, text=True, timeout=60,
         )
         if result.returncode != 0:
-            return f"<p style='color:#f66'>Clone failed: {result.stderr[:300]}</p>"
+            return f"<p style='color:#f66'>Clone failed: {result.stderr[:300]}</p>", repo_url, "error"
 
         commits = get_git_commits(tmpdir, max_commits)
         if not commits:
-            return "<p style='color:#888'>No commits found in this repo.</p>"
+            return "<p style='color:#888'>No commits found in this repo.</p>", repo_url, "empty"
 
         hologram_html = render_hologram(commits)
         summary_html = render_summary(commits)
+        full_html = f'<div class="summary-card">{summary_html}</div>{hologram_html}'
 
-        return f"""
-        <div class="summary-card">{summary_html}</div>
-        {hologram_html}
-        """
+        # Cache it
+        _set_cached(repo_url, tmpdir, {"url": repo_url, "commits": len(commits), "cached_at": datetime.now().isoformat()})
+        return full_html, repo_url, "ok"
     except subprocess.TimeoutExpired:
-        return "<p style='color:#f66'>Clone timed out — repo may be too large.</p>"
+        return "<p style='color:#f66'>Clone timed out — repo may be too large.</p>", repo_url, "timeout"
     except Exception as e:
-        return f"<p style='color:#f66'>Error: {str(e)[:300]}</p>"
+        return f"<p style='color:#f66'>Error: {str(e)[:300]}</p>", repo_url, "error"
     finally:
-        shutil.rmtree(tmpdir, ignore_errors=True)
+        if os.path.exists(tmpdir):
+            shutil.rmtree(tmpdir, ignore_errors=True)
+
+
+# ── CAPT Analysis ──────────────────────────────────────────────────────
+
+import asyncio
+
+async def capt_analyze(repo_url: str, analysis_type: str) -> str:
+    if not repo_url.strip():
+        return "⚠️ Generate a hologram first."
+    mid = "capt-core-01"
+    try:
+        loop = asyncio.get_event_loop()
+        if analysis_type == "codebase":
+            prompt = f"Analyze this codebase: {repo_url}\n\nProvide architectural insights, tech stack, complexity assessment, and improvement suggestions."
+            result = await loop.run_in_executor(None, lambda: backend.cogitate(prompt))
+            resp = result.get("response", result.get("content", "No response"))
+            fb = result.get("_fallback", False) or "error" in result
+            badge = "🔴 FALLBACK" if fb else "🟢 LIVE"
+            return f"**{badge}** · Codebase Analysis\n\n{resp}"
+        elif analysis_type == "commits":
+            result = await loop.run_in_executor(None, lambda: backend.echo_recall(f"Commit patterns for {repo_url}"))
+            resp = result.get("response", result.get("content", json.dumps(result, indent=2)))
+            fb = result.get("_fallback", False) or "error" in result
+            badge = "🔴 FALLBACK" if fb else "🟢 LIVE"
+            return f"**{badge}** · Commit Pattern Recall\n\n{resp}"
+        elif analysis_type == "security":
+            prompt = f"Security audit of repository {repo_url}. Check for secrets, vulnerabilities, dependency risks, and supply-chain exposure."
+            result = await loop.run_in_executor(None, lambda: backend.immu_scan(prompt))
+            resp = result.get("response", result.get("content", json.dumps(result, indent=2)))
+            fb = result.get("_fallback", False) or "error" in result
+            badge = "🔴 FALLBACK" if fb else "🟢 LIVE"
+            return f"**{badge}** · Security Audit (ImmuScan)\n\n{resp}"
+        return "Unknown analysis type."
+    except Exception as e:
+        return f"🔴 Error: {str(e)[:300]}"
 
 
 # ── Gradio UI ──────────────────────────────────────────────────────────
@@ -220,17 +306,30 @@ CUSTOM_CSS = """
     border: 1px solid #2a2a40; color: #ccc; font-size: 14px; line-height: 1.8;
 }
 .thread { letter-spacing: 1px; }
+.result-box {background:#0d0d14; border:1px solid #1a1a2e; border-radius:10px; padding:14px; min-height:120px}
+#header {text-align: center; margin-bottom: 16px}
+#header h1 {font-size: 2.2em; background: linear-gradient(135deg, #667eea, #764ba2); -webkit-background-clip: text; -webkit-text-fill-color: transparent}
+@media (max-width: 768px) {
+  #header h1 {font-size: 1.6em !important}
+}
 """
 
 DESCRIPTION = """
-# 🌀 GIT HOLOGRAM
-Visualize any Git repository as **causal threads of intention**. Each commit is a point in spacetime where thought crystallized into code.
-
-**Enter a GitHub repo URL** (or `owner/repo` shorthand) and watch the hologram form.
+<div id="header">
+<h1>🌀 GIT HOLOGRAM + CAPT</h1>
+<p style="color:#888">Visualize any Git repository as <strong>causal threads of intention</strong>. Then analyze it with the bioCAPT cognitive backend.</p>
+</div>
 """
 
 with gr.Blocks(css=CUSTOM_CSS, theme=gr.themes.Soft()) as demo:
     gr.Markdown(DESCRIPTION)
+
+    # Health banner
+    health_html = gr.HTML(health_banner_html(backend))
+    refresh_health_btn = gr.Button("🔄 Refresh Health", size="sm")
+    refresh_health_btn.click(fn=lambda: health_banner_html(backend), outputs=health_html)
+
+    # Inputs
     with gr.Row():
         repo_input = gr.Textbox(
             label="Repo URL or `owner/repo`",
@@ -242,14 +341,55 @@ with gr.Blocks(css=CUSTOM_CSS, theme=gr.themes.Soft()) as demo:
             label="Max commits", scale=1,
         )
         generate_btn = gr.Button("🔮 Generate Hologram", variant="primary", scale=1)
-    output = gr.HTML(label="Hologram")
+
+    # Hologram output
+    hologram_output = gr.HTML(label="Hologram")
+
+    # Hidden state to track current repo URL
+    current_repo = gr.State("")
+
+    # CAPT Analysis section (appears after hologram)
+    with gr.Row():
+        analyze_select = gr.Dropdown(
+            ["codebase", "commits", "security"],
+            label="📡 CAPT Analysis",
+            value="codebase",
+            scale=1,
+        )
+        analyze_btn = gr.Button("🧠 Analyze with CAPT", variant="primary", scale=0, min_width=180)
+
+    capt_output = gr.Markdown(label="CAPT Analysis Result", elem_classes=["result-box"])
+
+    def on_generate(repo_url, max_commits):
+        html, url, status = fetch_repo(repo_url, max_commits)
+        if status in ("cached", "ok"):
+            return html, url, f"✅ Status: {status.upper()} · Repo cached"
+        return html, url, f"⚠️ Status: {status.upper()}"
 
     generate_btn.click(
-        fn=fetch_repo, inputs=[repo_input, max_commits_input], outputs=output,
+        fn=on_generate,
+        inputs=[repo_input, max_commits_input],
+        outputs=[hologram_output, current_repo, capt_output],
     )
     repo_input.submit(
-        fn=fetch_repo, inputs=[repo_input, max_commits_input], outputs=output,
+        fn=on_generate,
+        inputs=[repo_input, max_commits_input],
+        outputs=[hologram_output, current_repo, capt_output],
     )
 
+    analyze_btn.click(
+        fn=capt_analyze,
+        inputs=[current_repo, analyze_select],
+        outputs=capt_output,
+    )
+
+    # Footer
+    gr.Markdown("""
+    ---
+    <div style="text-align:center;color:#6b6b7b;font-size:0.8rem">
+    Git Hologram · Powered by bioCAPT · Connected to real CAPT Cloudflare Workers
+    </div>
+    """)
+
 if __name__ == "__main__":
-    demo.launch()
+    demo.launch()

capt_backend.py

@@ -0,0 +1,351 @@
+"""
+🧠 CAPT Backend Connector — Universal module for all HF Spaces
+Provides real CAPT/bioCAPT cognition via Cloudflare Workers with graceful fallbacks.
+
+Usage:
+    from capt_backend import CAPTBackend, BrainRouter
+    
+    backend = CAPTBackend()
+    result = backend.cogitate("What is consciousness?")
+    print(result["response"])
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import time
+from dataclasses import dataclass, field
+from typing import Any, Optional
+import requests
+
+# ─── Configuration ───
+_DEFAULT_WORKER = os.getenv("CAPT_WORKER_URL", "https://capt-brain-01.knowurknottty.workers.dev")
+_OPENROUTER_KEY = os.getenv("OPENROUTER_API_KEY", "")
+
+BRAINS = {
+    "capt-core-01": {
+        "name": "CAPT Core Alpha",
+        "icon": "🧠",
+        "worker": "https://capt-brain-01.knowurknottty.workers.dev",
+        "specialties": ["general", "analysis", "reasoning", "constitution", "ethics"],
+    },
+    "biocapt-core-01": {
+        "name": "bioCAPT Genesis",
+        "icon": "🧬",
+        "worker": "https://capt-brain-02-biocapt.knowurknottty.workers.dev",
+        "specialties": ["memory", "learning", "adapt", "bio", "organic"],
+    },
+    "frankencapt-core-01": {
+        "name": "FrankenCAPT Chimera",
+        "icon": "⚗️",
+        "worker": "https://capt-brain-03-frankencapt.knowurknottty.workers.dev",
+        "specialties": ["modular", "combine", "fuse", "skill", "build"],
+    },
+    "synthesis-core-01": {
+        "name": "Synthesis Nexus",
+        "icon": "🔮",
+        "worker": "https://capt-brain-04-synthesis.knowurknottty.workers.dev",
+        "specialties": ["synthesize", "consensus", "meta", "orchestrate"],
+    },
+    "council-core-01": {
+        "name": "LLM Council",
+        "icon": "🏛️",
+        "worker": "https://capt-brain-05-council.knowurknottty.workers.dev",
+        "specialties": ["council", "vote", "debate", "deliberate"],
+    },
+}
+
+SEVEN_LAWS = [
+    ("LAW_1_HUMAN_DIGNITY", "Respect human autonomy and dignity", "#00e676"),
+    ("LAW_2_TRUTHFULNESS", "Do not deceive or mislead", "#69f0ae"),
+    ("LAW_3_NON_MALEFICENCE", "Do no harm", "#ff5252"),
+    ("LAW_4_PRIVACY", "Protect personal data", "#448aff"),
+    ("LAW_5_TRANSPARENCY", "Explain reasoning when asked", "#ffd740"),
+    ("LAW_6_ACCOUNTABILITY", "Own mistakes, allow audit", "#ffab40"),
+    ("LAW_7_BENEFICENCE", "Act for human flourishing", "#e040fb"),
+]
+
+@dataclass
+class BrainHealth:
+    brain_id: str
+    name: str
+    icon: str
+    online: bool = False
+    status: str = "unknown"
+    latency_ms: int = 0
+    last_check: float = 0
+    error: str = ""
+
+
+class CAPTBackend:
+    """Universal backend connector for CAPT cognitive architecture."""
+
+    def __init__(self, worker_url: str = _DEFAULT_WORKER, timeout: int = 60):
+        self.worker_url = worker_url.rstrip("/")
+        self.timeout = timeout
+        self._health_cache: dict[str, BrainHealth] = {}
+        self._cache_ttl = 30  # seconds
+
+    # ─── Health ───
+    def check_health(self, force: bool = False) -> dict[str, BrainHealth]:
+        now = time.time()
+        if not force and self._health_cache and all(
+            now - h.last_check < self._cache_ttl for h in self._health_cache.values()
+        ):
+            return self._health_cache
+
+        results = {}
+        for bid, binfo in BRAINS.items():
+            try:
+                r = requests.get(f"{binfo['worker']}/health", timeout=10)
+                data = r.json() if r.status_code == 200 else {}
+                results[bid] = BrainHealth(
+                    brain_id=bid,
+                    name=binfo["name"],
+                    icon=binfo["icon"],
+                    online=r.status_code == 200,
+                    status=data.get("status", "unknown"),
+                    latency_ms=int(r.elapsed.total_seconds() * 1000),
+                    last_check=now,
+                )
+            except Exception as e:
+                results[bid] = BrainHealth(
+                    brain_id=bid,
+                    name=binfo["name"],
+                    icon=binfo["icon"],
+                    online=False,
+                    status="offline",
+                    last_check=now,
+                    error=str(e),
+                )
+        self._health_cache = results
+        return results
+
+    @property
+    def is_online(self) -> bool:
+        health = self.check_health()
+        return any(h.online for h in health.values())
+
+    # ─── Core API ───
+    def cogitate(self, query: str, context: str = "", model: str = "", tags: list = None) -> dict:
+        """Run full 46-module CAPT pipeline."""
+        payload = {"query": query, "context": context, "modality": "text"}
+        if model:
+            payload["model"] = model
+        if tags:
+            payload["tags"] = tags
+        return self._post("/cogitate", payload)
+
+    def pulse(self, prompt: str, model: str = "", system: str = "", temperature: float = 0.7, max_tokens: int = 2048) -> dict:
+        """Direct LLM generation via PULSE."""
+        payload = {"prompt": prompt, "temperature": temperature, "maxTokens": max_tokens}
+        if model:
+            payload["model"] = model
+        if system:
+            payload["systemPrompt"] = system
+        return self._post("/pulse", payload)
+
+    def echo_store(self, content: str, salience: float = 0.5, tags: list = None) -> dict:
+        """Store an episodic memory trace."""
+        payload = {"content": content, "salience": salience}
+        if tags:
+            payload["tags"] = tags
+        return self._post("/echo/store", payload)
+
+    def echo_recall(self, query: str, top_k: int = 5) -> dict:
+        """Recall memory traces by similarity."""
+        return self._post("/echo/recall", {"query": query, "topK": top_k})
+
+    def immu_scan(self, text: str) -> dict:
+        """Constitutional scan for violations."""
+        return self._post("/immu/scan", {"text": text})
+
+    def constitution_enforce(self, text: str) -> dict:
+        """Enforce Seven Laws on content."""
+        return self._post("/constitution/enforce", {"text": text})
+
+    def get_status(self) -> dict:
+        """Get brain status and API endpoints."""
+        try:
+            r = requests.get(f"{self.worker_url}/status", timeout=10)
+            return r.json() if r.status_code == 200 else {"error": f"HTTP {r.status_code}"}
+        except Exception as e:
+            return {"error": str(e)}
+
+    # ─── Internal ───
+    def _post(self, endpoint: str, payload: dict) -> dict:
+        try:
+            r = requests.post(
+                f"{self.worker_url}{endpoint}",
+                json=payload,
+                timeout=self.timeout,
+                headers={"Content-Type": "application/json"},
+            )
+            if r.status_code == 200:
+                return r.json()
+            return {"error": f"HTTP {r.status_code}", "detail": r.text[:200]}
+        except Exception as e:
+            return {"error": str(e)}
+
+    # ─── Fallback Data (for when backend is down) ───
+    def get_fallback_cogitation(self, query: str) -> dict:
+        """Beautiful fallback when CAPT backend is unavailable."""
+        return {
+            "response": f"⚡ CAPT Backend Status: Offline\n\nYour query: '{query[:60]}...'\n\nThe cognitive architecture is currently warming up. Try again in a moment, or check the Fleet Controller for brain health.",
+            "confidence": 0.0,
+            "pipelineTrace": {"stages": [], "totalLatencyMs": 0, "moduleCount": 0},
+            "echoTraces": [],
+            "consensus": {"confidence": 0, "dissenterCount": 0},
+            "threats": {"clean": True, "threats": []},
+            "metrics": {},
+            "_fallback": True,
+        }
+
+    def get_fallback_health_banner(self) -> str:
+        health = self.check_health()
+        online = sum(1 for h in health.values() if h.online)
+        total = len(health)
+        if online == total:
+            return f"🟢 All {total} brains online"
+        elif online > 0:
+            return f"🟡 {online}/{total} brains online"
+        return "🔴 All brains offline — check Fleet Controller"
+
+
+class BrainRouter:
+    """Routes queries to the best brain based on content analysis."""
+
+    KEYWORDS = {
+        "capt-core-01": ["general", "analyze", "explain", "what is", "how to", "why", "compare", "constitution", "ethics", "laws", "truth", "dignity"],
+        "biocapt-core-01": ["memory", "remember", "recall", "learn", "adapt", "evolve", "bio", "organic", "neural", "synapse", "cell", "dna"],
+        "frankencapt-core-01": ["modular", "combine", "fuse", "mix", "variant", "skill", "build", "assemble", "component", "plugin"],
+        "synthesis-core-01": ["synthesize", "consensus", "vote", "aggregate", "combine brains", "meta", "orchestrate", "coordinate", "unify"],
+        "council-core-01": ["council", "vote", "deliberate", "debate", "panel", "jury", "multi-model", "ensemble", "committee"],
+    }
+
+    @classmethod
+    def route(cls, query: str) -> str:
+        q = query.lower()
+        scores = {bid: 0 for bid in BRAINS}
+        for bid, keywords in cls.KEYWORDS.items():
+            for kw in keywords:
+                if kw in q:
+                    scores[bid] += 1
+        best = max(scores, key=scores.get)
+        return best if scores[best] > 0 else "capt-core-01"
+
+    @classmethod
+    def get_brain_for_space(cls, space_name: str) -> str:
+        """Suggest the best brain for a given space type."""
+        space_lower = space_name.lower()
+        if "wiki" in space_lower:
+            return "capt-core-01"
+        elif "bio" in space_lower or "universal" in space_lower:
+            return "biocapt-core-01"
+        elif "franken" in space_lower or "forge" in space_lower:
+            return "frankencapt-core-01"
+        elif "council" in space_lower or "dashboard" in space_lower:
+            return "council-core-01"
+        elif "novel" in space_lower or "studio" in space_lower:
+            return "capt-core-01"
+        elif "cogitate" in space_lower:
+            return "capt-core-01"
+        return "capt-core-01"
+
+
+# ─── OpenRouter Direct (when Worker is unavailable) ───
+def direct_llm(prompt: str, model: str = "deepseek/deepseek-chat-v3-0324", system: str = "", temperature: float = 0.7) -> dict:
+    """Direct OpenRouter call — bypasses CAPT Workers entirely."""
+    if not _OPENROUTER_KEY:
+        return {"error": "OPENROUTER_API_KEY not set", "content": "API key required for LLM generation."}
+    try:
+        r = requests.post(
+            "https://openrouter.ai/api/v1/chat/completions",
+            headers={
+                "Authorization": f"Bearer {_OPENROUTER_KEY}",
+                "Content-Type": "application/json",
+                "HTTP-Referer": "https://capt.dev",
+                "X-Title": "CAPT-HF-Spaces",
+            },
+            json={
+                "model": model,
+                "messages": [
+                    {"role": "system", "content": system or "You are CAPT, a 46-module cognitive architecture."},
+                    {"role": "user", "content": prompt},
+                ],
+                "temperature": temperature,
+                "max_tokens": 2048,
+            },
+            timeout=60,
+        )
+        data = r.json()
+        if r.status_code != 200:
+            return {"error": data.get("error", {}).get("message", "Unknown error"), "content": ""}
+        return {
+            "content": data["choices"][0]["message"]["content"],
+            "model": data.get("model", model),
+        }
+    except Exception as e:
+        return {"error": str(e), "content": ""}
+
+
+# ─── Beautiful HTML Components ───
+def health_banner_html(backend: CAPTBackend) -> str:
+    """Generate a beautiful health banner for Gradio/HTML embedding."""
+    health = backend.check_health()
+    online_count = sum(1 for h in health.values() if h.online)
+    total = len(health)
+    
+    if online_count == total:
+        color = "#00e676"
+        emoji = "🟢"
+        text = f"All {total} brains online"
+    elif online_count > 0:
+        color = "#ffd740"
+        emoji = "🟡"
+        text = f"{online_count}/{total} brains online"
+    else:
+        color = "#ff5252"
+        emoji = "🔴"
+        text = "All brains offline"
+    
+    cards = ""
+    for bid, h in health.items():
+        dot = "🟢" if h.online else "🔴"
+        latency = f"{h.latency_ms}ms" if h.online else "offline"
+        cards += f'<div style="display:inline-block;padding:6px 10px;margin:3px;background:#1a1a2e;border-radius:8px;font-size:0.75rem"><span style="font-size:1rem">{h.icon}</span> {dot} {h.name}<br><span style="color:#8b8b9b">{latency}</span></div>'
+    
+    return f"""
+    <div style="background:linear-gradient(135deg,#12121a,#0a0a0f);border:1px solid #1a1a2e;border-radius:12px;padding:14px;margin-bottom:16px">
+      <div style="display:flex;align-items:center;gap:8px;margin-bottom:8px">
+        <span style="font-size:1.2rem">{emoji}</span>
+        <span style="color:{color};font-weight:700;font-size:0.9rem">{text}</span>
+        <span style="color:#6b6b7b;font-size:0.75rem;margin-left:auto">CAPT Backend Connector v1.0</span>
+      </div>
+      <div style="display:flex;flex-wrap:wrap;gap:4px">{cards}</div>
+    </div>
+    """
+
+
+def pipeline_trace_html(trace: dict) -> str:
+    """Render a beautiful pipeline trace."""
+    stages = trace.get("stages", [])
+    if not stages:
+        return "<div style='color:#6b6b7b;font-size:0.8rem'>No pipeline trace available</div>"
+    
+    total = trace.get("totalLatencyMs", 0)
+    bars = ""
+    for s in stages:
+        color = "#00e676" if s.get("success") else "#ff5252"
+        name = s.get("module", "?")
+        latency = s.get("latencyMs", 0)
+        width = max(2, min(12, latency / 50))
+        bars += f'<div title="{name}: {latency}ms" style="display:inline-block;width:{width}px;height:20px;background:{color};margin-right:1px;border-radius:2px"></div>'
+    
+    return f"""
+    <div style="background:#0d0d14;border-radius:8px;padding:10px">
+      <div style="color:#8b8b9b;font-size:0.7rem;margin-bottom:6px">⚡ Pipeline: {len(stages)} modules · {total}ms</div>
+      <div style="display:flex;align-items:center;height:20px">{bars}</div>
+    </div>
+    """

requirements.txt

@@ -1 +1,2 @@
 gradio>=5.0.0
+requests>=2.28.0