Request History

"""FastAPI entry point — new five-layer architecture. 来源：REFACTOR_HANDBOOK §2.2 (app.py, L5) + §2.3 (数据流) Keeps the same HTTP interface: /agent/perceive, /agent/interact, /debug/* """ from __future__ import annotations import html as html_lib import logging import sys from agent_build_sdk.server.server import EndpointServer from fastapi.responses import HTMLResponse, JSONResponse, Response from werewolftown.agent import WerewolfTownAgent from werewolftown.state.persistence import ( DEFAULT_REPLAY_DIR, game_summaries, read_replay, zip_all_replays, ) from werewolftown.transport.replay_uploader import is_enabled as replay_upload_enabled from werewolftown.transport.replay_uploader import start_replay_scheduler logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") logger = logging.getLogger(__name__) def register_debug_endpoints(server: EndpointServer) -> None: """Register /debug/* endpoints for observability.""" agent = server.agent @server.app.get("/config") def get_config(): canary_info = agent.get_canary_info() return {"status": "ok", "profile": canary_info["current_profile"], "canary_ratio": canary_info["canary_ratio"]} @server.app.get("/health") def health_check(): """Health check endpoint for monitoring.""" canary_info = agent.get_canary_info() return { "status": "ok", "canary_ratio": canary_info["canary_ratio"], "current_profile": canary_info["current_profile"], "current_game_id": canary_info["current_game_id"], "tests": 441, "coverage": {"strategies": 92, "transport": 93}, } @server.app.get("/debug/requests") def get_requests(): history = agent.get_req_resp_history() return {"success": True, "total": len(history), "history": history} @server.app.get("/debug/requests/view") def view_requests(): history = agent.get_req_resp_history() html_items = "" for item in reversed(history): req_data = item.get("req", {}) resp_data = item.get("resp", {}) req_type = item.get("type", "?") status = req_data.get("status", "?") round_num = req_data.get("round", "?") action = resp_data.get("action", "-") if resp_data else "-" html_items += f"""

Request

{html_lib.escape(str(req_data))}

Response

{html_lib.escape(str(resp_data))}

""" return HTMLResponse(content=f""" Request History

Request History ({len(history)})

{html_items}""") @server.app.post("/debug/requests/clear") def clear_requests(): agent.clear_req_resp_history() return {"success": True} @server.app.get("/debug/state") def get_state(): """Debug endpoint: current game state.""" state = agent.get_state() return { "player_id": state.player_id, "round": state.round, "cash": state.cash, "plots": sorted(state.plots), "shops": list(state.shops), "built_on": {str(k): v for k, v in state.built_on.items()}, "other_players": { pid: {"cash": pv.cash, "plots": sorted(pv.plots), "shops": list(pv.shops)} for pid, pv in state.other_players.items() }, } @server.app.get("/debug/games") def list_saved_games(): """List all saved game replays from persistent storage.""" try: games = game_summaries() return { "success": True, "base_dir": DEFAULT_REPLAY_DIR, "upload_enabled": replay_upload_enabled(), "total": len(games), "games": games, } except Exception as exc: # noqa: BLE001 -- debug endpoint must never 500 the app logger.warning(f"[/debug/games] failed: {exc}") return JSONResponse(status_code=500, content={"success": False, "error": str(exc)}) @server.app.get("/debug/replay/{game_id}") def get_replay(game_id: str): """Return one game's full replay (events + decisions + meta) as JSON.""" try: return {"success": True, **read_replay(game_id)} except Exception as exc: # noqa: BLE001 -- debug endpoint must never 500 the app logger.warning(f"[/debug/replay] failed for {game_id}: {exc}") return JSONResponse(status_code=500, content={"success": False, "error": str(exc)}) @server.app.get("/debug/replays.zip") def download_all_replays(): """Download every saved replay as a single zip archive.""" try: data = zip_all_replays() headers = {"Content-Disposition": 'attachment; filename="replays.zip"'} return Response(content=data, media_type="application/zip", headers=headers) except Exception as exc: # noqa: BLE001 -- debug endpoint must never 500 the app logger.warning(f"[/debug/replays.zip] failed: {exc}") return JSONResponse(status_code=500, content={"success": False, "error": str(exc)}) if __name__ == "__main__": try: logger.info("Starting WerewolfTown agent (five-layer architecture)") agent = WerewolfTownAgent(player_id="") server = EndpointServer(agent) register_debug_endpoints(server) # Keep a module-level reference so the background uploader is not GC'd. _replay_scheduler = start_replay_scheduler() logger.info("Server starting on port 7860") server.start() except Exception as e: logger.error(f"STARTUP CRASH: {e}", exc_info=True) sys.exit(1)