Upload folder using huggingface_hub

app.py

@@ -1,1393 +1,38 @@
-"""Fullstack Code Builder - Local AI-powered fullstack app generator.
+"""Fullstack Code Builder — entry point.
 
 Uses MiniCPM5-1B for local inference (no external APIs).
 Supports generating fullstack applications in any language.
 Can push generated projects to HuggingFace Hub.
 Web search via Google scraping (no API keys needed).
 Gradio app support for Python.
+
+Project structure:
+    code/
+    ├── config/constants.py         App constants, language options, system prompt
+    ├── model/loader.py             Model loading & status
+    ├── model/inference.py          Streaming model inference
+    ├── execution/code_extractor.py Code extraction & language normalization
+    ├── execution/python_runner.py  Sandboxed Python execution
+    ├── execution/gradio_runner.py  Gradio app subprocess runner
+    ├── websearch/google_scraper.py Google search scraping (no API)
+    ├── huggingface/push.py         HuggingFace Hub push & ZIP packaging
+    ├── server/chat_helpers.py      Chat history & prompt building
+    └── server/routes.py            FastAPI / Gradio server routes
 """
 
 from __future__ import annotations
 
-import html
-import json
 import logging
-import os
-import re
-import shutil
-import subprocess
-import sys
-import tempfile
-import textwrap
-import threading
-import time
-import urllib.parse
-import zipfile
-from collections.abc import Iterator
-from dataclasses import dataclass, field
-from pathlib import Path
-from typing import Any
-
-from gradio import Server
-from fastapi.responses import HTMLResponse, FileResponse
-
-APP_TITLE = "Fullstack Code Builder"
-MODEL_ID = "openbmb/MiniCPM5-1B"
-MODEL_URL = "https://huggingface.co/openbmb/MiniCPM5-1B"
-
-DEFAULT_TEMPERATURE = 0.6
-DEFAULT_MAX_TOKENS = 4096
-PY_TIMEOUT_S = 15
-GRADIO_TIMEOUT_S = 30
-PY_MEM_LIMIT_MB = 1024
-MAX_STDIO_CHARS = 16_000
-OUTPUT_PNG = "output.png"
 
-THINKING_BLOCK_RE = re.compile(r"<\s*think\s*>.*?<\s*/\s*think\s*>", re.IGNORECASE | re.DOTALL)
-CODE_BLOCK_RE = re.compile(r"```([a-zA-Z0-9_+.#-]*)\s*\n(.*?)```", re.DOTALL)
-FILE_BLOCK_RE = re.compile(r"@@FILE:\s*(.+?)@@\s*\n(.*?)(?=@@FILE:|@@END@@)", re.DOTALL)
+from code.model.loader import start_background_load
+from code.server.routes import get_app
 
-logger = logging.getLogger(__name__)
 logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
 
-# ─── Supported Languages & Frameworks ───────────────────────────────────
-
-LANGUAGE_OPTIONS = [
-    ("Python", ["Gradio", "Flask", "Django", "FastAPI", "Streamlit", "Plain Python"]),
-    ("JavaScript", ["React", "Vue.js", "Next.js", "Express.js", "Node.js", "Vanilla JS"]),
-    ("TypeScript", ["React", "Next.js", "Express.js", "NestJS"]),
-    ("HTML/CSS/JS", ["Tailwind CSS", "Bootstrap", "Vanilla"]),
-    ("Java", ["Spring Boot", "Maven", "Gradle"]),
-    ("Go", ["Gin", "Fiber", "Echo", "Plain Go"]),
-    ("Rust", ["Actix", "Axum", "Rocket"]),
-    ("PHP", ["Laravel", "Symfony", "Plain PHP"]),
-    ("Ruby", ["Rails", "Sinatra"]),
-    ("C#", ["ASP.NET", "Blazor"]),
-    ("Swift", ["Vapor", "SwiftUI"]),
-    ("Kotlin", ["Ktor", "Spring Boot"]),
-]
-
-LANGUAGE_MAP = {lang: frameworks for lang, frameworks in LANGUAGE_OPTIONS}
-
-SYSTEM_PROMPT = """You are a fullstack application code generator running locally. You help users build complete, runnable applications in any programming language and framework.
-
-When the user asks you to build an application:
-1. Generate complete, working code - not snippets or pseudocode
-2. Include all necessary files for the project to run
-3. Add proper error handling and comments
-4. For web apps, make the UI responsive and modern
-5. For Gradio apps, use gradio library and create a complete working app with gr.Interface or gr.Blocks
-
-FILE OUTPUT FORMAT - IMPORTANT:
-When generating multi-file projects, wrap each file in this format:
-@@FILE: path/to/file.ext@@
-(file content here)
-@@FILE: path/to/another/file.ext@@
-(another file content here)
-@@END@@
-
-For single-file code, use standard markdown fenced blocks:
-```python for Python
-```html for HTML/CSS/JS
-```javascript for JavaScript
-```typescript for TypeScript
-etc.
-
-When generating web apps with HTML/CSS/JS, return a single self-contained HTML document with all CSS and JavaScript inline. Make the page fully responsive: html/body at margin:0 and 100% width/height, use flexbox/grid layouts, and size any canvas to its container.
-
-When generating Gradio apps, create a complete app.py with:
-- import gradio as gr
-- Define the interface using gr.Interface() or gr.Blocks()
-- Call iface.launch(server_name="0.0.0.0", server_port=7860) at the end
-- Include all necessary processing logic inline
-
-For Python, prefer standard library or common packages. Do not use network calls, subprocesses, shell commands, or long-running loops in demo code (except Gradio apps which are server-based).
-
-If web search results are provided in the context, use them to inform your code generation. Incorporate relevant information from the search results into the generated code.
-"""
-
-# Curated starter prompts
-EXAMPLE_PROMPTS: list[tuple[str, str, str, str]] = [
-    (
-        "🎨 Gradio Image Filter",
-        "Create a Gradio app that lets users upload an image and apply filters like grayscale, blur, sepia, and edge detection using PIL. Show the original and filtered images side by side.",
-        "Python",
-        "Gradio",
-    ),
-    (
-        "🤖 Gradio Chat App",
-        "Build a Gradio chatbot app with gr.Blocks that has a chat interface, a text input, and a send button. Include a simple echo bot that repeats the user's message with a fun twist.",
-        "Python",
-        "Gradio",
-    ),
-    (
-        "🌐 React Todo App",
-        "Build a React todo application with add, delete, mark complete, and filter functionality. Use modern hooks and a clean responsive UI.",
-        "JavaScript",
-        "React",
-    ),
-    (
-        "🐍 Flask API",
-        "Create a Flask REST API for a book library with CRUD operations, in-memory storage, and proper error handling.",
-        "Python",
-        "Flask",
-    ),
-    (
-        "🎨 Landing Page",
-        "Build a modern landing page for a SaaS product with a hero section, features grid, pricing cards, and a footer. Use Tailwind-style CSS.",
-        "HTML/CSS/JS",
-        "Vanilla",
-    ),
-    (
-        "📊 Dashboard",
-        "Create an interactive data dashboard with charts (bar, line, pie), a sidebar navigation, and summary cards. All in a single HTML file.",
-        "HTML/CSS/JS",
-        "Vanilla",
-    ),
-]
-
-
-# ─── Web Search (Google Scraping — No API) ──────────────────────────────
-
-def web_search_google(query: str, num_results: int = 8) -> list[dict[str, str]]:
-    """Search Google by scraping the results page. No API key needed.
-
-    Returns a list of dicts with keys: title, url, snippet.
-    Uses requests with a browser-like User-Agent to avoid captchas.
-    """
-    try:
-        import requests
-        from bs4 import BeautifulSoup
-
-        encoded_query = urllib.parse.quote_plus(query)
-        url = f"https://www.google.com/search?q={encoded_query}&num={num_results + 2}&hl=en"
-
-        headers = {
-            "User-Agent": (
-                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
-                "AppleWebKit/537.36 (KHTML, like Gecko) "
-                "Chrome/120.0.0.0 Safari/537.36"
-            ),
-            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
-            "Accept-Language": "en-US,en;q=0.5",
-            "Accept-Encoding": "gzip, deflate",
-            "DNT": "1",
-            "Connection": "keep-alive",
-            "Upgrade-Insecure-Requests": "1",
-        }
-
-        resp = requests.get(url, headers=headers, timeout=10, allow_redirects=True)
-        resp.raise_for_status()
-
-        soup = BeautifulSoup(resp.text, "html.parser")
-        results: list[dict[str, str]] = []
-
-        # Parse Google search results
-        # Google uses various class names; we try multiple selectors
-        for g_div in soup.select("div.g, div[data-sokoban-container], div.yuRUbf"):
-            title_el = g_div.select_one("h3")
-            link_el = g_div.select_one("a[href]")
-            snippet_el = g_div.select_one("div.VwiC3b, span.aCOpRe, div[data-sncf]")
-
-            if not title_el or not link_el:
-                continue
-
-            href = link_el.get("href", "")
-            # Google sometimes prefixes URLs; extract the real URL
-            if href.startswith("/url?q="):
-                real_url = urllib.parse.parse_qs(urllib.parse.urlparse(href).query).get("q", [href])[0]
-            elif href.startswith("http"):
-                real_url = href
-            else:
-                continue
-
-            # Skip Google-internal URLs
-            if "google.com" in real_url or "googleusercontent.com" in real_url:
-                continue
-
-            title = title_el.get_text(strip=True)
-            snippet = snippet_el.get_text(strip=True) if snippet_el else ""
-
-            if title and real_url:
-                results.append({
-                    "title": title,
-                    "url": real_url,
-                    "snippet": snippet,
-                })
-
-            if len(results) >= num_results:
-                break
-
-        # Fallback: try parsing from <a> tags with data-ved attribute
-        if not results:
-            for a_tag in soup.select("a[data-ved]"):
-                href = a_tag.get("href", "")
-                if not href.startswith("http"):
-                    continue
-                if "google.com" in href:
-                    continue
-
-                title_el = a_tag.select_one("h3, span")
-                title = title_el.get_text(strip=True) if title_el else a_tag.get_text(strip=True)[:100]
-                snippet = ""
-
-                if title and href:
-                    results.append({
-                        "title": title,
-                        "url": href,
-                        "snippet": snippet,
-                    })
-
-                if len(results) >= num_results:
-                    break
-
-        logger.info("Web search for '%s' returned %d results", query, len(results))
-        return results
-
-    except ImportError:
-        logger.warning("requests or beautifulsoup4 not installed for web search")
-        return []
-    except Exception as exc:
-        logger.exception("Web search failed: %s", exc)
-        return []
-
-
-def format_search_results(results: list[dict[str, str]]) -> str:
-    """Format search results into a text block for model context."""
-    if not results:
-        return "No search results found."
-
-    parts = ["Here are the web search results for reference:\n"]
-    for i, r in enumerate(results, 1):
-        parts.append(f"{i}. {r['title']}")
-        parts.append(f"   URL: {r['url']}")
-        if r["snippet"]:
-            parts.append(f"   {r['snippet']}")
-        parts.append("")
-
-    return "\n".join(parts)
-
-
-# ─── Model Loading ──────────────────────────────────────────────────────
-
-_model = None
-_tokenizer = None
-_model_loaded = False
-_model_loading = False
-_load_error: str | None = None
-
-
-def load_model() -> None:
-    """Load MiniCPM5-1B model and tokenizer locally."""
-    global _model, _tokenizer, _model_loaded, _model_loading, _load_error
-
-    if _model_loaded or _model_loading:
-        return
-
-    _model_loading = True
-    _load_error = None
-
-    try:
-        from transformers import AutoModelForCausalLM, AutoTokenizer
-        import torch
-
-        logger.info("Loading MiniCPM5-1B model...")
-
-        dtype = torch.float16 if torch.cuda.is_available() else torch.float32
-        device_map = "auto" if torch.cuda.is_available() else None
-
-        _tokenizer = AutoTokenizer.from_pretrained(
-            MODEL_ID,
-            trust_remote_code=True,
-        )
-        _model = AutoModelForCausalLM.from_pretrained(
-            MODEL_ID,
-            torch_dtype=dtype,
-            device_map=device_map,
-            trust_remote_code=True,
-            low_cpu_mem_usage=True,
-        )
-
-        if device_map is None:
-            _model = _model.to("cpu")
-
-        _model.eval()
-        _model_loaded = True
-        logger.info("MiniCPM5-1B model loaded successfully.")
-
-    except Exception as exc:
-        _load_error = str(exc)
-        logger.exception("Failed to load model: %s", exc)
-    finally:
-        _model_loading = False
-
-
-# Start loading model in background thread
-_load_thread = threading.Thread(target=load_model, daemon=True)
-_load_thread.start()
-
-
-def get_model_status() -> dict[str, Any]:
-    """Return current model loading status."""
-    if _model_loaded:
-        return {"status": "ready", "message": "Model loaded and ready"}
-    if _model_loading:
-        return {"status": "loading", "message": "Model is loading... (this may take a few minutes on first run)"}
-    if _load_error:
-        return {"status": "error", "message": f"Model load error: {_load_error}"}
-    return {"status": "unknown", "message": "Model not initialized"}
-
-
-# ─── Model Inference ────────────────────────────────────────────────────
-
-def call_model(messages: list[dict[str, Any]], max_new_tokens: int = DEFAULT_MAX_TOKENS) -> Iterator[str]:
-    """Stream model text using local MiniCPM5-1B."""
-
-    if not _model_loaded:
-        status = get_model_status()
-        yield status["message"]
-        return
-
-    try:
-        from transformers import TextIteratorStreamer
-        import torch
-
-        # Build the prompt from messages
-        prompt_parts = []
-        for msg in messages:
-            role = msg.get("role", "user")
-            content = msg.get("content", "")
-            if role == "system":
-                prompt_parts.append(f"System: {content}")
-            elif role == "user":
-                prompt_parts.append(f"User: {content}")
-            elif role == "assistant":
-                prompt_parts.append(f"Assistant: {content}")
-        prompt_parts.append("Assistant:")
-        full_prompt = "\n\n".join(prompt_parts)
-
-        # Tokenize
-        inputs = _tokenizer(full_prompt, return_tensors="pt", truncation=True, max_length=4096)
-        if torch.cuda.is_available():
-            inputs = {k: v.to("cuda") for k, v in inputs.items()}
-
-        # Stream generation
-        streamer = TextIteratorStreamer(_tokenizer, skip_prompt=True, skip_special_tokens=True)
-
-        generation_kwargs = {
-            **inputs,
-            "streamer": streamer,
-            "max_new_tokens": max_new_tokens,
-            "temperature": DEFAULT_TEMPERATURE,
-            "do_sample": True,
-            "top_p": 0.9,
-            "repetition_penalty": 1.1,
-            "pad_token_id": _tokenizer.eos_token_id,
-        }
-
-        # Run generation in a separate thread
-        thread = threading.Thread(target=_model.generate, kwargs=generation_kwargs)
-        thread.start()
-
-        output = ""
-        for new_text in streamer:
-            output += new_text
-            yield output
-
-        thread.join()
-
-    except Exception as exc:
-        logger.exception("Error during model inference")
-        yield f"_Error during generation: {exc}_"
-
-
-def call_model_sync(messages: list[dict[str, Any]], max_new_tokens: int = DEFAULT_MAX_TOKENS) -> str:
-    """Non-streaming model call - returns complete response."""
-    result = ""
-    for chunk in call_model(messages, max_new_tokens):
-        result = chunk
-    return result
-
-
-# ─── Code Extraction ────────────────────────────────────────────────────
-
-def _strip_thinking_blocks(text: str) -> str:
-    return THINKING_BLOCK_RE.sub("", text).strip()
-
-
-def extract_code(response: str) -> tuple[str, str | None]:
-    """Return the first fenced code block and its language tag."""
-    visible_response = _strip_thinking_blocks(response)
-    match = CODE_BLOCK_RE.search(visible_response)
-    if not match:
-        return "", None
-    return match.group(2).strip(), (match.group(1).strip().lower() or None)
-
-
-def extract_multi_file(response: str) -> dict[str, str]:
-    """Extract multi-file project from @@FILE: format.
-
-    Returns dict of {filepath: content}.
-    """
-    files: dict[str, str] = {}
-    visible = _strip_thinking_blocks(response)
-
-    for match in FILE_BLOCK_RE.finditer(visible):
-        filepath = match.group(1).strip()
-        content = match.group(2).strip()
-        files[filepath] = content
-
-    # Fallback: if no @@FILE: blocks found, extract single code block
-    if not files:
-        code, lang = extract_code(response)
-        if code:
-            ext_map = {
-                "python": "main.py", "py": "main.py",
-                "javascript": "index.js", "js": "index.js",
-                "typescript": "index.ts", "ts": "index.ts",
-                "html": "index.html",
-                "css": "styles.css",
-                "java": "Main.java",
-                "go": "main.go",
-                "rust": "main.rs",
-                "php": "index.php",
-                "ruby": "main.rb",
-                "csharp": "Program.cs",
-                "swift": "main.swift",
-                "kotlin": "Main.kt",
-            }
-            filename = ext_map.get(lang or "", "code.txt")
-            files[filename] = code
-
-    return files
-
-
-def _normalize_language(target_language: str | None, fence_lang: str | None) -> str:
-    """Normalize language name to a canonical form."""
-    lang = (fence_lang or target_language or "python").lower()
-    if lang in {"python", "py"}:
-        return "python"
-    if lang in {"html", "web", "css"}:
-        return "web"
-    if lang in {"javascript", "js"}:
-        return "javascript"
-    if lang in {"typescript", "ts"}:
-        return "typescript"
-    if lang == "java":
-        return "java"
-    if lang == "go":
-        return "go"
-    if lang == "rust":
-        return "rust"
-    if lang == "php":
-        return "php"
-    if lang == "ruby":
-        return "ruby"
-    if lang in {"csharp", "c#"}:
-        return "csharp"
-    if lang == "swift":
-        return "swift"
-    if lang == "kotlin":
-        return "kotlin"
-    return lang
-
-
-def _is_gradio_code(code: str) -> bool:
-    """Detect if Python code is a Gradio app."""
-    return bool(re.search(r"import\s+gradio|from\s+gradio\s+import|gr\.\s*(Interface|Blocks|TabbedInterface|ChatInterface|App)", code))
-
-
-# ─── Python Execution ───────────────────────────────────────────────────
-
-@dataclass
-class PythonExecutionResult:
-    stdout: str
-    stderr: str
-    image_path: str | None
-    returncode: int | None
-    timed_out: bool = False
-
-
-def _apply_subprocess_limits() -> None:
-    import resource
-    mem_bytes = PY_MEM_LIMIT_MB * 1024 * 1024
-    resource.setrlimit(resource.RLIMIT_AS, (mem_bytes, mem_bytes))
-    resource.setrlimit(resource.RLIMIT_CPU, (PY_TIMEOUT_S, PY_TIMEOUT_S))
-
-
-def _python_runner_source() -> str:
-    return textwrap.dedent(
-        f"""
-        import os
-        import runpy
-        import sys
-        import traceback
-
-        os.environ.setdefault("MPLBACKEND", "Agg")
-        exit_code = 0
-        try:
-            runpy.run_path(os.path.join(os.getcwd(), "user_code.py"), run_name="__main__")
-        except SystemExit as exc:
-            code = exc.code
-            exit_code = code if isinstance(code, int) else 1
-        except Exception:
-            traceback.print_exc()
-            exit_code = 1
-        finally:
-            try:
-                import matplotlib
-                matplotlib.use("Agg", force=True)
-                import matplotlib.pyplot as plt
-                if plt.get_fignums():
-                    plt.savefig(os.environ["OUTPUT_PNG"], bbox_inches="tight")
-            except ModuleNotFoundError as exc:
-                if exc.name != "matplotlib":
-                    traceback.print_exc()
-            except Exception:
-                traceback.print_exc()
-
-        raise SystemExit(exit_code)
-        """
-    ).strip()
-
-
-def _truncate_output(text: str) -> str:
-    if len(text) <= MAX_STDIO_CHARS:
-        return text
-    remaining = len(text) - MAX_STDIO_CHARS
-    return text[:MAX_STDIO_CHARS] + f"\n\n... truncated {remaining} characters ..."
-
-
-def _decode_timeout_output(value: str | bytes | None) -> str:
-    if value is None:
-        return ""
-    if isinstance(value, bytes):
-        return value.decode("utf-8", errors="replace")
-    return value
-
-
-def run_python(code: str) -> PythonExecutionResult:
-    with tempfile.TemporaryDirectory(prefix="fullstack_run_") as tmp:
-        workdir = Path(tmp)
-        runner_path = workdir / "runner.py"
-        user_path = workdir / "user_code.py"
-        image_path = workdir / OUTPUT_PNG
-
-        runner_path.write_text(_python_runner_source(), encoding="utf-8")
-        user_path.write_text(code, encoding="utf-8")
-
-        env = {
-            "PATH": "/usr/bin:/bin",
-            "HOME": str(workdir),
-            "TMPDIR": str(workdir),
-            "MPLBACKEND": "Agg",
-            "MPLCONFIGDIR": str(workdir / ".matplotlib"),
-            "OUTPUT_PNG": str(image_path),
-            "PYTHONIOENCODING": "utf-8",
-            "PYTHONNOUSERSITE": "1",
-            "PYTHONUNBUFFERED": "1",
-            "LANG": "C.UTF-8",
-            "OPENBLAS_NUM_THREADS": "1",
-            "OMP_NUM_THREADS": "1",
-            "MKL_NUM_THREADS": "1",
-            "NUMEXPR_NUM_THREADS": "1",
-        }
-
-        try:
-            completed = subprocess.run(
-                [sys.executable, "-I", str(runner_path)],
-                cwd=workdir,
-                env=env,
-                capture_output=True,
-                text=True,
-                encoding="utf-8",
-                errors="replace",
-                timeout=PY_TIMEOUT_S,
-                preexec_fn=_apply_subprocess_limits if sys.platform == "linux" else None,
-                check=False,
-            )
-            stdout = _truncate_output(completed.stdout)
-            stderr = _truncate_output(completed.stderr)
-
-            if completed.returncode and not stderr:
-                stderr = f"Process exited with status {completed.returncode}."
-
-            saved_image: str | None = None
-            if image_path.exists() and image_path.stat().st_size > 0:
-                saved = tempfile.NamedTemporaryFile(
-                    prefix="fullstack_plot_", suffix=".png", delete=False
-                )
-                saved.close()
-                Path(saved.name).write_bytes(image_path.read_bytes())
-                saved_image = saved.name
-
-            return PythonExecutionResult(
-                stdout=stdout,
-                stderr=stderr,
-                image_path=saved_image,
-                returncode=completed.returncode,
-            )
-        except subprocess.TimeoutExpired as exc:
-            stdout = _truncate_output(_decode_timeout_output(exc.stdout))
-            stderr = _truncate_output(_decode_timeout_output(exc.stderr))
-            timeout_note = f"Timed out after {PY_TIMEOUT_S} seconds; the process was killed."
-            stderr = f"{stderr}\n{timeout_note}".strip()
-            return PythonExecutionResult(
-                stdout=stdout,
-                stderr=stderr,
-                image_path=None,
-                returncode=None,
-                timed_out=True,
-            )
-
-
-# ─── Gradio App Runner ─────────────────────────────────────────────────
-
-# Registry for running Gradio subprocesses
-_running_gradio_procs: dict[str, subprocess.Popen] = {}
-
-
-def run_gradio_app(code: str, port: int = 7861) -> dict[str, Any]:
-    """Launch a Gradio app as a subprocess and return its URL.
-
-    The Gradio app is run on the specified port. We modify the code
-    to ensure it launches on the correct port and is accessible.
-    """
-    # Kill any previously running Gradio app
-    for pid, proc in list(_running_gradio_procs.items()):
-        try:
-            proc.terminate()
-            proc.wait(timeout=3)
-        except Exception:
-            try:
-                proc.kill()
-            except Exception:
-                pass
-    _running_gradio_procs.clear()
-
-    # Patch the code: ensure launch uses correct server_name and server_port
-    patched_code = code
-
-    # Replace .launch() with correct params
-    patched_code = re.sub(
-        r"(\w+)\.launch\([^)]*\)",
-        f'\\1.launch(server_name="0.0.0.0", server_port={port}, share=False)',
-        patched_code,
-    )
-
-    # If no .launch() found, add one
-    if ".launch(" not in patched_code:
-        # Add launch at the end if missing
-        patched_code += f'\n\nif __name__ == "__main__":\n    iface.launch(server_name="0.0.0.0", server_port={port}, share=False)\n'
-
-    with tempfile.TemporaryDirectory(prefix="gradio_app_") as tmp:
-        app_path = Path(tmp) / "gradio_app.py"
-        app_path.write_text(patched_code, encoding="utf-8")
-
-        env = {
-            **os.environ,
-            "PYTHONUNBUFFERED": "1",
-            "GRADIO_SERVER_NAME": "0.0.0.0",
-            "GRADIO_SERVER_PORT": str(port),
-        }
-
-        try:
-            proc = subprocess.Popen(
-                [sys.executable, str(app_path)],
-                cwd=tmp,
-                env=env,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE,
-                text=True,
-            )
-
-            proc_id = f"gradio_{port}"
-            _running_gradio_procs[proc_id] = proc
-
-            # Wait a bit for the server to start
-            import time as _time
-            _time.sleep(3)
-
-            # Check if process is still running
-            poll = proc.poll()
-            if poll is not None:
-                stdout = proc.stdout.read() if proc.stdout else ""
-                stderr = proc.stderr.read() if proc.stderr else ""
-                return {
-                    "success": False,
-                    "url": "",
-                    "message": f"Gradio app exited with code {poll}",
-                    "stdout": stdout[-2000:] if stdout else "",
-                    "stderr": stderr[-2000:] if stderr else "",
-                }
-
-            gradio_url = f"http://localhost:{port}"
-            return {
-                "success": True,
-                "url": gradio_url,
-                "message": f"Gradio app running at {gradio_url}",
-                "port": port,
-            }
-
-        except Exception as exc:
-            logger.exception("Failed to launch Gradio app")
-            return {
-                "success": False,
-                "url": "",
-                "message": f"Failed to launch: {exc}",
-            }
-
-
-def stop_gradio_app() -> dict[str, Any]:
-    """Stop any running Gradio app subprocess."""
-    stopped = 0
-    for pid, proc in list(_running_gradio_procs.items()):
-        try:
-            proc.terminate()
-            proc.wait(timeout=3)
-            stopped += 1
-        except Exception:
-            try:
-                proc.kill()
-                stopped += 1
-            except Exception:
-                pass
-    _running_gradio_procs.clear()
-
-    return {"success": True, "message": f"Stopped {stopped} Gradio app(s)"}
-
-
-# ─── Web Document ───────────────────────────────────────────────────────
-
-def _web_document(code: str, fence_lang: str | None) -> str:
-    lang = (fence_lang or "").lower()
-    if lang in {"javascript", "js"}:
-        return f"<!doctype html><html><body><script>\n{code}\n</script></body></html>"
-    if lang == "css":
-        return f"<!doctype html><html><head><style>\n{code}\n</style></head><body></body></html>"
-    if re.search(r"<!doctype|<html[\s>]", code, flags=re.IGNORECASE):
-        return code
-    return f"<!doctype html><html><head><meta charset='utf-8'></head><body>\n{code}\n</body></html>"
-
-
-def build_iframe(code: str, fence_lang: str | None = None) -> str:
-    document = _web_document(code, fence_lang)
-    srcdoc = html.escape(document, quote=True)
-    return (
-        '<iframe class="web-frame" '
-        'sandbox="allow-scripts" '
-        'allow="fullscreen" '
-        "allowfullscreen "
-        f'srcdoc="{srcdoc}" '
-        'style="width:100%; min-height:680px; border:0; border-radius:14px; '
-        'background:white;"></iframe>'
-    )
-
-
-# ─── Project Packaging ──────────────────────────────────────────────────
-
-def create_project_zip(files: dict[str, str], project_name: str) -> str:
-    """Create a ZIP file from extracted project files."""
-    zip_dir = tempfile.mkdtemp(prefix="fullstack_project_")
-    zip_path = os.path.join(zip_dir, f"{project_name}.zip")
-
-    with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
-        for filepath, content in files.items():
-            zf.writestr(f"{project_name}/{filepath}", content)
-
-    return zip_path
-
-
-# ─── HuggingFace Hub Push ───────────────────────────────────────────────
-
-def push_to_huggingface(
-    files: dict[str, str],
-    project_name: str,
-    repo_name: str,
-    hf_token: str,
-    space_sdk: str = "static",
-    is_space: bool = True,
-) -> dict[str, Any]:
-    """Push generated project to HuggingFace Hub."""
-    try:
-        from huggingface_hub import HfApi, create_repo
-
-        api = HfApi(token=hf_token)
-
-        if "/" in repo_name:
-            namespace, name = repo_name.split("/", 1)
-        else:
-            user_info = api.whoami()
-            namespace = user_info["name"]
-            name = repo_name
-            repo_name = f"{namespace}/{name}"
-
-        try:
-            if is_space:
-                create_repo(
-                    repo_id=repo_name,
-                    repo_type="space",
-                    space_sdk=space_sdk,
-                    token=hf_token,
-                    exist_ok=True,
-                )
-            else:
-                create_repo(
-                    repo_id=repo_name,
-                    repo_type="model",
-                    token=hf_token,
-                    exist_ok=True,
-                )
-        except Exception as e:
-            logger.warning("Repo creation warning: %s", e)
-
-        with tempfile.TemporaryDirectory(prefix="hf_push_") as tmp_dir:
-            for filepath, content in files.items():
-                full_path = os.path.join(tmp_dir, filepath)
-                os.makedirs(os.path.dirname(full_path), exist_ok=True)
-                Path(full_path).write_text(content, encoding="utf-8")
-
-            # Add README if not present
-            readme_path = os.path.join(tmp_dir, "README.md")
-            if not os.path.exists(readme_path):
-                readme_content = f"""---
-title: {name}
-emoji: 🚀
-colorFrom: blue
-colorTo: purple
-sdk: {space_sdk}
-app_file: app.py
----
-
-# {name}
-
-Generated by Fullstack Code Builder using {MODEL_ID}.
-"""
-                Path(readme_path).write_text(readme_content, encoding="utf-8")
-
-            # Add requirements.txt for Python/Gradio projects
-            req_path = os.path.join(tmp_dir, "requirements.txt")
-            if not os.path.exists(req_path):
-                has_python = any(
-                    f.endswith(".py") for f in files.keys()
-                )
-                if has_python:
-                    reqs = ["gradio>=4.0.0"]
-                    # Detect common dependencies
-                    all_code = "\n".join(files.values())
-                    if "matplotlib" in all_code:
-                        reqs.append("matplotlib>=3.8")
-                    if "PIL" in all_code or "Pillow" in all_code:
-                        reqs.append("Pillow>=10.0")
-                    if "numpy" in all_code:
-                        reqs.append("numpy>=1.24")
-                    if "pandas" in all_code:
-                        reqs.append("pandas>=2.0")
-                    Path(req_path).write_text("\n".join(reqs) + "\n", encoding="utf-8")
-
-            api.upload_folder(
-                folder_path=tmp_dir,
-                repo_id=repo_name,
-                repo_type="space" if is_space else "model",
-                token=hf_token,
-            )
-
-        repo_url = f"https://huggingface.co/{repo_name}"
-        if is_space:
-            repo_url = f"https://huggingface.co/spaces/{repo_name}"
-
-        return {
-            "success": True,
-            "url": repo_url,
-            "repo_name": repo_name,
-            "message": f"Successfully pushed to {repo_url}",
-        }
-
-    except Exception as exc:
-        logger.exception("Failed to push to HuggingFace")
-        return {
-            "success": False,
-            "url": "",
-            "repo_name": repo_name,
-            "message": f"Failed to push: {str(exc)}",
-        }
-
-
-# ─── Chat Helpers ───────────────────────────────────────────────────────
-
-def _chat_history_to_messages(history: list[dict[str, str]]) -> list[dict[str, Any]]:
-    messages: list[dict[str, Any]] = [{"role": "system", "content": SYSTEM_PROMPT}]
-    for item in history:
-        role = item.get("role")
-        content = str(item.get("content") or "").strip()
-        if role not in {"user", "assistant"} or not content:
-            continue
-        if role == "assistant":
-            content = _strip_thinking_blocks(content)
-        messages.append({"role": role, "content": content})
-    return messages
-
-
-def _clip_context(text: str, limit: int = 4_000) -> str:
-    if len(text) <= limit:
-        return text
-    return text[:limit] + f"\n... truncated {len(text) - limit} characters ..."
-
-
-def _iteration_context(execution_context: dict[str, Any] | None) -> str:
-    if not execution_context or not execution_context.get("code"):
-        return ""
-
-    code = _clip_context(str(execution_context.get("code") or ""), 6_000)
-    target = str(execution_context.get("target") or "code")
-    fence_lang = str(execution_context.get("fence_lang") or target)
-    status = str(execution_context.get("status") or "")
-    stdout = _clip_context(str(execution_context.get("stdout") or ""), 2_000)
-    stderr = _clip_context(str(execution_context.get("stderr") or ""), 2_000)
-
-    parts = [
-        "Previous generated code and run result are available for iteration.",
-        f"Previous target: {target}",
-        f"Previous status: {status}",
-        f"Previous code:\n```{fence_lang}\n{code}\n```",
-    ]
-    if stdout:
-        parts.append(f"Previous stdout:\n{stdout}")
-    if stderr:
-        parts.append(f"Previous stderr / traceback:\n{stderr}")
-    parts.append("If the user asks to revise, debug, extend, or explain the prior code, use this context.")
-    return "\n\n".join(parts)
-
-
-def _targeted_prompt(
-    prompt: str,
-    target_language: str,
-    target_framework: str = "",
-    execution_context: dict[str, Any] | None = None,
-    search_context: str = "",
-) -> str:
-    iteration_context = _iteration_context(execution_context)
-    context_block = f"\n\n{iteration_context}" if iteration_context else ""
-
-    search_block = ""
-    if search_context:
-        search_block = f"\n\n{search_context}\n\nUse the above search results to inform your code generation if relevant."
-
-    framework_hint = f" using {target_framework}" if target_framework else ""
-
-    gradio_hint = ""
-    if target_framework == "Gradio":
-        gradio_hint = (
-            "\n\nIMPORTANT: This is a Gradio app. Create a complete Python script that:\n"
-            "- Imports gradio as gr\n"
-            "- Defines the UI using gr.Interface() or gr.Blocks()\n"
-            "- Includes all processing logic inline\n"
-            "- Calls .launch(server_name='0.0.0.0', server_port=7860) at the end\n"
-            "- Uses only standard library + gradio + common packages (PIL, matplotlib, numpy)\n"
-            "- Make the UI clean, modern, and functional"
-        )
-
-    return (
-        f"Target: {target_language}{framework_hint}. Generate a complete, runnable application. "
-        "If the user asks for a web app, include all HTML/CSS/JS. "
-        "If they ask for a backend, include the server code and any API definitions. "
-        "For single-file apps, use a single code block. For multi-file projects, use the @@FILE: format. "
-        "Make the code complete, working, and well-structured."
-        f"{gradio_hint}"
-        f"{search_block}"
-        f"{context_block}\n\n"
-        f"User request:\n{prompt}"
-    )
-
-
-# ─── Run Extracted Code ────────────────────────────────────────────────
-
-def _run_extracted_code(
-    code: str,
-    target: str,
-    framework: str = "",
-) -> tuple[str, str, str | None, str, str]:
-    """Run extracted code. For Gradio apps, launch as a subprocess server."""
-    if target == "python" and _is_gradio_code(code):
-        result = run_gradio_app(code)
-        if result["success"]:
-            return (
-                result.get("stdout", ""),
-                f"Gradio app running at {result['url']}",
-                None,
-                f"Gradio running at {result['url']}",
-                "success",
-            )
-        else:
-            return (
-                result.get("stdout", ""),
-                result.get("stderr", result.get("message", "Gradio launch failed")),
-                None,
-                "Gradio launch failed",
-                "error",
-            )
-
-    if target == "python":
-        result = run_python(code)
-        if result.timed_out:
-            return result.stdout, result.stderr, result.image_path, f"Timed out after {PY_TIMEOUT_S}s", "error"
-        if result.returncode:
-            return result.stdout, result.stderr, result.image_path, "Finished with errors", "error"
-        return result.stdout, result.stderr, result.image_path, "Ran successfully", "success"
-
-    return "", "", None, "Preview ready", "success"
-
-
-# ─── Served Files Registry ──────────────────────────────────────────────
-
-_served_files: dict[str, str] = {}
-
-# ─── FastAPI / Gradio Application ───────────────────────────────────────
-
-app = Server()
-
-
-@app.get("/", response_class=HTMLResponse)
-async def homepage():
-    html_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "index.html")
-    with open(html_path, "r", encoding="utf-8") as f:
-        content = f.read()
-
-    config = json.dumps({
-        "app_title": APP_TITLE,
-        "model_id": MODEL_ID,
-        "model_url": MODEL_URL,
-        "languages": LANGUAGE_OPTIONS,
-        "examples": [
-            {"label": label, "prompt": prompt, "language": lang, "framework": fw}
-            for label, prompt, lang, fw in EXAMPLE_PROMPTS
-        ],
-    })
-    content = content.replace("__RUNTIME_CONFIG__", config)
-    return content
-
-
-@app.get("/api/model-status")
-async def model_status_endpoint():
-    return get_model_status()
-
-
-@app.get("/images/{filename}")
-async def serve_image(filename: str):
-    path = _served_files.get(f"img:{filename}")
-    if path and os.path.exists(path):
-        return FileResponse(path, media_type="image/png")
-    return HTMLResponse("Not found", status_code=404)
-
-
-@app.get("/download/{filename}")
-async def serve_download(filename: str):
-    path = _served_files.get(f"dl:{filename}")
-    if path and os.path.exists(path):
-        return FileResponse(path, filename=filename, media_type="application/octet-stream")
-    return HTMLResponse("Not found", status_code=404)
-
-
-@app.api(name="web_search", concurrency_limit=4)
-def handle_web_search(query: str) -> str:
-    """Search the web using Google scraping. No API key needed."""
-    query = (query or "").strip()
-    if not query:
-        yield json.dumps({"success": False, "results": [], "message": "Empty search query"})
-        return
-
-    try:
-        results = web_search_google(query, num_results=8)
-        formatted = format_search_results(results)
-
-        yield json.dumps({
-            "success": True,
-            "results": results,
-            "formatted": formatted,
-            "message": f"Found {len(results)} results",
-        })
-    except Exception as exc:
-        logger.exception("Web search failed")
-        yield json.dumps({
-            "success": False,
-            "results": [],
-            "message": f"Search failed: {str(exc)}",
-        })
-
-
-@app.api(name="chat", concurrency_limit=2)
-def handle_chat(
-    prompt: str,
-    target_language: str,
-    target_framework: str,
-    history_json: str,
-    exec_context_json: str,
-    search_enabled: str = "false",
-) -> str:
-    """Stream chat responses with code execution. Yields JSON strings."""
-    history = json.loads(history_json) if history_json else []
-    execution_context = json.loads(exec_context_json) if exec_context_json else {}
-
-    prompt = (prompt or "").strip()
-    if not prompt:
-        yield json.dumps({
-            "type": "error",
-            "status_text": "Enter a prompt to get started.",
-            "status_state": "info",
-            "history": history,
-            "execution": execution_context,
-        })
-        return
-
-    # Check model status
-    model_status = get_model_status()
-    if model_status["status"] == "loading":
-        yield json.dumps({
-            "type": "error",
-            "status_text": model_status["message"],
-            "status_state": "working",
-            "history": history,
-            "execution": execution_context,
-        })
-        return
-    if model_status["status"] != "ready":
-        yield json.dumps({
-            "type": "error",
-            "status_text": model_status["message"],
-            "status_state": "error",
-            "history": history,
-            "execution": execution_context,
-        })
-        return
-
-    # Add user message and placeholder assistant message
-    history = list(history) + [
-        {"role": "user", "content": prompt},
-        {"role": "assistant", "content": ""},
-    ]
-    yield json.dumps({
-        "type": "status",
-        "status_text": "Thinking...",
-        "status_state": "working",
-        "history": history,
-        "execution": execution_context,
-    })
-
-    # Web search if enabled
-    search_context = ""
-    if search_enabled.lower() == "true":
-        yield json.dumps({
-            "type": "status",
-            "status_text": "Searching the web...",
-            "status_state": "working",
-            "history": history,
-            "execution": execution_context,
-        })
-        search_results = web_search_google(prompt, num_results=6)
-        if search_results:
-            search_context = format_search_results(search_results)
-            yield json.dumps({
-                "type": "search_results",
-                "status_text": f"Found {len(search_results)} results, generating code...",
-                "status_state": "working",
-                "history": history,
-                "execution": execution_context,
-                "search_results": search_results,
-            })
-
-    # Build messages for model
-    model_history = list(history[:-1])
-    model_history[-1] = {
-        "role": "user",
-        "content": _targeted_prompt(prompt, target_language, target_framework, execution_context, search_context),
-    }
-    messages = _chat_history_to_messages(model_history)
-
-    final_response = ""
-    for partial in call_model(messages):
-        final_response = partial
-        history[-1]["content"] = partial
-        yield json.dumps({
-            "type": "streaming",
-            "status_text": "Generating...",
-            "status_state": "working",
-            "history": history,
-            "execution": execution_context,
-        })
-
-    if not final_response:
-        history[-1]["content"] = "The model did not return a response."
-        yield json.dumps({
-            "type": "error",
-            "status_text": "No model response.",
-            "status_state": "error",
-            "history": history,
-            "execution": execution_context,
-        })
-        return
-
-    # Extract code from response
-    code, fence_lang = extract_code(final_response)
-    target = _normalize_language(target_language, fence_lang)
-
-    # Also try multi-file extraction
-    multi_files = extract_multi_file(final_response)
-
-    if not code and not multi_files:
-        yield json.dumps({
-            "type": "complete",
-            "status_text": "Answered without running code.",
-            "status_state": "info",
-            "history": history,
-            "execution": execution_context,
-        })
-        return
-
-    yield json.dumps({
-        "type": "status",
-        "status_text": "Running...",
-        "status_state": "working",
-        "history": history,
-        "execution": execution_context,
-    })
-
-    # Execute code
-    stdout, stderr, image_path, status_text, status_state = "", "", None, "Preview ready", "success"
-    is_gradio = False
-    gradio_url = None
-
-    if target == "python" and code:
-        if _is_gradio_code(code) or target_framework == "Gradio":
-            is_gradio = True
-            gradio_result = run_gradio_app(code)
-            if gradio_result["success"]:
-                gradio_url = gradio_result["url"]
-                status_text = f"Gradio app running at {gradio_url}"
-                status_state = "success"
-                stderr = f"Gradio app launched successfully at {gradio_url}"
-            else:
-                status_text = "Gradio launch failed"
-                status_state = "error"
-                stderr = gradio_result.get("stderr", gradio_result.get("message", "Launch failed"))
-        else:
-            stdout, stderr, image_path, status_text, status_state = _run_extracted_code(code, target, target_framework)
-
-    # Register image for serving
-    image_url = None
-    if image_path:
-        filename = os.path.basename(image_path)
-        _served_files[f"img:{filename}"] = image_path
-        image_url = f"/images/{filename}"
-
-    # Register code for download
-    download_url = None
-    project_files = multi_files if multi_files else {}
-
-    if project_files:
-        project_name = "generated-project"
-        zip_path = create_project_zip(project_files, project_name)
-        zip_filename = f"{project_name}.zip"
-        _served_files[f"dl:{zip_filename}"] = zip_path
-        download_url = f"/download/{zip_filename}"
-    elif code:
-        ext = "py" if target == "python" else "html"
-        dl_filename = f"generated.{ext}"
-        dl_dir = tempfile.mkdtemp(prefix="fullstack_dl_")
-        dl_path = os.path.join(dl_dir, dl_filename)
-        Path(dl_path).write_text(code, encoding="utf-8")
-        _served_files[f"dl:{dl_filename}"] = dl_path
-        download_url = f"/download/{dl_filename}"
-
-    # Determine if this is web previewable
-    is_web = target in {"web", "javascript", "typescript", "html"} or (fence_lang or "") in {"html", "web"}
-    web_code = code if is_web else None
-
-    execution_context = {
-        "code": code,
-        "target": target,
-        "fence_lang": fence_lang or target,
-        "stdout": stdout,
-        "stderr": stderr,
-        "image_url": image_url,
-        "image_path": image_path,
-        "status": status_text,
-        "language": fence_lang or target,
-        "suggested_tab": "preview" if (image_path or is_web or is_gradio) else "console",
-        "download_url": download_url,
-        "project_files": project_files,
-        "is_web": is_web,
-        "web_code": web_code,
-        "is_gradio": is_gradio,
-        "gradio_url": gradio_url,
-    }
-
-    yield json.dumps({
-        "type": "complete",
-        "status_text": status_text,
-        "status_state": status_state,
-        "history": history,
-        "execution": execution_context,
-    })
-
-
-@app.api(name="push_hf", concurrency_limit=1)
-def handle_push_hf(
-    exec_context_json: str,
-    repo_name: str,
-    hf_token: str,
-    space_sdk: str = "static",
-    is_space: str = "true",
-) -> str:
-    """Push generated project to HuggingFace Hub."""
-    try:
-        execution_context = json.loads(exec_context_json) if exec_context_json else {}
-        project_files = execution_context.get("project_files", {})
-
-        if not project_files:
-            code = execution_context.get("code", "")
-            if not code:
-                yield json.dumps({
-                    "success": False,
-                    "message": "No code to push. Generate some code first.",
-                    "url": "",
-                })
-                return
-
-            lang = execution_context.get("language", "python")
-            is_gradio = execution_context.get("is_gradio", False)
-            ext_map = {
-                "python": "app.py", "py": "app.py",
-                "javascript": "index.js", "js": "index.js",
-                "html": "index.html", "web": "index.html",
-                "typescript": "index.ts", "ts": "index.ts",
-            }
-            filename = ext_map.get(lang, "app.py")
-            project_files = {filename: code}
-
-            # Auto-detect SDK for Gradio apps
-            if is_gradio or _is_gradio_code(code):
-                space_sdk = "gradio"
-
-        project_name = repo_name.split("/")[-1] if "/" in repo_name else repo_name
-
-        result = push_to_huggingface(
-            files=project_files,
-            project_name=project_name,
-            repo_name=repo_name,
-            hf_token=hf_token,
-            space_sdk=space_sdk,
-            is_space=is_space.lower() == "true",
-        )
-
-        yield json.dumps(result)
-
-    except Exception as exc:
-        logger.exception("Push to HuggingFace failed")
-        yield json.dumps({
-            "success": False,
-            "message": f"Push failed: {str(exc)}",
-            "url": "",
-        })
-
+# Start loading model in background
+start_background_load()
 
-app.launch(show_error=True)
+# Launch the server
+application = get_app()
+application.launch(show_error=True)

code/__init__.py

@@ -0,0 +1 @@
+"""Fullstack Code Builder — modular package."""

code/config/__init__.py

@@ -0,0 +1 @@
+"""Configuration and constants."""

code/config/constants.py

@@ -0,0 +1,132 @@
+"""Application-wide constants, regex patterns, language options, and system prompt."""
+
+from __future__ import annotations
+
+import re
+
+# ─── App Identity ────────────────────────────────────────────────────────
+
+APP_TITLE = "Fullstack Code Builder"
+MODEL_ID = "openbmb/MiniCPM5-1B"
+MODEL_URL = "https://huggingface.co/openbmb/MiniCPM5-1B"
+
+# ─── Runtime Defaults ───────────────────────────────────────────────────
+
+DEFAULT_TEMPERATURE = 0.6
+DEFAULT_MAX_TOKENS = 4096
+PY_TIMEOUT_S = 15
+GRADIO_TIMEOUT_S = 30
+PY_MEM_LIMIT_MB = 1024
+MAX_STDIO_CHARS = 16_000
+OUTPUT_PNG = "output.png"
+
+# ─── Regex Patterns ─────────────────────────────────────────────────────
+
+THINKING_BLOCK_RE = re.compile(
+    r"<\s*think\s*>.*?<\s*/\s*think\s*>", re.IGNORECASE | re.DOTALL
+)
+CODE_BLOCK_RE = re.compile(
+    r"```([a-zA-Z0-9_+.#-]*)\s*\n(.*?)```", re.DOTALL
+)
+FILE_BLOCK_RE = re.compile(
+    r"@@FILE:\s*(.+?)@@\s*\n(.*?)(?=@@FILE:|@@END@@)", re.DOTALL
+)
+
+# ─── Supported Languages & Frameworks ───────────────────────────────────
+
+LANGUAGE_OPTIONS: list[tuple[str, list[str]]] = [
+    ("Python", ["Gradio", "Flask", "Django", "FastAPI", "Streamlit", "Plain Python"]),
+    ("JavaScript", ["React", "Vue.js", "Next.js", "Express.js", "Node.js", "Vanilla JS"]),
+    ("TypeScript", ["React", "Next.js", "Express.js", "NestJS"]),
+    ("HTML/CSS/JS", ["Tailwind CSS", "Bootstrap", "Vanilla"]),
+    ("Java", ["Spring Boot", "Maven", "Gradle"]),
+    ("Go", ["Gin", "Fiber", "Echo", "Plain Go"]),
+    ("Rust", ["Actix", "Axum", "Rocket"]),
+    ("PHP", ["Laravel", "Symfony", "Plain PHP"]),
+    ("Ruby", ["Rails", "Sinatra"]),
+    ("C#", ["ASP.NET", "Blazor"]),
+    ("Swift", ["Vapor", "SwiftUI"]),
+    ("Kotlin", ["Ktor", "Spring Boot"]),
+]
+
+LANGUAGE_MAP: dict[str, list[str]] = {lang: frameworks for lang, frameworks in LANGUAGE_OPTIONS}
+
+# ─── System Prompt ───────────────────────────────────────────────────────
+
+SYSTEM_PROMPT = """You are a fullstack application code generator running locally. You help users build complete, runnable applications in any programming language and framework.
+
+When the user asks you to build an application:
+1. Generate complete, working code - not snippets or pseudocode
+2. Include all necessary files for the project to run
+3. Add proper error handling and comments
+4. For web apps, make the UI responsive and modern
+5. For Gradio apps, use gradio library and create a complete working app with gr.Interface or gr.Blocks
+
+FILE OUTPUT FORMAT - IMPORTANT:
+When generating multi-file projects, wrap each file in this format:
+@@FILE: path/to/file.ext@@
+(file content here)
+@@FILE: path/to/another/file.ext@@
+(another file content here)
+@@END@@
+
+For single-file code, use standard markdown fenced blocks:
+```python for Python
+```html for HTML/CSS/JS
+```javascript for JavaScript
+```typescript for TypeScript
+etc.
+
+When generating web apps with HTML/CSS/JS, return a single self-contained HTML document with all CSS and JavaScript inline. Make the page fully responsive: html/body at margin:0 and 100% width/height, use flexbox/grid layouts, and size any canvas to its container.
+
+When generating Gradio apps, create a complete app.py with:
+- import gradio as gr
+- Define the interface using gr.Interface() or gr.Blocks()
+- Call iface.launch(server_name="0.0.0.0", server_port=7860) at the end
+- Include all necessary processing logic inline
+
+For Python, prefer standard library or common packages. Do not use network calls, subprocesses, shell commands, or long-running loops in demo code (except Gradio apps which are server-based).
+
+If web search results are provided in the context, use them to inform your code generation. Incorporate relevant information from the search results into the generated code.
+"""
+
+# ─── Example Prompts ────────────────────────────────────────────────────
+
+EXAMPLE_PROMPTS: list[tuple[str, str, str, str]] = [
+    (
+        "🎨 Gradio Image Filter",
+        "Create a Gradio app that lets users upload an image and apply filters like grayscale, blur, sepia, and edge detection using PIL. Show the original and filtered images side by side.",
+        "Python",
+        "Gradio",
+    ),
+    (
+        "🤖 Gradio Chat App",
+        "Build a Gradio chatbot app with gr.Blocks that has a chat interface, a text input, and a send button. Include a simple echo bot that repeats the user's message with a fun twist.",
+        "Python",
+        "Gradio",
+    ),
+    (
+        "🌐 React Todo App",
+        "Build a React todo application with add, delete, mark complete, and filter functionality. Use modern hooks and a clean responsive UI.",
+        "JavaScript",
+        "React",
+    ),
+    (
+        "🐍 Flask API",
+        "Create a Flask REST API for a book library with CRUD operations, in-memory storage, and proper error handling.",
+        "Python",
+        "Flask",
+    ),
+    (
+        "🎨 Landing Page",
+        "Build a modern landing page for a SaaS product with a hero section, features grid, pricing cards, and a footer. Use Tailwind-style CSS.",
+        "HTML/CSS/JS",
+        "Vanilla",
+    ),
+    (
+        "📊 Dashboard",
+        "Create an interactive data dashboard with charts (bar, line, pie), a sidebar navigation, and summary cards. All in a single HTML file.",
+        "HTML/CSS/JS",
+        "Vanilla",
+    ),
+]

code/execution/__init__.py

@@ -0,0 +1 @@
+"""Code execution engines."""

code/execution/code_extractor.py

@@ -0,0 +1,137 @@
+"""Code extraction from model responses.
+
+Extracts fenced code blocks and multi-file @@FILE: blocks.
+Normalizes language names and detects Gradio code.
+"""
+
+from __future__ import annotations
+
+import html
+import re
+
+from code.config.constants import (
+    CODE_BLOCK_RE,
+    FILE_BLOCK_RE,
+    THINKING_BLOCK_RE,
+)
+
+
+def strip_thinking_blocks(text: str) -> str:
+    """Remove <think/> blocks from model output."""
+    return THINKING_BLOCK_RE.sub("", text).strip()
+
+
+def extract_code(response: str) -> tuple[str, str | None]:
+    """Return the first fenced code block and its language tag."""
+    visible_response = strip_thinking_blocks(response)
+    match = CODE_BLOCK_RE.search(visible_response)
+    if not match:
+        return "", None
+    return match.group(2).strip(), (match.group(1).strip().lower() or None)
+
+
+def extract_multi_file(response: str) -> dict[str, str]:
+    """Extract multi-file project from @@FILE: format.
+
+    Returns dict of {filepath: content}.
+    """
+    files: dict[str, str] = {}
+    visible = strip_thinking_blocks(response)
+
+    for match in FILE_BLOCK_RE.finditer(visible):
+        filepath = match.group(1).strip()
+        content = match.group(2).strip()
+        files[filepath] = content
+
+    # Fallback: if no @@FILE: blocks found, extract single code block
+    if not files:
+        code, lang = extract_code(response)
+        if code:
+            ext_map = {
+                "python": "main.py", "py": "main.py",
+                "javascript": "index.js", "js": "index.js",
+                "typescript": "index.ts", "ts": "index.ts",
+                "html": "index.html",
+                "css": "styles.css",
+                "java": "Main.java",
+                "go": "main.go",
+                "rust": "main.rs",
+                "php": "index.php",
+                "ruby": "main.rb",
+                "csharp": "Program.cs",
+                "swift": "main.swift",
+                "kotlin": "Main.kt",
+            }
+            filename = ext_map.get(lang or "", "code.txt")
+            files[filename] = code
+
+    return files
+
+
+def normalize_language(target_language: str | None, fence_lang: str | None) -> str:
+    """Normalize language name to a canonical form."""
+    lang = (fence_lang or target_language or "python").lower()
+    if lang in {"python", "py"}:
+        return "python"
+    if lang in {"html", "web", "css"}:
+        return "web"
+    if lang in {"javascript", "js"}:
+        return "javascript"
+    if lang in {"typescript", "ts"}:
+        return "typescript"
+    if lang == "java":
+        return "java"
+    if lang == "go":
+        return "go"
+    if lang == "rust":
+        return "rust"
+    if lang == "php":
+        return "php"
+    if lang == "ruby":
+        return "ruby"
+    if lang in {"csharp", "c#"}:
+        return "csharp"
+    if lang == "swift":
+        return "swift"
+    if lang == "kotlin":
+        return "kotlin"
+    return lang
+
+
+def is_gradio_code(code: str) -> bool:
+    """Detect if Python code is a Gradio app."""
+    return bool(
+        re.search(
+            r"import\s+gradio|from\s+gradio\s+import|gr\.\s*(Interface|Blocks|TabbedInterface|ChatInterface|App)",
+            code,
+        )
+    )
+
+
+# ─── Web Document / Iframe Builder ─────────────────────────────────────
+
+def _web_document(code: str, fence_lang: str | None) -> str:
+    """Wrap code in an HTML document if needed."""
+    lang = (fence_lang or "").lower()
+    if lang in {"javascript", "js"}:
+        return f"<!doctype html><html><body><script>\n{code}\n</script></body></html>"
+    if lang == "css":
+        return f"<!doctype html><html><head><style>\n{code}\n</style></head><body></body></html>"
+    if re.search(r"<!doctype|<html[\s>]", code, flags=re.IGNORECASE):
+        return code
+    return f"<!doctype html><html><head><meta charset='utf-8'></head><body>\n{code}\n</body></html>"
+
+
+def build_iframe(code: str, fence_lang: str | None = None) -> str:
+    """Build a sandboxed iframe HTML string for web preview."""
+    document = _web_document(code, fence_lang)
+    srcdoc = html.escape(document, quote=True)
+    return (
+        '<iframe class="web-frame" '
+        'sandbox="allow-scripts" '
+        'allow="fullscreen" '
+        "allowfullscreen "
+        f'srcdoc="{srcdoc}" '
+        'style="width:100%; min-height:680px; border:0; border-radius:14px; '
+        'background:white;"></iframe>'
+    )

code/execution/gradio_runner.py

@@ -0,0 +1,129 @@
+"""Gradio app runner — launches Gradio apps as subprocess servers.
+
+Manages the lifecycle of Gradio app processes: start, status check, and stop.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import re
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# ─── Registry for running Gradio subprocesses ───────────────────────────
+
+_running_gradio_procs: dict[str, subprocess.Popen] = {}
+
+
+def run_gradio_app(code: str, port: int = 7861) -> dict[str, Any]:
+    """Launch a Gradio app as a subprocess and return its URL.
+
+    The Gradio app is run on the specified port. We modify the code
+    to ensure it launches on the correct port and is accessible.
+    """
+    # Kill any previously running Gradio app
+    _stop_all_procs()
+
+    # Patch the code: ensure launch uses correct server_name and server_port
+    patched_code = code
+
+    # Replace .launch() with correct params
+    patched_code = re.sub(
+        r"(\w+)\.launch\([^)]*\)",
+        f'\\1.launch(server_name="0.0.0.0", server_port={port}, share=False)',
+        patched_code,
+    )
+
+    # If no .launch() found, add one
+    if ".launch(" not in patched_code:
+        patched_code += (
+            f'\n\nif __name__ == "__main__":\n'
+            f'    iface.launch(server_name="0.0.0.0", server_port={port}, share=False)\n'
+        )
+
+    with tempfile.TemporaryDirectory(prefix="gradio_app_") as tmp:
+        app_path = Path(tmp) / "gradio_app.py"
+        app_path.write_text(patched_code, encoding="utf-8")
+
+        env = {
+            **os.environ,
+            "PYTHONUNBUFFERED": "1",
+            "GRADIO_SERVER_NAME": "0.0.0.0",
+            "GRADIO_SERVER_PORT": str(port),
+        }
+
+        try:
+            proc = subprocess.Popen(
+                [sys.executable, str(app_path)],
+                cwd=tmp,
+                env=env,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+            )
+
+            proc_id = f"gradio_{port}"
+            _running_gradio_procs[proc_id] = proc
+
+            # Wait a bit for the server to start
+            import time as _time
+            _time.sleep(3)
+
+            # Check if process is still running
+            poll = proc.poll()
+            if poll is not None:
+                stdout = proc.stdout.read() if proc.stdout else ""
+                stderr = proc.stderr.read() if proc.stderr else ""
+                return {
+                    "success": False,
+                    "url": "",
+                    "message": f"Gradio app exited with code {poll}",
+                    "stdout": stdout[-2000:] if stdout else "",
+                    "stderr": stderr[-2000:] if stderr else "",
+                }
+
+            gradio_url = f"http://localhost:{port}"
+            return {
+                "success": True,
+                "url": gradio_url,
+                "message": f"Gradio app running at {gradio_url}",
+                "port": port,
+            }
+
+        except Exception as exc:
+            logger.exception("Failed to launch Gradio app")
+            return {
+                "success": False,
+                "url": "",
+                "message": f"Failed to launch: {exc}",
+            }
+
+
+def stop_gradio_app() -> dict[str, Any]:
+    """Stop any running Gradio app subprocess."""
+    stopped = _stop_all_procs()
+    return {"success": True, "message": f"Stopped {stopped} Gradio app(s)"}
+
+
+def _stop_all_procs() -> int:
+    """Stop all running Gradio processes. Returns count of stopped procs."""
+    stopped = 0
+    for pid, proc in list(_running_gradio_procs.items()):
+        try:
+            proc.terminate()
+            proc.wait(timeout=3)
+            stopped += 1
+        except Exception:
+            try:
+                proc.kill()
+                stopped += 1
+            except Exception:
+                pass
+    _running_gradio_procs.clear()
+    return stopped

