109 lines
2.9 KiB
Python
109 lines
2.9 KiB
Python
"""FastAPI entry point for nvi.
|
|
|
|
Endpoints:
|
|
GET /healthz — liveness / readiness probe
|
|
POST /ask — submit a question, returns run_id
|
|
GET /runs/{run_id}/stream — SSE stream of node-state events
|
|
GET /runs/{run_id} — final state snapshot
|
|
|
|
Stubs only at this point — the Plan/Analyses/Tools/runtime are next.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import uuid
|
|
from contextlib import asynccontextmanager
|
|
from datetime import datetime, timezone
|
|
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import StreamingResponse
|
|
from pydantic import BaseModel
|
|
|
|
logger = logging.getLogger("nvi")
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")
|
|
|
|
|
|
# ── In-memory run store (single-process; replace with redis if we ever scale) ──
|
|
|
|
runs: dict[str, dict] = {}
|
|
RUN_TTL_SECONDS = 3600
|
|
RUN_CLEANUP_INTERVAL = 300
|
|
|
|
|
|
async def _cleanup_runs():
|
|
while True:
|
|
await asyncio.sleep(RUN_CLEANUP_INTERVAL)
|
|
now = datetime.now(timezone.utc).timestamp()
|
|
expired = [
|
|
rid for rid, r in runs.items()
|
|
if r["status"] in ("completed", "error")
|
|
and now - r.get("created_at", now) > RUN_TTL_SECONDS
|
|
]
|
|
for rid in expired:
|
|
del runs[rid]
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(_app: FastAPI):
|
|
task = asyncio.create_task(_cleanup_runs())
|
|
yield
|
|
task.cancel()
|
|
|
|
|
|
app = FastAPI(title="nvi", lifespan=lifespan)
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
|
|
@app.get("/healthz")
|
|
async def healthz():
|
|
return {"status": "ok", "runs_in_memory": len(runs)}
|
|
|
|
|
|
# ── Ask ──
|
|
|
|
class AskRequest(BaseModel):
|
|
question: str
|
|
|
|
|
|
@app.post("/ask")
|
|
async def ask(req: AskRequest):
|
|
run_id = uuid.uuid4().hex[:8]
|
|
now = datetime.now(timezone.utc)
|
|
runs[run_id] = {
|
|
"status": "pending",
|
|
"question": req.question,
|
|
"created_at": now.timestamp(),
|
|
"events": [],
|
|
}
|
|
# Real runtime invocation lands here once it exists.
|
|
logger.info("ask_received run_id=%s q=%r", run_id, req.question)
|
|
return {"run_id": run_id, "status": "pending"}
|
|
|
|
|
|
@app.get("/runs/{run_id}")
|
|
async def get_run(run_id: str):
|
|
if run_id not in runs:
|
|
raise HTTPException(404, detail=f"Run {run_id} not found")
|
|
return runs[run_id]
|
|
|
|
|
|
@app.get("/runs/{run_id}/stream")
|
|
async def stream_run(run_id: str):
|
|
if run_id not in runs:
|
|
raise HTTPException(404, detail=f"Run {run_id} not found")
|
|
|
|
async def event_stream():
|
|
# Placeholder: real implementation will tail the runtime's event queue.
|
|
yield f"event: hello\ndata: {{\"run_id\": \"{run_id}\"}}\n\n"
|
|
await asyncio.sleep(0.1)
|
|
yield "event: end\ndata: {}\n\n"
|
|
|
|
return StreamingResponse(event_stream(), media_type="text/event-stream")
|