fit-boilerplate-wox/web/_default_fastapi_htmx/app/main.py

"""Web layer default — FastAPI + Jinja2 + HTMX.

Estructura mínima viable para demos / control planes. Incluye:
- /              landing con embed WxO
- /traces        timeline observable de tool calls
- /api/traces    endpoint que recibe trazas del decorator @observable_tool
- /health        healthcheck para Coolify/Traefik

Reemplazá lo que necesites o usá esto como base para extender.
"""
from __future__ import annotations

import json
import os
import sqlite3
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates


BASE_DIR = Path(__file__).resolve().parent
TEMPLATES_DIR = BASE_DIR / "templates"
STATIC_DIR = BASE_DIR / "static"
TRACES_DB = os.environ.get("TRACES_DB_PATH", "/data/traces.db")

# WxO embed config (capturar tras deploy)
WXO_AGENT_ID = os.environ.get("WXO_AGENT_ID", "")
WXO_INSTANCE_URL = os.environ.get("WXO_INSTANCE_URL", "")


def _init_db():
    """Crea la tabla de trazas si no existe."""
    Path(TRACES_DB).parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(TRACES_DB)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS traces (
            trace_id TEXT PRIMARY KEY,
            tool TEXT,
            domain TEXT,
            agent_caller TEXT,
            correlation_id TEXT,
            started_at TEXT,
            duration_ms INTEGER,
            payload TEXT
        )
    """)
    conn.execute("CREATE INDEX IF NOT EXISTS ix_started ON traces (started_at)")
    conn.execute("CREATE INDEX IF NOT EXISTS ix_agent ON traces (agent_caller)")
    conn.commit()
    conn.close()


@asynccontextmanager
async def lifespan(app: FastAPI):
    _init_db()
    yield


app = FastAPI(title="Boilerplate Web Layer", lifespan=lifespan)
templates = Jinja2Templates(directory=str(TEMPLATES_DIR))

if STATIC_DIR.exists():
    app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")


# ─── Routes ─────────────────────────────────────────────────────────────────

@app.get("/health")
def health():
    return {"status": "ok"}


@app.get("/", response_class=HTMLResponse)
def landing(request: Request):
    return templates.TemplateResponse("index.html", {
        "request": request,
        "wxo_agent_id": WXO_AGENT_ID,
        "wxo_instance_url": WXO_INSTANCE_URL,
    })


@app.get("/traces", response_class=HTMLResponse)
def traces_page(request: Request):
    return templates.TemplateResponse("traces.html", {"request": request})


@app.get("/api/traces/recent")
def api_traces_recent(since: str | None = None, limit: int = 100):
    """JSON endpoint — devuelve las últimas N trazas, opcionalmente desde un ISO timestamp."""
    conn = sqlite3.connect(TRACES_DB)
    if since:
        rows = conn.execute(
            "SELECT payload FROM traces WHERE started_at > ? ORDER BY started_at DESC LIMIT ?",
            (since, limit)
        ).fetchall()
    else:
        rows = conn.execute(
            "SELECT payload FROM traces ORDER BY started_at DESC LIMIT ?",
            (limit,)
        ).fetchall()
    conn.close()
    return {"traces": [json.loads(r[0]) for r in rows]}


@app.post("/api/traces")
async def api_traces_post(request: Request):
    """Endpoint que recibe trazas del decorator @observable_tool con TRACE_SINK=http."""
    trace: dict[str, Any] = await request.json()
    conn = sqlite3.connect(TRACES_DB)
    conn.execute(
        "INSERT OR REPLACE INTO traces VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
        (
            trace.get("trace_id"), trace.get("tool"), trace.get("domain"),
            trace.get("agent_caller"), trace.get("correlation_id"),
            trace.get("started_at"), trace.get("duration_ms"),
            json.dumps(trace),
        ),
    )
    conn.commit()
    conn.close()
    return {"ok": True}


@app.get("/partials/timeline", response_class=HTMLResponse)
def partial_timeline(request: Request):
    """HTMX partial — la tabla de trazas para refrescar via hx-trigger."""
    conn = sqlite3.connect(TRACES_DB)
    rows = conn.execute(
        "SELECT payload FROM traces ORDER BY started_at DESC LIMIT 50"
    ).fetchall()
    conn.close()
    traces = [json.loads(r[0]) for r in rows]
    return templates.TemplateResponse("_timeline_rows.html", {
        "request": request, "traces": traces,
    })