rewrite: use gradio.Server + @app .api() + custom HTML (proper Gradio 6.x pattern)

app.py

@@ -1,24 +1,15 @@
 """SoniCoder v2 — Entry point.
 
-Optimized AI code writer agent with:
-- Event-driven agent protocol (from Gemini CLI)
-- ToolBuilder/ToolInvocation pattern (from Gemini CLI)
-- MCP support (from both Gemini CLI & Claude Code)
-- Multiple smart 1B models (Qwen2.5-Coder-1.5B, DeepSeek-Coder-1.3B, etc.)
-- Policy/permission engine (from Gemini CLI)
-- Context management
-- Professional web UI
+Uses gradio.Server (Gradio 6.x) to serve a custom HTML frontend
+with Gradio's queuing, concurrency, and API infrastructure.
 
-Usage:
-    python app.py
+Read: https://huggingface.co/blog/introducing-gradio-server
 """
 
 from __future__ import annotations
 
-import json
 import logging
 import os
-from pathlib import Path
 
 # Ensure workspace exists
 from sonicoder.config import WORKSPACE_ROOT, CONFIG_DIR
@@ -55,62 +46,14 @@ def main():
             else:
                 logger.info(f"MCP server '{name}' connected")
 
-    # Create the Gradio app with custom HTML
-    from sonicoder.server import get_app, api_get_config
+    # Create the server (gradio.Server — custom HTML + Gradio backend)
+    from sonicoder.server import create_app
+    app = create_app()
 
-    # Build config JSON for frontend injection
-    config_json = api_get_config()
-    config_data = json.loads(config_json)
-    config_str = json.dumps(config_data, ensure_ascii=False)
-
-    # Read index.html and inject config
-    index_path = Path(__file__).parent / "static" / "index.html"
-    custom_html = None
-    if index_path.exists():
-        html = index_path.read_text(encoding="utf-8")
-        html = html.replace(
-            "__RUNTIME_CONFIG__ || {}",
-            config_str + " || {}"
-        )
-        custom_html = html
-        logger.info("Custom HTML prepared")
-
-    application = get_app()
-
-    # Register custom root route after app is built
-    if custom_html:
-        try:
-            from fastapi import Response
-            from fastapi.responses import HTMLResponse
-
-            # Try multiple ways to access the underlying app in Gradio 6.x
-            underlying = None
-            for attr in ("app", "server_app", "_app", "app_obj"):
-                underlying = getattr(application, attr, None)
-                if underlying is not None:
-                    break
-
-            if underlying is not None and hasattr(underlying, "get"):
-                @underlying.get("/", response_class=HTMLResponse, include_in_schema=False)
-                async def serve_index():
-                    return Response(content=custom_html, media_type="text/html")
-                logger.info("Custom root route registered successfully")
-            else:
-                logger.warning("Could not register custom root route — using default Gradio UI")
-        except Exception as e:
-            logger.warning(f"Custom root route registration failed: {e}")
-
-    # Launch — HF Spaces sets PORT env var
+    # Launch
     port = int(os.environ.get("PORT", 7860))
-    logger.info(f"Launching web interface on port {port}")
-
-    application.launch(
-        show_error=True,
-        server_name="0.0.0.0",
-        server_port=port,
-        share=False,
-        theme=None,
-    )
+    logger.info(f"Launching on port {port}")
+    app.launch(show_error=True)
 
 
 if __name__ == "__main__":

sonicoder/server/__init__.py

@@ -1,12 +1,15 @@
-"""Server routes — FastAPI + Gradio API endpoints.
+"""Server routes — using gradio.Server (Gradio 6.x pattern).
 
