fix: game service api block when a game is running

This commit is contained in:
2026-05-13 21:26:46 +08:00
parent 3c027eae0b
commit 79dccde963
7 changed files with 378 additions and 73 deletions
+16 -2
View File
@@ -11,6 +11,7 @@
- 支持盲注、四条街下注、弃牌、过牌、跟注、下注、加注、全下、边池和摊牌结算。
- 支持本地 Agent 和 HTTP Agent。
- 支持 Human Agent 和 OpenAI-compatible AI Agent 的终端过程输出。
- 游戏运行中可以并发查询状态;查询返回上一手完成后的稳定快照。
## 运行服务
@@ -51,6 +52,12 @@ curl -X POST http://127.0.0.1:8000/games/demo/hands/run \
curl http://127.0.0.1:8000/games/demo
```
也可以使用单数别名:
```bash
curl http://127.0.0.1:8000/game/demo
```
## HTTP Agent 协议
玩家配置可以使用远程 HTTP Agent:
@@ -62,12 +69,19 @@ curl http://127.0.0.1:8000/games/demo
"agent": {
"type": "http",
"endpoint": "http://127.0.0.1:9101",
"timeout_seconds": 10
"timeout_seconds": 10,
"game_update_timeout_seconds": 3,
"retries": 2,
"retry_backoff_seconds": 0.25
}
}
```
服务会向 `endpoint` 发送当前行动玩家的观察 JSON。Agent 返回:
服务会向 `endpoint + /game` 推送每手开始时的游戏快照,向 `endpoint + /act` 发送当前行动玩家的观察 JSON。`endpoint` 也可以传入历史形式的 `/act``/game` 后缀,服务会归一化为 base URL。
同一个 HTTP Agent endpoint 不能同时被不同游戏占用;后创建的游戏会返回错误。服务会给 HTTP Agent 请求自动重试,`/act` 重试仍失败时,规则引擎会按 `check > call > fold` 选择默认动作,避免整桌卡死。
Agent 返回:
```json
{"action": "call"}
+60
View File
@@ -0,0 +1,60 @@
import json
import unittest
from unittest.mock import patch
from urllib.error import URLError
from texas_holdem.agents import HttpAgent, normalise_http_agent_endpoint
class FakeResponse:
def __init__(self, payload: dict[str, object]) -> None:
self.payload = payload
def read(self) -> bytes:
return json.dumps(self.payload).encode("utf-8")
def __enter__(self) -> "FakeResponse":
return self
def __exit__(self, *args: object) -> None:
return None
class AgentTests(unittest.TestCase):
def test_normalise_http_agent_endpoint_accepts_action_or_game_paths(self) -> None:
self.assertEqual(
normalise_http_agent_endpoint("http://127.0.0.1:9101/act"),
"http://127.0.0.1:9101",
)
self.assertEqual(
normalise_http_agent_endpoint("http://127.0.0.1:9101/game/"),
"http://127.0.0.1:9101",
)
def test_http_agent_post_retries_and_sets_player_header(self) -> None:
calls = []
def fake_urlopen(request, timeout): # type: ignore[no-untyped-def]
calls.append((request, timeout))
if len(calls) == 1:
raise URLError("temporary")
return FakeResponse({"ok": True})
agent = HttpAgent(
"http://agent.test/act",
player_id="p1",
retries=1,
retry_backoff_seconds=0,
)
with patch("texas_holdem.agents.urlopen", fake_urlopen):
payload = agent._post_json("/game", {"game_id": "g1"}, timeout_seconds=2)
self.assertEqual(payload, {"ok": True})
self.assertEqual(len(calls), 2)
self.assertEqual(calls[1][0].headers["X-player-id"], "p1")
self.assertEqual(calls[1][1], 2)
if __name__ == "__main__":
unittest.main()
+74 -1
View File
@@ -1,8 +1,26 @@
import unittest
from threading import Event, Thread
from texas_holdem.agents import PokerAgent
from texas_holdem.models import Observation, PlayerAction
from texas_holdem.service import GameManager
class BlockingAgent(PokerAgent):
def __init__(self, entered: Event, release: Event) -> None:
self.entered = entered
self.release = release
def decide(self, observation: Observation) -> PlayerAction:
self.entered.set()
if not self.release.wait(timeout=5):
raise RuntimeError("test timed out waiting to release blocking agent")
for action in observation.legal_actions:
if action["action"] == "check":
return PlayerAction("check")
return PlayerAction("call")
class ServiceTests(unittest.TestCase):
def test_create_and_run_game(self) -> None:
manager = GameManager()
@@ -23,7 +41,62 @@ class ServiceTests(unittest.TestCase):
hands = manager.run_hands(game.game_id, count=1)
self.assertEqual(len(hands), 1)
self.assertEqual(manager.get_game("demo").to_dict()["hand_number"], 1)
self.assertEqual(manager.get_game_state("demo")["hand_number"], 1)
def test_get_game_state_does_not_block_during_run(self) -> None:
manager = GameManager()
entered = Event()
release = Event()
game = manager.create_game(
{
"game_id": "blocking",
"seed": 13,
"starting_stack": 200,
"small_blind": 5,
"big_blind": 10,
"players": [
{"id": "a", "type": "calling"},
{"id": "b", "type": "calling"},
],
}
)
manager.run_hands("blocking", count=1)
game.agents["a"] = BlockingAgent(entered, release)
thread = Thread(target=lambda: manager.run_hands("blocking", count=1))
thread.start()
self.assertTrue(entered.wait(timeout=2))
state = manager.get_game_state("blocking")
release.set()
thread.join(timeout=2)
self.assertFalse(thread.is_alive())
self.assertEqual(state["hand_number"], 1)
self.assertEqual(len(state["hands"]), 1)
def test_duplicate_http_agent_endpoint_is_rejected_across_active_games(self) -> None:
manager = GameManager()
payload = {
"starting_stack": 200,
"small_blind": 5,
"big_blind": 10,
"players": [
{
"id": "ai",
"agent": {
"type": "http",
"endpoint": "http://127.0.0.1:9101/act",
},
},
{"id": "b", "type": "calling"},
],
}
manager.create_game({"game_id": "g1", **payload})
with self.assertRaisesRegex(ValueError, "already belongs to game g1"):
manager.create_game({"game_id": "g2", **payload})
if __name__ == "__main__":
+106 -31
View File
@@ -2,10 +2,11 @@ from __future__ import annotations
import json
import sys
import time
from abc import ABC, abstractmethod
from random import Random
from typing import IO, Any
from urllib.error import URLError
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
from texas_holdem.human_io import clear_screen, prompt_action, render_observation
@@ -54,6 +55,27 @@ class CallingStationAgent(PokerAgent):
return PlayerAction("fold")
def normalise_http_agent_endpoint(raw: str) -> str:
"""Return the canonical base URL for an HTTP agent endpoint."""
url = raw.rstrip("/")
if url.endswith("/act"):
url = url[: -len("/act")]
if url.endswith("/game"):
url = url[: -len("/game")]
return url
def http_agent_endpoint_from_spec(spec: dict[str, Any]) -> str | None:
"""Extract the canonical HTTP endpoint from an agent spec, if present."""
agent_type = str(spec.get("type", "calling")).lower()
if agent_type != "http":
return None
endpoint = spec.get("endpoint")
if not endpoint:
raise ValueError("http agent requires an endpoint")
return normalise_http_agent_endpoint(str(endpoint))
class HttpAgent(PokerAgent):
"""Remote agent that talks to a base URL exposing ``/act`` and ``/game``.
@@ -66,28 +88,36 @@ class HttpAgent(PokerAgent):
ACT_PATH = "/act"
GAME_PATH = "/game"
def __init__(self, endpoint: str, timeout_seconds: float = 10.0) -> None:
self.base_url = self._normalise_base_url(endpoint)
def __init__(
self,
endpoint: str,
timeout_seconds: float = 10.0,
player_id: str | None = None,
game_update_timeout_seconds: float | None = None,
retries: int = 2,
retry_backoff_seconds: float = 0.25,
) -> None:
self.base_url = normalise_http_agent_endpoint(endpoint)
self.timeout_seconds = timeout_seconds
@staticmethod
def _normalise_base_url(raw: str) -> str:
"""Strip a trailing slash so URL joins do not produce double slashes.
Centralising this also tolerates the legacy "endpoint already points
at /act" mistake by chopping off a redundant ``/act`` suffix.
"""
url = raw.rstrip("/")
if url.endswith("/act"):
url = url[: -len("/act")]
return url
self.player_id = player_id
self.game_update_timeout_seconds = (
float(game_update_timeout_seconds)
if game_update_timeout_seconds is not None
else min(timeout_seconds, 3.0)
)
self.retries = max(0, retries)
self.retry_backoff_seconds = max(0.0, retry_backoff_seconds)
def _url(self, path: str) -> str:
"""Compose a full URL by joining the base with a path component."""
return f"{self.base_url}{path}"
def decide(self, observation: Observation) -> PlayerAction:
payload = self._post_json(self.ACT_PATH, observation.to_dict())
payload = self._post_json(
self.ACT_PATH,
observation.to_dict(),
timeout_seconds=self.timeout_seconds,
)
if not isinstance(payload, dict):
raise RuntimeError("agent endpoint must return a JSON object")
return PlayerAction.from_dict(payload)
@@ -100,30 +130,54 @@ class HttpAgent(PokerAgent):
only by way of the raised exception bubbling to the engine guard.
"""
try:
self._post_json(self.GAME_PATH, game_state)
self._post_json(
self.GAME_PATH,
game_state,
timeout_seconds=self.game_update_timeout_seconds,
)
except RuntimeError:
# ``/game`` is informational; treat any HTTP error as a benign
# drop rather than reraising and aborting the hand loop.
return None
def _post_json(self, path: str, payload: dict[str, Any]) -> Any:
def _post_json(
self,
path: str,
payload: dict[str, Any],
timeout_seconds: float,
) -> Any:
"""POST ``payload`` as JSON to ``base_url + path`` and return parsed body.
Extracted as a tiny helper so ``decide`` and ``on_game_update`` share
identical transport semantics (timeout, error wrapping, content-type).
"""
body = json.dumps(payload).encode("utf-8")
request = Request(
self._url(path),
data=body,
headers={"Content-Type": "application/json"},
method="POST",
)
try:
with urlopen(request, timeout=self.timeout_seconds) as response:
raw = response.read().decode("utf-8")
except (OSError, URLError) as exc:
raise RuntimeError(f"agent endpoint failed: {self._url(path)}") from exc
last_error: BaseException | None = None
raw = ""
for attempt in range(self.retries + 1):
request = Request(
self._url(path),
data=body,
headers=self._headers(),
method="POST",
)
try:
with urlopen(request, timeout=timeout_seconds) as response:
raw = response.read().decode("utf-8")
break
except HTTPError as exc:
detail = exc.read().decode("utf-8", errors="replace")
last_error = RuntimeError(
f"agent endpoint failed with HTTP {exc.code}: "
f"{self._url(path)} {detail}"
)
except (OSError, URLError) as exc:
last_error = exc
if attempt < self.retries and self.retry_backoff_seconds > 0:
time.sleep(self.retry_backoff_seconds * (2**attempt))
else:
raise RuntimeError(f"agent endpoint failed: {self._url(path)}") from last_error
if not raw:
return None
try:
@@ -133,6 +187,12 @@ class HttpAgent(PokerAgent):
f"agent endpoint returned invalid JSON: {self._url(path)}"
) from exc
def _headers(self) -> dict[str, str]:
headers = {"Content-Type": "application/json", "Connection": "close"}
if self.player_id:
headers["X-Player-Id"] = self.player_id
return headers
class HumanAgent(PokerAgent):
"""Interactive CLI agent for debugging and manual play.
@@ -189,7 +249,11 @@ class HumanAgent(PokerAgent):
return line.rstrip("\n")
def build_agent(spec: dict[str, Any], rng: Random | None = None) -> PokerAgent:
def build_agent(
spec: dict[str, Any],
rng: Random | None = None,
player_id: str | None = None,
) -> PokerAgent:
agent_type = str(spec.get("type", "calling")).lower()
if agent_type == "random":
return RandomAgent(rng)
@@ -199,7 +263,18 @@ def build_agent(spec: dict[str, Any], rng: Random | None = None) -> PokerAgent:
endpoint = spec.get("endpoint")
if not endpoint:
raise ValueError("http agent requires an endpoint")
return HttpAgent(str(endpoint), float(spec.get("timeout_seconds", 10.0)))
return HttpAgent(
str(endpoint),
timeout_seconds=float(spec.get("timeout_seconds", 10.0)),
player_id=player_id,
game_update_timeout_seconds=(
float(spec["game_update_timeout_seconds"])
if "game_update_timeout_seconds" in spec
else None
),
retries=int(spec.get("retries", 2)),
retry_backoff_seconds=float(spec.get("retry_backoff_seconds", 0.25)),
)
if agent_type in {"human", "cli", "interactive"}:
return HumanAgent()
raise ValueError(f"unknown agent type: {agent_type}")
+63 -18
View File
@@ -1,6 +1,8 @@
from __future__ import annotations
from copy import deepcopy
from random import Random
from threading import RLock
from time import time
from texas_holdem.agents import PokerAgent
@@ -54,6 +56,7 @@ class TableGame:
self.small_blind = small_blind
self.big_blind = big_blind
self.rng = rng or Random()
self.lock = RLock()
self.hand_number = 0
self.button_index: int | None = None
self.board = []
@@ -64,6 +67,7 @@ class TableGame:
# first hand that played under those stakes, which makes it trivial
# to reconstruct the schedule from the outside.
self.blind_history: list[BlindLevel] = []
self._completed_snapshot: dict[str, object] = self._to_dict_unlocked()
@property
def is_complete(self) -> bool:
@@ -83,6 +87,14 @@ class TableGame:
:attr:`blind_history` whenever it changes (including the very first
hand) so external observers can replay the schedule.
"""
with self.lock:
return self._run_hand_locked(small_blind=small_blind, big_blind=big_blind)
def _run_hand_locked(
self,
small_blind: int | None = None,
big_blind: int | None = None,
) -> HandSummary:
if self.is_complete:
raise GameComplete("game is complete")
@@ -151,6 +163,7 @@ class TableGame:
finished_at=time(),
)
self.hand_summaries.append(summary)
self._completed_snapshot = deepcopy(self._to_dict_unlocked())
return summary
def run_hands(
@@ -166,24 +179,47 @@ class TableGame:
with the first hand of this call; subsequent calls can raise them
again. Leaving them ``None`` keeps the current level unchanged.
"""
if max_hands <= 0:
raise ValueError("max_hands must be positive")
summaries = []
for _ in range(max_hands):
if self.is_complete:
break
# Only the first hand of the batch needs to apply the blind
# override; after that the engine reuses the stored values.
summaries.append(
self.run_hand(small_blind=small_blind, big_blind=big_blind)
)
small_blind = None
big_blind = None
if until_one_left and self.is_complete:
break
return summaries
with self.lock:
if max_hands <= 0:
raise ValueError("max_hands must be positive")
summaries = []
for _ in range(max_hands):
if self.is_complete:
break
# Only the first hand of the batch needs to apply the blind
# override; after that the engine reuses the stored values.
summaries.append(
self._run_hand_locked(
small_blind=small_blind,
big_blind=big_blind,
)
)
small_blind = None
big_blind = None
if until_one_left and self.is_complete:
break
return summaries
def to_dict(self) -> dict[str, object]:
with self.lock:
return self._to_dict_unlocked()
def snapshot_completed(self) -> dict[str, object]:
"""Return a stable snapshot from the latest completed hand boundary.
If a hand is currently running under ``self.lock``, this method does
not block. It returns the most recent completed hand summary and
stacks captured in memory, which is exactly what status endpoints
need while a long-running HTTP-agent decision is in progress.
"""
if self.lock.acquire(blocking=False):
try:
return deepcopy(self._to_dict_unlocked())
finally:
self.lock.release()
return deepcopy(self._completed_snapshot)
def _to_dict_unlocked(self) -> dict[str, object]:
return {
"game_id": self.game_id,
"status": "complete" if self.is_complete else "running",
@@ -449,9 +485,18 @@ class TableGame:
try:
requested = agent.decide(observation)
except Exception:
requested = PlayerAction("fold")
requested = self._default_action(observation.legal_actions)
return self._coerce_action(requested, observation.legal_actions)
def _default_action(self, legal_actions: list[dict[str, object]]) -> PlayerAction:
by_action = {str(action["action"]): action for action in legal_actions}
for action_type in ("check", "call", "fold"):
if action_type in by_action:
legal = by_action[action_type]
return PlayerAction(action_type, int(legal.get("amount") or 0))
legal = legal_actions[0]
return PlayerAction(str(legal["action"]), int(legal.get("amount") or 0))
def _coerce_action(
self,
requested: PlayerAction,
@@ -581,7 +626,7 @@ class TableGame:
swallow individual exceptions so a flaky remote endpoint cannot
break the table flow.
"""
snapshot = self.to_dict()
snapshot = self._to_dict_unlocked()
for agent in self.agents.values():
try:
agent.on_game_update(snapshot)
+8 -8
View File
@@ -25,8 +25,8 @@ class PokerRequestHandler(BaseHTTPRequestHandler):
if path == ["games"]:
self._json({"games": MANAGER.list_games()})
return
if len(path) == 2 and path[0] == "games":
self._json(MANAGER.get_game(path[1]).to_dict())
if len(path) == 2 and path[0] in {"game", "games"}:
self._json(MANAGER.get_game_state(path[1]))
return
self._json({"error": "not found"}, HTTPStatus.NOT_FOUND)
except KeyError as exc:
@@ -35,11 +35,11 @@ class PokerRequestHandler(BaseHTTPRequestHandler):
def do_POST(self) -> None:
path = self._path_parts()
try:
if path == ["games"]:
if path in (["game"], ["games"]):
game = MANAGER.create_game(self._read_json())
self._json(game.to_dict(), HTTPStatus.CREATED)
self._json(game.snapshot_completed(), HTTPStatus.CREATED)
return
if len(path) == 3 and path[0] == "games" and path[2] == "hands":
if len(path) == 3 and path[0] in {"game", "games"} and path[2] == "hands":
body = self._read_json()
count = int(body.get("count", 1))
until_one_left = bool(body.get("until_one_left", False))
@@ -51,9 +51,9 @@ class PokerRequestHandler(BaseHTTPRequestHandler):
small_blind=small_blind,
big_blind=big_blind,
)
self._json({"hands": summaries, "game": MANAGER.get_game(path[1]).to_dict()})
self._json({"hands": summaries, "game": MANAGER.get_game_state(path[1])})
return
if len(path) == 4 and path[0] == "games" and path[2] == "hands" and path[3] == "run":
if len(path) == 4 and path[0] in {"game", "games"} and path[2] == "hands" and path[3] == "run":
body = self._read_json()
count = int(body.get("count", 1))
until_one_left = bool(body.get("until_one_left", False))
@@ -65,7 +65,7 @@ class PokerRequestHandler(BaseHTTPRequestHandler):
small_blind=small_blind,
big_blind=big_blind,
)
self._json({"hands": summaries, "game": MANAGER.get_game(path[1]).to_dict()})
self._json({"hands": summaries, "game": MANAGER.get_game_state(path[1])})
return
self._json({"error": "not found"}, HTTPStatus.NOT_FOUND)
except KeyError as exc:
+51 -13
View File
@@ -5,13 +5,14 @@ from threading import RLock
from typing import Any
from uuid import uuid4
from texas_holdem.agents import build_agent
from texas_holdem.agents import build_agent, http_agent_endpoint_from_spec
from texas_holdem.engine import TableGame
class GameManager:
def __init__(self) -> None:
self._games: dict[str, TableGame] = {}
self._http_endpoint_owners: dict[str, str] = {}
self._lock = RLock()
def create_game(self, payload: dict[str, Any]) -> TableGame:
@@ -29,12 +30,19 @@ class GameManager:
big_blind = int(payload.get("big_blind", 10))
specs = []
http_endpoints: set[str] = set()
for seat, raw_spec in enumerate(players):
if not isinstance(raw_spec, dict):
raise ValueError("each player must be an object")
player_id = str(raw_spec.get("id") or raw_spec.get("player_id") or f"p{seat + 1}")
name = str(raw_spec.get("name") or player_id)
agent = build_agent(raw_spec.get("agent", raw_spec), rng)
agent_spec = raw_spec.get("agent", raw_spec)
if not isinstance(agent_spec, dict):
raise ValueError("agent spec must be an object")
endpoint = http_agent_endpoint_from_spec(agent_spec)
if endpoint is not None:
http_endpoints.add(endpoint)
agent = build_agent(agent_spec, rng, player_id=player_id)
specs.append((player_id, name, agent))
game = TableGame(
@@ -46,9 +54,18 @@ class GameManager:
rng=rng,
)
with self._lock:
self._release_completed_http_endpoints_locked()
if game_id in self._games:
raise ValueError(f"game already exists: {game_id}")
for endpoint in http_endpoints:
owner = self._http_endpoint_owners.get(endpoint)
if owner is not None and owner != game_id:
raise ValueError(
f"http agent endpoint already belongs to game {owner}: {endpoint}"
)
self._games[game_id] = game
for endpoint in http_endpoints:
self._http_endpoint_owners[endpoint] = game_id
return game
def get_game(self, game_id: str) -> TableGame:
@@ -58,9 +75,13 @@ class GameManager:
except KeyError as exc:
raise KeyError(f"game not found: {game_id}") from exc
def get_game_state(self, game_id: str) -> dict[str, object]:
return self.get_game(game_id).snapshot_completed()
def list_games(self) -> list[dict[str, object]]:
with self._lock:
return [game.to_dict() for game in self._games.values()]
games = list(self._games.values())
return [game.snapshot_completed() for game in games]
def run_hands(
self,
@@ -78,13 +99,30 @@ class GameManager:
no-argument behaviour.
"""
game = self.get_game(game_id)
with self._lock:
return [
summary.to_dict()
for summary in game.run_hands(
count,
until_one_left=until_one_left,
small_blind=small_blind,
big_blind=big_blind,
)
]
summaries = [
summary.to_dict()
for summary in game.run_hands(
count,
until_one_left=until_one_left,
small_blind=small_blind,
big_blind=big_blind,
)
]
if game.is_complete:
with self._lock:
self._release_http_endpoints_for_game_locked(game_id)
return summaries
def _release_completed_http_endpoints_locked(self) -> None:
for game_id, game in list(self._games.items()):
if game.lock.acquire(blocking=False):
try:
if game.is_complete:
self._release_http_endpoints_for_game_locked(game_id)
finally:
game.lock.release()
def _release_http_endpoints_for_game_locked(self, game_id: str) -> None:
for endpoint, owner in list(self._http_endpoint_owners.items()):
if owner == game_id:
del self._http_endpoint_owners[endpoint]