texas_hold_x/texas_holdem/human_client.py

"""Standalone interactive HTTP Human Agent.

Run this as a process on the operator's machine to expose:

* ``POST /act``  - the server posts the current observation; we render it
  on the local terminal and block on stdin until the human picks a legal
  action, then return ``{"action": ..., "amount": N}``.
* ``POST /game`` - the server posts the full game snapshot at the end of
  every hand (same shape as ``GET /games/{id}``) so the operator sees how
  the table is evolving. The body of the response is empty.
* ``GET /health`` - liveness probe.

Start the client:

    python -m texas_holdem.human_client --host 127.0.0.1 --port 9001

Hook it up by passing the *base* URL when creating the game::

    {
      "id": "alice",
      "name": "Alice",
      "agent": {
        "type": "http",
        "endpoint": "http://127.0.0.1:9001",
        "timeout_seconds": 600
      }
    }

Design notes:
- The HTTP layer reuses :mod:`texas_holdem.human_io` so rendering and menu
  validation stay consistent with the in-process :class:`HumanAgent`.
- A :class:`threading.Lock` inside :class:`HumanClientConsole` serialises
  terminal access so concurrent ``/act`` and ``/game`` callbacks never
  interleave on the same TTY.
"""

from __future__ import annotations

import argparse
import json
import sys
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from threading import Lock
from typing import IO, Any

from texas_holdem.human_io import (
    clear_screen,
    prompt_action,
    render_game_state,
    render_observation,
)


class HumanClientConsole:
    """Encapsulates terminal IO with a lock to serialise prompts.

    Wrapping the streams in a tiny class keeps stream injection (handy for
    tests) and concurrency control in one place, instead of leaking through
    free functions.

    ``keep_history`` defaults to ``False`` so every ``/act`` callback wipes
    the terminal first; pass ``True`` to retain previous output (e.g. for
    log-style debugging or when the terminal does not support ANSI codes).
    """

    def __init__(
        self,
        input_stream: IO[str] | None = None,
        output_stream: IO[str] | None = None,
        keep_history: bool = False,
    ) -> None:
        self._input = input_stream if input_stream is not None else sys.stdin
        self._output = output_stream if output_stream is not None else sys.stdout
        # The lock guards every ``decide`` and ``announce_game`` call so two
        # concurrent server callbacks never interleave on the same TTY.
        self._lock = Lock()
        self._keep_history = keep_history

    def decide(self, observation: dict[str, Any]) -> dict[str, Any]:
        """Render an observation and return the operator's action dict."""
        with self._lock:
            # Clear-by-default keeps the focus on the current decision; only
            # opt-out callers see the entire history scrolling upwards.
            if not self._keep_history:
                clear_screen(self._write)
            self._write(render_observation(observation))
            return prompt_action(
                list(observation.get("legal_actions") or []),
                self._read_line,
                self._write,
            )

    def announce_game(self, game_state: dict[str, Any]) -> None:
        """Render an end-of-hand game snapshot to the operator's terminal.

        Separated from :meth:`decide` because it is purely informational and
        must never block on input; it just writes a digest under the same
        lock to avoid corrupting an in-progress prompt.
        """
        with self._lock:
            self._write(render_game_state(game_state))

    def _write(self, text: str) -> None:
        self._output.write(text)
        self._output.flush()

    def _read_line(self, prompt: str) -> str:
        self._write(prompt)
        line = self._input.readline()
        if line == "":
            raise EOFError("input stream closed while waiting for human action")
        return line.rstrip("\n")