code/execution/python_runner.py

@@ -0,0 +1,173 @@
+"""Sandboxed Python code execution.
+
+Runs user Python code in a subprocess with resource limits,
+captures stdout/stderr, and saves matplotlib figures.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import tempfile
+import textwrap
+from dataclasses import dataclass
+from pathlib import Path
+
+from code.config.constants import (
+    MAX_STDIO_CHARS,
+    OUTPUT_PNG,
+    PY_MEM_LIMIT_MB,
+    PY_TIMEOUT_S,
+)
+
+
+@dataclass
+class PythonExecutionResult:
+    """Result of a sandboxed Python execution."""
+    stdout: str
+    stderr: str
+    image_path: str | None
+    returncode: int | None
+    timed_out: bool = False
+
+
+def _apply_subprocess_limits() -> None:
+    """Set resource limits for the subprocess (Linux only)."""
+    import resource
+    mem_bytes = PY_MEM_LIMIT_MB * 1024 * 1024
+    resource.setrlimit(resource.RLIMIT_AS, (mem_bytes, mem_bytes))
+    resource.setrlimit(resource.RLIMIT_CPU, (PY_TIMEOUT_S, PY_TIMEOUT_S))
+
+
+def _python_runner_source() -> str:
+    """Return the source code of the runner script that wraps user code."""
+    return textwrap.dedent(
+        f"""
+        import os
+        import runpy
+        import sys
+        import traceback
+
+        os.environ.setdefault("MPLBACKEND", "Agg")
+        exit_code = 0
+        try:
+            runpy.run_path(os.path.join(os.getcwd(), "user_code.py"), run_name="__main__")
+        except SystemExit as exc:
+            code = exc.code
+            exit_code = code if isinstance(code, int) else 1
+        except Exception:
+            traceback.print_exc()
+            exit_code = 1
+        finally:
+            try:
+                import matplotlib
+                matplotlib.use("Agg", force=True)
+                import matplotlib.pyplot as plt
+                if plt.get_fignums():
+                    plt.savefig(os.environ["OUTPUT_PNG"], bbox_inches="tight")
+            except ModuleNotFoundError as exc:
+                if exc.name != "matplotlib":
+                    traceback.print_exc()
+            except Exception:
+                traceback.print_exc()
+
+        raise SystemExit(exit_code)
+        """
+    ).strip()
+
+
+def _truncate_output(text: str) -> str:
+    """Truncate output to MAX_STDIO_CHARS with a note."""
+    if len(text) <= MAX_STDIO_CHARS:
+        return text
+    remaining = len(text) - MAX_STDIO_CHARS
+    return text[:MAX_STDIO_CHARS] + f"\n\n... truncated {remaining} characters ..."
+
+
+def _decode_timeout_output(value: str | bytes | None) -> str:
+    """Safely decode subprocess output from timeout exceptions."""
+    if value is None:
+        return ""
+    if isinstance(value, bytes):
+        return value.decode("utf-8", errors="replace")
+    return value
+
+
+def run_python(code: str) -> PythonExecutionResult:
+    """Execute Python code in a sandboxed subprocess.
+
+    Returns a PythonExecutionResult with stdout, stderr, image path, and status.
+    """
+    with tempfile.TemporaryDirectory(prefix="fullstack_run_") as tmp:
+        workdir = Path(tmp)
+        runner_path = workdir / "runner.py"
+        user_path = workdir / "user_code.py"
+        image_path = workdir / OUTPUT_PNG
+
+        runner_path.write_text(_python_runner_source(), encoding="utf-8")
+        user_path.write_text(code, encoding="utf-8")
+
+        env = {
+            "PATH": "/usr/bin:/bin",
+            "HOME": str(workdir),
+            "TMPDIR": str(workdir),
+            "MPLBACKEND": "Agg",
+            "MPLCONFIGDIR": str(workdir / ".matplotlib"),
+            "OUTPUT_PNG": str(image_path),
+            "PYTHONIOENCODING": "utf-8",
+            "PYTHONNOUSERSITE": "1",
+            "PYTHONUNBUFFERED": "1",
+            "LANG": "C.UTF-8",
+            "OPENBLAS_NUM_THREADS": "1",
+            "OMP_NUM_THREADS": "1",
+            "MKL_NUM_THREADS": "1",
+            "NUMEXPR_NUM_THREADS": "1",
+        }
+
+        try:
+            completed = subprocess.run(
+                [sys.executable, "-I", str(runner_path)],
+                cwd=workdir,
+                env=env,
+                capture_output=True,
+                text=True,
+                encoding="utf-8",
+                errors="replace",
+                timeout=PY_TIMEOUT_S,
+                preexec_fn=_apply_subprocess_limits if sys.platform == "linux" else None,
+                check=False,
+            )
+            stdout = _truncate_output(completed.stdout)
+            stderr = _truncate_output(completed.stderr)
+
+            if completed.returncode and not stderr:
+                stderr = f"Process exited with status {completed.returncode}."
+
+            saved_image: str | None = None
+            if image_path.exists() and image_path.stat().st_size > 0:
+                saved = tempfile.NamedTemporaryFile(
+                    prefix="fullstack_plot_", suffix=".png", delete=False
+                )
+                saved.close()
+                Path(saved.name).write_bytes(image_path.read_bytes())
+                saved_image = saved.name
+
+            return PythonExecutionResult(
+                stdout=stdout,
+                stderr=stderr,
+                image_path=saved_image,
+                returncode=completed.returncode,
+            )
+        except subprocess.TimeoutExpired as exc:
+            stdout = _truncate_output(_decode_timeout_output(exc.stdout))
+            stderr = _truncate_output(_decode_timeout_output(exc.stderr))
+            timeout_note = f"Timed out after {PY_TIMEOUT_S} seconds; the process was killed."
+            stderr = f"{stderr}\n{timeout_note}".strip()
+            return PythonExecutionResult(
+                stdout=stdout,
+                stderr=stderr,
+                image_path=None,
+                returncode=None,
+                timed_out=True,
+            )

code/huggingface/__init__.py

@@ -0,0 +1 @@
+"""HuggingFace Hub push."""

code/huggingface/push.py

@@ -0,0 +1,147 @@
+"""HuggingFace Hub push and project ZIP packaging.
+
+Creates ZIP archives from extracted project files and pushes
+projects to HuggingFace Spaces or model repos.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import tempfile
+import zipfile
+from pathlib import Path
+from typing import Any
+
+from code.config.constants import MODEL_ID
+
+logger = logging.getLogger(__name__)
+
+
+def create_project_zip(files: dict[str, str], project_name: str) -> str:
+    """Create a ZIP file from extracted project files.
+
+    Returns the path to the created ZIP file.
+    """
+    zip_dir = tempfile.mkdtemp(prefix="fullstack_project_")
+    zip_path = os.path.join(zip_dir, f"{project_name}.zip")
+
+    with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
+        for filepath, content in files.items():
+            zf.writestr(f"{project_name}/{filepath}", content)
+
+    return zip_path
+
+
+def push_to_huggingface(
+    files: dict[str, str],
+    project_name: str,
+    repo_name: str,
+    hf_token: str,
+    space_sdk: str = "static",
+    is_space: bool = True,
+) -> dict[str, Any]:
+    """Push generated project to HuggingFace Hub.
+
+    Creates the repo if it doesn't exist, writes all files,
+    and adds README.md and requirements.txt as needed.
+    """
+    try:
+        from huggingface_hub import HfApi, create_repo
+
+        api = HfApi(token=hf_token)
+
+        if "/" in repo_name:
+            namespace, name = repo_name.split("/", 1)
+        else:
+            user_info = api.whoami()
+            namespace = user_info["name"]
+            name = repo_name
+            repo_name = f"{namespace}/{name}"
+
+        try:
+            if is_space:
+                create_repo(
+                    repo_id=repo_name,
+                    repo_type="space",
+                    space_sdk=space_sdk,
+                    token=hf_token,
+                    exist_ok=True,
+                )
+            else:
+                create_repo(
+                    repo_id=repo_name,
+                    repo_type="model",
+                    token=hf_token,
+                    exist_ok=True,
+                )
+        except Exception as e:
+            logger.warning("Repo creation warning: %s", e)
+
+        with tempfile.TemporaryDirectory(prefix="hf_push_") as tmp_dir:
+            for filepath, content in files.items():
+                full_path = os.path.join(tmp_dir, filepath)
+                os.makedirs(os.path.dirname(full_path), exist_ok=True)
+                Path(full_path).write_text(content, encoding="utf-8")
+
+            # Add README if not present
+            readme_path = os.path.join(tmp_dir, "README.md")
+            if not os.path.exists(readme_path):
+                readme_content = f"""---
+title: {name}
+emoji: 🚀
+colorFrom: blue
+colorTo: purple
+sdk: {space_sdk}
+app_file: app.py
+---
+
+# {name}
+
+Generated by Fullstack Code Builder using {MODEL_ID}.
+"""
+                Path(readme_path).write_text(readme_content, encoding="utf-8")
+
+            # Add requirements.txt for Python/Gradio projects
+            req_path = os.path.join(tmp_dir, "requirements.txt")
+            if not os.path.exists(req_path):
+                has_python = any(f.endswith(".py") for f in files.keys())
+                if has_python:
+                    reqs = ["gradio>=4.0.0"]
+                    all_code = "\n".join(files.values())
+                    if "matplotlib" in all_code:
+                        reqs.append("matplotlib>=3.8")
+                    if "PIL" in all_code or "Pillow" in all_code:
+                        reqs.append("Pillow>=10.0")
+                    if "numpy" in all_code:
+                        reqs.append("numpy>=1.24")
+                    if "pandas" in all_code:
+                        reqs.append("pandas>=2.0")
+                    Path(req_path).write_text("\n".join(reqs) + "\n", encoding="utf-8")
+
+            api.upload_folder(
+                folder_path=tmp_dir,
+                repo_id=repo_name,
+                repo_type="space" if is_space else "model",
+                token=hf_token,
+            )
+
+        repo_url = f"https://huggingface.co/{repo_name}"
+        if is_space:
+            repo_url = f"https://huggingface.co/spaces/{repo_name}"
+
+        return {
+            "success": True,
+            "url": repo_url,
+            "repo_name": repo_name,
+            "message": f"Successfully pushed to {repo_url}",
+        }
+
+    except Exception as exc:
+        logger.exception("Failed to push to HuggingFace")
+        return {
+            "success": False,
+            "url": "",
+            "repo_name": repo_name,
+            "message": f"Failed to push: {str(exc)}",
+        }

