Spaces:
Configuration error
Configuration error
File size: 4,276 Bytes
7b2a69c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 | """
FastAPI HTTP server for CodeReview OpenEnv.
Exposes the environment as a REST API for agents to interact with.
"""
from __future__ import annotations
from typing import Any, Dict, Optional
from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from env.environment import CodeReviewEnv, TASK_SPECS
from env.models import Action, ReviewCategory, ReviewComment, Severity
# ---------------------------------------------------------------------------
# App setup
# ---------------------------------------------------------------------------
app = FastAPI(
title="CodeReview OpenEnv",
description="An OpenEnv-compliant AI training environment for Python code review.",
version="1.0.0",
)
# In-memory session store
SESSIONS: Dict[str, CodeReviewEnv] = {}
# ---------------------------------------------------------------------------
# Request / Response schemas
# ---------------------------------------------------------------------------
class ResetRequest(BaseModel):
task_id: str = "task_1_easy"
session_id: str = "default"
class StepRequest(BaseModel):
session_id: str = "default"
action: Dict[str, Any]
# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------
import os
@app.get("/", response_class=HTMLResponse)
def landing_page():
"""HTML landing page."""
template_path = os.path.join(os.path.dirname(__file__), "templates", "index.html")
try:
with open(template_path, "r", encoding="utf-8") as f:
return f.read()
except FileNotFoundError:
return "<html><body><h1>Error: templates/index.html not found.</h1></body></html>"
@app.get("/health")
def health():
"""Health check endpoint."""
return {"status": "ok"}
@app.get("/tasks")
def list_tasks():
"""Return specs for all available tasks."""
return {
task_id: spec.model_dump()
for task_id, spec in TASK_SPECS.items()
}
@app.post("/reset")
def reset(req: ResetRequest):
"""Start or restart an episode for the given task and session."""
if req.task_id not in TASK_SPECS:
raise HTTPException(
status_code=400,
detail=f"Unknown task_id '{req.task_id}'. Choose from: {list(TASK_SPECS.keys())}",
)
env = CodeReviewEnv(task_id=req.task_id)
obs = env.reset()
SESSIONS[req.session_id] = env
return {"observation": obs.model_dump(), "session_id": req.session_id}
@app.post("/step")
def step(req: StepRequest):
"""Submit an action for the given session."""
env = SESSIONS.get(req.session_id)
if env is None:
raise HTTPException(
status_code=404,
detail=f"Session '{req.session_id}' not found. Call /reset first.",
)
# Parse the action dict into an Action model
action_dict = req.action
comments = []
for c in action_dict.get("comments", []):
try:
comments.append(ReviewComment(
line=c.get("line"),
category=ReviewCategory(c.get("category", "bug")),
severity=Severity(c.get("severity", "medium")),
message=c.get("message", ""),
suggestion=c.get("suggestion"),
))
except Exception:
pass # skip malformed comments
action = Action(
comments=comments,
summary=action_dict.get("summary"),
submit=action_dict.get("submit", False),
)
try:
result = env.step(action)
except RuntimeError as e:
raise HTTPException(status_code=400, detail=str(e))
return {
"observation": result.observation.model_dump(),
"reward": result.reward.model_dump(),
"done": result.done,
"info": result.info,
}
@app.get("/state")
def get_state(session_id: str = Query(default="default")):
"""Return full serialisable state for the given session."""
env = SESSIONS.get(session_id)
if env is None:
raise HTTPException(
status_code=404,
detail=f"Session '{session_id}' not found. Call /reset first.",
)
return env.state().model_dump()
|