Spaces:
Sleeping
Sleeping
File size: 3,947 Bytes
ef1fd72 | 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 | """Shared FastAPI application module for source and packaged entry points."""
from __future__ import annotations
import os
from typing import Any, Dict, Optional
from fastapi import FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from env import Action, CodeReviewEnv, Observation
app = FastAPI(
title="CodeReviewEnv",
description="OpenEnv environment for AI-driven code review and bug triage",
version="1.0.0",
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
_envs: Dict[str, CodeReviewEnv] = {}
def _get_env(task: str) -> CodeReviewEnv:
if task not in _envs:
_envs[task] = CodeReviewEnv(task_id=task)
return _envs[task]
class StepResult(BaseModel):
observation: Observation
reward: float
done: bool
info: Dict[str, Any]
class GradeResult(BaseModel):
task: str
score: float
found_issues: list
false_positives: int
final_decision: Optional[str]
steps_taken: int
@app.post("/reset", response_model=Observation)
def reset(task: str = Query("easy", description="Task difficulty: easy | medium | hard")):
"""Reset the environment and return the initial observation."""
env = _get_env(task)
try:
obs = env.reset()
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return obs
@app.post("/step", response_model=StepResult)
def step(
action: Action,
task: str = Query("easy", description="Task difficulty: easy | medium | hard"),
):
"""Execute one action and return observation, reward, done, info."""
env = _get_env(task)
if env._state is None:
raise HTTPException(status_code=400, detail="Call /reset first.")
try:
obs, reward, done, info = env.step(action)
except RuntimeError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return StepResult(observation=obs, reward=reward, done=done, info=info)
@app.get("/state")
def state(task: str = Query("easy")):
"""Return the full internal episode state."""
env = _get_env(task)
return env.state()
@app.get("/tasks")
def list_tasks():
"""List available task IDs with descriptions."""
return {
"tasks": [
{
"id": "easy",
"difficulty": "easy",
"description": "Find one obvious null-dereference bug in a user service.",
"num_issues": 1,
},
{
"id": "medium",
"difficulty": "medium",
"description": "Find an off-by-one boundary bug AND a missing thread-safety lock in a rate limiter.",
"num_issues": 2,
},
{
"id": "hard",
"difficulty": "hard",
"description": "Find a SQL injection vulnerability AND an unbounded memory leak in a report service.",
"num_issues": 2,
},
]
}
@app.post("/grade", response_model=GradeResult)
def grade(task: str = Query("easy")):
"""Compute normalised 0-1 score for the current (or just-finished) episode."""
env = _get_env(task)
if env._state is None:
raise HTTPException(status_code=400, detail="No episode to grade. Call /reset first.")
score = env.grade()
state_obj = env._state
return GradeResult(
task=task,
score=score,
found_issues=state_obj.found_issue_ids,
false_positives=state_obj.false_positives,
final_decision=state_obj.final_decision,
steps_taken=state_obj.step,
)
@app.get("/health")
def health():
return {"status": "ok", "env": "CodeReviewEnv", "version": "1.0.0"}
def main() -> None:
import uvicorn
port = int(os.getenv("PORT", 7860))
uvicorn.run(app, host="0.0.0.0", port=port)
|