code/model/__init__.py

@@ -0,0 +1 @@
+"""Model loading and inference."""

code/model/inference.py

@@ -0,0 +1,97 @@
+"""Model inference — streaming and synchronous generation.
+
+Uses TextIteratorStreamer for real-time token streaming.
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+from collections.abc import Iterator
+from typing import Any
+
+from code.config.constants import DEFAULT_TEMPERATURE, DEFAULT_MAX_TOKENS
+from code.model.loader import get_model, get_tokenizer, get_model_status, is_model_loaded
+
+logger = logging.getLogger(__name__)
+
+
+def call_model(
+    messages: list[dict[str, Any]],
+    max_new_tokens: int = DEFAULT_MAX_TOKENS,
+) -> Iterator[str]:
+    """Stream model text using local MiniCPM5-1B.
+
+    Yields progressively longer strings (full text so far).
+    """
+
+    if not is_model_loaded():
+        status = get_model_status()
+        yield status["message"]
+        return
+
+    model = get_model()
+    tokenizer = get_tokenizer()
+
+    try:
+        from transformers import TextIteratorStreamer
+        import torch
+
+        # Build the prompt from messages
+        prompt_parts: list[str] = []
+        for msg in messages:
+            role = msg.get("role", "user")
+            content = msg.get("content", "")
+            if role == "system":
+                prompt_parts.append(f"System: {content}")
+            elif role == "user":
+                prompt_parts.append(f"User: {content}")
+            elif role == "assistant":
+                prompt_parts.append(f"Assistant: {content}")
+        prompt_parts.append("Assistant:")
+        full_prompt = "\n\n".join(prompt_parts)
+
+        # Tokenize
+        inputs = tokenizer(full_prompt, return_tensors="pt", truncation=True, max_length=4096)
+        if torch.cuda.is_available():
+            inputs = {k: v.to("cuda") for k, v in inputs.items()}
+
+        # Stream generation
+        streamer = TextIteratorStreamer(tokenizer, skip_prompt=True, skip_special_tokens=True)
+
+        generation_kwargs = {
+            **inputs,
+            "streamer": streamer,
+            "max_new_tokens": max_new_tokens,
+            "temperature": DEFAULT_TEMPERATURE,
+            "do_sample": True,
+            "top_p": 0.9,
+            "repetition_penalty": 1.1,
+            "pad_token_id": tokenizer.eos_token_id,
+        }
+
+        # Run generation in a separate thread
+        thread = threading.Thread(target=model.generate, kwargs=generation_kwargs)
+        thread.start()
+
+        output = ""
+        for new_text in streamer:
+            output += new_text
+            yield output
+
+        thread.join()
+
+    except Exception as exc:
+        logger.exception("Error during model inference")
+        yield f"_Error during generation: {exc}_"
+
+
+def call_model_sync(
+    messages: list[dict[str, Any]],
+    max_new_tokens: int = DEFAULT_MAX_TOKENS,
+) -> str:
+    """Non-streaming model call — returns complete response."""
+    result = ""
+    for chunk in call_model(messages, max_new_tokens):
+        result = chunk
+    return result

code/model/loader.py

@@ -0,0 +1,101 @@
+"""Model loading and status management.
+
+Handles loading MiniCPM5-1B locally using transformers.
+The model is loaded in a background thread on startup.
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+from typing import Any
+
+from code.config.constants import MODEL_ID
+
+logger = logging.getLogger(__name__)
+
+# ─── Module-level state ─────────────────────────────────────────────────
+
+_model = None
+_tokenizer = None
+_model_loaded = False
+_model_loading = False
+_load_error: str | None = None
+
+
+def load_model() -> None:
+    """Load MiniCPM5-1B model and tokenizer locally."""
+    global _model, _tokenizer, _model_loaded, _model_loading, _load_error
+
+    if _model_loaded or _model_loading:
+        return
+
+    _model_loading = True
+    _load_error = None
+
+    try:
+        from transformers import AutoModelForCausalLM, AutoTokenizer
+        import torch
+
+        logger.info("Loading MiniCPM5-1B model...")
+
+        dtype = torch.float16 if torch.cuda.is_available() else torch.float32
+        device_map = "auto" if torch.cuda.is_available() else None
+
+        _tokenizer = AutoTokenizer.from_pretrained(
+            MODEL_ID,
+            trust_remote_code=True,
+        )
+        _model = AutoModelForCausalLM.from_pretrained(
+            MODEL_ID,
+            torch_dtype=dtype,
+            device_map=device_map,
+            trust_remote_code=True,
+            low_cpu_mem_usage=True,
+        )
+
+        if device_map is None:
+            _model = _model.to("cpu")
+
+        _model.eval()
+        _model_loaded = True
+        logger.info("MiniCPM5-1B model loaded successfully.")
+
+    except Exception as exc:
+        _load_error = str(exc)
+        logger.exception("Failed to load model: %s", exc)
+    finally:
+        _model_loading = False
+
+
+def start_background_load() -> threading.Thread:
+    """Start loading the model in a background daemon thread."""
+    thread = threading.Thread(target=load_model, daemon=True)
+    thread.start()
+    return thread
+
+
+def get_model_status() -> dict[str, Any]:
+    """Return current model loading status."""
+    if _model_loaded:
+        return {"status": "ready", "message": "Model loaded and ready"}
+    if _model_loading:
+        return {"status": "loading", "message": "Model is loading... (this may take a few minutes on first run)"}
+    if _load_error:
+        return {"status": "error", "message": f"Model load error: {_load_error}"}
+    return {"status": "unknown", "message": "Model not initialized"}
+
+
+def get_model():
+    """Return the loaded model instance (or None)."""
+    return _model
+
+
+def get_tokenizer():
+    """Return the loaded tokenizer instance (or None)."""
+    return _tokenizer
+
+
+def is_model_loaded() -> bool:
+    """Return True if the model has been loaded successfully."""
+    return _model_loaded

