"""FastAPI application for the DNS-Env OpenEnv environment. Exposes the DNS zone-file debugging environment over HTTP so that remote agents (or a HuggingFace Spaces front-end) can interact with it via the standard OpenEnv ``/reset``, ``/step``, ``/state`` endpoints. """ from __future__ import annotations from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse # --------------------------------------------------------------------------- # Dual-import pattern -- works as a sub-package *and* when run directly. # --------------------------------------------------------------------------- try: from .dns_environment import DNSEnvironment except ImportError: from dns_environment import DNSEnvironment # type: ignore[no-redef] try: from ..models import Action, Observation, State except ImportError: from models import Action, Observation, State # type: ignore[no-redef] try: from .tasks import TASK_IDS except ImportError: from tasks import TASK_IDS # type: ignore[no-redef] # --------------------------------------------------------------------------- # Application setup # --------------------------------------------------------------------------- app = FastAPI( title="DNS-Env", description="DNS Zone File Debugging Environment for OpenEnv", version="0.1.0", ) # Allow all origins so HuggingFace Spaces (and other frontends) can call us. app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # --------------------------------------------------------------------------- # Session management # --------------------------------------------------------------------------- environments: dict[str, DNSEnvironment] = {} def get_env(session_id: str = "default") -> DNSEnvironment: """Return the environment for *session_id*, creating one if needed.""" if session_id not in environments: environments[session_id] = DNSEnvironment() return environments[session_id] # --------------------------------------------------------------------------- # Endpoints # --------------------------------------------------------------------------- _LANDING_HTML = """\ DNS-Env | OpenEnv
Running

DNS-Env

DNS Zone File Debugging Environment for OpenEnv — Train AI agents to diagnose and fix real-world DNS misconfigurations.

openenv reinforcement-learning dns infrastructure devops
🎯 Tasks

fix_single_record

Easy

Fix broken records in example.com — missing trailing dots, invalid IPs, malformed MX. The classic DNS gotchas that cause real outages.

configure_mail

Medium

Set up complete email delivery for acme.co — MX records, SPF authorization, DMARC policy. 80% of domains get this wrong.

debug_delegation

Hard

Repair broken NS delegation across parent.org and dev.parent.org — fix glue records, NS consistency, SOA serials.

⚖ Reward Function
Structural validity
30%
Resolution checks
50%
No regressions
20%

Deterministic grading via zone-file parsing and record matching. Partial credit for incremental progress.

🔌 API Endpoints
MethodEndpointDescription
GET/healthLiveness probe
POST/resetStart a new episode with a task
POST/stepExecute an action (view, edit, dig, submit)
GET/stateCurrent episode state
GET/tasksList available tasks
GET/docsInteractive Swagger UI
⚡ Live Demo

Click to run a live interaction against this environment:

✅ OpenEnv Spec Compliance
✓ step(action) ✓ reset() ✓ state() ✓ openenv.yaml ✓ Typed Pydantic Models ✓ Docker Container ✓ 3 Graded Tasks ✓ Baseline Inference
""" @app.get("/", response_class=HTMLResponse) async def root(): """Landing page with environment info and live demo.""" return _LANDING_HTML @app.get("/health") async def health() -> dict: """Liveness / readiness probe.""" return {"status": "ok"} @app.post("/reset") async def reset(request: dict = {}) -> dict: # noqa: B006 -- mutable default is fine for FastAPI body """Reset the environment and start a new episode. Optional body fields -------------------- session_id : str Identifies the caller's session (default ``"default"``). seed : int | None RNG seed for reproducibility. episode_id : str | None Caller-supplied episode identifier. options : dict May contain ``task_id`` (e.g. ``"fix_single_record"``). """ session_id: str = request.get("session_id", "default") env = get_env(session_id) obs: Observation = env.reset( seed=request.get("seed"), episode_id=request.get("episode_id"), options=request.get("options", {}), ) return obs.model_dump() @app.post("/step") async def step(request: dict) -> dict: """Execute one agent action and return the observation. Body fields ----------- session_id : str Session identifier (default ``"default"``). action : dict Must contain ``command`` (str) and optionally ``args`` (dict) and ``metadata`` (dict). If the top-level dict already has a ``command`` key and no ``action`` wrapper, it is treated as the action directly for convenience. """ session_id: str = request.get("session_id", "default") # Accept either {"action": {…}} or a flat {command, args, …} body. action_data = request.get("action", request) action = Action(**action_data) env = get_env(session_id) obs: Observation = env.step(action) return obs.model_dump() @app.get("/state") async def get_state(session_id: str = "default") -> dict: """Return the current episode state (step count, task id, etc.).""" env = get_env(session_id) return env.state.model_dump() @app.get("/tasks") async def list_tasks() -> dict: """List the available task identifiers.""" return {"tasks": list(TASK_IDS)} # --------------------------------------------------------------------------- # Entry-point # --------------------------------------------------------------------------- def main() -> None: """Run the DNS-Env server.""" import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860) if __name__ == "__main__": main()