AI Developer Agent v1.0 backend

Dockerfile

@@ -0,0 +1,39 @@
+FROM python:3.11-slim
+
+ENV PYTHONUNBUFFERED=1 \
+    PIP_NO_CACHE_DIR=1 \
+    PIP_DISABLE_PIP_VERSION_CHECK=1 \
+    DEBIAN_FRONTEND=noninteractive \
+    HOME=/home/user \
+    PATH=/home/user/.local/bin:$PATH \
+    TASKS_DB_PATH=/data/tasks.db \
+    PYTHONPATH=/home/user/app
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+        build-essential git git-lfs curl ca-certificates \
+        libnss3 libatk1.0-0 libatk-bridge2.0-0 libcups2 libdrm2 libxkbcommon0 \
+        libxcomposite1 libxdamage1 libxrandr2 libgbm1 libpango-1.0-0 libcairo2 \
+        libasound2 libatspi2.0-0 \
+    && rm -rf /var/lib/apt/lists/* \
+    && git lfs install --system
+
+RUN useradd -m -u 1000 user
+WORKDIR /home/user/app
+
+COPY --chown=user:user backend/requirements.txt /home/user/app/backend/requirements.txt
+RUN pip install --no-cache-dir -r /home/user/app/backend/requirements.txt
+
+# Playwright Chromium (best-effort; skip if it fails so build still completes)
+RUN python -m playwright install --with-deps chromium || \
+    python -m playwright install chromium || \
+    echo "playwright install failed; continuing"
+
+RUN mkdir -p /data && chown -R user:user /data
+
+COPY --chown=user:user . /home/user/app
+RUN chown -R user:user /home/user/app
+
+USER user
+EXPOSE 7860
+
+CMD ["uvicorn", "backend.app:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,10 +1,21 @@
 ---
-title: Ai Developer Agent
-emoji: 📊
+title: AI Developer Agent
+emoji: 🤖
 colorFrom: indigo
-colorTo: gray
+colorTo: purple
 sdk: docker
+app_port: 7860
 pinned: false
+license: mit
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# AI Developer Agent (Backend)
+
+Persistent autonomous AI Developer Agent. FastAPI service that plans, executes,
+repairs and deploys software tasks end-to-end.
+
+See `/health`, `/api/runtime`, `/api/tasks`, `/api/tasks/{id}/stream`, `/api/chat`.
+
+Configure via Space Secrets:
+- `GEMINI_API_KEYS`, `SAMBANOVA_API_KEYS`, `OPENAI_API_KEYS`, `GITHUB_LLM_API_KEYS`, `OPENROUTER_API_KEYS`
+- `E2B_API_KEY`, `HF_TOKEN`, `VERCEL_TOKEN`, `GITHUB_TOKEN`

backend/__init__.py

@@ -0,0 +1,2 @@
+"""AI Developer Agent - backend package."""
+__version__ = "1.0.0"

backend/agent.py

@@ -0,0 +1,160 @@
+"""
+Agent orchestrator - ties planner + executor + retry + repair + browser.
+
+`run_task(task_id, title, description)` is a generator yielding event dicts
+suitable for SSE streaming. Persists everything to tasks.db.
+"""
+from __future__ import annotations
+
+import json
+import time
+import logging
+import traceback
+from typing import Any, Dict, Generator, List, Optional
+
+from . import tasks
+from .planner import plan_task, repair_plan
+from .executor import get_executor
+from .classifier import classify
+from .browser import run_browser_action
+from .llm_router import get_router
+
+logger = logging.getLogger("agent")
+
+
+def _event(task_id: str, kind: str, message: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+    payload = {
+        "task_id": task_id,
+        "kind": kind,
+        "message": message,
+        "ts": time.time(),
+        "data": data or {},
+    }
+    tasks.log_event(task_id, kind, json.dumps({"message": message, "data": data or {}})[:7000])
+    return payload
+
+
+def run_task(task_id: str, title: str, description: str) -> Generator[Dict[str, Any], None, None]:
+    """Generator yielding event dicts. Persists to SQLite."""
+    try:
+        tasks.update_state(task_id, "planning")
+        yield _event(task_id, "state", "planning")
+        yield _event(task_id, "thought", f"Planning task: {title}")
+
+        plan = plan_task(title, description)
+        yield _event(task_id, "plan", "Plan generated", {"steps": plan})
+
+        if not plan:
+            yield _event(task_id, "warn", "Empty plan – using fallback note")
+            plan = [{"type": "note", "msg": "No actions planned"}]
+
+        tasks.update_state(task_id, "executing")
+        yield _event(task_id, "state", "executing")
+
+        executor = get_executor()
+        for idx, step in enumerate(plan):
+            tasks.save_checkpoint(task_id, idx, {"plan": plan, "current": step})
+            stype = step.get("type", "note")
+            yield _event(task_id, "step.start", f"[{idx + 1}/{len(plan)}] {stype}", {"step": step})
+
+            try:
+                result = _execute_step(task_id, step)
+            except Exception as e:
+                tb = traceback.format_exc()
+                result = {"ok": False, "stderr": str(e), "traceback": tb}
+
+            yield _event(task_id, "step.result", "ok" if result.get("ok") else "fail", {"step_index": idx, "result": result})
+
+            # If failed, attempt one repair cycle
+            if not result.get("ok"):
+                tasks.update_state(task_id, "repairing")
+                yield _event(task_id, "state", "repairing")
+                err_text = (result.get("stderr") or "") + "\n" + (result.get("traceback") or "")
+                err_class = classify(err_text)
+                if err_class:
+                    yield _event(task_id, "diagnose", f"Detected: {err_class.category}", {"detail": err_class.detail, "fix": err_class.suggested_fix})
+                    repair_actions = repair_plan(err_class.category, err_class.detail)
+                    for ridx, ra in enumerate(repair_actions):
+                        yield _event(task_id, "repair.start", f"repair[{ridx + 1}]", {"action": ra})
+                        rresult = _execute_step(task_id, ra)
+                        yield _event(task_id, "repair.result", "ok" if rresult.get("ok") else "fail", {"result": rresult})
+                    # retry original step once
+                    tasks.update_state(task_id, "retrying")
+                    yield _event(task_id, "state", "retrying")
+                    tasks.record_retry(task_id, 1, err_text[:1000])
+                    try:
+                        retry_result = _execute_step(task_id, step)
+                    except Exception as e:
+                        retry_result = {"ok": False, "stderr": str(e)}
+                    yield _event(task_id, "retry.result", "ok" if retry_result.get("ok") else "fail", {"step_index": idx, "result": retry_result})
+                else:
+                    yield _event(task_id, "warn", "No automatic repair – continuing")
+                tasks.update_state(task_id, "executing")
+                yield _event(task_id, "state", "executing")
+
+        tasks.update_state(task_id, "completed")
+        yield _event(task_id, "state", "completed")
+        yield _event(task_id, "done", f"Task {task_id} completed")
+    except Exception as e:
+        logger.exception("run_task fatal")
+        tasks.update_state(task_id, "failed")
+        yield _event(task_id, "error", f"Fatal: {e}", {"traceback": traceback.format_exc()})
+
+
+def _execute_step(task_id: str, step: Dict[str, Any]) -> Dict[str, Any]:
+    executor = get_executor()
+    stype = step.get("type", "note")
+
+    if stype == "shell":
+        cmd = step.get("cmd", "")
+        if not cmd:
+            return {"ok": True, "stdout": "(empty cmd)"}
+        r = executor.shell(cmd, timeout=float(step.get("timeout", 120)))
+        return {"ok": r.ok, "stdout": r.stdout[-3000:], "stderr": r.stderr[-3000:], "exit_code": r.exit_code, "duration_ms": r.duration_ms}
+
+    if stype == "python":
+        code = step.get("code", "")
+        r = executor.python(code, timeout=float(step.get("timeout", 120)))
+        return {"ok": r.ok, "stdout": r.stdout[-3000:], "stderr": r.stderr[-3000:], "exit_code": r.exit_code, "duration_ms": r.duration_ms}
+
+    if stype == "browser":
+        br = run_browser_action(step)
+        return {
+            "ok": br.ok,
+            "stdout": (br.text or "")[:3000],
+            "stderr": br.error or "",
+            "exit_code": 0 if br.ok else 1,
+            "screenshot_b64": br.screenshot_b64[:500] if br.screenshot_b64 else "",
+            "url": br.url,
+        }
+
+    if stype == "git":
+        op = step.get("op", "status")
+        args = step.get("args", "")
+        cmd = f"git {op} {args}".strip()
+        r = executor.shell(cmd, timeout=120)
+        return {"ok": r.ok, "stdout": r.stdout, "stderr": r.stderr, "exit_code": r.exit_code}
+
+    if stype == "deploy":
+        # Real deploy is invoked via dedicated /deploy endpoints; here we just log.
+        target = step.get("target", "unknown")
+        msg = f"Deploy step requested: {target}. Use /deploy endpoints for real deployment."
+        return {"ok": True, "stdout": msg}
+
+    if stype == "note":
+        return {"ok": True, "stdout": step.get("msg", "")}
+
+    if stype == "sleep":
+        time.sleep(float(step.get("seconds", 1)))
+        return {"ok": True, "stdout": f"slept {step.get('seconds', 1)}s"}
+
+    if stype == "llm":
+        router = get_router()
+        prompt = step.get("prompt", "")
+        try:
+            out = router.chat([{"role": "user", "content": prompt}], temperature=0.2, max_tokens=800)
+            return {"ok": True, "stdout": out[:3000]}
+        except Exception as e:
+            return {"ok": False, "stderr": str(e)}
+
+    return {"ok": False, "stderr": f"unknown step type: {stype}"}

backend/app.py

@@ -0,0 +1,311 @@
+"""
+FastAPI backend - AI Developer Agent
+====================================
+Endpoints:
+    GET  /                       service info
+    GET  /health                 health check
+    GET  /api/runtime            runtime + provider telemetry
+    POST /api/tasks              create + run a task (sync queued)
+    GET  /api/tasks              list tasks
+    GET  /api/tasks/{id}         get task
+    GET  /api/tasks/{id}/events  list events (REST)
+    GET  /api/tasks/{id}/stream  SSE event stream (live)
+    POST /api/chat               one-shot chat (streams)
+    POST /api/llm/chat           chat (non-streaming JSON)
+    POST /api/deploy/huggingface push backend dir to HF Space
+    POST /api/deploy/vercel      deploy frontend dir to Vercel
+    POST /api/git/push           commit + push to GitHub branch
+
+All endpoints accept JSON bodies and return JSON unless documented otherwise.
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import threading
+import time
+from typing import Any, Dict, List, Optional
+
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse, StreamingResponse
+from pydantic import BaseModel
+
+from . import tasks
+from .agent import run_task
+from .llm_router import get_router
+from .executor import get_executor
+from . import deployers
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
+logger = logging.getLogger("app")
+
+app = FastAPI(title="AI Developer Agent", version="1.0.0")
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=os.getenv("CORS_ALLOW_ORIGINS", "*").split(","),
+    allow_credentials=False,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+# ---------------------------------------------------------------------------
+# In-memory task queue (background worker)
+# ---------------------------------------------------------------------------
+_task_queue: "asyncio.Queue" = asyncio.Queue()
+_active_subscribers: Dict[str, List[asyncio.Queue]] = {}
+
+
+def _publish(task_id: str, event: Dict[str, Any]) -> None:
+    for q in list(_active_subscribers.get(task_id, [])):
+        try:
+            q.put_nowait(event)
+        except Exception:
+            pass
+
+
+def _worker_run(task_id: str, title: str, description: str) -> None:
+    """Run the agent generator in a thread and publish events."""
+    try:
+        for ev in run_task(task_id, title, description):
+            _publish(task_id, ev)
+    except Exception as e:
+        logger.exception("worker crashed")
+        _publish(task_id, {"task_id": task_id, "kind": "error", "message": str(e), "ts": time.time(), "data": {}})
+
+
+# ---------------------------------------------------------------------------
+# Schemas
+# ---------------------------------------------------------------------------
+class CreateTaskBody(BaseModel):
+    title: str
+    description: str = ""
+    payload: Optional[Dict[str, Any]] = None
+
+
+class ChatBody(BaseModel):
+    messages: List[Dict[str, str]]
+    model: Optional[str] = None
+    temperature: float = 0.2
+    max_tokens: int = 1500
+    preferred_provider: Optional[str] = None
+
+
+class HFDeployBody(BaseModel):
+    repo_id: str
+    source_dir: str = "."
+    commit_message: str = "Update from AI Developer Agent"
+
+
+class VercelDeployBody(BaseModel):
+    project_name: str
+    source_dir: str
+    framework: Optional[str] = "nextjs"
+    target: str = "production"
+    install_command: Optional[str] = None
+    build_command: Optional[str] = None
+    env: Optional[Dict[str, str]] = None
+
+
+class GitPushBody(BaseModel):
+    repo_dir: str = "."
+    branch: str = "genspark_ai_developer"
+    commit_message: str = "AI Developer Agent commit"
+    remote_url: Optional[str] = None
+
+
+# ---------------------------------------------------------------------------
+# Routes
+# ---------------------------------------------------------------------------
+@app.get("/")
+def index():
+    return {
+        "service": "AI Developer Agent",
+        "version": "1.0.0",
+        "ok": True,
+        "endpoints": [
+            "/health", "/api/runtime", "/api/tasks", "/api/tasks/{id}/stream",
+            "/api/chat", "/api/llm/chat",
+            "/api/deploy/huggingface", "/api/deploy/vercel", "/api/git/push",
+        ],
+    }
+
+
+@app.get("/health")
+def health():
+    router = get_router()
+    return {
+        "ok": True,
+        "ts": time.time(),
+        "providers": list(router.telemetry().keys()),
+        "executor": "e2b" if (get_executor().sandbox and get_executor().sandbox.available) else "local",
+    }
+
+
+@app.get("/api/runtime")
+def runtime():
+    info = get_executor().inspect_runtime()
+    info["providers"] = get_router().telemetry()
+    info["db"] = tasks.DB_PATH
+    return info
+
+
+# ----- Tasks ---------------------------------------------------------------
+@app.post("/api/tasks")
+def create_task(body: CreateTaskBody):
+    task_id = tasks.create_task(body.title, body.description, body.payload or {})
+    t = threading.Thread(target=_worker_run, args=(task_id, body.title, body.description), daemon=True)
+    t.start()
+    return {"task_id": task_id, "title": body.title, "state": "queued"}
+
+
+@app.get("/api/tasks")
+def list_tasks(limit: int = 50):
+    return {"tasks": tasks.list_tasks(limit=limit)}
+
+
+@app.get("/api/tasks/{task_id}")
+def get_task(task_id: str):
+    t = tasks.get_task(task_id)
+    if not t:
+        raise HTTPException(404, "task not found")
+    return t
+
+
+@app.get("/api/tasks/{task_id}/events")
+def get_events(task_id: str, since_id: int = 0, limit: int = 1000):
+    return {"events": tasks.get_events(task_id, since_id=since_id, limit=limit)}
+
+
+@app.get("/api/tasks/{task_id}/stream")
+async def stream_events(task_id: str, request: Request):
+    """Server-Sent Events stream. Replays historical events then live events."""
+
+    async def gen():
+        # 1) Replay history
+        last_id = 0
+        history = tasks.get_events(task_id, since_id=0, limit=2000)
+        for ev in history:
+            last_id = ev["id"]
+            yield f"id: {ev['id']}\nevent: {ev['kind']}\ndata: {json.dumps(ev)}\n\n"
+
+        # 2) Subscribe for live events
+        q: asyncio.Queue = asyncio.Queue()
+        _active_subscribers.setdefault(task_id, []).append(q)
+        try:
+            while True:
+                if await request.is_disconnected():
+                    break
+                try:
+                    ev = await asyncio.wait_for(q.get(), timeout=15.0)
+                    yield f"event: {ev['kind']}\ndata: {json.dumps(ev)}\n\n"
+                    if ev["kind"] in ("done", "error") and ev.get("data", {}).get("final"):
+                        break
+                except asyncio.TimeoutError:
+                    # heartbeat
+                    yield ":keepalive\n\n"
+        finally:
+            try:
+                _active_subscribers.get(task_id, []).remove(q)
+            except ValueError:
+                pass
+
+    return StreamingResponse(gen(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})
+
+
+# ----- LLM endpoints -------------------------------------------------------
+@app.post("/api/llm/chat")
+def llm_chat(body: ChatBody):
+    router = get_router()
+    try:
+        text = router.chat(
+            body.messages, model=body.model, temperature=body.temperature,
+            max_tokens=body.max_tokens, preferred_provider=body.preferred_provider,
+        )
+        return {"ok": True, "text": text, "telemetry": router.telemetry()}
+    except Exception as e:
+        return JSONResponse({"ok": False, "error": str(e)}, status_code=500)
+
+
+@app.post("/api/chat")
+def chat_stream(body: ChatBody):
+    """SSE chat stream."""
+    router = get_router()
+
+    def gen():
+        for chunk in router.stream(
+            body.messages, model=body.model, temperature=body.temperature,
+            max_tokens=body.max_tokens, preferred_provider=body.preferred_provider,
+        ):
+            yield f"data: {json.dumps({'delta': chunk})}\n\n"
+        yield "data: [DONE]\n\n"
+
+    return StreamingResponse(gen(), media_type="text/event-stream",
+                             headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})
+
+
+# ----- Deploy endpoints ----------------------------------------------------
+@app.post("/api/deploy/huggingface")
+def deploy_hf(body: HFDeployBody):
+    src = os.path.abspath(body.source_dir)
+    if not os.path.isdir(src):
+        raise HTTPException(400, f"source_dir not found: {src}")
+    r = deployers.hf_push_space(source_dir=src, repo_id=body.repo_id, commit_message=body.commit_message)
+    if r.get("ok"):
+        tasks.record_deployment("", "huggingface", r.get("url", ""), "ok")
+    else:
+        tasks.record_deployment("", "huggingface", "", "failed")
+    return r
+
+
+@app.post("/api/deploy/vercel")
+def deploy_vercel(body: VercelDeployBody):
+    src = os.path.abspath(body.source_dir)
+    if not os.path.isdir(src):
+        raise HTTPException(400, f"source_dir not found: {src}")
+    files = deployers.collect_files_for_vercel(src)
+    r = deployers.vercel_deploy_via_api(
+        project_name=body.project_name, files=files, target=body.target,
+        env=body.env, framework=body.framework,
+        install_command=body.install_command, build_command=body.build_command,
+    )
+    if r.get("ok"):
+        tasks.record_deployment("", "vercel", r.get("url", ""), "ok")
+    return r
+
+
+@app.post("/api/git/push")
+def git_push(body: GitPushBody):
+    repo_dir = os.path.abspath(body.repo_dir)
+    if not os.path.isdir(repo_dir):
+        raise HTTPException(400, f"repo_dir not found: {repo_dir}")
+    return deployers.github_push(
+        repo_dir=repo_dir, branch=body.branch,
+        commit_message=body.commit_message, remote_url=body.remote_url,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Startup self-check
+# ---------------------------------------------------------------------------
+@app.on_event("startup")
+def startup_check():
+    logger.info("AI Developer Agent starting")
+    try:
+        tasks.init_db()
+        info = get_executor().inspect_runtime()
+        logger.info("Runtime: %s", info)
+        logger.info("Providers: %s", list(get_router().telemetry().keys()))
+    except Exception as e:
+        logger.warning("Startup check error: %s", e)
+
+
+# Allow running directly
+if __name__ == "__main__":
+    import uvicorn
+    port = int(os.getenv("PORT", "7860"))
+    uvicorn.run("apps.backend.app:app", host="0.0.0.0", port=port, log_level="info")

backend/browser.py

@@ -0,0 +1,148 @@
+"""
+Browser automation via Playwright (runs inside E2B sandbox when available,
+otherwise locally). Provides retry-safe, structured browser actions.
+"""
+from __future__ import annotations
+
+import base64
+import logging
+import os
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger("browser")
+
+
+@dataclass
+class BrowserResult:
+    ok: bool
+    action: str
+    url: str = ""
+    text: str = ""
+    screenshot_b64: str = ""
+    error: str = ""
+    meta: Dict[str, Any] = field(default_factory=dict)
+
+
+class BrowserController:
+    """Lightweight controller; lazy-initializes Playwright."""
+
+    def __init__(self) -> None:
+        self._playwright = None
+        self._browser = None
+        self._context = None
+        self._page = None
+        self._available: Optional[bool] = None
+
+    @property
+    def available(self) -> bool:
+        if self._available is None:
+            try:
+                import playwright  # noqa: F401
+                from playwright.sync_api import sync_playwright  # noqa: F401
+                self._available = True
+            except Exception as e:
+                logger.warning("Playwright not installed: %s", e)
+                self._available = False
+        return self._available
+
+    def _ensure(self):
+        if self._page is not None:
+            return self._page
+        from playwright.sync_api import sync_playwright
+        self._playwright = sync_playwright().start()
+        self._browser = self._playwright.chromium.launch(headless=True, args=["--no-sandbox", "--disable-dev-shm-usage"])
+        self._context = self._browser.new_context()
+        self._page = self._context.new_page()
+        return self._page
+
+    def navigate(self, url: str, timeout_ms: int = 30000) -> BrowserResult:
+        if not self.available:
+            return BrowserResult(ok=False, action="navigate", url=url, error="playwright not available")
+        try:
+            page = self._ensure()
+            page.goto(url, timeout=timeout_ms, wait_until="domcontentloaded")
+            return BrowserResult(ok=True, action="navigate", url=page.url, text=page.title())
+        except Exception as e:
+            logger.exception("navigate failed")
+            return BrowserResult(ok=False, action="navigate", url=url, error=str(e))
+
+    def click(self, selector: str, timeout_ms: int = 10000) -> BrowserResult:
+        if not self.available:
+            return BrowserResult(ok=False, action="click", error="playwright not available")
+        try:
+            page = self._ensure()
+            page.click(selector, timeout=timeout_ms)
+            return BrowserResult(ok=True, action="click", meta={"selector": selector})
+        except Exception as e:
+            return BrowserResult(ok=False, action="click", error=str(e))
+
+    def type_text(self, selector: str, text: str, timeout_ms: int = 10000) -> BrowserResult:
+        if not self.available:
+            return BrowserResult(ok=False, action="type", error="playwright not available")
+        try:
+            page = self._ensure()
+            page.fill(selector, text, timeout=timeout_ms)
+            return BrowserResult(ok=True, action="type", meta={"selector": selector})
+        except Exception as e:
+            return BrowserResult(ok=False, action="type", error=str(e))
+
+    def screenshot(self) -> BrowserResult:
+        if not self.available:
+            return BrowserResult(ok=False, action="screenshot", error="playwright not available")
+        try:
+            page = self._ensure()
+            png = page.screenshot(full_page=False)
+            b64 = base64.b64encode(png).decode("ascii")
+            return BrowserResult(ok=True, action="screenshot", url=page.url, screenshot_b64=b64)
+        except Exception as e:
+            return BrowserResult(ok=False, action="screenshot", error=str(e))
+
+    def scrape_text(self) -> BrowserResult:
+        if not self.available:
+            return BrowserResult(ok=False, action="scrape", error="playwright not available")
+        try:
+            page = self._ensure()
+            content = page.evaluate("() => document.body ? document.body.innerText : ''")
+            return BrowserResult(ok=True, action="scrape", url=page.url, text=(content or "")[:20000])
+        except Exception as e:
+            return BrowserResult(ok=False, action="scrape", error=str(e))
+
+    def close(self):
+        try:
+            if self._context: self._context.close()
+        except Exception: pass
+        try:
+            if self._browser: self._browser.close()
+        except Exception: pass
+        try:
+            if self._playwright: self._playwright.stop()
+        except Exception: pass
+        self._context = self._browser = self._page = self._playwright = None
+
+
+_browser: Optional[BrowserController] = None
+
+
+def get_browser() -> BrowserController:
+    global _browser
+    if _browser is None:
+        _browser = BrowserController()
+    return _browser
+
+
+def run_browser_action(action: Dict[str, Any]) -> BrowserResult:
+    """action: {"action": "navigate"|"click"|"type"|"screenshot"|"scrape", ...}"""
+    b = get_browser()
+    op = action.get("action", "")
+    if op == "navigate":
+        return b.navigate(action.get("url", ""))
+    if op == "click":
+        return b.click(action.get("selector", ""))
+    if op == "type":
+        return b.type_text(action.get("selector", ""), action.get("text", ""))
+    if op == "screenshot":
+        return b.screenshot()
+    if op == "scrape":
+        return b.scrape_text()
+    return BrowserResult(ok=False, action=op, error=f"unknown action: {op}")

backend/classifier.py

@@ -0,0 +1,98 @@
+"""
+Error classifier - identifies error categories from tracebacks/output.
+Used by repair engine to produce targeted repair plans.
+"""
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from typing import Optional
+
+
+@dataclass
+class ErrorClass:
+    category: str
+    detail: str
+    suggested_fix: str
+
+
+RULES = [
+    # python module / pip
+    (r"ModuleNotFoundError: No module named ['\"]([\w\.\-]+)['\"]", "missing_python_module"),
+    (r"ImportError: No module named ([\w\.\-]+)", "missing_python_module"),
+    (r"pip(?:3)?: command not found", "missing_pip"),
+    # node / npm
+    (r"command not found: (npm|node|npx)", "missing_node"),
+    (r"npm ERR! code E?ENOENT", "npm_failure"),
+    (r"npm ERR! code (E\w+)", "npm_failure"),
+    # playwright
+    (r"playwright[^\s]*: command not found", "missing_playwright"),
+    (r"Executable doesn't exist at .+(chrom|firefox|webkit)", "playwright_browsers_missing"),
+    (r"BrowserType\.launch:.*Host system is missing dependencies", "playwright_missing_deps"),
+    # git
+    (r"fatal: unable to auto-detect email address", "git_identity_missing"),
+    (r"Please tell me who you are\.", "git_identity_missing"),
+    (r"fatal: not a git repository", "not_a_git_repo"),
+    (r"Authentication failed", "git_auth_failed"),
+    # python version / build
+    (r"greenlet[^\n]*failed to build", "greenlet_build_failure"),
+    (r"Could not build wheels for ([\w\-]+)", "python_build_failure"),
+    (r"requires Python ['\"][^'\"]+['\"]", "python_version_mismatch"),
+    # network
+    (r"Could not resolve host", "network_failure"),
+    (r"Connection refused", "network_failure"),
+    (r"Read timed out", "network_failure"),
+    # http
+    (r"\b429\b", "rate_limited"),
+    (r"\b401\b|\b403\b", "auth_failure"),
+]
+
+
+def classify(output: str) -> Optional[ErrorClass]:
+    if not output:
+        return None
+    for pattern, category in RULES:
+        m = re.search(pattern, output)
+        if m:
+            detail = m.group(0)
+            return ErrorClass(category=category, detail=detail, suggested_fix=_fix_for(category, m))
+    # Generic exception detection
+    if "Traceback (most recent call last):" in output:
+        return ErrorClass(category="python_exception", detail="Unclassified Python exception", suggested_fix="inspect traceback and retry")
+    return None
+
+
+def _fix_for(category: str, m: re.Match) -> str:
+    if category == "missing_python_module":
+        return f"pip install {m.group(1)}"
+    if category == "missing_pip":
+        return "ensure python3-pip is installed"
+    if category == "missing_node":
+        return "install node + npm"
+    if category == "npm_failure":
+        return "delete node_modules and reinstall"
+    if category == "missing_playwright":
+        return "pip install playwright && python -m playwright install"
+    if category == "playwright_browsers_missing":
+        return "python -m playwright install chromium"
+    if category == "playwright_missing_deps":
+        return "python -m playwright install-deps chromium"
+    if category == "git_identity_missing":
+        return "git config user.email and user.name"
+    if category == "not_a_git_repo":
+        return "git init or cd into repo"
+    if category == "git_auth_failed":
+        return "verify GITHUB_TOKEN and remote URL"
+    if category == "greenlet_build_failure":
+        return "pin Python 3.11 and install build essentials"
+    if category == "python_build_failure":
+        return "install build-essential and retry"
+    if category == "python_version_mismatch":
+        return "use Python 3.11"
+    if category == "network_failure":
+        return "retry after backoff"
+    if category == "rate_limited":
+        return "rotate provider key or wait"
+    if category == "auth_failure":
+        return "rotate API key"
+    return "retry"

backend/deployers.py

@@ -0,0 +1,281 @@
+"""
+Deployment helpers for HuggingFace Spaces, Vercel, and GitHub.
+All real - no mocks. Each helper returns a structured result dict.
+"""
+from __future__ import annotations
+
+import os
+import json
+import logging
+import shutil
+import subprocess
+import tempfile
+import time
+from typing import Any, Dict, List, Optional
+
+import httpx
+
+logger = logging.getLogger("deployers")
+
+
+# ---------------------------------------------------------------------------
+# GitHub
+# ---------------------------------------------------------------------------
+def github_push(
+    repo_dir: str,
+    branch: str = "genspark_ai_developer",
+    commit_message: str = "AI Developer Agent commit",
+    token: Optional[str] = None,
+    remote_url: Optional[str] = None,
+) -> Dict[str, Any]:
+    token = token or os.getenv("GITHUB_TOKEN") or os.getenv("GITHUB_PAT")
+    if not token:
+        return {"ok": False, "error": "GITHUB_TOKEN missing"}
+    try:
+        env = os.environ.copy()
+        env["GIT_TERMINAL_PROMPT"] = "0"
+
+        def run(cmd: List[str], check: bool = True) -> subprocess.CompletedProcess:
+            r = subprocess.run(cmd, cwd=repo_dir, capture_output=True, text=True, env=env, timeout=120)
+            if check and r.returncode != 0:
+                raise RuntimeError(f"{' '.join(cmd)}: {r.stderr[:500]}")
+            return r
+
+        # ensure identity
+        run(["git", "config", "user.email", "ai-developer@genspark.ai"], check=False)
+        run(["git", "config", "user.name", "AI Developer Agent"], check=False)
+
+        # set token URL
+        if remote_url:
+            authed = remote_url.replace("https://", f"https://x-access-token:{token}@")
+            run(["git", "remote", "set-url", "origin", authed], check=False)
+
+        run(["git", "add", "-A"], check=False)
+        # commit may fail if nothing to commit; that's OK
+        commit = subprocess.run(["git", "commit", "-m", commit_message], cwd=repo_dir, capture_output=True, text=True, env=env)
+        run(["git", "checkout", "-B", branch], check=False)
+        push = subprocess.run(["git", "push", "-u", "origin", branch, "--force"], cwd=repo_dir, capture_output=True, text=True, env=env, timeout=180)
+        ok = push.returncode == 0
+        return {
+            "ok": ok,
+            "branch": branch,
+            "commit_out": commit.stdout + commit.stderr,
+            "push_out": push.stdout + push.stderr,
+        }
+    except Exception as e:
+        logger.exception("github_push failed")
+        return {"ok": False, "error": str(e)}
+
+
+# ---------------------------------------------------------------------------
+# Hugging Face Space
+# ---------------------------------------------------------------------------
+def hf_ensure_space(
+    repo_id: str,
+    token: Optional[str] = None,
+    sdk: str = "docker",
+    private: bool = False,
+) -> Dict[str, Any]:
+    """Create the Space if it doesn't exist (idempotent)."""
+    token = token or os.getenv("HF_TOKEN")
+    if not token:
+        return {"ok": False, "error": "HF_TOKEN missing"}
+    try:
+        headers = {"Authorization": f"Bearer {token}"}
+        info = httpx.get(f"https://huggingface.co/api/spaces/{repo_id}", headers=headers, timeout=30.0)
+        if info.status_code == 200:
+            return {"ok": True, "created": False, "url": f"https://huggingface.co/spaces/{repo_id}"}
+        # create
+        owner, name = repo_id.split("/", 1)
+        payload = {
+            "name": name,
+            "organization": None if owner == _hf_whoami(token) else owner,
+            "type": "space",
+            "sdk": sdk,
+            "private": private,
+        }
+        r = httpx.post(
+            "https://huggingface.co/api/repos/create",
+            headers={**headers, "Content-Type": "application/json"},
+            json=payload,
+            timeout=30.0,
+        )
+        if r.status_code >= 400:
+            return {"ok": False, "error": f"create failed: {r.status_code} {r.text[:300]}"}
+        return {"ok": True, "created": True, "url": f"https://huggingface.co/spaces/{repo_id}"}
+    except Exception as e:
+        return {"ok": False, "error": str(e)}
+
+
+def _hf_whoami(token: str) -> str:
+    try:
+        r = httpx.get("https://huggingface.co/api/whoami-v2", headers={"Authorization": f"Bearer {token}"}, timeout=15)
+        if r.status_code == 200:
+            return r.json().get("name", "")
+    except Exception:
+        pass
+    return ""
+
+
+def hf_push_space(
+    source_dir: str,
+    repo_id: str,
+    token: Optional[str] = None,
+    commit_message: str = "Update from AI Developer Agent",
+) -> Dict[str, Any]:
+    """Push contents of source_dir to a HuggingFace Space using git."""
+    token = token or os.getenv("HF_TOKEN")
+    if not token:
+        return {"ok": False, "error": "HF_TOKEN missing"}
+    try:
+        # First ensure space exists
+        ensure = hf_ensure_space(repo_id, token=token, sdk="docker")
+        if not ensure.get("ok"):
+            return {"ok": False, "error": f"ensure_space: {ensure.get('error')}"}
+
+        tmp = tempfile.mkdtemp(prefix="hfpush_")
+        try:
+            remote = f"https://user:{token}@huggingface.co/spaces/{repo_id}"
+            # Clone (may be empty)
+            clone = subprocess.run(["git", "clone", remote, tmp], capture_output=True, text=True, timeout=120)
+            if clone.returncode != 0:
+                # try init
+                subprocess.run(["git", "init"], cwd=tmp, capture_output=True, text=True)
+                subprocess.run(["git", "remote", "add", "origin", remote], cwd=tmp, capture_output=True, text=True)
+
+            # Copy source files into tmp (preserve .git)
+            for entry in os.listdir(source_dir):
+                if entry == ".git":
+                    continue
+                src = os.path.join(source_dir, entry)
+                dst = os.path.join(tmp, entry)
+                if os.path.isdir(src):
+                    if os.path.exists(dst):
+                        shutil.rmtree(dst, ignore_errors=True)
+                    shutil.copytree(src, dst)
+                else:
+                    shutil.copy2(src, dst)
+
+            subprocess.run(["git", "config", "user.email", "ai-developer@genspark.ai"], cwd=tmp, capture_output=True, text=True)
+            subprocess.run(["git", "config", "user.name", "AI Developer Agent"], cwd=tmp, capture_output=True, text=True)
+            subprocess.run(["git", "lfs", "install"], cwd=tmp, capture_output=True, text=True)
+            subprocess.run(["git", "add", "-A"], cwd=tmp, capture_output=True, text=True)
+            commit = subprocess.run(["git", "commit", "-m", commit_message], cwd=tmp, capture_output=True, text=True)
+            push = subprocess.run(["git", "push", "origin", "main", "--force"], cwd=tmp, capture_output=True, text=True, timeout=300)
+            ok = push.returncode == 0
+            return {
+                "ok": ok,
+                "url": f"https://huggingface.co/spaces/{repo_id}",
+                "commit_out": (commit.stdout + commit.stderr)[-800:],
+                "push_out": (push.stdout + push.stderr)[-800:],
+            }
+        finally:
+            shutil.rmtree(tmp, ignore_errors=True)
+    except Exception as e:
+        logger.exception("hf_push_space failed")
+        return {"ok": False, "error": str(e)}
+
+
+def hf_space_health(repo_id: str, path: str = "/health", timeout: float = 15.0) -> Dict[str, Any]:
+    """Check the live space URL for health."""
+    owner, name = repo_id.split("/", 1)
+    url = f"https://{owner}-{name}.hf.space{path}"
+    try:
+        r = httpx.get(url, timeout=timeout)
+        return {"ok": r.status_code < 500, "status": r.status_code, "url": url, "body": r.text[:500]}
+    except Exception as e:
+        return {"ok": False, "url": url, "error": str(e)}
+
+
+# ---------------------------------------------------------------------------
+# Vercel
+# ---------------------------------------------------------------------------
+def vercel_deploy_via_api(
+    project_name: str,
+    files: List[Dict[str, Any]],
+    token: Optional[str] = None,
+    target: str = "production",
+    env: Optional[Dict[str, str]] = None,
+    framework: Optional[str] = "nextjs",
+    install_command: Optional[str] = None,
+    build_command: Optional[str] = None,
+) -> Dict[str, Any]:
+    """
+    Deploy via Vercel HTTP API.
+
+    files: list of {"file": "path/in/repo", "data": "file contents"}
+    """
+    token = token or os.getenv("VERCEL_TOKEN")
+    if not token:
+        return {"ok": False, "error": "VERCEL_TOKEN missing"}
+    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
+    body: Dict[str, Any] = {
+        "name": project_name,
+        "files": files,
+        "target": target,
+        "projectSettings": {
+            "framework": framework,
+            "installCommand": install_command,
+            "buildCommand": build_command,
+        },
+    }
+    if env:
+        body["env"] = env
+        body["build"] = {"env": env}
+    try:
+        r = httpx.post("https://api.vercel.com/v13/deployments", headers=headers, json=body, timeout=180.0)
+        if r.status_code >= 400:
+            return {"ok": False, "status": r.status_code, "error": r.text[:1000]}
+        data = r.json()
+        url = data.get("url") or ""
+        full = f"https://{url}" if url and not url.startswith("http") else url
+        return {"ok": True, "url": full, "id": data.get("id"), "data": data}
+    except Exception as e:
+        logger.exception("vercel_deploy_via_api failed")
+        return {"ok": False, "error": str(e)}
+
+
+def vercel_deployment_status(deployment_id: str, token: Optional[str] = None) -> Dict[str, Any]:
+    token = token or os.getenv("VERCEL_TOKEN")
+    if not token:
+        return {"ok": False, "error": "VERCEL_TOKEN missing"}
+    try:
+        r = httpx.get(
+            f"https://api.vercel.com/v13/deployments/{deployment_id}",
+            headers={"Authorization": f"Bearer {token}"},
+            timeout=30.0,
+        )
+        if r.status_code >= 400:
+            return {"ok": False, "status": r.status_code, "error": r.text[:500]}
+        d = r.json()
+        return {"ok": True, "state": d.get("readyState"), "url": d.get("url"), "data": d}
+    except Exception as e:
+        return {"ok": False, "error": str(e)}
+
+
+def collect_files_for_vercel(root_dir: str) -> List[Dict[str, Any]]:
+    """Walk root_dir and produce list of {file, data} for Vercel API.
+
+    Skips node_modules, .git, .next, .vercel and other build artifacts.
+    """
+    SKIP_DIRS = {"node_modules", ".git", ".next", ".vercel", "dist", "build", "__pycache__"}
+    SKIP_EXTS = {".log"}
+    files: List[Dict[str, Any]] = []
+    for cur, dirs, fns in os.walk(root_dir):
+        dirs[:] = [d for d in dirs if d not in SKIP_DIRS]
+        for fn in fns:
+            if any(fn.endswith(e) for e in SKIP_EXTS):
+                continue
+            full = os.path.join(cur, fn)
+            rel = os.path.relpath(full, root_dir).replace("\\", "/")
+            try:
+                with open(full, "r", encoding="utf-8") as f:
+                    data = f.read()
+            except UnicodeDecodeError:
+                with open(full, "rb") as f:
+                    import base64 as _b
+                    data = _b.b64encode(f.read()).decode("ascii")
+                    files.append({"file": rel, "data": data, "encoding": "base64"})
+                    continue
+            files.append({"file": rel, "data": data})
+    return files

backend/executor.py

@@ -0,0 +1,238 @@
+"""
+Executor - runs plan actions. Uses E2B sandbox when E2B_API_KEY is set,
+otherwise falls back to local subprocess execution (with strict allowlist).
+"""
+from __future__ import annotations
+
+import os
+import shlex
+import subprocess
+import logging
+import threading
+import time
+from dataclasses import dataclass, field
+from typing import Any, Dict, Generator, List, Optional
+
+logger = logging.getLogger("executor")
+
+E2B_API_KEY = os.getenv("E2B_API_KEY", "")
+
+
+@dataclass
+class ExecutionResult:
+    ok: bool
+    stdout: str = ""
+    stderr: str = ""
+    exit_code: int = 0
+    duration_ms: float = 0.0
+    meta: Dict[str, Any] = field(default_factory=dict)
+
+
+# ---------------------------------------------------------------------------
+# E2B sandbox wrapper (with graceful fallback)
+# ---------------------------------------------------------------------------
+class E2BSandbox:
+    """Thin wrapper around e2b_code_interpreter. Created lazily."""
+
+    def __init__(self) -> None:
+        self._sbx = None
+        self._lock = threading.Lock()
+        self._available = False
+        if E2B_API_KEY:
+            try:
+                from e2b_code_interpreter import Sandbox  # type: ignore
+                self._Sandbox = Sandbox
+                self._available = True
+            except Exception as e:
+                logger.warning("E2B SDK not available: %s", e)
+                self._available = False
+
+    @property
+    def available(self) -> bool:
+        return self._available
+
+    def _ensure(self):
+        if self._sbx is None:
+            self._sbx = self._Sandbox(api_key=E2B_API_KEY)
+        return self._sbx
+
+    def run_shell(self, cmd: str, timeout: float = 120.0) -> ExecutionResult:
+        started = time.time()
+        with self._lock:
+            try:
+                sbx = self._ensure()
+                # Newer e2b SDKs use sbx.commands.run
+                try:
+                    cmd_result = sbx.commands.run(cmd, timeout=int(timeout))
+                    stdout = getattr(cmd_result, "stdout", "") or ""
+                    stderr = getattr(cmd_result, "stderr", "") or ""
+                    exit_code = getattr(cmd_result, "exit_code", 0) or 0
+                except AttributeError:
+                    # Fallback for legacy SDK
+                    res = sbx.run_code(f"import subprocess; r=subprocess.run({cmd!r}, shell=True, capture_output=True, text=True, timeout={timeout}); print(r.stdout); print(r.stderr)")
+                    stdout = "\n".join([str(getattr(r, "text", "")) for r in getattr(res, "logs", {}).get("stdout", []) or []])
+                    stderr = ""
+                    exit_code = 0
+                ok = exit_code == 0
+                return ExecutionResult(
+                    ok=ok, stdout=stdout, stderr=stderr, exit_code=exit_code,
+                    duration_ms=(time.time() - started) * 1000,
+                    meta={"engine": "e2b"},
+                )
+            except Exception as e:
+                logger.exception("E2B run_shell failed")
+                return ExecutionResult(
+                    ok=False, stderr=str(e), exit_code=1,
+                    duration_ms=(time.time() - started) * 1000,
+                    meta={"engine": "e2b", "error": True},
+                )
+
+    def run_python(self, code: str, timeout: float = 120.0) -> ExecutionResult:
+        started = time.time()
+        with self._lock:
+            try:
+                sbx = self._ensure()
+                try:
+                    res = sbx.run_code(code, timeout=int(timeout))
+                    stdout_logs = []
+                    stderr_logs = []
+                    if hasattr(res, "logs"):
+                        for entry in getattr(res.logs, "stdout", []) or []:
+                            stdout_logs.append(str(entry))
+                        for entry in getattr(res.logs, "stderr", []) or []:
+                            stderr_logs.append(str(entry))
+                    return ExecutionResult(
+                        ok=True,
+                        stdout="\n".join(stdout_logs),
+                        stderr="\n".join(stderr_logs),
+                        exit_code=0,
+                        duration_ms=(time.time() - started) * 1000,
+                        meta={"engine": "e2b"},
+                    )
+                except Exception as e:
+                    return ExecutionResult(
+                        ok=False, stderr=str(e), exit_code=1,
+                        duration_ms=(time.time() - started) * 1000,
+                        meta={"engine": "e2b", "error": True},
+                    )
+            except Exception as e:
+                return ExecutionResult(
+                    ok=False, stderr=str(e), exit_code=1,
+                    duration_ms=(time.time() - started) * 1000,
+                    meta={"engine": "e2b", "error": True},
+                )
+
+    def close(self):
+        with self._lock:
+            try:
+                if self._sbx is not None:
+                    self._sbx.kill()
+            except Exception:
+                pass
+            self._sbx = None
+
+
+# ---------------------------------------------------------------------------
+# Local subprocess fallback - LIMITED commands only
+# ---------------------------------------------------------------------------
+_DISALLOWED_PATTERNS = [
+    "rm -rf /",
+    ":(){:|:&};:",
+    "mkfs",
+    "> /dev/sda",
+]
+
+
+def _local_run_shell(cmd: str, timeout: float = 120.0) -> ExecutionResult:
+    started = time.time()
+    if any(p in cmd for p in _DISALLOWED_PATTERNS):
+        return ExecutionResult(ok=False, stderr="Disallowed command", exit_code=126,
+                               duration_ms=(time.time() - started) * 1000)
+    try:
+        res = subprocess.run(
+            cmd, shell=True, capture_output=True, text=True, timeout=timeout,
+        )
+        return ExecutionResult(
+            ok=res.returncode == 0,
+            stdout=res.stdout or "",
+            stderr=res.stderr or "",
+            exit_code=res.returncode,
+            duration_ms=(time.time() - started) * 1000,
+            meta={"engine": "local"},
+        )
+    except subprocess.TimeoutExpired as e:
+        return ExecutionResult(ok=False, stderr=f"timeout: {e}", exit_code=124,
+                               duration_ms=(time.time() - started) * 1000,
+                               meta={"engine": "local"})
+    except Exception as e:
+        return ExecutionResult(ok=False, stderr=str(e), exit_code=1,
+                               duration_ms=(time.time() - started) * 1000,
+                               meta={"engine": "local"})
+
+
+def _local_run_python(code: str, timeout: float = 120.0) -> ExecutionResult:
+    # Run python in subprocess for isolation
+    started = time.time()
+    try:
+        res = subprocess.run(
+            ["python3", "-c", code], capture_output=True, text=True, timeout=timeout,
+        )
+        return ExecutionResult(
+            ok=res.returncode == 0,
+            stdout=res.stdout or "",
+            stderr=res.stderr or "",
+            exit_code=res.returncode,
+            duration_ms=(time.time() - started) * 1000,
+            meta={"engine": "local"},
+        )
+    except subprocess.TimeoutExpired as e:
+        return ExecutionResult(ok=False, stderr=f"timeout: {e}", exit_code=124,
+                               duration_ms=(time.time() - started) * 1000)
+    except Exception as e:
+        return ExecutionResult(ok=False, stderr=str(e), exit_code=1,
+                               duration_ms=(time.time() - started) * 1000)
+
+
+# ---------------------------------------------------------------------------
+# Executor singleton
+# ---------------------------------------------------------------------------
+class Executor:
+    def __init__(self) -> None:
+        self.sandbox = E2BSandbox() if E2B_API_KEY else None
+
+    def shell(self, cmd: str, timeout: float = 120.0) -> ExecutionResult:
+        if self.sandbox and self.sandbox.available:
+            return self.sandbox.run_shell(cmd, timeout=timeout)
+        return _local_run_shell(cmd, timeout=timeout)
+
+    def python(self, code: str, timeout: float = 120.0) -> ExecutionResult:
+        if self.sandbox and self.sandbox.available:
+            return self.sandbox.run_python(code, timeout=timeout)
+        return _local_run_python(code, timeout=timeout)
+
+    def inspect_runtime(self) -> Dict[str, str]:
+        info: Dict[str, str] = {}
+        for label, cmd in [
+            ("python", "python3 --version"),
+            ("node", "node --version"),
+            ("npm", "npm --version"),
+            ("git", "git --version"),
+            ("playwright", "python3 -c 'import playwright; print(playwright.__version__)' 2>/dev/null || echo 'not installed'"),
+        ]:
+            r = self.shell(cmd, timeout=15)
+            info[label] = (r.stdout or r.stderr).strip()[:200]
+        return info
+
+    def close(self):
+        if self.sandbox:
+            self.sandbox.close()
+
+
+_executor: Optional[Executor] = None
+
+
+def get_executor() -> Executor:
+    global _executor
+    if _executor is None:
+        _executor = Executor()
+    return _executor

backend/llm_router.py

@@ -0,0 +1,407 @@
+"""
+Multi-Provider LLM Router with key rotation, cooldowns, health tracking,
+and adaptive provider selection.
+
+Supported providers:
+- OpenAI (chat completions, OpenAI compatible)
+- Gemini (Google Generative AI)
+- SambaNova (OpenAI-compatible API)
+- GitHub Models (OpenAI-compatible /inference endpoint)
+- OpenRouter (OpenAI-compatible)
+
+Environment variables (comma-separated key lists):
+    OPENAI_API_KEYS
+    GEMINI_API_KEYS
+    SAMBANOVA_API_KEYS
+    GITHUB_LLM_API_KEYS
+    OPENROUTER_API_KEYS
+
+Public API:
+    router = LLMRouter()
+    text = router.chat(messages, model=None, temperature=0.2)
+    for chunk in router.stream(messages, model=None): ...
+"""
+from __future__ import annotations
+
+import json
+import os
+import random
+import time
+import threading
+import logging
+from dataclasses import dataclass, field
+from typing import Generator, Iterable, List, Optional, Dict, Any
+
+import httpx
+
+logger = logging.getLogger("llm_router")
+
+
+def _split_env(name: str) -> List[str]:
+    raw = os.getenv(name, "")
+    return [k.strip() for k in raw.split(",") if k.strip()]
+
+
+# ---------------------------------------------------------------------------
+# Key registry
+# ---------------------------------------------------------------------------
+@dataclass
+class KeyState:
+    key: str
+    provider: str
+    cooldown_until: float = 0.0
+    failures: int = 0
+    requests: int = 0
+    last_latency_ms: float = 0.0
+    last_error: str = ""
+
+    def healthy(self) -> bool:
+        return time.time() >= self.cooldown_until
+
+
+@dataclass
+class ProviderConfig:
+    name: str
+    default_model: str
+    base_url: str
+    api_style: str  # "openai" | "gemini"
+    # Optional extra headers (e.g. for GitHub Models)
+    extra_headers: Dict[str, str] = field(default_factory=dict)
+
+
+PROVIDER_CONFIGS: Dict[str, ProviderConfig] = {
+    "openai": ProviderConfig(
+        name="openai",
+        default_model="gpt-4o-mini",
+        base_url="https://api.openai.com/v1",
+        api_style="openai",
+    ),
+    "gemini": ProviderConfig(
+        name="gemini",
+        default_model="gemini-1.5-flash-latest",
+        base_url="https://generativelanguage.googleapis.com/v1beta",
+        api_style="gemini",
+    ),
+    "sambanova": ProviderConfig(
+        name="sambanova",
+        default_model="Meta-Llama-3.3-70B-Instruct",
+        base_url="https://api.sambanova.ai/v1",
+        api_style="openai",
+    ),
+    "github": ProviderConfig(
+        name="github",
+        default_model="gpt-4o-mini",
+        base_url="https://models.github.ai/inference",
+        api_style="openai",
+    ),
+    "openrouter": ProviderConfig(
+        name="openrouter",
+        default_model="openai/gpt-4o-mini",
+        base_url="https://openrouter.ai/api/v1",
+        api_style="openai",
+        extra_headers={"HTTP-Referer": "https://github.com/ai-developer-agent", "X-Title": "AI Developer Agent"},
+    ),
+}
+
+
+class LLMRouter:
+    """Thread-safe, multi-provider LLM router with rotation + failover."""
+
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._registry: Dict[str, List[KeyState]] = {}
+        self._rr_index: Dict[str, int] = {}
+        self._provider_priority: List[str] = []
+        self._load_keys()
+
+    # -------------------- key registry --------------------
+    def _load_keys(self) -> None:
+        mapping = [
+            ("openai", "OPENAI_API_KEYS"),
+            ("gemini", "GEMINI_API_KEYS"),
+            ("sambanova", "SAMBANOVA_API_KEYS"),
+            ("github", "GITHUB_LLM_API_KEYS"),
+            ("openrouter", "OPENROUTER_API_KEYS"),
+        ]
+        for provider, env_name in mapping:
+            keys = _split_env(env_name)
+            if keys:
+                self._registry[provider] = [KeyState(k, provider) for k in keys]
+                self._rr_index[provider] = 0
+                self._provider_priority.append(provider)
+
+        if not self._registry:
+            logger.warning("LLMRouter: no provider keys configured.")
+
+    def telemetry(self) -> Dict[str, Any]:
+        out: Dict[str, Any] = {}
+        with self._lock:
+            for provider, states in self._registry.items():
+                out[provider] = {
+                    "keys": len(states),
+                    "healthy_keys": sum(1 for s in states if s.healthy()),
+                    "total_requests": sum(s.requests for s in states),
+                    "failures": sum(s.failures for s in states),
+                    "last_error": next((s.last_error for s in states if s.last_error), ""),
+                }
+        return out
+
+    def _pick_key(self, provider: str) -> Optional[KeyState]:
+        states = self._registry.get(provider, [])
+        if not states:
+            return None
+        idx = self._rr_index.get(provider, 0)
+        n = len(states)
+        for i in range(n):
+            s = states[(idx + i) % n]
+            if s.healthy():
+                self._rr_index[provider] = (idx + i + 1) % n
+                return s
+        return None
+
+    def _cooldown(self, state: KeyState, seconds: float, error: str = "") -> None:
+        with self._lock:
+            state.cooldown_until = time.time() + seconds
+            state.failures += 1
+            state.last_error = error[:300]
+
+    # -------------------- public API --------------------
+    def chat(
+        self,
+        messages: List[Dict[str, str]],
+        model: Optional[str] = None,
+        temperature: float = 0.2,
+        max_tokens: int = 1500,
+        timeout: float = 60.0,
+        preferred_provider: Optional[str] = None,
+    ) -> str:
+        chunks: List[str] = []
+        for chunk in self.stream(
+            messages,
+            model=model,
+            temperature=temperature,
+            max_tokens=max_tokens,
+            timeout=timeout,
+            preferred_provider=preferred_provider,
+        ):
+            chunks.append(chunk)
+        return "".join(chunks)
+
+    def stream(
+        self,
+        messages: List[Dict[str, str]],
+        model: Optional[str] = None,
+        temperature: float = 0.2,
+        max_tokens: int = 1500,
+        timeout: float = 60.0,
+        preferred_provider: Optional[str] = None,
+    ) -> Generator[str, None, None]:
+        """Try providers in priority order, with key rotation per provider."""
+        order = list(self._provider_priority)
+        if preferred_provider and preferred_provider in order:
+            order.remove(preferred_provider)
+            order.insert(0, preferred_provider)
+
+        if not order:
+            yield "[LLMRouter] No providers configured. Returning placeholder.\n"
+            yield self._offline_placeholder(messages)
+            return
+
+        last_error = "no providers tried"
+        for provider in order:
+            tried = 0
+            max_keys = len(self._registry.get(provider, []))
+            while tried < max_keys:
+                key_state = self._pick_key(provider)
+                if key_state is None:
+                    break
+                tried += 1
+                try:
+                    started = time.time()
+                    yielded_any = False
+                    for chunk in self._stream_provider(
+                        provider,
+                        key_state,
+                        messages,
+                        model=model,
+                        temperature=temperature,
+                        max_tokens=max_tokens,
+                        timeout=timeout,
+                    ):
+                        yielded_any = True
+                        yield chunk
+                    if yielded_any:
+                        with self._lock:
+                            key_state.requests += 1
+                            key_state.last_latency_ms = (time.time() - started) * 1000
+                        return
+                    else:
+                        last_error = f"{provider}: empty response"
+                        self._cooldown(key_state, 15, last_error)
+                except httpx.HTTPStatusError as e:
+                    code = e.response.status_code if e.response else 0
+                    body = ""
+                    try:
+                        body = e.response.text[:300] if e.response else ""
+                    except Exception:
+                        body = ""
+                    last_error = f"{provider}/{code}: {body}"
+                    logger.warning(last_error)
+                    if code == 429:
+                        self._cooldown(key_state, 60, last_error)
+                    elif 500 <= code < 600:
+                        self._cooldown(key_state, 30, last_error)
+                    elif code in (401, 403):
+                        # bad key – cool it down for a long time
+                        self._cooldown(key_state, 3600, last_error)
+                    else:
+                        self._cooldown(key_state, 10, last_error)
+                except Exception as e:
+                    last_error = f"{provider} error: {e}"
+                    logger.exception("Provider error")
+                    self._cooldown(key_state, 20, last_error)
+
+        yield f"[LLMRouter] All providers failed. Last error: {last_error}\n"
+        yield self._offline_placeholder(messages)
+
+    def _offline_placeholder(self, messages: List[Dict[str, str]]) -> str:
+        last_user = next((m["content"] for m in reversed(messages) if m.get("role") == "user"), "")
+        return (
+            "OFFLINE_PLAN_FALLBACK\n"
+            "I could not reach any LLM provider. Producing a heuristic response.\n"
+            f"User intent: {last_user[:200]}\n"
+        )
+
+    # -------------------- provider impls --------------------
+    def _stream_provider(
+        self,
+        provider: str,
+        key_state: KeyState,
+        messages: List[Dict[str, str]],
+        model: Optional[str],
+        temperature: float,
+        max_tokens: int,
+        timeout: float,
+    ) -> Generator[str, None, None]:
+        cfg = PROVIDER_CONFIGS[provider]
+        if cfg.api_style == "openai":
+            yield from self._stream_openai_compatible(cfg, key_state, messages, model, temperature, max_tokens, timeout)
+        elif cfg.api_style == "gemini":
+            yield from self._stream_gemini(cfg, key_state, messages, model, temperature, max_tokens, timeout)
+        else:
+            raise RuntimeError(f"Unknown api_style: {cfg.api_style}")
+
+    def _stream_openai_compatible(
+        self,
+        cfg: ProviderConfig,
+        key_state: KeyState,
+        messages: List[Dict[str, str]],
+        model: Optional[str],
+        temperature: float,
+        max_tokens: int,
+        timeout: float,
+    ) -> Generator[str, None, None]:
+        url = f"{cfg.base_url}/chat/completions"
+        headers = {
+            "Authorization": f"Bearer {key_state.key}",
+            "Content-Type": "application/json",
+        }
+        headers.update(cfg.extra_headers)
+        body = {
+            "model": model or cfg.default_model,
+            "messages": messages,
+            "temperature": temperature,
+            "max_tokens": max_tokens,
+            "stream": True,
+        }
+        with httpx.Client(timeout=timeout) as client:
+            with client.stream("POST", url, json=body, headers=headers) as r:
+                if r.status_code >= 400:
+                    text = r.read().decode("utf-8", "ignore")
+                    raise httpx.HTTPStatusError(text[:300], request=r.request, response=r)
+                for line in r.iter_lines():
+                    if not line:
+                        continue
+                    if isinstance(line, bytes):
+                        line = line.decode("utf-8", "ignore")
+                    if line.startswith("data:"):
+                        line = line[5:].strip()
+                    if not line or line == "[DONE]":
+                        continue
+                    try:
+                        obj = json.loads(line)
+                        delta = obj.get("choices", [{}])[0].get("delta", {})
+                        content = delta.get("content") or ""
+                        if content:
+                            yield content
+                    except Exception:
+                        continue
+
+    def _stream_gemini(
+        self,
+        cfg: ProviderConfig,
+        key_state: KeyState,
+        messages: List[Dict[str, str]],
+        model: Optional[str],
+        temperature: float,
+        max_tokens: int,
+        timeout: float,
+    ) -> Generator[str, None, None]:
+        model_name = model or cfg.default_model
+        # Use streamGenerateContent
+        url = f"{cfg.base_url}/models/{model_name}:streamGenerateContent?alt=sse&key={key_state.key}"
+        contents = []
+        sys_prompt = ""
+        for m in messages:
+            role = m.get("role", "user")
+            content = m.get("content", "")
+            if role == "system":
+                sys_prompt += content + "\n"
+                continue
+            mapped = "user" if role == "user" else "model"
+            contents.append({"role": mapped, "parts": [{"text": content}]})
+        body: Dict[str, Any] = {
+            "contents": contents,
+            "generationConfig": {
+                "temperature": temperature,
+                "maxOutputTokens": max_tokens,
+            },
+        }
+        if sys_prompt:
+            body["systemInstruction"] = {"parts": [{"text": sys_prompt.strip()}]}
+        headers = {"Content-Type": "application/json"}
+        with httpx.Client(timeout=timeout) as client:
+            with client.stream("POST", url, json=body, headers=headers) as r:
+                if r.status_code >= 400:
+                    text = r.read().decode("utf-8", "ignore")
+                    raise httpx.HTTPStatusError(text[:300], request=r.request, response=r)
+                for line in r.iter_lines():
+                    if not line:
+                        continue
+                    if isinstance(line, bytes):
+                        line = line.decode("utf-8", "ignore")
+                    if line.startswith("data:"):
+                        line = line[5:].strip()
+                    if not line or line == "[DONE]":
+                        continue
+                    try:
+                        obj = json.loads(line)
+                        for cand in obj.get("candidates", []):
+                            parts = cand.get("content", {}).get("parts", [])
+                            for p in parts:
+                                t = p.get("text")
+                                if t:
+                                    yield t
+                    except Exception:
+                        continue
+
+
+# Singleton getter
+_router_singleton: Optional[LLMRouter] = None
+
+
+def get_router() -> LLMRouter:
+    global _router_singleton
+    if _router_singleton is None:
+        _router_singleton = LLMRouter()
+    return _router_singleton

backend/planner.py

@@ -0,0 +1,130 @@
+"""
+Autonomous task planner.
+
+Produces structured action plans (list of dicts):
+    {"type": "shell"|"python"|"browser"|"git"|"deploy"|"note", ...}
+
+Strategy:
+1. Try LLM-based planning with strict JSON output.
+2. If LLM fails or returns invalid JSON, use heuristic fallback.
+3. Always produces non-empty plan.
+"""
+from __future__ import annotations
+
+import json
+import logging
+import re
+from typing import Any, Dict, List
+
+from .llm_router import get_router
+
+logger = logging.getLogger("planner")
+
+PLANNER_SYSTEM = (
+    "You are an autonomous AI Developer Agent's planner. "
+    "Decompose a user task into a JSON list of concrete actions. "
+    "Each action is an object with a 'type' field and supporting fields. "
+    "Supported types: shell (cmd), python (code), browser (action,url,...), "
+    "git (op,args), deploy (target), note (msg). "
+    "Return ONLY a JSON array, no commentary. Keep plan under 12 steps."
+)
+
+
+def plan_task(title: str, description: str, context: Dict[str, Any] | None = None) -> List[Dict[str, Any]]:
+    """Generate a concrete action plan."""
+    router = get_router()
+    user_prompt = (
+        f"TASK TITLE: {title}\n"
+        f"DESCRIPTION:\n{description}\n\n"
+        f"CONTEXT:\n{json.dumps(context or {}, indent=2)[:2000]}\n\n"
+        "Output a JSON array of action objects. Example:\n"
+        '[{"type":"shell","cmd":"echo hi"},{"type":"note","msg":"done"}]'
+    )
+    messages = [
+        {"role": "system", "content": PLANNER_SYSTEM},
+        {"role": "user", "content": user_prompt},
+    ]
+    try:
+        raw = router.chat(messages, temperature=0.1, max_tokens=1200, timeout=45.0)
+    except Exception as e:
+        logger.warning("Planner LLM call failed: %s", e)
+        raw = ""
+
+    plan = _parse_plan_json(raw)
+    if plan:
+        return plan
+    logger.info("Planner falling back to heuristic plan")
+    return heuristic_plan(title, description)
+
+
+def _parse_plan_json(raw: str) -> List[Dict[str, Any]]:
+    if not raw:
+        return []
+    # Try direct
+    try:
+        obj = json.loads(raw)
+        if isinstance(obj, list):
+            return [a for a in obj if isinstance(a, dict) and "type" in a]
+    except Exception:
+        pass
+    # Find first JSON array in text
+    m = re.search(r"\[[\s\S]*\]", raw)
+    if m:
+        try:
+            obj = json.loads(m.group(0))
+            if isinstance(obj, list):
+                return [a for a in obj if isinstance(a, dict) and "type" in a]
+        except Exception:
+            return []
+    return []
+
+
+def heuristic_plan(title: str, description: str) -> List[Dict[str, Any]]:
+    """Always-valid fallback plan."""
+    text = (title + "\n" + description).lower()
+    plan: List[Dict[str, Any]] = []
+
+    if any(k in text for k in ["deploy", "deployment", "huggingface", "hf space", "vercel"]):
+        plan.append({"type": "note", "msg": "Deployment task detected"})
+        if "vercel" in text:
+            plan.append({"type": "deploy", "target": "vercel"})
+        if "huggingface" in text or "hf" in text:
+            plan.append({"type": "deploy", "target": "huggingface"})
+        return plan
+
+    if any(k in text for k in ["git", "github", "commit", "push", "pr ", "pull request"]):
+        plan.append({"type": "git", "op": "status"})
+        plan.append({"type": "note", "msg": "GitHub task detected"})
+        return plan
+
+    if any(k in text for k in ["browser", "scrape", "navigate", "click", "open url", "http://", "https://"]):
+        url_match = re.search(r"https?://[\w\.\-/?=&%#]+", description)
+        url = url_match.group(0) if url_match else "https://example.com"
+        plan.append({"type": "browser", "action": "navigate", "url": url})
+        plan.append({"type": "browser", "action": "screenshot"})
+        return plan
+
+    if any(k in text for k in ["run python", "python script", "execute python"]):
+        plan.append({"type": "python", "code": "print('Hello from AI Developer Agent')"})
+        return plan
+
+    if any(k in text for k in ["install", "pip ", "npm "]):
+        # Try to extract a package name
+        m = re.search(r"(?:install|add)\s+([\w\.\-]+)", text)
+        pkg = m.group(1) if m else ""
+        if "npm" in text:
+            plan.append({"type": "shell", "cmd": f"npm install {pkg}".strip()})
+        else:
+            plan.append({"type": "shell", "cmd": f"pip install {pkg}".strip()})
+        return plan
+
+    # Generic fallback: echo the task back as an inspection action.
+    plan.append({"type": "note", "msg": f"Plan-fallback for: {title}"})
+    plan.append({"type": "shell", "cmd": "uname -a && python3 --version && node --version 2>/dev/null || true"})
+    return plan
+
+
+def repair_plan(error_category: str, detail: str = "") -> List[Dict[str, Any]]:
+    from .classifier import ErrorClass
+    from .repair import repair_actions
+    return repair_actions(ErrorClass(category=error_category, detail=detail, suggested_fix=""))

backend/repair.py

@@ -0,0 +1,70 @@
+"""
+Self-repair engine: converts an ErrorClass into actionable shell commands.
+"""
+from __future__ import annotations
+
+from typing import List
+from .classifier import ErrorClass
+
+
+def repair_actions(err: ErrorClass) -> List[dict]:
+    """Return list of structured actions (type=shell|git|python)."""
+    actions: List[dict] = []
+    c = err.category
+
+    if c == "missing_python_module":
+        # err.detail might be 'ModuleNotFoundError: No module named "x"'
+        import re
+        m = re.search(r"['\"]([\w\.\-]+)['\"]", err.detail)
+        pkg = m.group(1) if m else err.detail.split()[-1].strip("'\"")
+        actions.append({"type": "shell", "cmd": f"pip install --no-cache-dir {pkg}"})
+
+    elif c == "missing_pip":
+        actions.append({"type": "shell", "cmd": "python -m ensurepip --upgrade || apt-get install -y python3-pip"})
+
+    elif c == "missing_node":
+        actions.append({"type": "shell", "cmd": "apt-get update && apt-get install -y nodejs npm"})
+
+    elif c == "npm_failure":
+        actions.append({"type": "shell", "cmd": "rm -rf node_modules package-lock.json && npm install"})
+
+    elif c == "missing_playwright":
+        actions.append({"type": "shell", "cmd": "pip install playwright && python -m playwright install --with-deps chromium"})
+
+    elif c == "playwright_browsers_missing":
+        actions.append({"type": "shell", "cmd": "python -m playwright install chromium"})
+
+    elif c == "playwright_missing_deps":
+        actions.append({"type": "shell", "cmd": "python -m playwright install-deps chromium"})
+
+    elif c == "git_identity_missing":
+        actions.append({"type": "shell", "cmd": "git config --global user.email 'ai-developer@genspark.ai' && git config --global user.name 'AI Developer Agent'"})
+
+    elif c == "not_a_git_repo":
+        actions.append({"type": "shell", "cmd": "git init"})
+
+    elif c == "git_auth_failed":
+        actions.append({"type": "note", "msg": "ensure GITHUB_TOKEN is set and remote uses token URL"})
+
+    elif c == "greenlet_build_failure":
+        actions.append({"type": "shell", "cmd": "pip install --upgrade pip setuptools wheel && pip install greenlet --only-binary :all:"})
+
+    elif c == "python_build_failure":
+        actions.append({"type": "shell", "cmd": "apt-get install -y build-essential python3-dev && pip install --upgrade pip setuptools wheel"})
+
+    elif c == "python_version_mismatch":
+        actions.append({"type": "note", "msg": "Need Python 3.11 - check Dockerfile base image"})
+
+    elif c == "network_failure":
+        actions.append({"type": "sleep", "seconds": 5})
+
+    elif c == "rate_limited":
+        actions.append({"type": "sleep", "seconds": 30})
+
+    elif c == "auth_failure":
+        actions.append({"type": "note", "msg": "rotate API key"})
+
+    elif c == "python_exception":
+        actions.append({"type": "note", "msg": "no automatic repair; will retry once"})
+
+    return actions

backend/requirements.txt

@@ -0,0 +1,9 @@
+fastapi==0.115.0
+uvicorn[standard]==0.30.6
+pydantic==2.9.2
+httpx==0.27.2
+python-multipart==0.0.9
+sse-starlette==2.1.3
+e2b-code-interpreter==1.0.4
+playwright==1.47.0
+greenlet==3.0.3

backend/retry.py

@@ -0,0 +1,45 @@
+"""
+Retry wrapper - executes a callable with exponential backoff and
+optional classification-based repair callback.
+"""
+from __future__ import annotations
+
+import time
+import logging
+from typing import Callable, Optional, Any
+
+from .classifier import classify
+
+logger = logging.getLogger("retry")
+
+
+def retry_call(
+    fn: Callable[[], Any],
+    max_attempts: int = 3,
+    base_delay: float = 1.5,
+    on_error: Optional[Callable[[Exception, int], None]] = None,
+    repair_cb: Optional[Callable[[str], None]] = None,
+) -> Any:
+    last_exc: Optional[Exception] = None
+    for attempt in range(1, max_attempts + 1):
+        try:
+            return fn()
+        except Exception as e:
+            last_exc = e
+            logger.warning("retry attempt %s failed: %s", attempt, e)
+            if on_error:
+                try:
+                    on_error(e, attempt)
+                except Exception:
+                    pass
+            if repair_cb:
+                try:
+                    err_class = classify(str(e))
+                    if err_class:
+                        repair_cb(err_class.category)
+                except Exception:
+                    pass
+            if attempt < max_attempts:
+                time.sleep(base_delay * (2 ** (attempt - 1)))
+    assert last_exc is not None
+    raise last_exc

backend/tasks.py

@@ -0,0 +1,254 @@
+"""
+SQLite task persistence.
+
+Tables:
+    tasks(id, title, description, state, created_at, updated_at, payload_json)
+    task_events(id, task_id, ts, kind, message)
+    retries(id, task_id, attempt, error, ts)
+    deployments(id, task_id, target, url, status, ts)
+    sandboxes(id, task_id, sandbox_id, status, ts)
+    provider_usage(id, provider, ts, ok, latency_ms, error)
+    checkpoints(id, task_id, step_index, state_json, ts)
+"""
+from __future__ import annotations
+
+import json
+import os
+import sqlite3
+import threading
+import time
+import uuid
+from typing import Any, Dict, List, Optional
+
+DB_PATH = os.getenv("TASKS_DB_PATH", os.path.join(os.path.dirname(__file__), "tasks.db"))
+_LOCK = threading.RLock()
+
+SCHEMA = """
+CREATE TABLE IF NOT EXISTS tasks (
+    id TEXT PRIMARY KEY,
+    title TEXT,
+    description TEXT,
+    state TEXT,
+    created_at REAL,
+    updated_at REAL,
+    payload_json TEXT
+);
+CREATE TABLE IF NOT EXISTS task_events (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    task_id TEXT,
+    ts REAL,
+    kind TEXT,
+    message TEXT
+);
+CREATE INDEX IF NOT EXISTS idx_task_events_task ON task_events(task_id);
+CREATE TABLE IF NOT EXISTS retries (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    task_id TEXT,
+    attempt INTEGER,
+    error TEXT,
+    ts REAL
+);
+CREATE TABLE IF NOT EXISTS deployments (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    task_id TEXT,
+    target TEXT,
+    url TEXT,
+    status TEXT,
+    ts REAL
+);
+CREATE TABLE IF NOT EXISTS sandboxes (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    task_id TEXT,
+    sandbox_id TEXT,
+    status TEXT,
+    ts REAL
+);
+CREATE TABLE IF NOT EXISTS provider_usage (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    provider TEXT,
+    ts REAL,
+    ok INTEGER,
+    latency_ms REAL,
+    error TEXT
+);
+CREATE TABLE IF NOT EXISTS checkpoints (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    task_id TEXT,
+    step_index INTEGER,
+    state_json TEXT,
+    ts REAL
+);
+"""
+
+# Valid task states
+TASK_STATES = [
+    "queued", "planning", "thinking", "executing",
+    "repairing", "retrying", "deploying", "completed", "failed",
+]
+
+
+def _conn() -> sqlite3.Connection:
+    c = sqlite3.connect(DB_PATH, check_same_thread=False, timeout=30.0)
+    c.row_factory = sqlite3.Row
+    c.execute("PRAGMA journal_mode=WAL")
+    return c
+
+
+def init_db() -> None:
+    with _LOCK:
+        c = _conn()
+        try:
+            c.executescript(SCHEMA)
+            c.commit()
+        finally:
+            c.close()
+
+
+# ---------------------------------------------------------------------------
+# Task CRUD
+# ---------------------------------------------------------------------------
+def create_task(title: str, description: str, payload: Optional[Dict[str, Any]] = None) -> str:
+    task_id = uuid.uuid4().hex[:12]
+    now = time.time()
+    with _LOCK:
+        c = _conn()
+        try:
+            c.execute(
+                "INSERT INTO tasks(id, title, description, state, created_at, updated_at, payload_json) VALUES (?,?,?,?,?,?,?)",
+                (task_id, title, description, "queued", now, now, json.dumps(payload or {})),
+            )
+            c.commit()
+        finally:
+            c.close()
+    log_event(task_id, "create", f"Task created: {title}")
+    return task_id
+
+
+def update_state(task_id: str, state: str) -> None:
+    with _LOCK:
+        c = _conn()
+        try:
+            c.execute("UPDATE tasks SET state=?, updated_at=? WHERE id=?", (state, time.time(), task_id))
+            c.commit()
+        finally:
+            c.close()
+    log_event(task_id, "state", state)
+
+
+def get_task(task_id: str) -> Optional[Dict[str, Any]]:
+    with _LOCK:
+        c = _conn()
+        try:
+            row = c.execute("SELECT * FROM tasks WHERE id=?", (task_id,)).fetchone()
+            return dict(row) if row else None
+        finally:
+            c.close()
+
+
+def list_tasks(limit: int = 50) -> List[Dict[str, Any]]:
+    with _LOCK:
+        c = _conn()
+        try:
+            rows = c.execute(
+                "SELECT * FROM tasks ORDER BY updated_at DESC LIMIT ?", (limit,)
+            ).fetchall()
+            return [dict(r) for r in rows]
+        finally:
+            c.close()
+
+
+def log_event(task_id: str, kind: str, message: str) -> None:
+    with _LOCK:
+        c = _conn()
+        try:
+            c.execute(
+                "INSERT INTO task_events(task_id, ts, kind, message) VALUES (?,?,?,?)",
+                (task_id, time.time(), kind, message[:8000]),
+            )
+            c.commit()
+        finally:
+            c.close()
+
+
+def get_events(task_id: str, since_id: int = 0, limit: int = 1000) -> List[Dict[str, Any]]:
+    with _LOCK:
+        c = _conn()
+        try:
+            rows = c.execute(
+                "SELECT * FROM task_events WHERE task_id=? AND id>? ORDER BY id ASC LIMIT ?",
+                (task_id, since_id, limit),
+            ).fetchall()
+            return [dict(r) for r in rows]
+        finally:
+            c.close()
+
+
+def record_retry(task_id: str, attempt: int, error: str) -> None:
+    with _LOCK:
+        c = _conn()
+        try:
+            c.execute(
+                "INSERT INTO retries(task_id, attempt, error, ts) VALUES (?,?,?,?)",
+                (task_id, attempt, error[:4000], time.time()),
+            )
+            c.commit()
+        finally:
+            c.close()
+
+
+def record_deployment(task_id: str, target: str, url: str, status: str) -> None:
+    with _LOCK:
+        c = _conn()
+        try:
+            c.execute(
+                "INSERT INTO deployments(task_id, target, url, status, ts) VALUES (?,?,?,?,?)",
+                (task_id, target, url, status, time.time()),
+            )
+            c.commit()
+        finally:
+            c.close()
+
+
+def save_checkpoint(task_id: str, step_index: int, state: Dict[str, Any]) -> None:
+    with _LOCK:
+        c = _conn()
+        try:
+            c.execute(
+                "INSERT INTO checkpoints(task_id, step_index, state_json, ts) VALUES (?,?,?,?)",
+                (task_id, step_index, json.dumps(state)[:64000], time.time()),
+            )
+            c.commit()
+        finally:
+            c.close()
+
+
+def latest_checkpoint(task_id: str) -> Optional[Dict[str, Any]]:
+    with _LOCK:
+        c = _conn()
+        try:
+            row = c.execute(
+                "SELECT * FROM checkpoints WHERE task_id=? ORDER BY id DESC LIMIT 1", (task_id,)
+            ).fetchone()
+            if not row:
+                return None
+            d = dict(row)
+            d["state"] = json.loads(d["state_json"] or "{}")
+            return d
+        finally:
+            c.close()
+
+
+def record_provider_usage(provider: str, ok: bool, latency_ms: float, error: str = "") -> None:
+    with _LOCK:
+        c = _conn()
+        try:
+            c.execute(
+                "INSERT INTO provider_usage(provider, ts, ok, latency_ms, error) VALUES (?,?,?,?,?)",
+                (provider, time.time(), 1 if ok else 0, latency_ms, error[:1000]),
+            )
+            c.commit()
+        finally:
+            c.close()
+
+
+init_db()