-All HTTP and Gradio API endpoints for the application.
+Read the blog: https://huggingface.co/blog/introducing-gradio-server
+gradio.Server extends FastAPI. @app.api() gives queuing + concurrency.
+@app.get("/") serves custom HTML frontend directly.
 """
 
 from __future__ import annotations
 
 import json
 import logging
+import os
 import uuid
 from pathlib import Path
 from typing import Any, Generator
@@ -48,14 +51,12 @@ def _get_registry() -> ToolRegistry:
     if _registry is None:
         _registry = get_tool_registry()
         register_builtin_tools(_registry)
-        # Apply filters
         settings = _get_settings()
         _registry.set_filters(settings.enabled_tools, settings.disabled_tools)
     return _registry
 
 
 def _get_or_create_session(session_id: str = "") -> tuple[str, AgentSession]:
-    """Get existing session or create new one."""
     if not session_id:
         session_id = str(uuid.uuid4())[:12]
     if session_id not in _sessions:
@@ -66,14 +67,14 @@ def _get_or_create_session(session_id: str = "") -> tuple[str, AgentSession]:
     return session_id, _sessions[session_id]
 
 
-# ── Gradio API Functions ──────────────────────────────────────────────
+# ── API Handlers ───────────────────────────────────────────────────────
 
 def api_chat(
     message: str,
     session_id: str = "",
     image_url: str | None = None,
 ) -> Generator[str, None, None]:
-    """Main chat/agent endpoint — streams events as JSON lines."""
+    """Main chat endpoint — streams events as JSON lines."""
     if not message or not message.strip():
         yield json.dumps({"type": "error", "error": "Empty message"})
         return
@@ -88,27 +89,16 @@ def api_chat(
         yield json.dumps({"type": "error", "error": str(e)})
 
 
-def api_chat_sync(message: str, session_id: str = "") -> str:
-    """Non-streaming chat for simple calls."""
-    events = list(api_chat(message, session_id))
-    if not events:
-        return json.dumps({"type": "error", "error": "No response"})
-    return events[-1]  # Return last event (complete)
-
-
 def api_model_status() -> str:
-    """Get model loading status."""
     return json.dumps(get_model_status())
 
 
 def api_switch_model(model_key: str) -> str:
-    """Switch to a different model."""
     switch_model(model_key)
     return json.dumps({"status": "switching", "model": model_key})
 
 
 def api_list_models() -> str:
-    """List all available models."""
     models = get_available_models()
     current = get_model_status().get("model_key", "")
     result = []
@@ -125,8 +115,6 @@ def api_list_models() -> str:
 
 
 def api_workspace_tree(max_depth: int = 3) -> str:
-    """Get workspace directory tree."""
-    import os
     tree = []
 
     def walk(path: Path, prefix: str = "", depth: int = 0):
@@ -136,7 +124,6 @@ def api_workspace_tree(max_depth: int = 3) -> str:
             items = sorted(path.iterdir(), key=lambda p: (not p.is_dir(), p.name.lower()))
         except PermissionError:
             return
-
         skip = {".git", "node_modules", "__pycache__", ".venv", "venv", "dist", "build", ".sonicoder"}
         for i, item in enumerate(items):
             if item.name.startswith(".") or item.name in skip:
@@ -160,7 +147,6 @@ def api_workspace_tree(max_depth: int = 3) -> str:
 
 
 def api_workspace_read(path: str, offset: int = 1, limit: int = 500) -> str:
-    """Read a workspace file."""
     from ..tools.fs import _resolve_safe
     try:
         resolved = _resolve_safe(path)
@@ -183,7 +169,6 @@ def api_workspace_read(path: str, offset: int = 1, limit: int = 500) -> str:
 
 
 def api_workspace_write(path: str, content: str) -> str:
-    """Write a workspace file."""
     from ..tools.fs import _resolve_safe
     try:
         resolved = _resolve_safe(path)
@@ -195,7 +180,6 @@ def api_workspace_write(path: str, content: str) -> str:
 
 
 def api_workspace_bash(command: str) -> str:
-    """Execute a bash command in workspace."""
     from ..tools.bash import BashBuilder
     builder = BashBuilder()
     errors = builder.validate({"command": command})
@@ -207,7 +191,6 @@ def api_workspace_bash(command: str) -> str:
 
 
 def api_web_search(query: str) -> str:
-    """Search the web."""
     from ..tools.web import WebSearchBuilder
     builder = WebSearchBuilder()
     invocation = builder.create_invocation({"query": query})
@@ -216,29 +199,21 @@ def api_web_search(query: str) -> str:
 
 
 def api_mcp_status() -> str:
-    """Get MCP server status."""
-    manager = get_mcp_manager()
-    return json.dumps(manager.get_status())
+    return json.dumps(get_mcp_manager().get_status())
 
 
 def api_policy_status() -> str:
-    """Get policy engine status."""
     engine = get_policy_engine()
-    return json.dumps({
-        "mode": engine.approval_mode,
-        "rules": engine.get_rules(),
-    })
+    return json.dumps({"mode": engine.approval_mode, "rules": engine.get_rules()})
 
 
 def api_reset_session(session_id: str = "") -> str:
-    """Reset a chat session."""
     if session_id in _sessions:
         _sessions[session_id].clear_history()
     return json.dumps({"success": True})
 
 
 def api_get_config() -> str:
-    """Get runtime configuration for the frontend."""
     models = get_available_models()
     model_list = []
     for key, config in models.items():
@@ -249,7 +224,6 @@ def api_get_config() -> str:
             "size_gb": config.size_gb,
             "description": config.description,
         })
-
     return json.dumps({
         "models": model_list,
         "default_model": _get_settings().default_model,
@@ -258,123 +232,90 @@ def api_get_config() -> str:
     }, ensure_ascii=False)
 
 
-# ── Create Gradio App ──────────────────────────────────────────────────
+# ── Build App (gradio.Server) ─────────────────────────────────────────
+
+def create_app():
+    """Create and return a gradio.Server with custom HTML frontend.
 