class HumanRequestHandler(BaseHTTPRequestHandler):
    """HTTP entry point for the standalone human agent.

    Routes:

    * ``GET  /health`` - liveness probe.
    * ``POST /act``    - decision request (blocks on stdin).
    * ``POST /game``   - end-of-hand snapshot (non-blocking).
    """

    server_version = "TexasHoldemHumanClient/0.2"

    # Injected by :func:`create_server` on the underlying server class so
    # every handler shares the same terminal console (and lock).
    console: HumanClientConsole  # type: ignore[assignment]

    def do_GET(self) -> None:
        if self.path == "/health":
            self._json({"ok": True})
            return
        self._json({"error": "not found"}, HTTPStatus.NOT_FOUND)

    def do_POST(self) -> None:
        # Dispatch table keeps add/remove of routes mechanical and avoids
        # the deeply-nested if/elif ladder common in BaseHTTPRequestHandler.
        routes = {
            "/act": self._handle_act,
            "/game": self._handle_game,
        }
        handler = routes.get(self.path)
        if handler is None:
            self._json({"error": "not found"}, HTTPStatus.NOT_FOUND)
            return

        try:
            payload = self._read_json()
        except ValueError as exc:
            self._json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
            return

        try:
            handler(payload)
        except EOFError as exc:
            # The operator closed stdin (Ctrl-D); surface as 503 so the
            # server can fall back to its default coercion (fold).
            self._json({"error": str(exc)}, HTTPStatus.SERVICE_UNAVAILABLE)
        except Exception as exc:  # pragma: no cover - defensive guard
            self._json({"error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR)

    def _handle_act(self, payload: dict[str, Any]) -> None:
        action = self.console.decide(payload)
        self._json(action)

    def _handle_game(self, payload: dict[str, Any]) -> None:
        # The /game callback is informational; reply with an empty 204 so
        # the calling engine knows we received it but does not parse a body.
        self.console.announce_game(payload)
        self._empty(HTTPStatus.NO_CONTENT)

    # Silence the default access log so it does not interleave with prompts.
    def log_message(self, format: str, *args: Any) -> None:  # noqa: A002
        return

    def _read_json(self) -> dict[str, Any]:
        length = int(self.headers.get("Content-Length", "0"))
        if length <= 0:
            raise ValueError("request body is required")
        try:
            payload = json.loads(self.rfile.read(length).decode("utf-8"))
        except json.JSONDecodeError as exc:
            raise ValueError("request body must be valid JSON") from exc
        if not isinstance(payload, dict):
            raise ValueError("request body must be a JSON object")
        return payload

    def _json(
        self,
        payload: dict[str, Any],
        status: HTTPStatus = HTTPStatus.OK,
    ) -> None:
        body = json.dumps(payload, ensure_ascii=True).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _empty(self, status: HTTPStatus) -> None:
        """Write a header-only response (used for ``204 No Content``)."""
        self.send_response(status)
        self.send_header("Content-Length", "0")
        self.end_headers()


def create_server(
    host: str,
    port: int,
    console: HumanClientConsole | None = None,
) -> ThreadingHTTPServer:
    """Build a server with a shared :class:`HumanClientConsole`.

    Exposed as a function so tests (or callers wiring custom IO streams)
    can construct the server without touching ``main``.
    """
    server = ThreadingHTTPServer((host, port), HumanRequestHandler)
    HumanRequestHandler.console = console or HumanClientConsole()
    return server


def main() -> None:
    parser = argparse.ArgumentParser(
        description=(
            "Run an interactive HTTP Human Agent that exposes "
            "POST /act and POST /game."
        ),
    )
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", default=9001, type=int)
    # Default behaviour clears the terminal on every /act so the operator
    # always sees a fresh view. Opt-in flag restores the historical
    # "append forever" behaviour for log-style debugging.
    parser.add_argument(
        "--keep-history",
        action="store_true",
        help=(
            "Keep previous terminal output when a new /act request arrives "
            "instead of clearing the screen."
        ),
    )
    args = parser.parse_args()

    console = HumanClientConsole(keep_history=args.keep_history)
    server = create_server(args.host, args.port, console=console)
    print(
        f"Human HTTP agent listening on http://{args.host}:{args.port}\n"
        f"  POST /act   - decision request\n"
        f"  POST /game  - end-of-hand snapshot\n"
        f"  clear-screen: {'off (keep history)' if args.keep_history else 'on'}\n"
        "Pass the base URL above as the 'endpoint' field of an 'http' agent spec.",
        file=sys.stderr,
        flush=True,
    )
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()


if __name__ == "__main__":
    main()