Choose mode on reset: ticket for legacy triage or incident for EICC workflow.
In sandbox mode, optional drill mode injects deterministic mid-episode failures.
Use Get state to inspect available_actions; unavailable actions are auto-disabled after reset/step.
Step-by-step action
quick actions for current phase
Action Timeline
No actions yet. Run Reset, then Execute step.
observation
—
reward (response)
—
done
—
info
—
"""
# ── routes ───────────────────────────────────────────────────────────────────
@app.get(
"/",
response_class=HTMLResponse,
tags=["Interface"],
summary="Web Command Center",
)
def ui() -> HTMLResponse:
"""Browser debug UI (HTML). Use **GET /** in the browser; API clients should use JSON endpoints below."""
return HTMLResponse(content=_DEBUG_UI_HTML)
@app.get("/health", tags=["System"], summary="Health check")
async def health() -> dict[str, str]:
"""Return `{\"status\": \"ok\"}` when the server is up."""
return {"status": "ok"}
@app.post(
"/reset",
response_model=EnvResponse,
tags=["Environment"],
summary="Reset episode",
response_description="Initial observation, zero reward, `done` false.",
responses={
400: {"description": "Bad request (e.g. empty ticket pool for the given filter)."},
},
)
async def reset(
req: ResetRequest | None = None,
x_session_id: str | None = Header(default=None, alias="X-Session-ID"),
) -> dict[str, Any]:
"""Start a new episode with a deterministically selected ticket. Body is optional (`seed=0`, all difficulties)."""
if req is None:
req = ResetRequest()
session_id = _resolve_session_id(req.session_id, x_session_id)
env = await _get_env(session_id, create_if_missing=True)
try:
if isinstance(env, SandboxEnv):
result = await env.reset(
seed=req.seed,
difficulty=req.difficulty,
mode=req.mode,
drill_mode=req.drill_mode,
drill_seed=req.drill_seed,
)
else:
result = await env.reset(seed=req.seed, difficulty=req.difficulty, mode=req.mode)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return _result_to_dict(result, session_id=session_id)
@app.post(
"/step",
response_model=EnvResponse,
tags=["Environment"],
summary="Step environment",
responses={
422: {"description": "Action JSON does not match the documented action schemas."},
},
)
async def step(
req: StepRequest,
x_session_id: str | None = Header(default=None, alias="X-Session-ID"),
) -> dict[str, Any]:
"""Apply one agent action. Invalid **phase** actions still return HTTP 200 with a penalty in `reward`."""
session_id = _resolve_session_id(req.session_id, x_session_id)
env = await _get_env(session_id, create_if_missing=True)
try:
result = await env.step(req.action)
except Exception as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return _result_to_dict(result, session_id=session_id)
@app.get(
"/state",
response_model=EnvResponse,
tags=["Environment"],
summary="Get state",
)
async def state(
session_id: str | None = None,
x_session_id: str | None = Header(default=None, alias="X-Session-ID"),
) -> dict[str, Any]:
"""Read the current observation without advancing time. If `reset` was never called, `observation` is `null`."""
resolved_session_id = _resolve_session_id(session_id, x_session_id)
env = await _get_env(resolved_session_id, create_if_missing=True)
result = await env.state()
if result is None:
return {
"observation": None,
"reward": 0.0,
"done": False,
"info": {"session_id": resolved_session_id},
}
return _result_to_dict(result, session_id=resolved_session_id)
@app.post("/close", tags=["Environment"], summary="Close episode")
async def close(
session_id: str | None = None,
x_session_id: str | None = Header(default=None, alias="X-Session-ID"),
) -> dict[str, str]:
"""Release the current episode; call **reset** before stepping again."""
resolved_session_id = _resolve_session_id(session_id, x_session_id)
await _close_session(resolved_session_id)
return {"status": "closed", "session_id": resolved_session_id}
@app.post(
"/inference",
response_model=InferenceResponse,
tags=["System"],
summary="Run inference benchmark",
)
async def inference_endpoint() -> dict[str, Any]:
"""Execute the bundled LLM inference loop once. Requires server-side env (e.g. `HF_TOKEN`); see project README."""
import inference
buf = io.StringIO()
with redirect_stdout(buf):
await inference.run()
stdout = buf.getvalue()
score = 0.0
success = False
for line in stdout.strip().splitlines():
if line.startswith("[END]"):
for part in line.split():
if part.startswith("score="):
score = float(part.split("=")[1])
if part.startswith("success="):
success = part.split("=")[1] == "true"
return {"stdout": stdout, "score": score, "success": success}
# ── entrypoint ───────────────────────────────────────────────────────────────
def main() -> None:
import uvicorn
workers = int(os.environ.get("WEB_CONCURRENCY", "1"))
uvicorn.run("server.app:app", host="0.0.0.0", port=7860, workers=max(1, workers))
if __name__ == "__main__":
main()