-def get_app():
-    """Create and return the Gradio application."""
+    Uses the Gradio 6.x Server pattern:
+    - @app.api() for backend endpoints (gets queuing + concurrency)
+    - @app.get("/") to serve custom index.html
+    - No gr.Blocks, no hidden Textbox hacks
+    """
     import gradio as gr
-    from fastapi.responses import FileResponse
-    from pathlib import Path
-
-    # Create the Gradio app
-    app = gr.Blocks(
-        title="SoniCoder v2 — AI Code Agent",
-    )
-
-    with app:
-        # Hidden components for API
-        api_chat_output = gr.Textbox(visible=False)
-        api_status_output = gr.Textbox(visible=False)
-        api_json_output = gr.Textbox(visible=False)
-
-        # API endpoints
-        gr.Interface(
-            fn=api_chat,
-            inputs=[
-                gr.Textbox(label="message"),
-                gr.Textbox(label="session_id", value=""),
-                gr.Textbox(label="image_url", value=""),
-            ],
-            outputs=[api_chat_output],
-            api_name="chat",
-        )
+    from fastapi.responses import HTMLResponse
 
-        gr.Interface(
-            fn=api_model_status,
-            inputs=[],
-            outputs=[api_json_output],
-            api_name="model_status",
-        )
+    app = gr.Server()
 
-        gr.Interface(
-            fn=api_switch_model,
-            inputs=[gr.Textbox(label="model_key")],
-            outputs=[api_json_output],
-            api_name="switch_model",
-        )
+    # ── Backend API endpoints (with Gradio queuing) ──
 
-        gr.Interface(
-            fn=api_list_models,
-            inputs=[],
-            outputs=[api_json_output],
-            api_name="list_models",
-        )
+    @app.api(name="chat")
+    def chat(message: str, session_id: str = "", image_url: str = ""):
+        """Stream chat events as newline-delimited JSON."""
+        return api_chat(message, session_id, image_url or None)
 
-        gr.Interface(
-            fn=api_workspace_tree,
-            inputs=[gr.Number(label="max_depth", value=3)],
-            outputs=[api_json_output],
-            api_name="workspace_tree",
-        )
+    @app.api(name="model_status")
+    def model_status():
+        return api_model_status()
 
-        gr.Interface(
-            fn=api_workspace_read,
-            inputs=[
-                gr.Textbox(label="path"),
-                gr.Number(label="offset", value=1),
-                gr.Number(label="limit", value=500),
-            ],
-            outputs=[api_json_output],
-            api_name="workspace_read",
-        )
+    @app.api(name="switch_model")
+    def switch_model_api(model_key: str):
+        return api_switch_model(model_key)
 
-        gr.Interface(
-            fn=api_workspace_write,
-            inputs=[gr.Textbox(label="path"), gr.Textbox(label="content", lines=10)],
-            outputs=[api_json_output],
-            api_name="workspace_write",
-        )
+    @app.api(name="list_models")
+    def list_models():
+        return api_list_models()
 
-        gr.Interface(
-            fn=api_workspace_bash,
-            inputs=[gr.Textbox(label="command")],
-            outputs=[api_json_output],
-            api_name="workspace_bash",
-        )
+    @app.api(name="workspace_tree")
+    def workspace_tree(max_depth: int = 3):
+        return api_workspace_tree(max_depth)
 
-        gr.Interface(
-            fn=api_web_search,
-            inputs=[gr.Textbox(label="query")],
-            outputs=[api_json_output],
-            api_name="web_search",
-        )
+    @app.api(name="workspace_read")
+    def workspace_read(path: str, offset: int = 1, limit: int = 500):
+        return api_workspace_read(path, offset, limit)
 
-        gr.Interface(
-            fn=api_mcp_status,
-            inputs=[],
-            outputs=[api_json_output],
-            api_name="mcp_status",
-        )
+    @app.api(name="workspace_write")
+    def workspace_write(path: str, content: str):
+        return api_workspace_write(path, content)
 
-        gr.Interface(
-            fn=api_policy_status,
-            inputs=[],
-            outputs=[api_json_output],
-            api_name="policy_status",
-        )
+    @app.api(name="workspace_bash")
+    def workspace_bash(command: str):
+        return api_workspace_bash(command)
 
-        gr.Interface(
-            fn=api_reset_session,
-            inputs=[gr.Textbox(label="session_id", value="")],
-            outputs=[api_json_output],
-            api_name="reset_session",
-        )
+    @app.api(name="web_search")
+    def web_search(query: str):
+        return api_web_search(query)
 
-        gr.Interface(
-            fn=api_get_config,
-            inputs=[],
-            outputs=[api_json_output],
-            api_name="get_config",
-        )
+    @app.api(name="mcp_status")
+    def mcp_status():
+        return api_mcp_status()
+
+    @app.api(name="policy_status")
+    def policy_status():
+        return api_policy_status()
+
+    @app.api(name="reset_session")
+    def reset_session(session_id: str = ""):
+        return api_reset_session(session_id)
+
+    @app.api(name="get_config")
+    def get_config():
+        return api_get_config()
+
+    # ── Serve custom HTML frontend ──
+
+    @app.get("/", response_class=HTMLResponse)
+    async def homepage():
+        """Serve the custom SPA frontend."""
+        # Build config JSON to inject
+        config_str = api_get_config()
+
+        html_path = Path(__file__).parent.parent.parent / "static" / "index.html"
+        if not html_path.exists():
+            return HTMLResponse("<h1>SoniCoder v2</h1><p>index.html not found</p>")
+
+        html = html_path.read_text(encoding="utf-8")
+        html = html.replace("__RUNTIME_CONFIG__ || {}", config_str + " || {}")
+        return HTMLResponse(content=html)
 
     return app