""" FastAPI application for the hardened NovaTech OpenEnv environment. """ from __future__ import annotations import os from typing import Any, Dict, Optional from fastapi import Body, FastAPI, HTTPException, Query, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, RedirectResponse from pydantic import BaseModel import uvicorn from env.environment import DEBUG_STATE_ENABLED, store from env.models import Action, Observation, Reward app = FastAPI( title="NovaTech Incident Command", description="Seeded, session-safe OpenEnv environment for incident response under partial observability.", version="3.0.0", ) app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) class ResetRequest(BaseModel): task_id: str = "easy" seed: Optional[int] = None class StepResponse(BaseModel): observation: Dict[str, Any] reward: Dict[str, Any] done: bool info: Dict[str, Any] def _root_payload() -> Dict[str, Any]: return { "name": "NovaTech Incident Command", "version": "3.0.0", "debug_state_enabled": DEBUG_STATE_ENABLED, "endpoints": { "POST /reset": "Create an episode and return the initial observation.", "POST /step": "Apply an action using a session_id.", "GET /state": "Return public, non-leaking session state.", "GET /health": "Liveness probe.", }, "action_schema": Action.model_json_schema(), "observation_schema": Observation.model_json_schema(), "reward_schema": Reward.model_json_schema(), } @app.get("/") def root(request: Request): if "text/html" in (request.headers.get("accept") or "").lower(): return RedirectResponse(url="/playground", status_code=307) return _root_payload() @app.get("/playground", response_class=HTMLResponse) def playground() -> str: return """ NovaTech Incident Command
Live OpenEnv Ops Console

NovaTech Incident Command

Run a full incident workflow from one place: shape your search space, surface the most credible evidence, lock in a structured causal hypothesis, and pressure-test the final report before submission.

Mode
Seeded, Partial
Sessions
None
Last Reward
-

Mission Control

Start a seeded episode, track session health, and jump into common action patterns without writing boilerplate from scratch.

No active session yet. Reset an episode to begin.
Session ID
-
Task
-
Step
-
Done
-
Quick Templates

Action Composer

Work directly against the typed API. The current session id is auto-injected when missing, so you can focus on the action payload itself.

Tip: keep evidence grounded. The grader now rejects unseen log ids and penalizes contradictions across service, impact, and containment.

Situation Room

Read the live incident summary, then switch between raw JSON and a cleaner operator view to understand what changed after each step.

Incident Snapshot

No active incident briefing yet.

Operational Constraints

  • Reset an episode to load task-specific constraints.

Suspected Services

None 
No data yet.
""" @app.get("/health") def health() -> Dict[str, str]: return {"status": "ok"} @app.post("/reset", response_model=Dict[str, Any]) def reset(request: Optional[ResetRequest] = Body(default=None)) -> Dict[str, Any]: try: payload = request or ResetRequest() observation = store.reset(task_id=payload.task_id, seed=payload.seed) except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc return observation.model_dump() @app.post("/step", response_model=StepResponse) def step(action: Action) -> StepResponse: try: observation, reward, done, info = store.step(action) except (RuntimeError, ValueError) as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc return StepResponse( observation=observation.model_dump(), reward=reward.model_dump(), done=done, info=info, ) @app.get("/state", response_model=Dict[str, Any]) def state(session_id: Optional[str] = Query(default=None)) -> Dict[str, Any]: try: return store.public_state(session_id=session_id) except RuntimeError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc @app.get("/debug_state", response_model=Dict[str, Any]) def debug_state(session_id: Optional[str] = Query(default=None)) -> Dict[str, Any]: try: return store.debug_state(session_id=session_id) except PermissionError as exc: raise HTTPException(status_code=403, detail=str(exc)) from exc except RuntimeError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc def main() -> None: port = int(os.getenv("PORT", "7860")) uvicorn.run("app:app", host="0.0.0.0", port=port)