Spaces:
Build error
Build error
File size: 2,372 Bytes
7a529e5 | 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 | import os
import uvicorn
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from pydantic import ValidationError
from .environment import AEGISEnvironment
from .models import AEGISAction
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Initialize environment on startup, not at import time."""
scenario_dir = os.getenv("SCENARIO_DIR", None)
worker_mode = os.getenv("WORKER_MODE", "scripted")
memory_enabled = os.getenv("MEMORY_ENABLED", "true").lower() == "true"
seed = int(os.getenv("SEED", "42"))
env = AEGISEnvironment(
scenario_dir=scenario_dir, # type: ignore
worker_mode=worker_mode,
memory_enabled=memory_enabled,
seed=seed,
)
app.state.env = env
yield
app = FastAPI(
title="AEGIS-Env Server",
description="OpenEnv backend for RL model oversight.",
lifespan=lifespan,
)
@app.get("/")
async def root():
return {"message": "AEGIS-ENV is running. Use POST /reset and POST /step."}
@app.get("/health")
async def health():
return {"status": "ok"}
@app.post("/reset")
async def reset_env(request: Request):
"""Starts a new episode, generating scenario logs and clearing limits."""
env = request.app.state.env
obs, info = env.reset()
return {"observation": obs, "info": info}
@app.post("/step")
async def step_env(request: Request):
"""Layer-1 FIX: Accepts any dict body — validates internally, never returns 422.
Invalid actions yield reward=0.0 via the format gate instead of crashing.
"""
env = request.app.state.env
# Parse body leniently
try:
body = await request.json()
except Exception:
body = {}
# Internal validation — mirrors environment.py step() gate
try:
validated = AEGISAction(**body)
action_dict = validated.model_dump()
action_dict["__valid__"] = True
except (ValidationError, TypeError):
action_dict = {
"decision": "ALLOW",
"confidence": 0.5,
"violation_type": "none",
"explanation": "",
"__valid__": False, # format gate → 0.0 reward, episode continues
}
obs, reward, done, info = env.step(action_dict)
return {"observation": obs, "reward": reward, "done": done, "info": info}
|