code/server/__init__.py

@@ -0,0 +1 @@
+"""FastAPI / Gradio server routes."""

code/server/chat_helpers.py

@@ -0,0 +1,110 @@
+"""Chat helper functions — history conversion, prompt building, iteration context."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from code.config.constants import SYSTEM_PROMPT
+from code.execution.code_extractor import strip_thinking_blocks
+
+
+def chat_history_to_messages(history: list[dict[str, str]]) -> list[dict[str, Any]]:
+    """Convert chat history list to messages format for the model.
+
+    Prepends the system prompt and strips thinking blocks from
+    assistant messages.
+    """
+    messages: list[dict[str, Any]] = [{"role": "system", "content": SYSTEM_PROMPT}]
+    for item in history:
+        role = item.get("role")
+        content = str(item.get("content") or "").strip()
+        if role not in {"user", "assistant"} or not content:
+            continue
+        if role == "assistant":
+            content = strip_thinking_blocks(content)
+        messages.append({"role": role, "content": content})
+    return messages
+
+
+def clip_context(text: str, limit: int = 4_000) -> str:
+    """Truncate text to a character limit with a note."""
+    if len(text) <= limit:
+        return text
+    return text[:limit] + f"\n... truncated {len(text) - limit} characters ..."
+
+
+def iteration_context(execution_context: dict[str, Any] | None) -> str:
+    """Build a context string from previous execution results.
+
+    This allows the model to reference prior code, stdout, and stderr
+    when the user asks to iterate or debug.
+    """
+    if not execution_context or not execution_context.get("code"):
+        return ""
+
+    code = clip_context(str(execution_context.get("code") or ""), 6_000)
+    target = str(execution_context.get("target") or "code")
+    fence_lang = str(execution_context.get("fence_lang") or target)
+    status = str(execution_context.get("status") or "")
+    stdout = clip_context(str(execution_context.get("stdout") or ""), 2_000)
+    stderr = clip_context(str(execution_context.get("stderr") or ""), 2_000)
+
+    parts = [
+        "Previous generated code and run result are available for iteration.",
+        f"Previous target: {target}",
+        f"Previous status: {status}",
+        f"Previous code:\n```{fence_lang}\n{code}\n```",
+    ]
+    if stdout:
+        parts.append(f"Previous stdout:\n{stdout}")
+    if stderr:
+        parts.append(f"Previous stderr / traceback:\n{stderr}")
+    parts.append(
+        "If the user asks to revise, debug, extend, or explain the prior code, use this context."
+    )
+    return "\n\n".join(parts)
+
+
+def targeted_prompt(
+    prompt: str,
+    target_language: str,
+    target_framework: str = "",
+    execution_context: dict[str, Any] | None = None,
+    search_context: str = "",
+) -> str:
+    """Build the full user prompt with language, framework, search, and iteration context."""
+    iter_ctx = iteration_context(execution_context)
+    context_block = f"\n\n{iter_ctx}" if iter_ctx else ""
+
+    search_block = ""
+    if search_context:
+        search_block = (
+            f"\n\n{search_context}\n\n"
+            "Use the above search results to inform your code generation if relevant."
+        )
+
+    framework_hint = f" using {target_framework}" if target_framework else ""
+
+    gradio_hint = ""
+    if target_framework == "Gradio":
+        gradio_hint = (
+            "\n\nIMPORTANT: This is a Gradio app. Create a complete Python script that:\n"
+            "- Imports gradio as gr\n"
+            "- Defines the UI using gr.Interface() or gr.Blocks()\n"
+            "- Includes all processing logic inline\n"
+            "- Calls .launch(server_name='0.0.0.0', server_port=7860) at the end\n"
+            "- Uses only standard library + gradio + common packages (PIL, matplotlib, numpy)\n"
+            "- Make the UI clean, modern, and functional"
+        )
+
+    return (
+        f"Target: {target_language}{framework_hint}. Generate a complete, runnable application. "
+        "If the user asks for a web app, include all HTML/CSS/JS. "
+        "If they ask for a backend, include the server code and any API definitions. "
+        "For single-file apps, use a single code block. For multi-file projects, use the @@FILE: format. "
+        "Make the code complete, working, and well-structured."
+        f"{gradio_hint}"
+        f"{search_block}"
+        f"{context_block}\n\n"
+        f"User request:\n{prompt}"
+    )

code/server/routes.py

@@ -0,0 +1,428 @@
+"""FastAPI / Gradio Server routes.
+
+Defines all HTTP and API endpoints:
+- GET /            → serves the index.html frontend
+- GET /api/model-status → model loading status
+- GET /images/{f}  → serve generated plot images
+- GET /download/{f} → serve project ZIP downloads
+- API web_search   → Google search scraping
+- API chat         → streaming chat with code execution
+- API push_hf      → push to HuggingFace Hub
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import tempfile
+from pathlib import Path
+from typing import Any
+
+from fastapi.responses import HTMLResponse, FileResponse
+from gradio import Server
+
+from code.config.constants import (
+    APP_TITLE,
+    EXAMPLE_PROMPTS,
+    LANGUAGE_OPTIONS,
+    MODEL_ID,
+    MODEL_URL,
+    PY_TIMEOUT_S,
+)
+from code.execution.code_extractor import (
+    build_iframe,
+    extract_code,
+    extract_multi_file,
+    is_gradio_code,
+    normalize_language,
+)
+from code.execution.gradio_runner import run_gradio_app, stop_gradio_app
+from code.execution.python_runner import run_python
+from code.huggingface.push import create_project_zip, push_to_huggingface
+from code.model.loader import get_model_status, is_model_loaded
+from code.model.inference import call_model
+from code.server.chat_helpers import chat_history_to_messages, targeted_prompt
+from code.websearch.google_scraper import web_search_google, format_search_results
+
+logger = logging.getLogger(__name__)
+
+# ─── Served Files Registry ──────────────────────────────────────────────
+
+_served_files: dict[str, str] = {}
+
+# ─── Server Instance ────────────────────────────────────────────────────
+
+app = Server()
+
+
+# ─── HTTP Routes ────────────────────────────────────────────────────────
+
+
+@app.get("/", response_class=HTMLResponse)
+async def homepage():
+    """Serve the index.html frontend with runtime config injected."""
+    html_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..", "index.html")
+    with open(html_path, "r", encoding="utf-8") as f:
+        content = f.read()
+
+    config = json.dumps({
+        "app_title": APP_TITLE,
+        "model_id": MODEL_ID,
+        "model_url": MODEL_URL,
+        "languages": LANGUAGE_OPTIONS,
+        "examples": [
+            {"label": label, "prompt": prompt, "language": lang, "framework": fw}
+            for label, prompt, lang, fw in EXAMPLE_PROMPTS
+        ],
+    })
+    content = content.replace("__RUNTIME_CONFIG__", config)
+    return content
+
+
+@app.get("/api/model-status")
+async def model_status_endpoint():
+    """Return the current model loading status."""
+    return get_model_status()
+
+
+@app.get("/images/{filename}")
+async def serve_image(filename: str):
+    """Serve a generated plot image by filename."""
+    path = _served_files.get(f"img:{filename}")
+    if path and os.path.exists(path):
+        return FileResponse(path, media_type="image/png")
+    return HTMLResponse("Not found", status_code=404)
+
+
+@app.get("/download/{filename}")
+async def serve_download(filename: str):
+    """Serve a project ZIP download by filename."""
+    path = _served_files.get(f"dl:{filename}")
+    if path and os.path.exists(path):
+        return FileResponse(path, filename=filename, media_type="application/octet-stream")
+    return HTMLResponse("Not found", status_code=404)
+
+
+# ─── Gradio API Endpoints ──────────────────────────────────────────────
+
+
+@app.api(name="web_search", concurrency_limit=4)
+def handle_web_search(query: str) -> str:
+    """Search the web using Google scraping. No API key needed."""
+    query = (query or "").strip()
+    if not query:
+        yield json.dumps({"success": False, "results": [], "message": "Empty search query"})
+        return
+
+    try:
+        results = web_search_google(query, num_results=8)
+        formatted = format_search_results(results)
+
+        yield json.dumps({
+            "success": True,
+            "results": results,
+            "formatted": formatted,
+            "message": f"Found {len(results)} results",
+        })
+    except Exception as exc:
+        logger.exception("Web search failed")
+        yield json.dumps({
+            "success": False,
+            "results": [],
+            "message": f"Search failed: {str(exc)}",
+        })
+
+
+@app.api(name="chat", concurrency_limit=2)
+def handle_chat(
+    prompt: str,
+    target_language: str,
+    target_framework: str,
+    history_json: str,
+    exec_context_json: str,
+    search_enabled: str = "false",
+) -> str:
+    """Stream chat responses with code execution. Yields JSON strings."""
+    history = json.loads(history_json) if history_json else []
+    execution_context = json.loads(exec_context_json) if exec_context_json else {}
+
+    prompt = (prompt or "").strip()
+    if not prompt:
+        yield json.dumps({
+            "type": "error",
+            "status_text": "Enter a prompt to get started.",
+            "status_state": "info",
+            "history": history,
+            "execution": execution_context,
+        })
+        return
+
+    # Check model status
+    model_status = get_model_status()
+    if model_status["status"] == "loading":
+        yield json.dumps({
+            "type": "error",
+            "status_text": model_status["message"],
+            "status_state": "working",
+            "history": history,
+            "execution": execution_context,
+        })
+        return
+    if model_status["status"] != "ready":
+        yield json.dumps({
+            "type": "error",
+            "status_text": model_status["message"],
+            "status_state": "error",
+            "history": history,
+            "execution": execution_context,
+        })
+        return
+
+    # Add user message and placeholder assistant message
+    history = list(history) + [
+        {"role": "user", "content": prompt},
+        {"role": "assistant", "content": ""},
+    ]
+    yield json.dumps({
+        "type": "status",
+        "status_text": "Thinking...",
+        "status_state": "working",
+        "history": history,
+        "execution": execution_context,
+    })
+
+    # Web search if enabled
+    search_context = ""
+    if search_enabled.lower() == "true":
+        yield json.dumps({
+            "type": "status",
+            "status_text": "Searching the web...",
+            "status_state": "working",
+            "history": history,
+            "execution": execution_context,
+        })
+        search_results = web_search_google(prompt, num_results=6)
+        if search_results:
+            search_context = format_search_results(search_results)
+            yield json.dumps({
+                "type": "search_results",
+                "status_text": f"Found {len(search_results)} results, generating code...",
+                "status_state": "working",
+                "history": history,
+                "execution": execution_context,
+                "search_results": search_results,
+            })
+
+    # Build messages for model
+    model_history = list(history[:-1])
+    model_history[-1] = {
+        "role": "user",
+        "content": targeted_prompt(
+            prompt, target_language, target_framework, execution_context, search_context
+        ),
+    }
+    messages = chat_history_to_messages(model_history)
+
+    final_response = ""
+    for partial in call_model(messages):
+        final_response = partial
+        history[-1]["content"] = partial
+        yield json.dumps({
+            "type": "streaming",
+            "status_text": "Generating...",
+            "status_state": "working",
+            "history": history,
+            "execution": execution_context,
+        })
+
+    if not final_response:
+        history[-1]["content"] = "The model did not return a response."
+        yield json.dumps({
+            "type": "error",
+            "status_text": "No model response.",
+            "status_state": "error",
+            "history": history,
+            "execution": execution_context,
+        })
+        return
+
+    # Extract code from response
+    code, fence_lang = extract_code(final_response)
+    target = normalize_language(target_language, fence_lang)
+
+    # Also try multi-file extraction
+    multi_files = extract_multi_file(final_response)
+
+    if not code and not multi_files:
+        yield json.dumps({
+            "type": "complete",
+            "status_text": "Answered without running code.",
+            "status_state": "info",
+            "history": history,
+            "execution": execution_context,
+        })
+        return
+
+    yield json.dumps({
+        "type": "status",
+        "status_text": "Running...",
+        "status_state": "working",
+        "history": history,
+        "execution": execution_context,
+    })
+
+    # Execute code
+    stdout, stderr, image_path, status_text, status_state = "", "", None, "Preview ready", "success"
+    is_gradio = False
+    gradio_url = None
+
+    if target == "python" and code:
+        if is_gradio_code(code) or target_framework == "Gradio":
+            is_gradio = True
+            gradio_result = run_gradio_app(code)
+            if gradio_result["success"]:
+                gradio_url = gradio_result["url"]
+                status_text = f"Gradio app running at {gradio_url}"
+                status_state = "success"
+                stderr = f"Gradio app launched successfully at {gradio_url}"
+            else:
+                status_text = "Gradio launch failed"
+                status_state = "error"
+                stderr = gradio_result.get("stderr", gradio_result.get("message", "Launch failed"))
+        else:
+            result = run_python(code)
+            if result.timed_out:
+                stdout, stderr, image_path = result.stdout, result.stderr, result.image_path
+                status_text = f"Timed out after {PY_TIMEOUT_S}s"
+                status_state = "error"
+            elif result.returncode:
+                stdout, stderr, image_path = result.stdout, result.stderr, result.image_path
+                status_text = "Finished with errors"
+                status_state = "error"
+            else:
+                stdout, stderr, image_path = result.stdout, result.stderr, result.image_path
+                status_text = "Ran successfully"
+                status_state = "success"
+
+    # Register image for serving
+    image_url = None
+    if image_path:
+        filename = os.path.basename(image_path)
+        _served_files[f"img:{filename}"] = image_path
+        image_url = f"/images/{filename}"
+
+    # Register code for download
+    download_url = None
+    project_files = multi_files if multi_files else {}
+
+    if project_files:
+        project_name = "generated-project"
+        zip_path = create_project_zip(project_files, project_name)
+        zip_filename = f"{project_name}.zip"
+        _served_files[f"dl:{zip_filename}"] = zip_path
+        download_url = f"/download/{zip_filename}"
+    elif code:
+        ext = "py" if target == "python" else "html"
+        dl_filename = f"generated.{ext}"
+        dl_dir = tempfile.mkdtemp(prefix="fullstack_dl_")
+        dl_path = os.path.join(dl_dir, dl_filename)
+        Path(dl_path).write_text(code, encoding="utf-8")
+        _served_files[f"dl:{dl_filename}"] = dl_path
+        download_url = f"/download/{dl_filename}"
+
+    # Determine if this is web previewable
+    is_web = target in {"web", "javascript", "typescript", "html"} or (fence_lang or "") in {"html", "web"}
+    web_code = code if is_web else None
+
+    execution_context = {
+        "code": code,
+        "target": target,
+        "fence_lang": fence_lang or target,
+        "stdout": stdout,
+        "stderr": stderr,
+        "image_url": image_url,
+        "image_path": image_path,
+        "status": status_text,
+        "language": fence_lang or target,
+        "suggested_tab": "preview" if (image_path or is_web or is_gradio) else "console",
+        "download_url": download_url,
+        "project_files": project_files,
+        "is_web": is_web,
+        "web_code": web_code,
+        "is_gradio": is_gradio,
+        "gradio_url": gradio_url,
+    }
+
+    yield json.dumps({
+        "type": "complete",
+        "status_text": status_text,
+        "status_state": status_state,
+        "history": history,
+        "execution": execution_context,
+    })
+
+
+@app.api(name="push_hf", concurrency_limit=1)
+def handle_push_hf(
+    exec_context_json: str,
+    repo_name: str,
+    hf_token: str,
+    space_sdk: str = "static",
+    is_space: str = "true",
+) -> str:
+    """Push generated project to HuggingFace Hub."""
+    try:
+        execution_context = json.loads(exec_context_json) if exec_context_json else {}
+        project_files = execution_context.get("project_files", {})
+
+        if not project_files:
+            code = execution_context.get("code", "")
+            if not code:
+                yield json.dumps({
+                    "success": False,
+                    "message": "No code to push. Generate some code first.",
+                    "url": "",
+                })
+                return
+
+            lang = execution_context.get("language", "python")
+            is_gradio = execution_context.get("is_gradio", False)
+            ext_map = {
+                "python": "app.py", "py": "app.py",
+                "javascript": "index.js", "js": "index.js",
+                "html": "index.html", "web": "index.html",
+                "typescript": "index.ts", "ts": "index.ts",
+            }
+            filename = ext_map.get(lang, "app.py")
+            project_files = {filename: code}
+
+            # Auto-detect SDK for Gradio apps
+            if is_gradio or is_gradio_code(code):
+                space_sdk = "gradio"
+
+        project_name = repo_name.split("/")[-1] if "/" in repo_name else repo_name
+
+        result = push_to_huggingface(
+            files=project_files,
+            project_name=project_name,
+            repo_name=repo_name,
+            hf_token=hf_token,
+            space_sdk=space_sdk,
+            is_space=is_space.lower() == "true",
+        )
+
+        yield json.dumps(result)
+
+    except Exception as exc:
+        logger.exception("Push to HuggingFace failed")
+        yield json.dumps({
+            "success": False,
+            "message": f"Push failed: {str(exc)}",
+            "url": "",
+        })
+
+
+def get_app() -> Server:
+    """Return the configured Gradio Server app instance."""
+    return app

