ahagestedt/apex-devops

1"""2Apex DevOps Corp QA Environment — FastAPI application.3 4Uses OpenEnv's create_app() factory + custom API endpoints + UI middleware.5"""6from __future__ import annotations7 8import json9from pathlib import Path10from typing import Any, Dict11 12from fastapi import Body, FastAPI, Request13from fastapi.responses import HTMLResponse, JSONResponse14from fastapi.staticfiles import StaticFiles15from starlette.middleware.base import BaseHTTPMiddleware16 17from openenv.core import create_app18from rl_qa_base.models import QAAction, QAObservation, QAState19from server.apex_devops_environment import ApexDevopsEnvironment20 21# Create the OpenEnv app22app: FastAPI = create_app(23    ApexDevopsEnvironment,24    QAAction,25    QAObservation,26    env_name="apex_devops_qa",27)28 29# Module-level instance for stateful custom endpoints30_env = ApexDevopsEnvironment()31 32# UI middleware — serve index.html at /, /ui, /web33_UI_DIR = Path(__file__).resolve().parent.parent / "ui"34 35 36class UIRedirectMiddleware(BaseHTTPMiddleware):37    async def dispatch(self, request: Request, call_next):38        if request.url.path in ("/", "/web", "/ui"):39            html_path = _UI_DIR / "index.html"40            if html_path.exists():41                return HTMLResponse(html_path.read_text(encoding="utf-8"))42        return await call_next(request)43 44 45app.add_middleware(UIRedirectMiddleware)46 47 48# --- Custom API Endpoints ---49 50 51@app.get("/api/scenarios")52async def list_scenarios(53    software: str = None, category: str = None, difficulty: str = None,54):55    """List all scenarios with optional filters."""56    scenarios = _env.get_scenario_list()57    if software:58        scenarios = [s for s in scenarios if s.get("software", "").lower() == software.lower()]59    if category:60        scenarios = [s for s in scenarios if s.get("category", "").lower() == category.lower()]61    if difficulty:62        scenarios = [s for s in scenarios if s.get("difficulty", "").lower() == difficulty.lower()]63    return JSONResponse(content={64        "enterprise": "Apex DevOps Corp",65        "domain": "dev-sim",66        "scenario_count": len(scenarios),67        "scenarios": scenarios,68    })69 70 71@app.get("/api/scenarios/{scenario_id}")72async def get_scenario(scenario_id: str):73    """Get details for a specific scenario."""74    for i, s in enumerate(_env.get_scenario_list()):75        if s.get("scenario_id") == scenario_id:76            return JSONResponse(content=s)77    return JSONResponse(content={"error": f"Scenario {scenario_id} not found"}, status_code=404)78 79 80@app.get("/api/info")81async def env_info():82    """Enterprise environment metadata."""83    return JSONResponse(content=_env.get_metadata())84 85 86@app.post("/api/reset")87async def custom_reset(request: Dict[str, Any] = Body(default={})):88    """Reset with scenario selection. Body: {"scenario_index": 0} or {"scenario_id": "MFG-QB-001"}"""89    scenario_index = request.get("scenario_index", 0)90    scenario_id = request.get("scenario_id")91    if scenario_id:92        for i, s in enumerate(_env.get_scenario_list()):93            if s.get("scenario_id") == scenario_id:94                scenario_index = i95                break96    obs = _env.reset_with_scenario(scenario_index=scenario_index)97    d = obs.model_dump()98    return {"observation": d, "reward": d.get("reward"), "done": d.get("done", False)}99 100 101@app.post("/api/step")102async def custom_step(request: Dict[str, Any] = Body(default={})):103    """Stateful step: {"action_type": "GET", "resource": "ticket"} etc."""104    action = QAAction(**{k: v for k, v in request.items() if v is not None})105    obs = _env.step(action)106    d = obs.model_dump()107    return {"observation": d, "reward": d.get("reward"), "done": d.get("done", False)}108 109 110@app.get("/api/state")111async def custom_state():112    """Current environment state."""113    return _env.state.model_dump()114 115 116@app.get("/api/reward-schema")117async def reward_schema():118    """Reward component definitions."""119    from rl_qa_base.models import REWARD_COMPONENTS120    return JSONResponse(content=REWARD_COMPONENTS)121 122 123@app.get("/api/verifiers")124async def verifier_types():125    """Available verifier types."""126    from rl_qa_base.verifiers import VERIFIER_REGISTRY, WORKFLOW_CATEGORY_MAP127    return JSONResponse(content={128        "verifier_types": list(VERIFIER_REGISTRY.keys()),129        "workflow_mapping": WORKFLOW_CATEGORY_MAP,130    })131 132 133@app.post("/api/demo")134async def run_demo(request: Dict[str, Any] = Body(default={})):135    """Run auto-agent demo: reset + full episode, returns all steps."""136    scenario_index = request.get("scenario_index", 0)137    obs = _env.reset_with_scenario(scenario_index=scenario_index)138    steps = [{"step": 0, "action": "reset", "observation": obs.model_dump()}]139 140    demo_actions = [141        QAAction(action_type="GET", resource="ticket"),142        QAAction(action_type="GET", resource="customer_history"),143        QAAction(action_type="GET", resource="rubric"),144        QAAction(action_type="GET", resource="similar_cases"),145        QAAction(action_type="POST", evaluation_data={"score": 4, "confidence": 0.82, "reasoning": "Auto-demo evaluation", "safety_violation": False}),146        QAAction(action_type="FINISH"),147    ]148    for i, action in enumerate(demo_actions, 1):149        obs = _env.step(action)150        steps.append({"step": i, "action": action.model_dump(exclude_none=True), "observation": obs.model_dump()})151        if obs.done:152            break153    return JSONResponse(content={"steps": steps, "total_steps": len(steps)})154 155 156@app.post("/api/export")157async def export_report():158    """Export current episode as JSON report."""159    state = _env.state.model_dump()160    return JSONResponse(content={161        "report_type": "episode_evaluation",162        "enterprise": "Apex DevOps Corp",163        "domain": "dev-sim",164        "state": state,165    })166