code/websearch/__init__.py

@@ -0,0 +1 @@
+"""Web search (Google scraping, no API)."""

code/websearch/google_scraper.py

@@ -0,0 +1,132 @@
+"""Web search via Google scraping — no API key needed.
+
+Uses requests with a browser-like User-Agent and BeautifulSoup
+to parse Google search result pages.
+"""
+
+from __future__ import annotations
+
+import logging
+import urllib.parse
+
+logger = logging.getLogger(__name__)
+
+
+def web_search_google(query: str, num_results: int = 8) -> list[dict[str, str]]:
+    """Search Google by scraping the results page. No API key needed.
+
+    Returns a list of dicts with keys: title, url, snippet.
+    Uses requests with a browser-like User-Agent to avoid captchas.
+    """
+    try:
+        import requests
+        from bs4 import BeautifulSoup
+
+        encoded_query = urllib.parse.quote_plus(query)
+        url = f"https://www.google.com/search?q={encoded_query}&num={num_results + 2}&hl=en"
+
+        headers = {
+            "User-Agent": (
+                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+                "AppleWebKit/537.36 (KHTML, like Gecko) "
+                "Chrome/120.0.0.0 Safari/537.36"
+            ),
+            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
+            "Accept-Language": "en-US,en;q=0.5",
+            "Accept-Encoding": "gzip, deflate",
+            "DNT": "1",
+            "Connection": "keep-alive",
+            "Upgrade-Insecure-Requests": "1",
+        }
+
+        resp = requests.get(url, headers=headers, timeout=10, allow_redirects=True)
+        resp.raise_for_status()
+
+        soup = BeautifulSoup(resp.text, "html.parser")
+        results: list[dict[str, str]] = []
+
+        # Parse Google search results
+        for g_div in soup.select("div.g, div[data-sokoban-container], div.yuRUbf"):
+            title_el = g_div.select_one("h3")
+            link_el = g_div.select_one("a[href]")
+            snippet_el = g_div.select_one("div.VwiC3b, span.aCOpRe, div[data-sncf]")
+
+            if not title_el or not link_el:
+                continue
+
+            href = link_el.get("href", "")
+            # Google sometimes prefixes URLs; extract the real URL
+            if href.startswith("/url?q="):
+                real_url = urllib.parse.parse_qs(
+                    urllib.parse.urlparse(href).query
+                ).get("q", [href])[0]
+            elif href.startswith("http"):
+                real_url = href
+            else:
+                continue
+
+            # Skip Google-internal URLs
+            if "google.com" in real_url or "googleusercontent.com" in real_url:
+                continue
+
+            title = title_el.get_text(strip=True)
+            snippet = snippet_el.get_text(strip=True) if snippet_el else ""
+
+            if title and real_url:
+                results.append({
+                    "title": title,
+                    "url": real_url,
+                    "snippet": snippet,
+                })
+
+            if len(results) >= num_results:
+                break
+
+        # Fallback: try parsing from <a> tags with data-ved attribute
+        if not results:
+            for a_tag in soup.select("a[data-ved]"):
+                href = a_tag.get("href", "")
+                if not href.startswith("http"):
+                    continue
+                if "google.com" in href:
+                    continue
+
+                title_el = a_tag.select_one("h3, span")
+                title = title_el.get_text(strip=True) if title_el else a_tag.get_text(strip=True)[:100]
+                snippet = ""
+
+                if title and href:
+                    results.append({
+                        "title": title,
+                        "url": href,
+                        "snippet": snippet,
+                    })
+
+                if len(results) >= num_results:
+                    break
+
+        logger.info("Web search for '%s' returned %d results", query, len(results))
+        return results
+
+    except ImportError:
+        logger.warning("requests or beautifulsoup4 not installed for web search")
+        return []
+    except Exception as exc:
+        logger.exception("Web search failed: %s", exc)
+        return []
+
+
+def format_search_results(results: list[dict[str, str]]) -> str:
+    """Format search results into a text block for model context."""
+    if not results:
+        return "No search results found."
+
+    parts = ["Here are the web search results for reference:\n"]
+    for i, r in enumerate(results, 1):
+        parts.append(f"{i}. {r['title']}")
+        parts.append(f"   URL: {r['url']}")
+        if r["snippet"]:
+            parts.append(f"   {r['snippet']}")
+        parts.append("")
+
+    return "\n".join(parts)