Session

The session layer plays any game move by move and returns semantic frames: the board by meaning, the decision with its checks and evidence, the panel and the outcome. The command line and the web showcase both run on it, and it speaks the versioned protocol described on Session protocol; recorded games use the replay format. Open this page when you drive a game from Python or write a front end.

Everything below is importable from grounded_decision.session and its modules. The game modules' contract is described in Agents and checks by game and the web plan.

session

The session layer: play any game move by move and get semantic frames.

One pure API for every front end. The command line (grounded-decision play, its terminal views and --view json), the web showcase's browser worker and a future server all open a :class:Session from a :class:SessionSpec and speak the versioned protocol in :mod:~grounded_decision.session.protocol. Nothing here does I/O, starts threads, handles signals or reads the wall clock, so it runs unchanged in Pyodide.

Example::

from grounded_decision.session import Session, SessionSpec

session = Session.open(SessionSpec("tictactoe", agents=("you", "gated")))
session.legal_moves()[4].notation      # "5"
frame = session.submit("5")             # X takes the centre
frame = session.step()                  # the gated player answers
frame.decision.verdict, frame.board.cells

Modules and what they own:

:mod:~grounded_decision.session.spec :class:SessionSpec: what a session plays, validated as play does. :mod:~grounded_decision.session.session :class:Session, its :class:~.session.Seat and :class:~.session.SubmittedMoves, and :class:SessionError. :mod:~grounded_decision.session.frame :class:Frame and its parts (board, decision, panel, outcome). :mod:~grounded_decision.session.builder :class:~.builder.FrameBuilder: frames from positions and moves. :mod:~grounded_decision.session.notation :class:~.notation.MoveNotation: typed moves, and each legal move's text. :mod:~grounded_decision.session.catalogue :class:~.catalogue.Catalogue: what describe returns. :mod:~grounded_decision.session.protocol The messages, :data:PROTOCOL_VERSION, the :class:~.protocol.Endpoint that answers them and the :class:~.protocol.ProtocolSchema. :mod:~grounded_decision.session.wire JSON encoding, validated decoding and JSON Schema from the dataclasses. :mod:~grounded_decision.session.compatibility :class:~.compatibility.SchemaChanges: what a schema change breaks. :mod:~grounded_decision.session.tokens :class:~.tokens.DesignTokens: colours and glyphs from the terminal palette. :mod:~grounded_decision.session.jsonlines --view json: the frames' JSON-lines projection. :mod:~grounded_decision.session.replay :class:~.replay.Replay: the .gdr replay format, recorded, encoded, decoded and verified. :mod:~grounded_decision.session.gallery The curated replays make replays writes (the site's gallery and the replay package's golden files). :mod:~grounded_decision.session.artefacts :class:~.artefacts.Artefacts: the generated files make protocol writes.

PROTOCOL_VERSION module-attribute

PROTOCOL_VERSION = '1.11.0'

The protocol's semantic version: major.minor.patch (see docs/reference/protocol.md).

Frame dataclass

Frame(
    index: int,
    to_move: int | None,
    awaiting_input: bool,
    decision: MoveDecision | None,
    board: Board,
    transition: tuple[TransitionStill, ...],
    move_cells: tuple[Point, ...],
    proposal_cells: tuple[Point, ...],
    event: EventLine | None,
    panel: Panel,
    outcome: Outcome,
    legal_moves: tuple[LegalMove, ...],
    assumption: str,
    arrows: tuple[MoveArrow, ...] = (),
)

One position of a session, with everything a front end draws for it.

Attributes:

Name	Type	Description
index	int	Moves played to reach it (0 is the starting position).
to_move	int | None	The seat whose turn it is, or null once the game is over.
awaiting_input	bool	Whether that seat is a person (submit a move).
decision	MoveDecision | None	The move that led here and how it was chosen; null at the start.
board	Board	The position.
transition	tuple[TransitionStill, ...]	In-between pictures of the move, in order (may be empty).
move_cells	tuple[Point, ...]	The cells the move touched, origin first.
proposal_cells	tuple[Point, ...]	On an override, the cells the rejected proposal would have touched; else empty.
event	EventLine | None	What the move did in a few words, or null.
panel	Panel	The panel's contents.
outcome	Outcome	Whether the game is over and why.
legal_moves	tuple[LegalMove, ...]	When a person is to move, every legal move with its notation; else empty.
assumption	str	When the world is not random and an agent bets on the odds, why its probabilistic statements do not apply; else empty.
arrows	tuple[MoveArrow, ...]	The move as arrows (protocol 1.7): the played move and, on an override, the rejected proposal, each checked against the previous frame's board; empty when the move is not one piece going from one cell to another (a placement, a drop, a slide of the whole board).

LegalMove dataclass

LegalMove(
    notation: str,
    description: str,
    cells: tuple[Point, ...],
    certified: bool | None,
)

A move a person may submit in the current position.

Attributes:

Name	Type	Description
notation	str	The text submit accepts for exactly this move.
description	str	The move as the game describes it.
cells	tuple[Point, ...]	The cells it touches, origin first and destination last.
certified	bool | None	For a guarded seat, whether the gate would pass it; else null.

SessionInfo dataclass

SessionInfo(
    game: str,
    title: str,
    seed: int,
    seats: tuple[SeatInfo, ...],
    world: str | None,
    panel: PanelSlots,
    max_steps: int | None,
    controls: RealtimeControls | None = None,
)

What stays the same for a whole session.

Attributes:

Name	Type	Description
game	str	The game's name.
title	str	Its title.
seed	int	The seed that fixes every random choice.
seats	tuple[SeatInfo, ...]	Who plays, in turn order.
world	str | None	The world (who controls the randomness), or null for a game without worlds.
panel	PanelSlots	The panel's fixed rows.
max_steps	int | None	The move budget, or null for none.
controls	RealtimeControls | None	Real-time keys (1.5), or null.

Endpoint dataclass

Endpoint(
    session: Session | None = None,
    comparison: ComparisonSession | None = None,
    audit: AuditSession | None = None,
    _catalogue: Catalogue | None = None,
    _demos: Demos = Demos(),
    _encoder: WireEncoder = WireEncoder(),
)

Answers one front end's requests, holding its open session, comparison and audit.

:meth:handle takes a decoded JSON message and always returns a reply (an error reply rather than an exception), so a transport only moves JSON; :meth:handle_text does the same for JSON text.

Attributes:

Name	Type	Description
session	Session | None	The open session, or None before open.
comparison	ComparisonSession | None	The open comparison, or None before compare.
audit	AuditSession | None	The open audit, or None before audit.

handle

handle(message: JsonValue) -> JsonObject

The reply to one request, as a JSON object.

Source code in src/grounded_decision/session/protocol.py

def handle(self, message: JsonValue) -> JsonObject:
    """The reply to one request, as a JSON object."""

    identifier = message.get("id") if isinstance(message, dict) else None
    reply_id = (
        identifier
        if isinstance(identifier, int) and not isinstance(identifier, bool)
        else None
    )
    try:
        reply = self._answer(self._request(message))
    except WireError as error:
        reply = ErrorReply(reply_id, "message", str(error))
    except PlayError as error:
        reply = ErrorReply(reply_id, "spec", str(error))
    except SessionError as error:
        reply = ErrorReply(reply_id, "session", str(error))
    except DemoError as error:
        reply = ErrorReply(reply_id, "spec", str(error))
    except Exception as error:  # noqa: BLE001 - a reply, never a crashed worker
        reply = ErrorReply(reply_id, "internal", f"{type(error).__name__}: {error}")
    encoded = self._encoder.encode(reply)
    assert isinstance(encoded, dict)
    return encoded

handle_text

handle_text(text: str) -> str

The reply to one request given as JSON text, as compact JSON text.

What a transport carries (the browser worker passes these strings between JavaScript and Python). Text that is not JSON gets the message error a non-object gets.

Source code in src/grounded_decision/session/protocol.py

def handle_text(self, text: str) -> str:
    """The reply to one request given as JSON text, as compact JSON text.

    What a transport carries (the browser worker passes these strings
    between JavaScript and Python). Text that is not JSON gets the
    ``message`` error a non-object gets.
    """

    try:
        message: JsonValue = json.loads(text)
    except ValueError:
        message = text
    return json.dumps(
        self.handle(message),
        separators=(",", ":"),
        ensure_ascii=False,
        allow_nan=False,
    )

Session dataclass

Session(
    spec: SessionSpec,
    match: Match[Any, Any],
    seats: tuple[Seat, ...],
    builder: FrameBuilder,
    cut_off: bool = False,
    _frame: Frame | None = None,
)

One match of a :class:~.spec.SessionSpec, stepped move by move.

Open one with :meth:open. :meth:step plays the agent to move; :meth:submit plays a person's move; :meth:run steps until a person must move or the game ends. :attr:frame is the current frame.

Attributes:

Name	Type	Description
spec	SessionSpec	What is played.
match	Match[Any, Any]	The game, the agents and the position.
seats	tuple[Seat, ...]	Who sits where.
builder	FrameBuilder	Builds frames, keeping the panel's running state.
cut_off	bool	Whether the move budget has stopped the game.

game property

game: Any

The rules, bound to this session's agents.

controls property

controls: RealtimeControls | None

The game's real-time controls (controls), or None for a turn-based game.

over property

over: bool

No more moves: the game is finished or the move budget is spent.

awaiting_input property

awaiting_input: bool

Whether a person must :meth:submit the next move.

frame property

frame: Frame

The current frame (the starting position until a move is played).

open classmethod

open(
    spec: SessionSpec,
    *,
    human: Callable[[], HumanProposer] | None = None,
    clock: Clock | None = None,
) -> Session

A session at the starting position of spec.

Parameters:

Name	Type	Description	Default
spec	SessionSpec	The game, seats, rules, world and seed.	required
human	Callable[[], HumanProposer] | None	Makes each person's seat (the command line's terminal prompt); by default a :class:SubmittedMoves, fed by :meth:submit.	None
clock	Clock | None	Times each decision; None (the default) reads no clock.	None

Source code in src/grounded_decision/session/session.py

@classmethod
def open(
    cls,
    spec: SessionSpec,
    *,
    human: Callable[[], HumanProposer] | None = None,
    clock: Clock | None = None,
) -> Session:
    """A session at the starting position of ``spec``.

    Args:
        spec: The game, seats, rules, world and seed.
        human: Makes each person's seat (the command line's terminal
            prompt); by default a :class:`SubmittedMoves`, fed by
            :meth:`submit`.
        clock: Times each decision; ``None`` (the default) reads no clock.
    """

    make_seat: Callable[[], HumanProposer] = human or SubmittedMoves
    proposers: list[HumanProposer] = []

    def seat() -> HumanProposer:
        proposer = make_seat()
        proposers.append(proposer)
        return proposer

    agents = spec.build_agents(seat)
    game = spec.rules_game
    world = spec.entry.check_world(game, agents)
    match = Match[Any, Any].start(game, agents, seed=spec.seed, clock=clock)
    people = iter(proposers)
    seats = tuple(
        Seat(
            index,
            str(match.game.player_names[index]),
            agent,
            next(people) if name == HUMAN else None,
        )
        for index, (name, agent) in enumerate(zip(spec.seats, agents, strict=True))
    )
    builder = FrameBuilder.start(
        match.game,
        match.state,
        checks=spec.entry.check_names_for(match.game),
        note_rows=spec.entry.note_rows,
        max_steps=spec.max_steps,
        assumption=assumption_note(world, agents) if world is not None else "",
    )
    return cls(spec, match, seats, builder)

describe staticmethod

describe() -> Catalogue

Every game, its agents, checks, worlds, variants and sizes.

Source code in src/grounded_decision/session/session.py

@staticmethod
def describe() -> Catalogue:
    """Every game, its agents, checks, worlds, variants and sizes."""

    return Catalogue.build()

step

step() -> Frame

The agent to move decides and plays; the frame after its move.

Raises:

Type	Description
SessionError	The game is over, or a person must move.

Source code in src/grounded_decision/session/session.py

def step(self) -> Frame:
    """The agent to move decides and plays; the frame after its move.

    Raises:
        SessionError: The game is over, or a person must move.
    """

    if self.over:
        raise SessionError("the game is over")
    if self.awaiting_input:
        seat = self.seat_to_move
        assert seat is not None
        raise SessionError(f"{seat.player} is a person: submit their move")
    return self._advance()

run

run(count: int = 1) -> tuple[Frame, ...]

Up to count moves, stopping early when a person must move or the game ends.

Source code in src/grounded_decision/session/session.py

def run(self, count: int = 1) -> tuple[Frame, ...]:
    """Up to ``count`` moves, stopping early when a person must move or the game ends."""

    frames: list[Frame] = []
    while len(frames) < count and not self.over and not self.awaiting_input:
        frames.append(self._advance())
    return tuple(frames)

submit

submit(text: str) -> Frame

Play a person's move, typed in the game's notation; the frame after it.

With a guarded seat the move is only a proposal: the gate checks it and plays its safe fallback if a check fails (the frame's decision says why).

Raises:

Type	Description
SessionError	Not a person's turn, the text names no move, or (unguarded) the move is not legal.

Source code in src/grounded_decision/session/session.py

def submit(self, text: str) -> Frame:
    """Play a person's move, typed in the game's notation; the frame after it.

    With a guarded seat the move is only a proposal: the gate checks
    it and plays its safe fallback if a check fails (the frame's
    decision says why).

    Raises:
        SessionError: Not a person's turn, the text names no move, or
            (unguarded) the move is not legal.
    """

    seat = self.seat_to_move
    if seat is None:
        raise SessionError("the game is over")
    submitted = seat.submitted
    if submitted is None:
        raise SessionError(
            f"it is {seat.player}'s turn ({seat.agent.name}), not yours"
        )
    view = agent_input(self.game, seat.agent, self.match.state)
    notation = MoveNotation(self.game)
    legal = list(self.game.legal_moves(view))
    move = notation.parse(view, text.strip(), legal)
    if move is None:
        raise SessionError(f"I did not understand {text!r}")
    if move not in legal and not seat.guarded:
        raise SessionError(f"{notation.describe(view, move)} is not legal here")
    submitted.pending = move
    try:
        return self._advance()
    finally:
        submitted.pending = None

legal_moves

legal_moves() -> tuple[LegalMove, ...]

Every legal move of the seat to move, with the text that submits it.

Moves are listed as the mover sees the position (a hidden-information game shows a person only what they may know); for a guarded seat, each says whether the gate would pass it.

Source code in src/grounded_decision/session/session.py

def legal_moves(self) -> tuple[LegalMove, ...]:
    """Every legal move of the seat to move, with the text that submits it.

    Moves are listed as the mover sees the position (a hidden-information
    game shows a person only what they may know); for a guarded seat,
    each says whether the gate would pass it.
    """

    seat = self.seat_to_move
    if seat is None:
        return ()
    view = agent_input(self.game, seat.agent, self.match.state)
    legal = list(self.game.legal_moves(view))
    texts = MoveNotation(self.game).notations(view, legal)
    cells_hook = cast(
        "Callable[[Any, Any], Sequence[tuple[int, int]]] | None",
        getattr(self.game, "move_cells", None),
    )
    moves: list[LegalMove] = []
    for move, text in zip(legal, texts, strict=True):
        if text is None:
            continue
        cells = Point.path(cells_hook(view, move)) if callable(cells_hook) else ()
        moves.append(self._legal_move(seat, view, move, text, cells))
    return tuple(moves)

select

select(piece: Point) -> PieceSelection

A person picks the piece on piece: its targets and the move to each.

Games with select(state, text) and cell_name(state, position) hooks (Hexxagon) highlight their own targets, as the terminal prompt does; for any other game the targets are the legal moves that start on piece.

Raises:

Type	Description
SessionError	Not a person's turn, or no move starts on piece.

Source code in src/grounded_decision/session/session.py

def select(self, piece: Point) -> PieceSelection:
    """A person picks the piece on ``piece``: its targets and the move to each.

    Games with ``select(state, text)`` and ``cell_name(state, position)``
    hooks (Hexxagon) highlight their own targets, as the terminal
    prompt does; for any other game the targets are the legal moves
    that start on ``piece``.

    Raises:
        SessionError: Not a person's turn, or no move starts on ``piece``.
    """

    seat = self.seat_to_move
    if seat is None:
        raise SessionError("the game is over")
    if seat.submitted is None:
        raise SessionError(
            f"it is {seat.player}'s turn ({seat.agent.name}), not yours"
        )
    view = agent_input(self.game, seat.agent, self.match.state)
    select = getattr(self.game, "select", None)
    name = getattr(self.game, "cell_name", None)
    if callable(select) and callable(name):
        return self._game_selection(seat, view, piece)
    moves = tuple(
        move
        for move in self.legal_moves()
        if len(move.cells) >= 2
        and move.cells[0] == piece
        and move.cells[-1] != piece
    )
    if not moves:
        raise SessionError(
            f"no move starts on row {piece.row + 1}, column {piece.column + 1}"
        )
    targets = list(dict.fromkeys(move.cells[-1] for move in moves))
    overlays = (
        Mark(piece.row, piece.column, Overlay.SELECTED),
        *(Mark(cell.row, cell.column, Overlay.TARGET_NEAR) for cell in targets),
    )
    return PieceSelection(
        piece, overlays, "Pick one of the highlighted targets.", moves
    )

play

play(
    observers: Sequence[Observer[Any, Any]] = (),
    *,
    detail: bool = True,
    keep_steps: bool = False,
) -> MatchRecord[Any, Any]

Play the match to the end, telling observers (the terminal views) every move.

This is how the command line runs a session: no frames are built, so a huge board runs at full speed. detail is what agents see from :func:~grounded_decision.examples.games.engine.detail_wanted.

Source code in src/grounded_decision/session/session.py

def play(
    self,
    observers: Sequence[Observer[Any, Any]] = (),
    *,
    detail: bool = True,
    keep_steps: bool = False,
) -> MatchRecord[Any, Any]:
    """Play the match to the end, telling ``observers`` (the terminal views) every move.

    This is how the command line runs a session: no frames are built,
    so a huge board runs at full speed. ``detail`` is what agents see
    from :func:`~grounded_decision.examples.games.engine.detail_wanted`.
    """

    self.match.detail = detail
    record = run_match(
        self.match,
        seed=self.spec.seed,
        max_steps=self.spec.budget,
        observers=observers,
        keep_steps=keep_steps,
    )
    self.cut_off = record.cut_off
    self._frame = None
    return record

record

record() -> MatchRecord[Any, Any]

The match so far (moves, overrides, outcome), without its steps.

Source code in src/grounded_decision/session/session.py

def record(self) -> MatchRecord[Any, Any]:
    """The match so far (moves, overrides, outcome), without its steps."""

    return self.match.record([], seed=self.spec.seed, cut_off=self.cut_off)

SessionError

Bases: Exception

The session cannot do that now (not your turn, an illegal move, the game is over).

SessionSpec dataclass

SessionSpec(
    game: str,
    agents: tuple[str, ...] = (),
    world: str | None = None,
    seed: int = 0,
    size: str | None = None,
    players: int | None = None,
    until_length: int | None = None,
    rules: Mapping[str, RuleValue] = dict[str, RuleValue](),
    you: bool = False,
    you_play: str | None = None,
    guarded: bool = False,
    max_steps: int | None = None,
)

Everything that fixes a match: game, seats, rules, world and seed.

Construction validates it all, exactly as grounded-decision play does, and raises :class:~grounded_decision.examples.games.registry.PlayError with the command line's message otherwise.

Attributes:

Name	Type	Description
game	str	The game's name (tictactoe, snake, 2048 …).
agents	tuple[str, ...]	One agent name per seat, in seat order; empty for the game's default line-up; one name fills every seat; you is a person (alone, you takes one seat beside the defaults).
world	str | None	Who controls the randomness (random, adversarial, MODE:p …), for games with worlds; null for the default.
seed	int	Fixes every random choice; the same spec always plays the same game.
size	str | None	WxH or one of the game's named sizes; null for the default.
players	int | None	Number of players, for games with a choice (Chinese checkers).
until_length	int | None	Snake's length goal; null for a full board.
rules	Mapping[str, RuleValue]	Rule options by name, as the game's configure takes them (mines, preview, board, start, layout, holes, stall_plies …).
you	bool	Put a person in a seat (you_play picks which).
you_play	str | None	The person's seat, by player name or number (1 = first).
guarded	bool	The game's gate checks the person's moves.
max_steps	int | None	The move budget; null (or 0) for the default (10,000,000).

entry class-attribute instance-attribute

entry: GameModule = field(
    init=False, repr=False, compare=False, metadata=_DERIVED
)

The game's module (derived).

rules_game class-attribute instance-attribute

rules_game: TurnGame[Any, Any] = field(
    init=False, repr=False, compare=False, metadata=_DERIVED
)

The configured game (derived; not yet bound to agents).

seats class-attribute instance-attribute

seats: tuple[str, ...] = field(
    init=False, repr=False, compare=False, metadata=_DERIVED
)

One agent name per seat, defaults filled in (derived).

budget property

budget: int

The move budget (max_steps, else the default).

build_agents

build_agents(
    human: Callable[[], HumanProposer],
) -> list[Agent[Any, Any]]

Fresh agents for every seat; person seats are human(), guarded if asked.

Raises:

Type	Description
PlayError	An unknown agent, or guarded for a game without a gate.

Source code in src/grounded_decision/session/spec.py

def build_agents(self, human: Callable[[], HumanProposer]) -> list[Agent[Any, Any]]:
    """Fresh agents for every seat; person seats are ``human()``, guarded if asked.

    Raises:
        PlayError: An unknown agent, or ``guarded`` for a game without a gate.
    """

    return self.entry.seat_agents(
        self.rules_game,
        self.seats,
        seed=self.seed,
        guarded=self.guarded,
        human=human,
    )

spec

What a session plays: the :class:SessionSpec.

A spec names everything that changes the game itself (the game, its size and rules, the world, the seats and the seed), never how it is shown. grounded-decision play builds one from its options and the web front end sends one in an open message; both are validated here by :class:~grounded_decision.examples.games.registry.GameModule, the code play has always used, so the command line and a session can never disagree about what is allowed.

SessionSpec dataclass

SessionSpec(
    game: str,
    agents: tuple[str, ...] = (),
    world: str | None = None,
    seed: int = 0,
    size: str | None = None,
    players: int | None = None,
    until_length: int | None = None,
    rules: Mapping[str, RuleValue] = dict[str, RuleValue](),
    you: bool = False,
    you_play: str | None = None,
    guarded: bool = False,
    max_steps: int | None = None,
)

Everything that fixes a match: game, seats, rules, world and seed.

Construction validates it all, exactly as grounded-decision play does, and raises :class:~grounded_decision.examples.games.registry.PlayError with the command line's message otherwise.

Attributes:

Name	Type	Description
game	str	The game's name (tictactoe, snake, 2048 …).
agents	tuple[str, ...]	One agent name per seat, in seat order; empty for the game's default line-up; one name fills every seat; you is a person (alone, you takes one seat beside the defaults).
world	str | None	Who controls the randomness (random, adversarial, MODE:p …), for games with worlds; null for the default.
seed	int	Fixes every random choice; the same spec always plays the same game.
size	str | None	WxH or one of the game's named sizes; null for the default.
players	int | None	Number of players, for games with a choice (Chinese checkers).
until_length	int | None	Snake's length goal; null for a full board.
rules	Mapping[str, RuleValue]	Rule options by name, as the game's configure takes them (mines, preview, board, start, layout, holes, stall_plies …).
you	bool	Put a person in a seat (you_play picks which).
you_play	str | None	The person's seat, by player name or number (1 = first).
guarded	bool	The game's gate checks the person's moves.
max_steps	int | None	The move budget; null (or 0) for the default (10,000,000).

entry class-attribute instance-attribute

entry: GameModule = field(
    init=False, repr=False, compare=False, metadata=_DERIVED
)

The game's module (derived).

rules_game class-attribute instance-attribute

rules_game: TurnGame[Any, Any] = field(
    init=False, repr=False, compare=False, metadata=_DERIVED
)

The configured game (derived; not yet bound to agents).

seats class-attribute instance-attribute

seats: tuple[str, ...] = field(
    init=False, repr=False, compare=False, metadata=_DERIVED
)

One agent name per seat, defaults filled in (derived).

budget property

budget: int

The move budget (max_steps, else the default).

build_agents

build_agents(
    human: Callable[[], HumanProposer],
) -> list[Agent[Any, Any]]

Fresh agents for every seat; person seats are human(), guarded if asked.

Raises:

Type	Description
PlayError	An unknown agent, or guarded for a game without a gate.

Source code in src/grounded_decision/session/spec.py

def build_agents(self, human: Callable[[], HumanProposer]) -> list[Agent[Any, Any]]:
    """Fresh agents for every seat; person seats are ``human()``, guarded if asked.

    Raises:
        PlayError: An unknown agent, or ``guarded`` for a game without a gate.
    """

    return self.entry.seat_agents(
        self.rules_game,
        self.seats,
        seed=self.seed,
        guarded=self.guarded,
        human=human,
    )

session

A match you can step, feed moves into and draw: the :class:Session.

open(spec) builds the game and its seats; :meth:Session.step lets the agent to move decide and play; :meth:Session.submit plays a person's move; every call returns a :class:~.frame.Frame. The command line drives the same session with :meth:Session.play and its terminal views as observers, so both run one code path (:class:~grounded_decision.examples.games.engine.Match).

The session layer is pure: no I/O, threads, signals or wall clock (a test imports it with those blocked), so it runs unchanged in a browser (Pyodide) and on a server. A clock may be injected for timing.

Concepts and the classes that own them:

:class:Session One match in progress and its frames. :class:Seat Who sits in one seat: the agent, and for a person how their moves arrive and whether the gate guards them. :class:SubmittedMoves A person's seat whose moves arrive through :meth:Session.submit. :class:SessionError A call the session cannot honour now (a move when it is not your turn, an unreadable or illegal move, a finished game).

SessionError

Bases: Exception

The session cannot do that now (not your turn, an illegal move, the game is over).

SubmittedMoves dataclass

SubmittedMoves(
    name: str = HUMAN,
    certify: Callable[[Any, Any], bool] | None = None,
    pending: Any = None,
    predictable: bool = False,
)

A person's seat whose moves come from :meth:Session.submit.

Attributes:

Name	Type	Description
name	str	The agent name (you).
certify	Callable[[Any, Any], bool] | None	Set when guarded: whether the gate would pass a move.
pending	Any	The move to play next, set by submit.
predictable	bool	Always False: no world may simulate a person.

Seat dataclass

Seat(
    index: int,
    player: str,
    agent: Agent[Any, Any],
    proposer: HumanProposer | None,
)

Who sits in one seat.

Attributes:

Name	Type	Description
index	int	The seat, in turn order.
player	str	The player's name.
agent	Agent[Any, Any]	The agent that moves for it (a gate around a person when guarded).
proposer	HumanProposer | None	A person's own seat object (the terminal prompt or :class:SubmittedMoves), or None for an agent.

submitted property

submitted: SubmittedMoves | None

The seat's :class:SubmittedMoves, if its moves come from submit.

Session dataclass

Session(
    spec: SessionSpec,
    match: Match[Any, Any],
    seats: tuple[Seat, ...],
    builder: FrameBuilder,
    cut_off: bool = False,
    _frame: Frame | None = None,
)

One match of a :class:~.spec.SessionSpec, stepped move by move.

Open one with :meth:open. :meth:step plays the agent to move; :meth:submit plays a person's move; :meth:run steps until a person must move or the game ends. :attr:frame is the current frame.

Attributes:

Name	Type	Description
spec	SessionSpec	What is played.
match	Match[Any, Any]	The game, the agents and the position.
seats	tuple[Seat, ...]	Who sits where.
builder	FrameBuilder	Builds frames, keeping the panel's running state.
cut_off	bool	Whether the move budget has stopped the game.

game property

game: Any

The rules, bound to this session's agents.

controls property

controls: RealtimeControls | None

The game's real-time controls (controls), or None for a turn-based game.

over property

over: bool

No more moves: the game is finished or the move budget is spent.

awaiting_input property

awaiting_input: bool

Whether a person must :meth:submit the next move.

frame property

frame: Frame

The current frame (the starting position until a move is played).

open classmethod

open(
    spec: SessionSpec,
    *,
    human: Callable[[], HumanProposer] | None = None,
    clock: Clock | None = None,
) -> Session

A session at the starting position of spec.

Parameters:

Name	Type	Description	Default
spec	SessionSpec	The game, seats, rules, world and seed.	required
human	Callable[[], HumanProposer] | None	Makes each person's seat (the command line's terminal prompt); by default a :class:SubmittedMoves, fed by :meth:submit.	None
clock	Clock | None	Times each decision; None (the default) reads no clock.	None

Source code in src/grounded_decision/session/session.py

@classmethod
def open(
    cls,
    spec: SessionSpec,
    *,
    human: Callable[[], HumanProposer] | None = None,
    clock: Clock | None = None,
) -> Session:
    """A session at the starting position of ``spec``.

    Args:
        spec: The game, seats, rules, world and seed.
        human: Makes each person's seat (the command line's terminal
            prompt); by default a :class:`SubmittedMoves`, fed by
            :meth:`submit`.
        clock: Times each decision; ``None`` (the default) reads no clock.
    """

    make_seat: Callable[[], HumanProposer] = human or SubmittedMoves
    proposers: list[HumanProposer] = []

    def seat() -> HumanProposer:
        proposer = make_seat()
        proposers.append(proposer)
        return proposer

    agents = spec.build_agents(seat)
    game = spec.rules_game
    world = spec.entry.check_world(game, agents)
    match = Match[Any, Any].start(game, agents, seed=spec.seed, clock=clock)
    people = iter(proposers)
    seats = tuple(
        Seat(
            index,
            str(match.game.player_names[index]),
            agent,
            next(people) if name == HUMAN else None,
        )
        for index, (name, agent) in enumerate(zip(spec.seats, agents, strict=True))
    )
    builder = FrameBuilder.start(
        match.game,
        match.state,
        checks=spec.entry.check_names_for(match.game),
        note_rows=spec.entry.note_rows,
        max_steps=spec.max_steps,
        assumption=assumption_note(world, agents) if world is not None else "",
    )
    return cls(spec, match, seats, builder)

describe staticmethod

describe() -> Catalogue

Every game, its agents, checks, worlds, variants and sizes.

Source code in src/grounded_decision/session/session.py

@staticmethod
def describe() -> Catalogue:
    """Every game, its agents, checks, worlds, variants and sizes."""

    return Catalogue.build()

step

step() -> Frame

The agent to move decides and plays; the frame after its move.

Raises:

Type	Description
SessionError	The game is over, or a person must move.

Source code in src/grounded_decision/session/session.py

def step(self) -> Frame:
    """The agent to move decides and plays; the frame after its move.

    Raises:
        SessionError: The game is over, or a person must move.
    """

    if self.over:
        raise SessionError("the game is over")
    if self.awaiting_input:
        seat = self.seat_to_move
        assert seat is not None
        raise SessionError(f"{seat.player} is a person: submit their move")
    return self._advance()

run

run(count: int = 1) -> tuple[Frame, ...]

Up to count moves, stopping early when a person must move or the game ends.

Source code in src/grounded_decision/session/session.py

def run(self, count: int = 1) -> tuple[Frame, ...]:
    """Up to ``count`` moves, stopping early when a person must move or the game ends."""

    frames: list[Frame] = []
    while len(frames) < count and not self.over and not self.awaiting_input:
        frames.append(self._advance())
    return tuple(frames)

submit

submit(text: str) -> Frame

Play a person's move, typed in the game's notation; the frame after it.

With a guarded seat the move is only a proposal: the gate checks it and plays its safe fallback if a check fails (the frame's decision says why).

Raises:

Type	Description
SessionError	Not a person's turn, the text names no move, or (unguarded) the move is not legal.

Source code in src/grounded_decision/session/session.py

def submit(self, text: str) -> Frame:
    """Play a person's move, typed in the game's notation; the frame after it.

    With a guarded seat the move is only a proposal: the gate checks
    it and plays its safe fallback if a check fails (the frame's
    decision says why).

    Raises:
        SessionError: Not a person's turn, the text names no move, or
            (unguarded) the move is not legal.
    """

    seat = self.seat_to_move
    if seat is None:
        raise SessionError("the game is over")
    submitted = seat.submitted
    if submitted is None:
        raise SessionError(
            f"it is {seat.player}'s turn ({seat.agent.name}), not yours"
        )
    view = agent_input(self.game, seat.agent, self.match.state)
    notation = MoveNotation(self.game)
    legal = list(self.game.legal_moves(view))
    move = notation.parse(view, text.strip(), legal)
    if move is None:
        raise SessionError(f"I did not understand {text!r}")
    if move not in legal and not seat.guarded:
        raise SessionError(f"{notation.describe(view, move)} is not legal here")
    submitted.pending = move
    try:
        return self._advance()
    finally:
        submitted.pending = None

legal_moves

legal_moves() -> tuple[LegalMove, ...]

Every legal move of the seat to move, with the text that submits it.

Moves are listed as the mover sees the position (a hidden-information game shows a person only what they may know); for a guarded seat, each says whether the gate would pass it.

Source code in src/grounded_decision/session/session.py

def legal_moves(self) -> tuple[LegalMove, ...]:
    """Every legal move of the seat to move, with the text that submits it.

    Moves are listed as the mover sees the position (a hidden-information
    game shows a person only what they may know); for a guarded seat,
    each says whether the gate would pass it.
    """

    seat = self.seat_to_move
    if seat is None:
        return ()
    view = agent_input(self.game, seat.agent, self.match.state)
    legal = list(self.game.legal_moves(view))
    texts = MoveNotation(self.game).notations(view, legal)
    cells_hook = cast(
        "Callable[[Any, Any], Sequence[tuple[int, int]]] | None",
        getattr(self.game, "move_cells", None),
    )
    moves: list[LegalMove] = []
    for move, text in zip(legal, texts, strict=True):
        if text is None:
            continue
        cells = Point.path(cells_hook(view, move)) if callable(cells_hook) else ()
        moves.append(self._legal_move(seat, view, move, text, cells))
    return tuple(moves)

select

select(piece: Point) -> PieceSelection

A person picks the piece on piece: its targets and the move to each.

Games with select(state, text) and cell_name(state, position) hooks (Hexxagon) highlight their own targets, as the terminal prompt does; for any other game the targets are the legal moves that start on piece.

Raises:

Type	Description
SessionError	Not a person's turn, or no move starts on piece.

Source code in src/grounded_decision/session/session.py

def select(self, piece: Point) -> PieceSelection:
    """A person picks the piece on ``piece``: its targets and the move to each.

    Games with ``select(state, text)`` and ``cell_name(state, position)``
    hooks (Hexxagon) highlight their own targets, as the terminal
    prompt does; for any other game the targets are the legal moves
    that start on ``piece``.

    Raises:
        SessionError: Not a person's turn, or no move starts on ``piece``.
    """

    seat = self.seat_to_move
    if seat is None:
        raise SessionError("the game is over")
    if seat.submitted is None:
        raise SessionError(
            f"it is {seat.player}'s turn ({seat.agent.name}), not yours"
        )
    view = agent_input(self.game, seat.agent, self.match.state)
    select = getattr(self.game, "select", None)
    name = getattr(self.game, "cell_name", None)
    if callable(select) and callable(name):
        return self._game_selection(seat, view, piece)
    moves = tuple(
        move
        for move in self.legal_moves()
        if len(move.cells) >= 2
        and move.cells[0] == piece
        and move.cells[-1] != piece
    )
    if not moves:
        raise SessionError(
            f"no move starts on row {piece.row + 1}, column {piece.column + 1}"
        )
    targets = list(dict.fromkeys(move.cells[-1] for move in moves))
    overlays = (
        Mark(piece.row, piece.column, Overlay.SELECTED),
        *(Mark(cell.row, cell.column, Overlay.TARGET_NEAR) for cell in targets),
    )
    return PieceSelection(
        piece, overlays, "Pick one of the highlighted targets.", moves
    )

play

play(
    observers: Sequence[Observer[Any, Any]] = (),
    *,
    detail: bool = True,
    keep_steps: bool = False,
) -> MatchRecord[Any, Any]

Play the match to the end, telling observers (the terminal views) every move.

This is how the command line runs a session: no frames are built, so a huge board runs at full speed. detail is what agents see from :func:~grounded_decision.examples.games.engine.detail_wanted.

Source code in src/grounded_decision/session/session.py

def play(
    self,
    observers: Sequence[Observer[Any, Any]] = (),
    *,
    detail: bool = True,
    keep_steps: bool = False,
) -> MatchRecord[Any, Any]:
    """Play the match to the end, telling ``observers`` (the terminal views) every move.

    This is how the command line runs a session: no frames are built,
    so a huge board runs at full speed. ``detail`` is what agents see
    from :func:`~grounded_decision.examples.games.engine.detail_wanted`.
    """

    self.match.detail = detail
    record = run_match(
        self.match,
        seed=self.spec.seed,
        max_steps=self.spec.budget,
        observers=observers,
        keep_steps=keep_steps,
    )
    self.cut_off = record.cut_off
    self._frame = None
    return record

record

record() -> MatchRecord[Any, Any]

The match so far (moves, overrides, outcome), without its steps.

Source code in src/grounded_decision/session/session.py

def record(self) -> MatchRecord[Any, Any]:
    """The match so far (moves, overrides, outcome), without its steps."""

    return self.match.record([], seed=self.spec.seed, cut_off=self.cut_off)

frame

What a front end draws after each move: the :class:Frame and its parts.

A frame is semantic, like the game hooks it comes from: cells are :class:~grounded_decision.examples.games.views.CellKind values, never glyphs or colours, and highlights are :class:~grounded_decision.examples.games.views.Overlay values. The terminal views draw the same meaning with characters; the web board draws it with the shared design tokens. Every class here is a frozen value that :func:~grounded_decision.session.wire.encode turns into JSON, and the protocol's JSON Schema is generated from these classes.

Concepts and the classes that own them:

:class:Frame One position of a session: the board, the move that led to it and how it was decided, the panel, the outcome. :class:Board The picture of a position: cells, per-cell details, ordered paths, overlays, hex offsets and coordinates (:class:CellNote, :class:BoardPath, :class:Mark, :class:Label, :class:Point). :class:TransitionStill One in-between picture of a move, for the player to animate. :class:MoveArrow A piece's move as an arrow, over the position before the move (the rule every arrow obeys is :class:~.arrows.ArrowRule). :class:MoveDecision The move and how it was chosen: the proposal, every check with its evidence (:class:Check), the verdict, notes and metrics. :class:Panel The fixed panel's contents (:class:Counters, :class:StandingRow, :class:EventLine, :class:Sparkline); its fixed rows are a :class:PanelSlots, sent once per session. :class:Outcome Whether the game is over, who won and why. :class:LegalMove A move a person may submit, with the text that submits it. :class:PieceSelection A person's selected piece, its highlighted targets and their moves.

Scalar module-attribute

Scalar = bool | int | float | str

A metric or stat value: what a panel prints.

RuleValue module-attribute

RuleValue = bool | int | float | str | None

A rule option's value (None turns an option off, as --stall-plies 0 does).

Point dataclass

Point(row: int, column: int)

A cell of the board, in the board's cells coordinates.

Attributes:

Name	Type	Description
row	int	Row, 0 at the top.
column	int	Column, 0 at the left.

CellNote dataclass

CellNote(
    row: int,
    column: int,
    order: float | None,
    connections: tuple[str, ...],
    heading: str | None,
    label: str | None,
    tint: int | None,
    extent: tuple[float, ...] | None = None,
)

Extra detail for one cell beyond its kind (CellDetail in the views).

Attributes:

Name	Type	Description
row	int	The cell's row.
column	int	The cell's column.
order	float | None	0.0 for the head or newest … 1.0 for the tail or oldest (colour gradients; a 2048 tile's colour level), or null.
connections	tuple[str, ...]	Neighbours this cell is joined to (up, down, left, right and the diagonals up_left …), sorted.
heading	str | None	Direction of travel for an arrow (up …), or null.
label	str | None	Short text drawn on the cell (a Minesweeper count), or null.
tint	int | None	The seat whose territory the cell is (a home arm), or null.
extent	tuple[float, ...] | None	Part of the cell covered, LTRB (1.5).

BoardPath dataclass

BoardPath(
    kind: PathKind,
    cells: tuple[Point, ...],
    heading: str | None = None,
)

An ordered path of cells (CellPath in the views), from which a front end derives details.

The cells' notes are not listed in the board's details: a reader derives them, as the terminal does. Cell i of n gets order i / (n - 1) (0.0 for a one-cell path) and connections to cells i - 1 and i + 1 when they are neighbours (diagonals included); the first cell gets heading: the path's own when it has one (1.8: the snake's last move, which a one-cell snake cannot show), else the direction from the second cell to it. A note the board's details gives for the same cell replaces the derived one. New kinds may be added in a minor protocol version.

Attributes:

Name	Type	Description
kind	PathKind	What the path is: body, a connected body such as the snake's.
cells	tuple[Point, ...]	The cells, head (or newest) first.
heading	str | None	Where the head faces (up …), or null to derive it.

Mark dataclass

Mark(row: int, column: int, overlay: Overlay)

A highlight on one cell for this frame.

Attributes:

Name	Type	Description
row	int	The cell's row.
column	int	The cell's column.
overlay	Overlay	What the highlight means.

Label dataclass

Label(row: int, column: int, text: str)

A coordinate drawn on one (empty) cell, at most two characters.

Attributes:

Name	Type	Description
row	int	The cell's row.
column	int	The cell's column.
text	str	The label.

Board dataclass

Board(
    layout: Layout,
    rows: int,
    columns: int,
    cells: tuple[tuple[CellKind, ...], ...],
    offsets: tuple[int, ...],
    details: tuple[CellNote, ...],
    overlays: tuple[Mark, ...],
    labels: tuple[Label, ...],
    row_labels: tuple[str, ...],
    column_labels: tuple[str, ...],
    focus: Point | None,
    paths: tuple[BoardPath, ...] = (),
)

The picture of one position, by meaning: what each cell holds and what is highlighted.

Attributes:

Name	Type	Description
layout	Layout	How cells sit on screen: square, pointy hex rows or flat hex columns.
rows	int	Number of rows of cells.
columns	int	Number of columns of every row.
cells	tuple[tuple[CellKind, ...], ...]	Rows of cell kinds, top to bottom.
offsets	tuple[int, ...]	Hex boards: per-row (pointy) or per-column (flat) indent in half cells; empty for square boards.
details	tuple[CellNote, ...]	Extra detail for some cells.
overlays	tuple[Mark, ...]	Highlights: the last move, a gate override's rejected and chosen targets, pieces that changed hands, targets.
labels	tuple[Label, ...]	Coordinates on cells.
row_labels	tuple[str, ...]	Coordinates of the rows (may be empty).
column_labels	tuple[str, ...]	Coordinates of the columns (may be empty).
focus	Point | None	The cell to keep in view on a board bigger than the screen (the snake's head), or null.
paths	tuple[BoardPath, ...]	Ordered paths of cells (the snake's body, head first), whose cells' details are derived rather than listed in details (protocol 1.2; absent means none).

of classmethod

of(
    game: Any,
    state: Any,
    overlays: Mapping[Position, Overlay] | None = None,
) -> Board

The whole board of state, as the terminal views collect it.

Uses :func:~grounded_decision.examples.games.views.board_image (the views' own code path) on a window that is the whole board, with coordinates. overlays are the move's highlights; the game's own overlays(state) are merged in as the views do. The game's cell_paths travel as :class:BoardPath values, not expanded into details. The game's move_path_details (the terminal's line along the last move) are left out: the frame's arrows show the move.

Source code in src/grounded_decision/session/frame.py

@classmethod
def of(
    cls, game: Any, state: Any, overlays: Mapping[Position, Overlay] | None = None
) -> Board:
    """The whole board of ``state``, as the terminal views collect it.

    Uses :func:`~grounded_decision.examples.games.views.board_image`
    (the views' own code path) on a window that is the whole board,
    with coordinates. ``overlays`` are the move's highlights; the
    game's own ``overlays(state)`` are merged in as the views do. The
    game's ``cell_paths`` travel as :class:`BoardPath` values, not
    expanded into ``details``. The game's ``move_path_details`` (the
    terminal's line along the last move) are left out: the frame's
    ``arrows`` show the move.
    """

    layout = game_layout(game)
    if not has_cells(game):
        return cls(layout, 0, 0, (), (), (), (), (), (), (), None)
    rows, columns = board_shape(game, state)
    window = Window(0, 0, rows, columns, rows, columns)
    image = board_image(
        game,
        state,
        window,
        overlays,
        coords=True,
        expand_paths=False,
        move_path=False,
    )
    focus_hook = _hook(game, "focus")
    focus = focus_hook(state) if focus_hook is not None else None
    return cls(
        layout=image.layout,
        rows=image.rows,
        columns=image.columns,
        cells=tuple(tuple(row) for row in image.cells),
        offsets=tuple(image.offsets),
        details=tuple(
            CellNote.of(position, image.details[position])
            for position in sorted(image.details)
        ),
        overlays=tuple(
            Mark(row, column, image.overlays[(row, column)])
            for row, column in sorted(image.overlays)
        ),
        labels=tuple(
            Label(row, column, image.labels[(row, column)])
            for row, column in sorted(image.labels)
        ),
        row_labels=tuple(image.row_labels),
        column_labels=tuple(image.column_labels),
        focus=None if focus is None else Point(int(focus[0]), int(focus[1])),
        paths=tuple(BoardPath.of(path) for path in image.paths),
    )

TransitionStill dataclass

TransitionStill(board: Board, weight: float)

One in-between picture of a move (a piece landing before pieces flip).

Attributes:

Name	Type	Description
board	Board	The intermediate position, with this picture's overlays.
weight	float	Its share of the move's animation time (relative).

ArrowRole

Bases: StrEnum

What a :class:MoveArrow shows.

PLAYED class-attribute instance-attribute

PLAYED = 'played'

The move that was played.

PROPOSED class-attribute instance-attribute

PROPOSED = 'proposed'

On a gate override, the proposal the gate turned down.

MoveArrow dataclass

MoveArrow(
    role: ArrowRole, seat: int, cells: tuple[Point, ...]
)

A piece going from one cell to another, drawn over the position before the move.

The frame builder emits an arrow only when :class:~grounded_decision.session.arrows.ArrowRule holds on the previous frame's board: the origin holds a piece of seat and every later cell is free. A reader draws it over the previous frame's board (where the piece still stands) or while animating into this frame, never over this frame's board, where the origin may be empty.

Attributes:

Name	Type	Description
role	ArrowRole	played (the move made) or proposed (an override's rejected proposal).
seat	int	The seat whose piece moves.
cells	tuple[Point, ...]	The path, origin first and destination last (a jump's landings in between); two or more cells.

Check dataclass

Check(name: str, passed: bool, explanation: str)

One check the gate ran on a proposal.

Attributes:

Name	Type	Description
name	str	The check's name (a key of the game's checks in describe).
passed	bool	Whether the proposal satisfied it.
explanation	str	The evidence, in one sentence.

MoveDecision dataclass

MoveDecision(
    seat: int,
    player: str,
    agent: str,
    move: str,
    proposal: str | None,
    verdict: Verdict | None,
    checks: tuple[Check, ...],
    notes: tuple[str, ...],
    metrics: Mapping[str, Scalar],
    regime: Regime | None,
    lapse: str = "",
)

The move that led to a frame and how it was chosen.

Attributes:

Name	Type	Description
seat	int	The seat that moved.
player	str	That seat's name (X, Black).
agent	str	The agent in that seat.
move	str	The move played, as the game describes it.
proposal	str | None	For a gated agent, the proposer's move; else null.
verdict	Verdict | None	act (the proposal was played), degrade (a check failed and the safe fallback was played), abstain, or null for an ungated agent.
checks	tuple[Check, ...]	Every check run on the proposal.
notes	tuple[str, ...]	Extra lines from the agent (captures, search depth).
metrics	Mapping[str, Scalar]	Numbers from the agent (nodes searched, evaluation).
regime	Regime | None	The agent's declared information regime, or null.
lapse	str	Why the gate cannot certify this move (no move meets what its checks protect), so its guarantee lapses here (protocol 1.10); empty while it holds.

reason property

reason: str

Why a proposal was overridden: the first failed check's evidence.

of classmethod

of(game: Any, step: Step[Any, Any]) -> MoveDecision

The decision of step, with moves described from the position before it.

Source code in src/grounded_decision/session/frame.py

@classmethod
def of(cls, game: Any, step: Step[Any, Any]) -> MoveDecision:
    """The decision of ``step``, with moves described from the position before it."""

    decision = step.decision
    notation = MoveNotation(game)
    regime = decision.regime
    return cls(
        seat=step.mover,
        player=str(game.player_names[step.mover]),
        agent=decision.agent,
        move=notation.describe(step.before, decision.move),
        proposal=None
        if decision.proposal is None
        else notation.describe(step.before, decision.proposal),
        verdict=decision.verdict,
        checks=tuple(
            Check(check.name, check.passed, check.explanation)
            for check in decision.checks
        ),
        notes=tuple(decision.notes),
        metrics=dict(decision.metrics),
        regime=regime if isinstance(regime, Regime) else None,
        lapse=decision.lapse,
    )

EventLine dataclass

EventLine(number: int, text: str, size: int, big: bool)

What a move did, in a few words (the panel keeps the last few).

Attributes:

Name	Type	Description
number	int	The move's number (1 for the first move).
text	str	Pearl jumps c3→e5 · infects 6: d4 d5 e4 e6 f4 f5.
size	int	Pieces taken, converted or cleared (0 for a quiet move).
big	bool	Whether to draw it prominently.

StandingRow dataclass

StandingRow(
    seat: int,
    name: str,
    kind: CellKind,
    value: float,
    maximum: float,
    text: str,
    history: tuple[float, ...],
    shared: bool = False,
)

One player's progress bar in the panel.

Attributes:

Name	Type	Description
seat	int	The player.
name	str	The player's name.
kind	CellKind	A cell kind whose glyph and colour stand for the player.
value	float	Progress, between 0 and maximum.
maximum	float	A full bar.
text	str	Short text after the bar.
history	tuple[float, ...]	Recent values, oldest first (the sparkline).
shared	bool	Every player's value is a share of the same maximum (one board), so the rows may be drawn as one stacked bar.

Counters dataclass

Counters(
    moves: int, acts: int, overrides: int, ungated: int
)

Running totals over the moves so far.

Attributes:

Name	Type	Description
moves	int	Moves played.
acts	int	Gated moves whose proposal was played.
overrides	int	Gated moves overridden by the safe fallback.
ungated	int	Moves by agents without a gate.

Sparkline dataclass

Sparkline(label: str, values: tuple[float, ...])

The panel's sparkline: one number followed over the recent moves.

Attributes:

Name	Type	Description
label	str	What it follows (a stat such as length, or score).
values	tuple[float, ...]	Recent values, oldest first.

PanelSlots dataclass

PanelSlots(
    checks: tuple[str, ...],
    note_rows: int,
    stats_keys: tuple[str, ...],
    standing_rows: int,
    event_rows: int,
    sparkline: bool,
    gated: bool,
)

The panel's fixed rows for a whole session (PanelLayout in the views).

The panel never grows or shrinks between moves: each slot has as many rows as the game can ever fill, and a slot with nothing this move stays empty.

Attributes:

Name	Type	Description
checks	tuple[str, ...]	Every check the gate can run, one row each, in order.
note_rows	int	Rows for a decision's notes.
stats_keys	tuple[str, ...]	The stats shown, in order (from the starting position or the game's declared schema).
standing_rows	int	One row per player's standing (0 without standings).
event_rows	int	Rows of recent move events (0 if the game has none).
sparkline	bool	Whether a sparkline row is shown (games without standings).
gated	bool	Whether check rows are reserved.

Panel dataclass

Panel(
    counters: Counters,
    stats: Mapping[str, Scalar] | None,
    standings: tuple[StandingRow, ...],
    events: tuple[EventLine, ...],
    sparkline: Sparkline | None,
)

The panel's contents for one frame (its rows are the session's :class:PanelSlots).

Attributes:

Name	Type	Description
counters	Counters	Moves, acts, overrides and ungated moves so far.
stats	Mapping[str, Scalar] | None	The game's running stats for this position (all of them; the panel shows stats_keys), or null for a game without stats.
standings	tuple[StandingRow, ...]	One row per player, or empty.
events	tuple[EventLine, ...]	Recent move events, newest first.
sparkline	Sparkline | None	The sparkline, or null when the panel has none.

Outcome dataclass

Outcome(
    finished: bool,
    winner: int | None,
    reason: str,
    scores: tuple[float, ...],
    cut_off: bool,
    placings: tuple[int, ...],
)

Where the game stands.

Attributes:

Name	Type	Description
finished	bool	Whether the game is over.
winner	int | None	The winning seat, or null (unfinished, a draw, or no winner).
reason	str	One sentence (X completed the top row); empty while playing.
scores	tuple[float, ...]	Per-player scores, if the game keeps them.
cut_off	bool	Whether the session's move budget ended it.
placings	tuple[int, ...]	Place per seat (1 = first) once a game of several players is over, if the game ranks them; else empty.

over property

over: bool

Finished, or stopped by the move budget: no more moves.

LegalMove dataclass

LegalMove(
    notation: str,
    description: str,
    cells: tuple[Point, ...],
    certified: bool | None,
)

A move a person may submit in the current position.

Attributes:

Name	Type	Description
notation	str	The text submit accepts for exactly this move.
description	str	The move as the game describes it.
cells	tuple[Point, ...]	The cells it touches, origin first and destination last.
certified	bool | None	For a guarded seat, whether the gate would pass it; else null.

PieceSelection dataclass

PieceSelection(
    piece: Point,
    overlays: tuple[Mark, ...],
    hint: str,
    moves: tuple[LegalMove, ...],
)

A person picked a piece: where it may go, before they pick the target.

The web form of the terminal's selection prompt (Hexxagon: type a piece, see its clone and jump targets, type the target). Games with a select hook draw their own highlights; for any other game the targets are the legal moves that start on the piece.

Attributes:

Name	Type	Description
piece	Point	The selected cell.
overlays	tuple[Mark, ...]	selected on the piece; target_near (a one-step target, a Hexxagon clone) or target_far (a two-step target, a jump) on each target.
hint	str	One line on what the highlights mean.
moves	tuple[LegalMove, ...]	One legal move per target, origin first and the target last in cells; submit its notation to play it.

Frame dataclass

Frame(
    index: int,
    to_move: int | None,
    awaiting_input: bool,
    decision: MoveDecision | None,
    board: Board,
    transition: tuple[TransitionStill, ...],
    move_cells: tuple[Point, ...],
    proposal_cells: tuple[Point, ...],
    event: EventLine | None,
    panel: Panel,
    outcome: Outcome,
    legal_moves: tuple[LegalMove, ...],
    assumption: str,
    arrows: tuple[MoveArrow, ...] = (),
)

One position of a session, with everything a front end draws for it.

Attributes:

Name	Type	Description
index	int	Moves played to reach it (0 is the starting position).
to_move	int | None	The seat whose turn it is, or null once the game is over.
awaiting_input	bool	Whether that seat is a person (submit a move).
decision	MoveDecision | None	The move that led here and how it was chosen; null at the start.
board	Board	The position.
transition	tuple[TransitionStill, ...]	In-between pictures of the move, in order (may be empty).
move_cells	tuple[Point, ...]	The cells the move touched, origin first.
proposal_cells	tuple[Point, ...]	On an override, the cells the rejected proposal would have touched; else empty.
event	EventLine | None	What the move did in a few words, or null.
panel	Panel	The panel's contents.
outcome	Outcome	Whether the game is over and why.
legal_moves	tuple[LegalMove, ...]	When a person is to move, every legal move with its notation; else empty.
assumption	str	When the world is not random and an agent bets on the odds, why its probabilistic statements do not apply; else empty.
arrows	tuple[MoveArrow, ...]	The move as arrows (protocol 1.7): the played move and, on an override, the rejected proposal, each checked against the previous frame's board; empty when the move is not one piece going from one cell to another (a placement, a drop, a slide of the whole board).

SeatInfo dataclass

SeatInfo(
    seat: int,
    player: str,
    agent: str,
    human: bool,
    guarded: bool,
    regime: Regime | None,
)

Who sits in one seat.

Attributes:

Name	Type	Description
seat	int	The seat's index, in turn order.
player	str	The player's name (X, Ruby).
agent	str	The agent's name (you for a person).
human	bool	Whether a person plays it (moves come from submit).
guarded	bool	Whether the game's gate checks the person's moves.
regime	Regime | None	The agent's declared information regime, or null.

SessionInfo dataclass

SessionInfo(
    game: str,
    title: str,
    seed: int,
    seats: tuple[SeatInfo, ...],
    world: str | None,
    panel: PanelSlots,
    max_steps: int | None,
    controls: RealtimeControls | None = None,
)

What stays the same for a whole session.

Attributes:

Name	Type	Description
game	str	The game's name.
title	str	Its title.
seed	int	The seed that fixes every random choice.
seats	tuple[SeatInfo, ...]	Who plays, in turn order.
world	str | None	The world (who controls the randomness), or null for a game without worlds.
panel	PanelSlots	The panel's fixed rows.
max_steps	int | None	The move budget, or null for none.
controls	RealtimeControls | None	Real-time keys (1.5), or null.

protocol

The protocol between a front end and the engine: versioned messages over JSON.

A front end (the web page's worker, later a WebSocket client) sends a request and gets one reply. Requests: describe, open, step, submit, select, and (protocol 1.3) compare, audit and analyse. Replies: describe (the catalogue), frames (the session's info and new frames), selection (a picked piece's targets), comparison (finished games and the table so far), audit (judged decisions and the report so far), analysis (the session's position analysed) and error. Protocol 1.4 adds the demos beyond games (:mod:.demos): learn, monitor and runtime, answered by learned, monitored and decisions. Protocol 1.5 adds real-time play (SessionInfo.controls: the clock and the keys of a game played one tick at a time) and sub-cell positions (CellNote.extent), for Breakout; 1.8 adds RealtimeControls.repeat_last (a tick with no key repeats the last move: Snake played in real time) and BoardPath.heading (where a path's head faces, which a one-cell snake cannot show); 1.9 adds GameInfo.size_offers (every size a game's board may take, its default and usual sizes and the rule in words, per rule option where it differs; empty for a fixed size); 1.10 adds MoveDecision.lapse (why a gate cannot certify the move it played, so its guarantee lapses there); 1.11 adds GameInfo.rules (each rule option's label, kind, range, default, meaning and the option values it needs). Every message is a JSON object with a type; a request may carry an id, which its reply repeats. The JSON Schema of all of them is generated from the classes below (make protocol) and versioned by :data:PROTOCOL_VERSION (semver; docs/reference/protocol.md has the rules, and a test fails a breaking change without a major bump).

Concepts and the classes that own them:

:class:DescribeRequest, :class:OpenRequest, :class:StepRequest, :class:SubmitRequest, :class:SelectRequest, :class:CompareRequest, :class:AuditRequest, :class:AnalyseRequest What a front end may ask. :class:DescribeReply, :class:FramesReply, :class:SelectionReply, :class:ComparisonReply, :class:AuditReply, :class:AnalysisReply, :class:ErrorReply What the engine answers. :data:DOCUMENTS Files the engine's types describe but no message carries (a lesson). :class:Endpoint Answers requests for one front end, holding its open session, comparison and audit. :class:ProtocolSchema The JSON Schema of every message.

PROTOCOL_VERSION module-attribute

PROTOCOL_VERSION = '1.11.0'

The protocol's semantic version: major.minor.patch (see docs/reference/protocol.md).

MAX_STEPS_PER_REQUEST module-attribute

MAX_STEPS_PER_REQUEST = 10000

The most moves one step request may ask for.

DOCUMENTS module-attribute

DOCUMENTS: tuple[type[Any], ...] = (Lesson,)

Types of files the site reads that no message carries: a lesson (make lessons).

DescribeRequest dataclass

DescribeRequest(
    type: Literal["describe"] = "describe",
    id: int | None = None,
)

Ask for every game, its agents, checks, worlds, variants and sizes.

Attributes:

Name	Type	Description
type	Literal['describe']	describe.
id	int | None	Repeated in the reply, to match replies to requests.

OpenRequest dataclass

OpenRequest(
    spec: SessionSpec,
    type: Literal["open"] = "open",
    id: int | None = None,
)

Start a session (replacing any open one); the reply holds its first frame.

Attributes:

Name	Type	Description
type	Literal['open']	open.
spec	SessionSpec	What to play.
id	int | None	Repeated in the reply.

StepRequest dataclass

StepRequest(
    type: Literal["step"] = "step",
    count: int = 1,
    id: int | None = None,
)

Let the agents play up to count moves (stopping when a person must move).

Attributes:

Name	Type	Description
type	Literal['step']	step.
count	int	Most moves to play, 1 to 10,000.
id	int | None	Repeated in the reply.

SubmitRequest dataclass

SubmitRequest(
    move: str,
    type: Literal["submit"] = "submit",
    id: int | None = None,
)

Play a person's move, in the notation of the frame's legal_moves.

Attributes:

Name	Type	Description
type	Literal['submit']	submit.
move	str	The move's notation.
id	int | None	Repeated in the reply.

SelectRequest dataclass

SelectRequest(
    cell: Point,
    type: Literal["select"] = "select",
    id: int | None = None,
)

A person picks a piece (protocol 1.1): the reply lists where it may go.

Attributes:

Name	Type	Description
type	Literal['select']	select.
cell	Point	The piece's cell, in the frame's board coordinates.
id	int | None	Repeated in the reply.

CompareRequest dataclass

CompareRequest(
    spec: CompareSpec,
    type: Literal["compare"] = "compare",
    games: tuple[int, ...] = (),
    results: tuple[ComparedGame, ...] = (),
    id: int | None = None,
)

Play games of a comparison (protocol 1.3): the reply has them and the table so far.

A spec equal to the open comparison's continues it, so several requests (or several engines, each told the others' results) fill in one table; any other spec starts a new comparison.

Attributes:

Name	Type	Description
type	Literal['compare']	compare.
spec	CompareSpec	The comparison.
games	tuple[int, ...]	The games to play now, by index (at most 100; may be empty, to ask only for the table). A game already known is not played again.
results	tuple[ComparedGame, ...]	Games finished elsewhere (by another engine), to count in the table.
id	int | None	Repeated in the reply.

AuditRequest dataclass

AuditRequest(
    spec: AuditSpec,
    type: Literal["audit"] = "audit",
    moves: int = 100,
    id: int | None = None,
)

Play on an audited match (protocol 1.3): the reply judges the audited agent's decisions.

A spec equal to the open audit's continues it; any other spec starts a new audit from the first move.

Attributes:

Name	Type	Description
type	Literal['audit']	audit.
spec	AuditSpec	The audit.
moves	int	Most moves (all players) to play now, 1 to 10,000; none once the match is over.
id	int | None	Repeated in the reply.

AnalyseRequest dataclass

AnalyseRequest(
    type: Literal["analyse"] = "analyse",
    id: int | None = None,
)

Analyse the open session's position with the game's own solvers (protocol 1.3).

Attributes:

Name	Type	Description
type	Literal['analyse']	analyse.
id	int | None	Repeated in the reply.

DescribeReply dataclass

DescribeReply(
    id: int | None,
    version: str,
    catalogue: Catalogue,
    type: Literal["describe"] = "describe",
)

Every game that can be played.

Attributes:

Name	Type	Description
type	Literal['describe']	describe.
id	int | None	The request's id, or null.
version	str	The protocol version the engine speaks.
catalogue	Catalogue	The games.

FramesReply dataclass

FramesReply(
    id: int | None,
    session: SessionInfo,
    frames: tuple[Frame, ...],
    type: Literal["frames"] = "frames",
)

The open session and its new frames (oldest first).

Attributes:

Name	Type	Description
type	Literal['frames']	frames.
id	int | None	The request's id, or null.
session	SessionInfo	What stays the same for the session.
frames	tuple[Frame, ...]	The new frames: the first position after open, one per move after step and submit (none if no move could be played).

SelectionReply dataclass

SelectionReply(
    id: int | None,
    selection: PieceSelection,
    type: Literal["selection"] = "selection",
)

A picked piece's targets, to highlight before the person picks one (protocol 1.1).

Attributes:

Name	Type	Description
type	Literal['selection']	selection.
id	int | None	The request's id, or null.
selection	PieceSelection	The piece, its highlights and the move to each target.

ErrorReply dataclass

ErrorReply(
    id: int | None,
    code: ErrorCode,
    message: str,
    type: Literal["error"] = "error",
)

A request that could not be honoured; the session (if any) is unchanged.

Attributes:

Name	Type	Description
type	Literal['error']	error.
id	int | None	The request's id, or null.
code	ErrorCode	message (malformed request), spec (the game, seats, size, rules or world are not allowed, or a demo cannot run with these settings or data), session (not now: no session, not your turn, an illegal move, the game is over) or internal (a bug).
message	str	What went wrong, in one sentence.

ComparisonReply dataclass

ComparisonReply(
    id: int | None,
    plan: ComparePlan,
    games: tuple[ComparedGame, ...],
    table: CompareTable,
    type: Literal["comparison"] = "comparison",
)

The games a compare request played and the table of every game known (protocol 1.3).

Attributes:

Name	Type	Description
type	Literal['comparison']	comparison.
id	int | None	The request's id, or null.
plan	ComparePlan	What stays fixed for the comparison.
games	tuple[ComparedGame, ...]	The games asked for, finished, in the order asked.
table	CompareTable	The ranked table of every finished game this engine knows.

AuditReply dataclass

AuditReply(
    id: int | None,
    info: AuditInfo,
    decisions: tuple[AuditedMove, ...],
    summary: AuditSummary,
    type: Literal["audit"] = "audit",
)

The audited agent's decisions in the moves just played, and the report so far (protocol 1.3).

Attributes:

Name	Type	Description
type	Literal['audit']	audit.
id	int | None	The request's id, or null.
info	AuditInfo	What stays fixed for the audit.
decisions	tuple[AuditedMove, ...]	The audited agent's decisions among the moves just played, judged, oldest first.
summary	AuditSummary	The report of every decision so far.

AnalysisReply dataclass

AnalysisReply(
    id: int | None,
    index: int,
    analysis: Analysis | None,
    type: Literal["analysis"] = "analysis",
)

The open session's position, analysed (protocol 1.3).

Attributes:

Name	Type	Description
type	Literal['analysis']	analysis.
id	int | None	The request's id, or null.
index	int	The frame analysed (moves played to reach it).
analysis	Analysis | None	The analysis, or null where the game has none for this position.

Endpoint dataclass

Endpoint(
    session: Session | None = None,
    comparison: ComparisonSession | None = None,
    audit: AuditSession | None = None,
    _catalogue: Catalogue | None = None,
    _demos: Demos = Demos(),
    _encoder: WireEncoder = WireEncoder(),
)

Answers one front end's requests, holding its open session, comparison and audit.

:meth:handle takes a decoded JSON message and always returns a reply (an error reply rather than an exception), so a transport only moves JSON; :meth:handle_text does the same for JSON text.

Attributes:

Name	Type	Description
session	Session | None	The open session, or None before open.
comparison	ComparisonSession | None	The open comparison, or None before compare.
audit	AuditSession | None	The open audit, or None before audit.

handle

handle(message: JsonValue) -> JsonObject

The reply to one request, as a JSON object.

Source code in src/grounded_decision/session/protocol.py

def handle(self, message: JsonValue) -> JsonObject:
    """The reply to one request, as a JSON object."""

    identifier = message.get("id") if isinstance(message, dict) else None
    reply_id = (
        identifier
        if isinstance(identifier, int) and not isinstance(identifier, bool)
        else None
    )
    try:
        reply = self._answer(self._request(message))
    except WireError as error:
        reply = ErrorReply(reply_id, "message", str(error))
    except PlayError as error:
        reply = ErrorReply(reply_id, "spec", str(error))
    except SessionError as error:
        reply = ErrorReply(reply_id, "session", str(error))
    except DemoError as error:
        reply = ErrorReply(reply_id, "spec", str(error))
    except Exception as error:  # noqa: BLE001 - a reply, never a crashed worker
        reply = ErrorReply(reply_id, "internal", f"{type(error).__name__}: {error}")
    encoded = self._encoder.encode(reply)
    assert isinstance(encoded, dict)
    return encoded

handle_text

handle_text(text: str) -> str

The reply to one request given as JSON text, as compact JSON text.

What a transport carries (the browser worker passes these strings between JavaScript and Python). Text that is not JSON gets the message error a non-object gets.

Source code in src/grounded_decision/session/protocol.py

def handle_text(self, text: str) -> str:
    """The reply to one request given as JSON text, as compact JSON text.

    What a transport carries (the browser worker passes these strings
    between JavaScript and Python). Text that is not JSON gets the
    ``message`` error a non-object gets.
    """

    try:
        message: JsonValue = json.loads(text)
    except ValueError:
        message = text
    return json.dumps(
        self.handle(message),
        separators=(",", ":"),
        ensure_ascii=False,
        allow_nan=False,
    )

ProtocolSchema dataclass

ProtocolSchema(version: str = PROTOCOL_VERSION)

The JSON Schema (draft 2020-12) of every protocol message.

Attributes:

Name	Type	Description
version	str	The protocol version it describes.

document

document() -> JsonObject

The whole schema: $defs for every type, and the request and reply unions.

Source code in src/grounded_decision/session/protocol.py

def document(self) -> JsonObject:
    """The whole schema: ``$defs`` for every type, and the request and reply unions."""

    builder = SchemaBuilder()
    requests = [builder.reference(kind) for kind in REQUESTS]
    replies = [builder.reference(kind) for kind in REPLIES]
    for document in DOCUMENTS:
        builder.reference(document)
    definitions: JsonObject = dict(builder.definitions)
    definitions["Request"] = {
        "title": "Request",
        "description": "A message from a front end to the engine.",
        "oneOf": list[JsonValue](requests),
    }
    definitions["Reply"] = {
        "title": "Reply",
        "description": "A message from the engine to a front end.",
        "oneOf": list[JsonValue](replies),
    }
    return {
        "$schema": "https://json-schema.org/draft/2020-12/schema",
        "$id": self.ID,
        "title": "grounded-decision session protocol",
        "description": (
            "Messages between a front end and the grounded-decision engine. "
            "Generated from grounded_decision.session by `make protocol`; "
            "do not edit."
        ),
        "version": self.version,
        "oneOf": [{"$ref": "#/$defs/Request"}, {"$ref": "#/$defs/Reply"}],
        "$defs": definitions,
    }

notation

How a person names a move: parsing typed text, and the text for each legal move.

The terminal prompt (:class:~grounded_decision.examples.games.play.HumanAgent) and :meth:grounded_decision.session.Session.submit parse through the same :class:MoveNotation, so a move typed at the command line and one sent by the web front end mean the same thing.

MoveNotation dataclass

MoveNotation(game: Any)

A game's move notation: its parse_move, or numbers into the legal moves.

Games with parse_move(state, text) take their own notation (e5, 11-15, w); others take the 1-based number of a move in the list of legal moves.

Attributes:

Name	Type	Description
game	Any	The rules (any object with legal_moves and describe_move).

numbered property

numbered: bool

Whether moves are typed as numbers (the game has no parse_move).

parse

parse(
    state: Any,
    text: str,
    legal: Sequence[Any] | None = None,
) -> Any

The move text names, or None; it may be illegal (the gate judges it).

Source code in src/grounded_decision/session/notation.py

def parse(self, state: Any, text: str, legal: Sequence[Any] | None = None) -> Any:
    """The move ``text`` names, or ``None``; it may be illegal (the gate judges it)."""

    parser = self.parser
    if parser is not None:
        return parser(state, text)
    if not text.strip().isdigit():
        return None
    moves = list(self.game.legal_moves(state)) if legal is None else legal
    index = int(text) - 1
    return moves[index] if 0 <= index < len(moves) else None

describe

describe(state: Any, move: Any) -> str

The game's description of move, or str(move) if describing fails.

Source code in src/grounded_decision/session/notation.py

def describe(self, state: Any, move: Any) -> str:
    """The game's description of ``move``, or ``str(move)`` if describing fails."""

    try:
        return str(self.game.describe_move(state, move))
    except Exception:  # noqa: BLE001 - a description must never stop a match
        return str(move)

notations

notations(
    state: Any, legal: Sequence[Any]
) -> tuple[str | None, ...]

For each legal move, a text :meth:parse turns back into exactly that move.

A game's optional move_notation(state, move) hook is used when present; otherwise the move's description, or the first part of it, or one of its words, whichever parses back. Every text is checked by parsing it, so None (no text found) never names the wrong move.

Source code in src/grounded_decision/session/notation.py

def notations(self, state: Any, legal: Sequence[Any]) -> tuple[str | None, ...]:
    """For each legal move, a text :meth:`parse` turns back into exactly that move.

    A game's optional ``move_notation(state, move)`` hook is used when
    present; otherwise the move's description, or the first part of
    it, or one of its words, whichever parses back. Every text is
    checked by parsing it, so ``None`` (no text found) never names
    the wrong move.
    """

    if self.numbered:
        return tuple(str(number) for number in range(1, len(legal) + 1))
    hook = getattr(self.game, "move_notation", None)
    found: list[str | None] = []
    for move in legal:
        candidates: list[str] = []
        if callable(hook):
            candidates.append(str(hook(state, move)))
        candidates.extend(self._candidates(self.describe(state, move)))
        found.append(
            next(
                (text for text in candidates if self._names(state, text, move)),
                None,
            )
        )
    return tuple(found)

catalogue

What can be played: the :class:Catalogue returned by describe.

Everything comes from the game modules' contract constants (AGENT_DESCRIPTIONS, AGENT_DETAILS, CHECK_DESCRIPTIONS, WORLDS, BOARDS …), the same text that play GAME --list, --explain and docs/reference/agents.md print, so the web front end and the command line describe every game in the same words.

COUNT_FLAGS module-attribute

COUNT_FLAGS: dict[str, RuleFlag] = {
    "players": RuleFlag(
        "players",
        "the number of players",
        kind="int",
        label="Players",
    ),
    "target_length": RuleFlag(
        "until_length",
        "snake: stop at this length",
        kind="int",
        label="Stop at length",
    ),
}

The two counts set by their own flags (--players, --until-length), not rule flags.

Named dataclass

Named(name: str, description: str)

A name and one sentence about it (a check, a world, a board variant).

Attributes:

Name	Type	Description
name	str	The name to send back (in a spec or a check result).
description	str	One sentence.

AgentInfo dataclass

AgentInfo(
    name: str,
    description: str,
    details: str,
    regime: Regime | None,
    gated: bool,
)

One agent a seat can hold.

Attributes:

Name	Type	Description
name	str	The name to put in a spec's agents.
description	str	One line.
details	str	A paragraph on how it decides and what it guarantees.
regime	Regime | None	Its information regime, or null for a baseline.
gated	bool	Whether it is the game's gated agent (propose, check, act or fall back).

SizePreset dataclass

SizePreset(name: str, width: int, height: int)

A named board size.

Attributes:

Name	Type	Description
name	str	The name a spec's size accepts.
width	int	Columns.
height	int	Rows.

RuleCondition dataclass

RuleCondition(option: str, value: str)

A rule option's value another option needs (Tetris's level: mode realtime).

Attributes:

Name	Type	Description
option	str	The configure keyword (mode).
value	str	Its value, as the command line writes it (realtime).

RuleInfo dataclass

RuleInfo(
    name: str,
    flag: str,
    label: str,
    kind: str,
    choices: tuple[str, ...] = (),
    span: Span | None = None,
    values: str = "",
    default: str = "",
    help: str = "",
    requires: tuple[RuleCondition, ...] = (),
    zero_is_off: bool = False,
)

One rule option a game takes, as a front end offers it (protocol 1.11).

Attributes:

Name	Type	Description
name	str	The configure keyword (lock_delay).
flag	str	The command-line flag without dashes (lock-delay), also the URL parameter.
label	str	A few words naming it in a form.
kind	str	int, text, switch (on or off), choice (one of choices) or file (a path: not for a browser).
choices	tuple[str, ...]	The values a choice takes, or the only numbers an int takes (2048's winning tiles).
span	Span | None	The whole numbers an int takes, when it has a range.
values	str	What it takes, in a few words (1 to 9), for a hint; empty for free text or a switch.
default	str	configure's default as the command line writes it; empty when it depends on the rest (Minesweeper's mines).
help	str	What it means, in the words of --help.
requires	tuple[RuleCondition, ...]	Option values it applies under; a front end offers it only then.
zero_is_off	bool	0 turns the rule off (--stall-plies 0): the spec's rules carry null for it.

all_of classmethod

all_of(entry: GameModule) -> tuple[RuleInfo, ...]

Every option of entry's configure a front end can set: the counts, then the rule flags in --help order (the order a command lists them).

Source code in src/grounded_decision/session/catalogue.py

@classmethod
def all_of(cls, entry: GameModule) -> tuple[RuleInfo, ...]:
    """Every option of ``entry``'s ``configure`` a front end can set: the counts,
    then the rule flags in ``--help`` order (the order a command lists them)."""

    rules = entry.option_rules
    defaults = entry.option_defaults
    variants = {
        "board": entry.variants("BOARDS"),
        "start": entry.variants("STARTS"),
        "layout": entry.variants("LAYOUTS"),
    }
    found: list[RuleInfo] = []
    order = [*COUNT_FLAGS, *FLAGS_BY_NAME]
    taken = set(entry.configure_options)
    for name in [name for name in order if name in taken]:
        flag = COUNT_FLAGS.get(name) or FLAGS_BY_NAME.get(name)
        if flag is None:
            continue  # not a command-line option (the world has its own setting)
        rule = rules.get(name, OptionRule())
        named = tuple(variants.get(name, {}))
        choices = rule.choices or flag.choices or named
        default = defaults.get(name)
        if not rule.offered:
            kind = "file"
        elif flag.kind in ("switch", "int"):
            kind = flag.kind
        else:
            kind = "choice" if choices else "text"
        found.append(
            cls(
                name=name,
                flag=flag.flag.removeprefix("--"),
                label=flag.label or name.replace("_", " ").capitalize(),
                kind=kind,
                choices=choices,
                span=rule.span,
                values=rule.values(flag),
                default=""
                if default is None or isinstance(default, bool)
                else str(default),
                help=flag.help,
                requires=tuple(
                    RuleCondition(option, value) for option, value in rule.requires
                ),
                zero_is_off=flag.zero_is_none,
            )
        )
    return tuple(found)

SizeOffer dataclass

SizeOffer(
    rule: SizeRule,
    description: str,
    option: str = "",
    value: str = "",
)

The sizes a front end offers for a game's board, under one option value (protocol 1.9).

Attributes:

Name	Type	Description
rule	SizeRule	Every size the board may take, its default and the usual ones (SIZES under option = value; no cases).
description	str	The rule in one line (Lines on a side: 2 to 25), for a hint beside the setting.
option	str	The rule option (a configure keyword, board) whose value this offer is for; empty for the offer without one.
value	str	That option's value (donut).

all_of classmethod

all_of(rule: SizeRule | None) -> tuple[SizeOffer, ...]

The offer without an option, then one per case; empty for a fixed size.

Source code in src/grounded_decision/session/catalogue.py

@classmethod
def all_of(cls, rule: SizeRule | None) -> tuple[SizeOffer, ...]:
    """The offer without an option, then one per case; empty for a fixed size."""

    if rule is None:
        return ()
    offers = [cls(rule.effective(), rule.describe())]
    for case in rule.cases:
        narrowed = rule.effective({case.option: case.value})
        offers.append(cls(narrowed, narrowed.describe(), case.option, case.value))
    return tuple(offers)

GameInfo dataclass

GameInfo(
    name: str,
    title: str,
    players: tuple[str, ...],
    agents: tuple[AgentInfo, ...],
    default_agents: tuple[str, ...],
    default_mode: str,
    checks: tuple[Named, ...],
    worlds: tuple[Named, ...],
    boards: tuple[Named, ...],
    starts: tuple[Named, ...],
    layouts: tuple[Named, ...],
    sizes: tuple[SizePreset, ...],
    options: tuple[str, ...],
    guardable: bool,
    notation: bool,
    oracles: tuple[Named, ...] = (),
    size_offers: tuple[SizeOffer, ...] = (),
    rules: tuple[RuleInfo, ...] = (),
)

One game, as a front end offers it.

Attributes:

Name	Type	Description
name	str	The name a spec's game takes.
title	str	A one-line title.
players	tuple[str, ...]	The default seats' player names, in turn order.
agents	tuple[AgentInfo, ...]	Every agent, in display order.
default_agents	tuple[str, ...]	The default line-up, one per seat.
default_mode	str	watch (agents play) or you (a person plays the first seat).
checks	tuple[Named, ...]	Every check the gated agent runs on a proposal.
worlds	tuple[Named, ...]	Who may control the randomness (empty: nothing random to control); the first is the default.
boards	tuple[Named, ...]	Board variants (rule board); the first is the default.
starts	tuple[Named, ...]	Opening positions (rule start).
layouts	tuple[Named, ...]	Screen layouts (rule layout).
sizes	tuple[SizePreset, ...]	Named sizes.
options	tuple[str, ...]	Rule options the game's configuration accepts.
guardable	bool	Whether a person's moves can be guarded by the gate.
notation	bool	Whether a person types moves in the game's notation (else the number of a legal move).
oracles	tuple[Named, ...]	The judges audit can use (protocol 1.3), the default first; empty for a game without audits.
size_offers	tuple[SizeOffer, ...]	Every size the board may take, the default, the usual ones and the rule in words (protocol 1.9): the first offer holds unless a later one's option has its value (a Hexxagon donut needs side 4). Empty for a fixed size, which a front end does not offer to change.
rules	tuple[RuleInfo, ...]	Every option of options a front end can set, with its label, kind, range, default and meaning (protocol 1.11).

RegimeInfo dataclass

RegimeInfo(name: Regime, label: str, description: str)

An information regime, in the words play GAME --explain uses (protocol 1.2).

Attributes:

Name	Type	Description
name	Regime	The regime, as agents and seats report it.
label	str	Its name for people (worst-case, clairvoyant (benchmark)).
description	str	What an agent in it may know about the future, and so what it can promise.

Catalogue dataclass

Catalogue(
    games: tuple[GameInfo, ...],
    regimes: tuple[RegimeInfo, ...],
)

Every game that can be played.

Attributes:

Name	Type	Description
games	tuple[GameInfo, ...]	The games, in the command line's order.
regimes	tuple[RegimeInfo, ...]	Every information regime an agent can declare (protocol 1.2).

jsonlines

play --view json: one JSON object per line, a projection of the session's frames.

The lines are start, one step per move and end. A step line is :meth:StepEvent.line of the move's decision and the stats of the position after it, and :meth:StepEvent.of_frame builds the same event from a session :class:~.frame.Frame, so the command line's JSON output equals the session layer's frames in this documented canonical form (docs/reference/protocol.md; a golden test compares them byte for byte).

Concepts and the classes that own them:

:class:StartEvent, :class:StepEvent, :class:EndEvent The three kinds of line, each with payload() and line(). :class:JsonLinesView The engine observer that writes them (--view json).

StartEvent dataclass

StartEvent(
    game: str,
    players: tuple[str, ...],
    agents: tuple[str, ...],
    world: str | None,
)

The first line: the game, its players and agents, and its world.

Attributes:

Name	Type	Description
game	str	The game's name.
players	tuple[str, ...]	Player names, in turn order.
agents	tuple[str, ...]	Agent names, in seat order.
world	str | None	The game's world, or None for a game without worlds.

StepEvent dataclass

StepEvent(
    index: int,
    decision: MoveDecision,
    stats: Mapping[str, Scalar] | None,
)

One move: who played what, the gate's verdict and checks, and the stats after it.

Attributes:

Name	Type	Description
index	int	The move's index (0 for the first move).
decision	MoveDecision	The move and how it was chosen.
stats	Mapping[str, Scalar] | None	The game's stats after the move, or None without stats.

of_frame classmethod

of_frame(frame: Frame) -> StepEvent

The event of the move that led to frame.

Raises:

Type	Description
ValueError	frame is a starting position (no move led to it).

Source code in src/grounded_decision/session/jsonlines.py

@classmethod
def of_frame(cls, frame: Frame) -> StepEvent:
    """The event of the move that led to ``frame``.

    Raises:
        ValueError: ``frame`` is a starting position (no move led to it).
    """

    if frame.decision is None:
        raise ValueError("the starting frame has no move")
    return cls(frame.index - 1, frame.decision, frame.panel.stats)

EndEvent dataclass

EndEvent(summary: Mapping[str, Any])

The last line: moves, overrides, whether the budget cut it off, the outcome.

Attributes:

Name	Type	Description
summary	Mapping[str, Any]	:meth:~grounded_decision.examples.games.engine.MatchRecord.summary.

JsonLinesView

JsonLinesView(write: Write)

One JSON object per line: start, one step per move, end.

Parameters:

Name	Type	Description	Default
write	Write	Where text goes.	required

Source code in src/grounded_decision/session/jsonlines.py

def __init__(self, write: Write) -> None:
    self.write = write

replay

Recorded games: the .gdr replay format (version 2), written and read.

A replay is a whole session's frames, so a front end can play a game back without running Python (the web plan's phase 2). The file is gzip of UTF-8 text, one JSON value per line:

1. the header (:class:ReplayHeader): format and protocol versions, the package version, the game, the :class:~.spec.SessionSpec and seed that reproduce it, the session's fixed info, the frame count, the keyframe interval, the SHA-256 of the canonical frame stream and the outcome;
2. one record per frame: a keyframe (the whole frame) every keyframe_interval frames, starting with frame 0, and in between a delta against the previous frame (:class:FrameDelta), so a reader seeks to any frame by decoding from the keyframe before it.

The reference is docs/reference/replay-format.md; the TypeScript reader is web/packages/replay. Everything here is pure (bytes in, bytes out); the command line does the file I/O.

Concepts and the classes that own them:

:class:FrameStream A session's frames in order, their canonical text and its hash. :class:JsonPatch The difference between two JSON values as set operations (:class:SetOperation) and array splices (:class:SpliceOperation), and applying it. :class:FrameDelta One frame against the previous one: a patch of everything but the transition, and each transition picture patched against the picture before it. :class:ReplayHeader The first line of a replay. :class:Replay A header and its frames: recorded from a spec, encoded to .gdr bytes, decoded and verified. :class:ReplayError Bytes that are not a valid replay, or a replay that fails a check.

FORMAT module-attribute

FORMAT = 'gdr'

The format field of every replay header.

FORMAT_VERSION module-attribute

FORMAT_VERSION = 2

The .gdr format version this module writes and reads.

DEFAULT_KEYFRAME_INTERVAL module-attribute

DEFAULT_KEYFRAME_INTERVAL = 256

A whole frame every this many frames: seeking applies at most this many frames' changes, and a larger interval saves little (a gallery snake: 40 KB every 32, 35 KB every 256, 34 KB every 1,024).

Same module-attribute

Same = Callable[[JsonValue, JsonValue], bool]

How a diff compares two values: :func:_equal (fast) or :func:_same (exact).

ReplayError

Bases: ValueError

Not a valid .gdr replay, or a replay that fails verification.

FrameStream dataclass

FrameStream(frames: tuple[Frame, ...])

A session's frames, in order, and the hash that identifies them.

The canonical stream is each frame's wire JSON as :func:canonical_json, one per line, each line ending in \n; its SHA-256 is the replay's content hash. It depends only on the frames, never on how a file encodes them (keyframe interval, deltas, compression).

Attributes:

Name	Type	Description
frames	tuple[Frame, ...]	The frames, frame 0 (the starting position) first.

wire

wire() -> Iterator[JsonObject]

Each frame's wire JSON.

Source code in src/grounded_decision/session/replay.py

def wire(self) -> Iterator[JsonObject]:
    """Each frame's wire JSON."""

    for frame in self.frames:
        yield cast("JsonObject", encode(frame))

content_hash

content_hash() -> str

sha256: and the hex digest of the canonical frame stream.

Source code in src/grounded_decision/session/replay.py

def content_hash(self) -> str:
    """``sha256:`` and the hex digest of the canonical frame stream."""

    return self.hash_of(self.wire())

hash_of staticmethod

hash_of(frames: Iterable[JsonValue]) -> str

The content hash of frames in their wire JSON (no protocol objects built).

Source code in src/grounded_decision/session/replay.py

@staticmethod
def hash_of(frames: Iterable[JsonValue]) -> str:
    """The content hash of frames in their wire JSON (no protocol objects built)."""

    digest = hashlib.sha256()
    for value in frames:
        digest.update((canonical_json(value) + "\n").encode("utf-8"))
    return HASH_PREFIX + digest.hexdigest()

SetOperation dataclass

SetOperation(path: Path, value: JsonValue)

Replace what is at path with value (the empty path replaces the whole value).

Its file form is [path, value].

SpliceOperation dataclass

SpliceOperation(
    path: Path,
    start: int,
    delete: int,
    items: tuple[JsonValue, ...],
)

In the array at path, replace delete items from start with items.

A list that shifts (a snake's body gains a head and loses its tail, a sparkline's window moves on, a newest-first event list takes a new event) is two small splices instead of the whole list. Its file form is [path, start, delete, items] (format 2).

AddOperation dataclass

AddOperation(path: Path, amount: int)

Add amount to the whole number at path.

A count that goes up by one every move (moves played, acts) is the same operation every time, which gzip stores in a few bits, where setting each new value is new text every move. Its file form is [path, "+", amount] (format 2).

JsonPatch dataclass

JsonPatch(operations: tuple[Operation, ...])

How one JSON value becomes another: set and splice operations, in order.

A path is a list of object keys and array indices from the root. A set replaces what is at its path; a splice replaces part of the array at its path (:class:SpliceOperation). :meth:between writes the shortest it finds: it descends into objects with the same keys, and for an array chooses the shortest of its items' own changes (same length), one splice around the common start and end, two splices for a list shifted by a few places, or the whole array.

Attributes:

Name	Type	Description
operations	tuple[Operation, ...]	The operations, applied in order.

SHIFTS class-attribute

SHIFTS: int = 4

How far an array may have moved (in places, either way) for :meth:between to splice it.

MATCHED class-attribute

MATCHED: int = 4000

The most items (old and new together) :meth:between matches one by one.

between classmethod

between(old: JsonValue, new: JsonValue) -> JsonPatch

The patch from old to new, exact to the JSON text.

The diff compares with == (at C speed), which cannot tell 1, 1.0 and true apart; the result is then proved once, on the whole value (applied to old, it gives new's exact text), and only a value where == hid such a difference is compared exactly (:func:_same) all the way down.

Source code in src/grounded_decision/session/replay.py

@classmethod
def between(cls, old: JsonValue, new: JsonValue) -> JsonPatch:
    """The patch from ``old`` to ``new``, exact to the JSON text.

    The diff compares with ``==`` (at C speed), which cannot tell ``1``,
    ``1.0`` and ``true`` apart; the result is then proved once, on the
    whole value (applied to ``old``, it gives ``new``'s exact text), and only
    a value where ``==`` hid such a difference is compared exactly
    (:func:`_same`) all the way down.
    """

    operations: list[Operation] = []
    cls._diff(old, new, (), operations, _equal)
    patch = cls(tuple(operations))
    if canonical_json(patch.apply(old)) == canonical_json(new):
        return patch
    operations = []
    cls._diff(old, new, (), operations, _same)
    return cls(tuple(operations))

apply

apply(value: JsonValue) -> JsonValue

value with every operation applied; value itself is never changed.

Source code in src/grounded_decision/session/replay.py

def apply(self, value: JsonValue) -> JsonValue:
    """``value`` with every operation applied; ``value`` itself is never changed."""

    for operation in self.operations:
        value = operation.apply(value)
    return value

at classmethod

at(
    value: JsonValue,
    path: Path,
    change: Callable[[JsonValue], JsonValue],
) -> JsonValue

value with what is at path replaced by change of it (copy on write).

Source code in src/grounded_decision/session/replay.py

@classmethod
def at(
    cls, value: JsonValue, path: Path, change: Callable[[JsonValue], JsonValue]
) -> JsonValue:
    """``value`` with what is at ``path`` replaced by ``change`` of it (copy on write)."""

    if not path:
        return change(value)
    head, rest = path[0], path[1:]
    if isinstance(value, list) and isinstance(head, int) and 0 <= head < len(value):
        copied = list(value)
        copied[head] = cls.at(value[head], rest, change)
        return copied
    if isinstance(value, dict) and isinstance(head, str) and head in value:
        copied_object = dict(value)
        copied_object[head] = cls.at(value[head], rest, change)
        return copied_object
    raise ReplayError(
        f"a delta changes {list(path)!r}, which the frame does not have"
    )

of_wire classmethod

of_wire(value: JsonValue) -> JsonPatch

A patch from its JSON form, checked: [path, value] sets and [path, start, delete, items] splices.

Source code in src/grounded_decision/session/replay.py

@classmethod
def of_wire(cls, value: JsonValue) -> JsonPatch:
    """A patch from its JSON form, checked: ``[path, value]`` sets and
    ``[path, start, delete, items]`` splices."""

    if not isinstance(value, list):
        raise ReplayError("a delta's operations must be an array")
    return cls(tuple(cls.operation_of_wire(item) for item in value))

operation_of_wire staticmethod

operation_of_wire(item: JsonValue) -> Operation

One operation from its JSON form, checked.

Source code in src/grounded_decision/session/replay.py

@staticmethod
def operation_of_wire(item: JsonValue) -> Operation:
    """One operation from its JSON form, checked."""

    if not (isinstance(item, list) and len(item) in (2, 3, 4)):
        raise ReplayError(
            'each delta operation is [path, value], [path, "+", amount] '
            "or [path, start, delete, items]"
        )
    path = item[0]
    if not isinstance(path, list) or not all(
        isinstance(part, str | int) and not isinstance(part, bool) for part in path
    ):
        raise ReplayError("a delta path is an array of keys and indices")
    steps = tuple(cast("list[str | int]", path))
    if len(item) == 2:
        return SetOperation(steps, item[1])
    if len(item) == 3:
        amount = item[2]
        if (
            item[1] != "+"
            or not isinstance(amount, int)
            or isinstance(amount, bool)
        ):
            raise ReplayError('an add is [path, "+", whole number]')
        return AddOperation(steps, amount)
    start, delete, items = item[1], item[2], item[3]
    if not (
        isinstance(start, int)
        and not isinstance(start, bool)
        and isinstance(delete, int)
        and not isinstance(delete, bool)
        and isinstance(items, list)
    ):
        raise ReplayError(
            "a splice is [path, start, delete, items] with whole numbers"
        )
    return SpliceOperation(steps, start, delete, tuple(items))

FrameDelta dataclass

FrameDelta(
    patch: JsonPatch,
    transition: tuple[tuple[float, JsonPatch], ...],
)

One frame as a change to the frame before it.

patch turns the previous frame (without its transition, and with its index one higher, as frames are consecutive) into this one (without its transition): the index is never stored. The transition's pictures are usually small changes of the previous position, so each picture's board is a patch of the board before it: the first against the previous frame's board, each next one against the picture before.

Attributes:

Name	Type	Description
patch	JsonPatch	Everything but the transition.
transition	tuple[tuple[float, JsonPatch], ...]	Each transition picture: its weight and its board's patch.

Channel dataclass

Channel(
    keys: tuple[str, ...],
    changes: tuple[tuple[int, Operation], ...],
)

Every change to one part of the frame, in order (one line of the file).

A part is named by the leading object keys of the paths it changes, at most two (board.cells, decision.metrics, arrows): the same kind of change lies together, which compresses far better than each frame's changes side by side. Its line is {"c": keys, "o": changes}, each change [gap, operation]: frames since the channel's last change, then the operation with its path after keys.

Attributes:

Name	Type	Description
keys	tuple[str, ...]	The leading keys every path here starts with.
changes	tuple[tuple[int, Operation], ...]	(frame index, operation) with whole paths, in frame order.

DEPTH class-attribute

DEPTH: int = 2

At most this many leading keys name a channel.

ReplayBody dataclass

ReplayBody(
    keyframes: tuple[JsonObject, ...],
    channels: tuple[Channel, ...],
    transitions: tuple[
        tuple[int, tuple[tuple[float, JsonPatch], ...]], ...
    ],
)

A replay's frames as the file stores them (format 2), after the header.

Lines, in this order: every keyframe ({"k": frame}, frame 0 and each multiple of the keyframe interval), one line per :class:Channel, and the transitions ({"t": [[gap, pictures], ...]}, each picture [weight, operations] against the board before it). Every frame that is not a keyframe is the frame before it, with the next index and no transition, changed by its operations in every channel and given its pictures.

The writer never changes, in one frame, both a path and a path inside it (:meth:JsonPatch.between either sets a value or changes its parts), so a frame's operations from different channels can be applied in any order; within a channel they keep theirs.

Attributes:

Name	Type	Description
keyframes	tuple[JsonObject, ...]	The keyframes, in frame order.
channels	tuple[Channel, ...]	Every channel, by its keys.
transitions	tuple[tuple[int, tuple[tuple[float, JsonPatch], ...]], ...]	(frame index, pictures) for every frame not a keyframe whose transition is not empty.

frames

frames(count: int, interval: int) -> Iterator[JsonObject]

The count frames, in order, as wire JSON.

Source code in src/grounded_decision/session/replay.py

def frames(self, count: int, interval: int) -> Iterator[JsonObject]:
    """The ``count`` frames, in order, as wire JSON."""

    expected = (count + interval - 1) // interval
    if len(self.keyframes) != expected:
        raise ReplayError(
            f"{count} frames every {interval} need {expected} keyframes; "
            f"the file has {len(self.keyframes)}"
        )
    operations: dict[int, list[Operation]] = {}
    for channel in self.channels:
        for frame, operation in channel.changes:
            if not 0 < frame < count or frame % interval == 0:
                raise ReplayError(
                    f"channel {'.'.join(channel.keys)} changes frame {frame}, "
                    "which is a keyframe or past the last frame"
                )
            operations.setdefault(frame, []).append(operation)
    pictures = dict(self.transitions)
    value: JsonObject = self.keyframes[0]
    for index in range(count):
        if index % interval == 0:
            value = self.keyframes[index // interval]
        else:
            delta = FrameDelta(
                JsonPatch(tuple(operations.get(index, ()))), pictures.get(index, ())
            )
            value = delta.apply(value)
        yield value

of_lines classmethod

of_lines(lines: Sequence[str], first: int) -> ReplayBody

The body from its lines; first is the first line's number in the file.

Source code in src/grounded_decision/session/replay.py

@classmethod
def of_lines(cls, lines: Sequence[str], first: int) -> ReplayBody:
    """The body from its lines; ``first`` is the first line's number in the file."""

    keyframes: list[JsonObject] = []
    channels: list[Channel] = []
    transitions: list[tuple[int, tuple[tuple[float, JsonPatch], ...]]] = []
    for number, line in enumerate(lines, start=first):
        record = Replay.json_line(line, number)
        if not isinstance(record, dict):
            raise ReplayError(f"line {number}: a record must be an object")
        try:
            if "k" in record:
                keyframe = record["k"]
                if not isinstance(keyframe, dict) or channels or transitions:
                    raise ReplayError("keyframes are objects, before every channel")
                keyframes.append(keyframe)
            elif "c" in record:
                channels.append(Channel.of_wire(record))
            elif "t" in record:
                transitions = cls._transitions(record["t"])
            else:
                raise ReplayError(
                    "a record is a keyframe, a channel or the transitions"
                )
        except ReplayError as error:
            raise ReplayError(f"line {number}: {error}") from None
    if not keyframes:
        raise ReplayError("a replay needs its first frame")
    return cls(tuple(keyframes), tuple(channels), tuple(transitions))

ReplayHeader dataclass

ReplayHeader(
    format_version: int,
    protocol_version: str,
    package_version: str,
    game: str,
    spec: SessionSpec,
    seed: int,
    session: SessionInfo,
    frame_count: int,
    keyframe_interval: int,
    content_hash: str,
    outcome: Outcome,
)

The first line of a .gdr file: what was played and how to check it.

Attributes:

Name	Type	Description
format_version	int	The file format's version (:data:FORMAT_VERSION).
protocol_version	str	The protocol version of the frames.
package_version	str	The grounded-decision version that recorded it.
game	str	The game's name.
spec	SessionSpec	Everything that fixes the match; recording it again gives the same frames.
seed	int	The seed (spec.seed).
session	SessionInfo	What stays the same for the session: seats, world, the panel's fixed rows.
frame_count	int	Number of frames (moves played + 1).
keyframe_interval	int	A whole frame every this many frames.
content_hash	str	sha256: and the hex digest of the canonical frame stream (:meth:FrameStream.content_hash).
outcome	Outcome	The last frame's outcome.

of_wire classmethod

of_wire(value: JsonValue) -> ReplayHeader

A header from its JSON line, validated; unknown fields are ignored.

Raises:

Type	Description
ReplayError	Not a .gdr header, a format version this reader does not know, or a field of the wrong type.

Source code in src/grounded_decision/session/replay.py

@classmethod
def of_wire(cls, value: JsonValue) -> ReplayHeader:
    """A header from its JSON line, validated; unknown fields are ignored.

    Raises:
        ReplayError: Not a ``.gdr`` header, a format version this
            reader does not know, or a field of the wrong type.
    """

    if not isinstance(value, dict) or value.get("format") != FORMAT:
        raise ReplayError("not a .gdr replay (no gdr header)")
    version = value.get("format_version")
    if version != FORMAT_VERSION or isinstance(version, bool):
        raise ReplayError(
            f"replay format version {version!r}; this reader reads {FORMAT_VERSION} "
            "(record the game again with this version)"
        )
    return cls(
        format_version=FORMAT_VERSION,
        protocol_version=cls._text(value, "protocol_version"),
        package_version=cls._text(value, "package_version"),
        game=cls._text(value, "game"),
        spec=cls._record(SessionSpec, value, "spec"),
        seed=cls._integer(value, "seed"),
        session=cls._record(SessionInfo, value, "session"),
        frame_count=cls._integer(value, "frame_count"),
        keyframe_interval=cls._integer(value, "keyframe_interval"),
        content_hash=cls._text(value, "content_hash"),
        outcome=cls._record(Outcome, value, "outcome"),
    )

Replay dataclass

Replay(header: ReplayHeader, frames: tuple[Frame, ...])

A recorded session: its header and every frame.

Make one with :meth:record (play a spec to the end) or :meth:decode (read .gdr bytes); :meth:encode writes the bytes.

Attributes:

Name	Type	Description
header	ReplayHeader	What was played, and the hash and outcome of its frames.
frames	tuple[Frame, ...]	Every frame, frame 0 first.

record classmethod

record(
    spec: SessionSpec,
    *,
    keyframe_interval: int = DEFAULT_KEYFRAME_INTERVAL,
    package_version: str | None = None,
) -> Replay

Play spec to its end (or its move budget) and keep every frame.

Raises:

Type	Description
ReplayError	A seat is a person (a replay records agents only).

Source code in src/grounded_decision/session/replay.py

@classmethod
def record(
    cls,
    spec: SessionSpec,
    *,
    keyframe_interval: int = DEFAULT_KEYFRAME_INTERVAL,
    package_version: str | None = None,
) -> Replay:
    """Play ``spec`` to its end (or its move budget) and keep every frame.

    Raises:
        ReplayError: A seat is a person (a replay records agents only).
    """

    if spec.you or "you" in spec.seats:
        raise ReplayError(
            "a replay records agents only: remove `you` from the seats"
        )
    session = Session.open(spec)
    frames = [session.frame]
    while not session.over:
        frames.extend(session.run(1_000))
    return cls.of_frames(
        spec,
        session.info,
        frames,
        keyframe_interval=keyframe_interval,
        package_version=package_version,
    )

lines

lines() -> Iterator[str]

The uncompressed text, line by line: the header, then the body (:class:ReplayBody).

Source code in src/grounded_decision/session/replay.py

def lines(self) -> Iterator[str]:
    """The uncompressed text, line by line: the header, then the body (:class:`ReplayBody`)."""

    # The version this writer writes, whatever version the replay was read from.
    header = replace(self.header, format_version=FORMAT_VERSION)
    yield canonical_json(header.wire()) + "\n"
    yield from self.body().lines()

body

body() -> ReplayBody

The frames as the file stores them: keyframes, channels and transitions.

Source code in src/grounded_decision/session/replay.py

def body(self) -> ReplayBody:
    """The frames as the file stores them: keyframes, channels and transitions."""

    return ReplayBody.of_frames(
        list(FrameStream(self.frames).wire()), self.header.keyframe_interval
    )

encode

encode() -> bytes

The .gdr bytes: gzip (level 9, no timestamp or name) of :meth:text.

Source code in src/grounded_decision/session/replay.py

def encode(self) -> bytes:
    """The ``.gdr`` bytes: gzip (level 9, no timestamp or name) of :meth:`text`."""

    return _gzip(self.text())

decode classmethod

decode(data: bytes) -> Replay

A replay from .gdr bytes.

Raises:

Type	Description
ReplayError	Not gzip, not a replay, an unknown format version, a malformed record, or a frame that does not fit the protocol.

Source code in src/grounded_decision/session/replay.py

@classmethod
def decode(cls, data: bytes) -> Replay:
    """A replay from ``.gdr`` bytes.

    Raises:
        ReplayError: Not gzip, not a replay, an unknown format version,
            a malformed record, or a frame that does not fit the protocol.
    """

    try:
        text = gzip.decompress(data).decode("utf-8")
    except (OSError, EOFError, zlib.error, UnicodeDecodeError) as error:
        raise ReplayError(f"not a .gdr replay ({error})") from None
    return cls.of_text(text)

json_line staticmethod

json_line(line: str, number: int) -> JsonValue

Line number of a file, parsed.

Source code in src/grounded_decision/session/replay.py

@staticmethod
def json_line(line: str, number: int) -> JsonValue:
    """Line ``number`` of a file, parsed."""

    try:
        return cast("JsonValue", json.loads(line))
    except json.JSONDecodeError as error:
        raise ReplayError(f"line {number}: not JSON ({error.msg})") from None

problems

problems() -> tuple[str, ...]

What is wrong with this replay (empty when sound): its frames against its header (:meth:hash_problems), then every arrow that breaks :class:~grounded_decision.session.arrows.ArrowRule.

Source code in src/grounded_decision/session/replay.py

def problems(self) -> tuple[str, ...]:
    """What is wrong with this replay (empty when sound): its frames against
    its header (:meth:`hash_problems`), then every arrow that breaks
    :class:`~grounded_decision.session.arrows.ArrowRule`."""

    return self.hash_problems() + tuple(
        f"move {report.index}: {problem}"
        for report in self.arrow_reports()
        for problem in report.problems
    )

hash_problems

hash_problems() -> tuple[str, ...]

What is wrong with this replay's frames against its header (empty when sound).

Source code in src/grounded_decision/session/replay.py

def hash_problems(self) -> tuple[str, ...]:
    """What is wrong with this replay's frames against its header (empty when sound)."""

    problems: list[str] = []
    actual = FrameStream(self.frames).content_hash()
    if actual != self.header.content_hash:
        problems.append(
            f"content hash {actual} does not match the header's {self.header.content_hash}"
        )
    if self.frames[-1].outcome != self.header.outcome:
        problems.append("the last frame's outcome does not match the header's")
    return tuple(problems)

byte_problems classmethod

byte_problems(data: bytes) -> tuple[str, ...]

What is wrong with .gdr bytes against their own header (empty when sound), without building the frames' protocol objects: the frames the body decodes to must hash to the header's content hash, and the last one's outcome must be the header's. A quick check of a file just written; :meth:decode then :meth:problems checks the frames too.

Raises:

Type	Description
ReplayError	The bytes are not a readable replay.

Source code in src/grounded_decision/session/replay.py

@classmethod
def byte_problems(cls, data: bytes) -> tuple[str, ...]:
    """What is wrong with ``.gdr`` bytes against their own header (empty when
    sound), without building the frames' protocol objects: the frames the
    body decodes to must hash to the header's content hash, and the last
    one's outcome must be the header's. A quick check of a file just
    written; :meth:`decode` then :meth:`problems` checks the frames too.

    Raises:
        ReplayError: The bytes are not a readable replay.
    """

    try:
        text = gzip.decompress(data).decode("utf-8")
    except (OSError, EOFError, zlib.error, UnicodeDecodeError) as error:
        raise ReplayError(f"not a .gdr replay ({error})") from None
    lines = text.splitlines()
    if not lines:
        raise ReplayError("not a .gdr replay (empty)")
    header = ReplayHeader.of_wire(cls.json_line(lines[0], 1))
    frames = list(
        ReplayBody.of_lines(lines[1:], first=2).frames(
            header.frame_count, header.keyframe_interval
        )
    )
    problems: list[str] = []
    actual = FrameStream.hash_of(frames)
    if actual != header.content_hash:
        problems.append(
            f"content hash {actual} does not match the header's {header.content_hash}"
        )
    if frames[-1].get("outcome") != encode(header.outcome):
        problems.append("the last frame's outcome does not match the header's")
    return tuple(problems)

arrow_reports

arrow_reports() -> tuple[ArrowReport, ...]

Every move with its arrows, checked against the position before it.

Source code in src/grounded_decision/session/replay.py

def arrow_reports(self) -> tuple[ArrowReport, ...]:
    """Every move with its arrows, checked against the position before it."""

    return ArrowReport.all(self.frames)

reproduce

reproduce() -> Replay

The header's spec recorded again, with this replay's keyframe interval.

Source code in src/grounded_decision/session/replay.py

def reproduce(self) -> Replay:
    """The header's spec recorded again, with this replay's keyframe interval."""

    return Replay.record(
        self.header.spec,
        keyframe_interval=self.header.keyframe_interval,
        package_version=self.header.package_version,
    )

reproduced

reproduced() -> bool

Whether recording the header's spec again gives exactly these frames (by hash).

Source code in src/grounded_decision/session/replay.py

def reproduced(self) -> bool:
    """Whether recording the header's spec again gives exactly these frames (by hash)."""

    return self.reproduce().header.content_hash == self.header.content_hash

BytePart dataclass

BytePart(name: str, text: int, compressed: int)

One part of a replay's records and its bytes.

Attributes:

Name	Type	Description
name	str	header, keyframes, transition, or a channel's keys (board.paths, decision, panel.counters): each a line of the file but the keyframes, which are one line each.
text	int	Its uncompressed text, in bytes.
compressed	int	That text gzipped on its own, in bytes (the file is gzipped whole, so these add up to a little more than it).

ByteReport dataclass

ByteReport(
    size: int, text: int, parts: tuple[BytePart, ...]
)

Where a replay's bytes go, part by part (replay-info --bytes).

Attributes:

Name	Type	Description
size	int	The replay's .gdr bytes as this package writes it.
text	int	Its uncompressed text's bytes.
parts	tuple[BytePart, ...]	Every part, the most compressed bytes first.

lines

lines() -> list[str]

The report as text: one line per part.

Source code in src/grounded_decision/session/replay.py

def lines(self) -> list[str]:
    """The report as text: one line per part."""

    return [
        f"  {'part':<22} {'compressed':>12} {'text':>14}",
        *(
            f"  {part.name:<22} {part.compressed:>12,} {part.text:>14,}"
            for part in self.parts
        ),
        f"  {'file':<22} {self.size:>12,} {self.text:>14,}",
    ]

canonical_json

canonical_json(value: JsonValue) -> str

A JSON value as compact text with sorted keys: the form hashed and stored.

Source code in src/grounded_decision/session/replay.py

def canonical_json(value: JsonValue) -> str:
    """A JSON value as compact text with sorted keys: the form hashed and stored."""

    return json.dumps(
        value,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
        allow_nan=False,
    )

gallery

The curated replays: the site's gallery and the replay package's golden files.

make replays (python -m grounded_decision.session --replays) records them. They are committed, because the site is built without Python and the TypeScript tests read them; a test records every one again and fails when a committed file no longer matches (a rule or agent changed), so the set is always current.

Concepts and the classes that own them:

:class:RecordCommand The grounded-decision record command line that records a spec: shown on the site ("run it yourself") and parsed back by a test. :class:CuratedReplay One replay to keep: a name, a title, a spec and a keyframe interval. :class:ReplaySet A directory of curated replays and the files it holds: the .gdr files, and a manifest (the gallery) or each replay's frames as JSON (the golden files the TypeScript decoder is compared with).

GALLERY module-attribute

GALLERY = ReplaySet(
    GALLERY_DIRECTORY,
    (
        CuratedReplay(
            "tictactoe",
            "Tic-tac-toe: the rule of thumb against the gated player",
            "The gate checks every proposal against the exact solver; the game is a draw.",
            SessionSpec(
                "tictactoe",
                agents=("rule_of_thumb", "grounded"),
                seed=1,
            ),
        ),
        CuratedReplay(
            "snake",
            "Snake: the certified snake heads for each apple",
            "Apples land at random; every move is checked safe, and the snake grows.",
            SessionSpec(
                "snake",
                world="random",
                size="10x10",
                seed=1,
            ),
        ),
        CuratedReplay(
            "snake-adversarial",
            "Snake against partly adversarial apples",
            "Three apples in ten land where they hurt most. The snake's safety proof holds whatever the apples do; only its speed bets on random apples.",
            SessionSpec(
                "snake",
                world="adversarial:0.3",
                size="10x10",
                seed=1,
            ),
        ),
        CuratedReplay(
            "checkers",
            "Checkers: the gated player against the search",
            "Captures, kings and the gate's checks on every move.",
            SessionSpec("checkers", seed=1),
        ),
        CuratedReplay(
            "chinese-checkers-6",
            "Chinese checkers for six",
            "Six arms of the star race for the opposite corner; placings at the end.",
            SessionSpec(
                "chinese_checkers", players=6, seed=1
            ),
        ),
        CuratedReplay(
            "go-9x9",
            "Go on 9 x 9",
            "Captures and territory on the small board, to two passes.",
            SessionSpec("go", size="9x9", seed=1),
        ),
        CuratedReplay(
            "tetris-sz",
            "Tetris against S and Z pieces only",
            "The worst pieces there are; every placement checked against the preview.",
            SessionSpec("tetris", world="sz", seed=1),
        ),
        CuratedReplay(
            "hexxagon-ring",
            "Hexxagon: the ring trap",
            "The ring opening sets a trap on move 1; pieces flip on every capture.",
            SessionSpec(
                "hexxagon", rules={"start": "ring"}, seed=1
            ),
        ),
        CuratedReplay(
            "minesweeper-expert",
            "Minesweeper, expert board",
            "30 x 16 with 99 mines: proved-safe clicks first, the least risky guess otherwise.",
            SessionSpec(
                "minesweeper", size="expert", seed=1
            ),
        ),
        CuratedReplay(
            "2048-evil",
            "2048 in the evil world",
            "Every new tile lands where it hurts most.",
            SessionSpec("2048", world="evil", seed=1),
        ),
        CuratedReplay(
            "connect4-greedy-vs-gated",
            "Connect Four: greedy against the same player, gated",
            "Ungated, O loses in 7 moves. Gated, the gate turns down three moves that would lose, and O wins in 26.",
            SessionSpec(
                "connect4",
                agents=("greedy", "gated"),
                seed=15,
            ),
        ),
        CuratedReplay(
            "breakout-adversarial-spin",
            "Breakout against adversarial spin",
            "The spin works against the paddle. The gate keeps each bounce spin-proof while it can, and says when it cannot, a flight before a ball is lost.",
            SessionSpec(
                "breakout",
                world="adversarial",
                rules={
                    "epsilon": 32,
                    "paddle_speed": 12,
                    "ball_speed": 4,
                },
                seed=1,
            ),
        ),
    ),
)

The site's gallery: every game, with interesting worlds, sizes and openings.

GOLDEN module-attribute

GOLDEN = ReplaySet(
    GOLDEN_DIRECTORY,
    tuple(
        CuratedReplay(
            name,
            name,
            "A golden replay for the decoder's tests.",
            spec,
            4,
        )
        for name, spec in (
            (
                "tictactoe",
                SessionSpec(
                    "tictactoe",
                    agents=("rule_of_thumb", "grounded"),
                    seed=1,
                ),
            ),
            (
                "snake",
                SessionSpec(
                    "snake",
                    size="6x6",
                    seed=2,
                    max_steps=12,
                ),
            ),
            (
                "checkers",
                SessionSpec(
                    "checkers", seed=0, max_steps=4
                ),
            ),
            (
                "chinese-checkers",
                SessionSpec(
                    "chinese_checkers",
                    players=3,
                    seed=0,
                    max_steps=3,
                ),
            ),
            (
                "go",
                SessionSpec(
                    "go", size="9x9", seed=0, max_steps=3
                ),
            ),
            (
                "tetris",
                SessionSpec(
                    "tetris",
                    world="adversarial",
                    seed=0,
                    max_steps=2,
                ),
            ),
            (
                "hexxagon",
                SessionSpec(
                    "hexxagon", seed=1, max_steps=3
                ),
            ),
            (
                "minesweeper",
                SessionSpec(
                    "minesweeper", seed=6, max_steps=8
                ),
            ),
            (
                "2048",
                SessionSpec(
                    "2048",
                    size="3x3",
                    world="evil",
                    seed=3,
                    max_steps=5,
                ),
            ),
            (
                "connect4",
                SessionSpec(
                    "connect4", seed=0, max_steps=8
                ),
            ),
            (
                "breakout",
                SessionSpec(
                    "breakout",
                    rules={"epsilon": 8},
                    seed=0,
                    max_steps=12,
                ),
            ),
        )
    ),
    golden=True,
)

Small replays of every game (a keyframe every 4 frames) and their frames as JSON.

RecordCommand dataclass

RecordCommand(
    spec: SessionSpec, keyframe_interval: int | None = None
)

The grounded-decision record arguments that record a spec.

Attributes:

Name	Type	Description
spec	SessionSpec	What is recorded (no person may sit in it).
keyframe_interval	int | None	--keyframe-interval, or None for the default.

arguments

arguments(output: str) -> list[str]

The argument list after grounded-decision.

Raises:

Type	Description
ValueError	A rule option record has no flag for.

Source code in src/grounded_decision/session/gallery.py

def arguments(self, output: str) -> list[str]:
    """The argument list after ``grounded-decision``.

    Raises:
        ValueError: A rule option ``record`` has no flag for.
    """

    spec = self.spec
    words = ["record", spec.game]
    if spec.agents:
        words += ["--agents", ",".join(spec.agents)]
    if spec.world is not None:
        words += ["--world", spec.world]
    words += ["--seed", str(spec.seed)]
    for flag, value in (
        ("--size", spec.size),
        ("--players", spec.players),
        ("--until-length", spec.until_length),
        ("--max-steps", spec.max_steps),
    ):
        if value is not None:
            words += [flag, str(value)]
    for name, value in sorted(spec.rules.items()):
        flag = FLAGS_BY_NAME.get(name)
        if flag is None:
            raise ValueError(f"record has no option for the rule {name}={value!r}")
        words += flag.words(value)
    if self.keyframe_interval is not None:
        words += ["--keyframe-interval", str(self.keyframe_interval)]
    return [*words, "-o", output]

text

text(output: str) -> str

The whole command line, as a person would type it.

Source code in src/grounded_decision/session/gallery.py

def text(self, output: str) -> str:
    """The whole command line, as a person would type it."""

    return " ".join(["grounded-decision", *self.arguments(output)])

CuratedReplay dataclass

CuratedReplay(
    name: str,
    title: str,
    description: str,
    spec: SessionSpec,
    keyframe_interval: int = DEFAULT_KEYFRAME_INTERVAL,
)

One committed replay.

Attributes:

Name	Type	Description
name	str	The file's name without .gdr.
title	str	A short title for the gallery.
description	str	What to watch for, in a sentence.
spec	SessionSpec	What is recorded.
keyframe_interval	int	A whole frame every this many frames.

entry

entry(replay: Replay) -> JsonObject

This replay's line in a manifest (the gallery's or the ambient library's).

Source code in src/grounded_decision/session/gallery.py

def entry(self, replay: Replay) -> JsonObject:
    """This replay's line in a manifest (the gallery's or the ambient library's)."""

    header = replay.header
    return {
        "name": self.name,
        "file": self.file,
        "title": self.title,
        "description": self.description,
        "game": header.game,
        "command": self.command.text(self.file),
        "frame_count": header.frame_count,
        "content_hash": header.content_hash,
        "outcome": header.outcome.reason,
    }

ReplaySet dataclass

ReplaySet(
    directory: str,
    replays: tuple[CuratedReplay, ...],
    golden: bool = False,
)

A directory of curated replays.

Attributes:

Name	Type	Description
directory	str	Where the files go, relative to the repository root.
replays	tuple[CuratedReplay, ...]	The replays.
golden	bool	Also write each replay's frames as JSON (NAME.frames.json), the expected output of the TypeScript decoder; otherwise write the gallery's index.json manifest.

files

files(
    versions: dict[str, str] | None = None,
) -> dict[str, bytes]

Every file of the set, as {repository-relative path: bytes}.

versions maps a replay's name to the package version to put in its header (the committed one, when checking the set is current).

Source code in src/grounded_decision/session/gallery.py

def files(self, versions: dict[str, str] | None = None) -> dict[str, bytes]:
    """Every file of the set, as ``{repository-relative path: bytes}``.

    ``versions`` maps a replay's name to the package version to put in
    its header (the committed one, when checking the set is current).
    """

    versions = versions or {}
    files: dict[str, bytes] = {}
    entries: list[JsonValue] = []
    for curated in self.replays:
        replay = curated.record(versions.get(curated.name))
        files[self.path(curated.file)] = replay.encode()
        if self.golden:
            lines = ",\n".join(
                canonical_json(encode(frame)) for frame in replay.frames
            )
            files[self.path(f"{curated.name}.frames.json")] = (
                f"[\n{lines}\n]\n".encode()
            )
        entries.append(curated.entry(replay))
    if not self.golden:
        manifest: JsonObject = {
            "description": "The gallery's replays; written by `make replays`.",
            "replays": entries,
        }
        text = json.dumps(manifest, indent=2, sort_keys=True, ensure_ascii=False)
        files[self.path(self.MANIFEST)] = (text + "\n").encode()
    return files

payload staticmethod

payload(data: bytes) -> str

A .gdr file's text (compressed bytes differ between zlib builds; text does not).

Source code in src/grounded_decision/session/gallery.py

@staticmethod
def payload(data: bytes) -> str:
    """A ``.gdr`` file's text (compressed bytes differ between zlib builds; text does not)."""

    return gzip.decompress(data).decode("utf-8")

wire

The protocol's wire form: dataclasses to JSON values, back, and to JSON Schema.

The protocol's messages are frozen dataclasses (:mod:.frame, :mod:.protocol); this module is the one place that knows how their Python types look in JSON, so encoding, decoding and the published schema cannot disagree.

Concepts and the classes that own them:

:class:WireEncoder Protocol values to JSON values, with each dataclass's wire fields looked up once (:func:encode is the one-off form). :class:WireType One Python type annotation, as the wire sees it: how to encode a value of it, decode (and validate) a JSON value into it, and describe it in JSON Schema. Built from an annotation by :meth:WireType.of. :class:SchemaBuilder Collects the $defs of every dataclass and enumeration reachable from the messages, in a stable order. :class:FieldDocs Field descriptions read from a dataclass's Attributes: docstring section, so the schema (and the TypeScript generated from it) carries the same documentation as the Python. :class:WireError A JSON value that does not fit its type, with the path to the problem.

The supported types are what the protocol needs and no more: str, int, float, bool, None, unions of those and None, Literal strings, string enumerations, tuple[X, ...] (an array), Mapping[str, X] (an object) and frozen dataclasses. Anything else is a programming error and fails when the schema is built, which a test does for every message.

JsonValue module-attribute

JsonValue = (
    None
    | bool
    | int
    | float
    | str
    | list["JsonValue"]
    | dict[str, "JsonValue"]
)

A value json.loads can return.

WIRE_EXCLUDE module-attribute

WIRE_EXCLUDE = 'wire_exclude'

Field metadata key: a field derived at construction, never sent.

RecordEncoder module-attribute

RecordEncoder = Callable[
    [Any, Callable[[object], "JsonValue"]], "JsonObject"
]

Encodes one dataclass value, given the encoder for its nested values.

WireError

WireError(path: str, problem: str)

Bases: ValueError

A JSON value does not fit the protocol type expected at path.

Source code in src/grounded_decision/session/wire.py

def __init__(self, path: str, problem: str) -> None:
    super().__init__(f"{path or 'message'}: {problem}")
    self.path = path
    self.problem = problem

WireEncoder dataclass

WireEncoder()

Encodes protocol values as JSON values (what :func:encode does), planned per class.

Each dataclass gets an encoding function made once from its fields' wire types (:meth:record_encoder, cached per class: a pure function of the type): a field whose values are already JSON (a scalar, a string enumeration, a union of those) is copied as it is, a tuple or mapping of such values is copied in one call, and only the rest is encoded recursively. The function is one dictionary display, compiled the way :mod:dataclasses compiles __init__. A 20 x 20 snake frame holds about sixty dataclasses and four hundred cell kinds, so this is a large part of what the browser's engine pays per move.

encode

encode(value: object) -> JsonValue

A protocol value (dataclass, enum, tuple, mapping, scalar) as a JSON value.

Source code in src/grounded_decision/session/wire.py

def encode(self, value: object) -> JsonValue:
    """A protocol value (dataclass, enum, tuple, mapping, scalar) as a JSON value."""

    kind = type(value)
    if kind in _SCALARS:
        return cast("JsonValue", value)
    if getattr(kind, "__dataclass_fields__", None) is not None:
        return self.record_encoder(kind)(value, self.encode)
    if kind is tuple or kind is list:
        return self._sequence(cast("tuple[object, ...] | list[object]", value))
    if isinstance(value, str):
        return value  # a string enumeration is its own value
    if isinstance(value, enum.Enum):
        return cast("JsonValue", value.value)
    if isinstance(value, bool | int | float):
        return value
    if isinstance(value, Mapping):
        mapping = cast("Mapping[object, object]", value)
        return {str(key): self.encode(item) for key, item in mapping.items()}
    if isinstance(value, tuple | list):
        return self._sequence(cast("tuple[object, ...] | list[object]", value))
    raise TypeError(f"cannot put a {kind.__name__} on the wire")

record_encoder cached staticmethod

record_encoder(record: type[Any]) -> RecordEncoder

The function that encodes a value of the dataclass record.

Source code in src/grounded_decision/session/wire.py

@staticmethod
@functools.cache
def record_encoder(record: type[Any]) -> RecordEncoder:
    """The function that encodes a value of the dataclass ``record``."""

    names = [item.name for item in wire_fields(record)]
    if not all(name.isidentifier() for name in names):
        raise TypeError(f"{record.__name__} has a field that is not an identifier")
    try:
        hints = WireType.hints(record)
        ways = [
            WireEncoder._expression(
                WireType.of(hints[name]),
                f"value.{name}",
                inside=frozenset({record}),
            )
            for name in names
        ]
    except (NameError, TypeError, KeyError):
        ways = [f"encode(value.{name})" for name in names]
    entries = ", ".join(
        f"{name!r}: {way}" for name, way in zip(names, ways, strict=True)
    )
    namespace: dict[str, Any] = {}
    source = f"def encode_record(value, encode):\n    return {{{entries}}}\n"
    exec(source, namespace)  # noqa: S102 - our own field names, checked above
    return cast("RecordEncoder", namespace["encode_record"])

FieldDocs dataclass

FieldDocs(summary: str, fields: Mapping[str, str])

A dataclass's summary and per-field descriptions, from its docstring.

The summary is the docstring's first paragraph; field descriptions come from a Google-style Attributes: section (name: text, continued on more-indented lines).

WireType dataclass

WireType(
    kind: str,
    python: type[Any] | None = None,
    items: tuple[WireType, ...] = (),
    literals: tuple[str, ...] = (),
)

One Python type annotation, as the protocol puts it on the wire.

Attributes:

Name	Type	Description
kind	str	primitive, literal, enum, array, mapping, record (a dataclass) or union.
python	type[Any] | None	The primitive's or enumeration's or dataclass's class.
items	tuple[WireType, ...]	Element type (array), value type (mapping) or the members (union).
literals	tuple[str, ...]	The allowed strings (literal).

discriminator property

discriminator: bool

A single string literal (a message's type): always sent, always required.

of classmethod

of(annotation: object) -> WireType

The wire type of an annotation (already resolved, not a string).

Raises:

Type	Description
TypeError	A type the protocol does not support.

Source code in src/grounded_decision/session/wire.py

@classmethod
def of(cls, annotation: object) -> WireType:
    """The wire type of an annotation (already resolved, not a string).

    Raises:
        TypeError: A type the protocol does not support.
    """

    value = getattr(annotation, "__value__", None)
    if isinstance(annotation, typing.TypeAliasType) and value is not None:
        return cls.of(value)
    origin = typing.get_origin(annotation)
    arguments = typing.get_args(annotation)
    if origin is typing.Literal:
        if not all(isinstance(item, str) for item in arguments):
            raise TypeError(f"only string literals are supported: {annotation}")
        return cls("literal", literals=tuple(cast("tuple[str, ...]", arguments)))
    if origin in (types.UnionType, typing.Union):
        return cls("union", items=tuple(cls.of(item) for item in arguments))
    if origin is tuple:
        if len(arguments) == 2 and arguments[1] is Ellipsis:
            return cls("array", items=(cls.of(arguments[0]),))
        raise TypeError(f"use tuple[X, ...] or a dataclass, not {annotation}")
    if origin in (Mapping, dict, typing.Mapping):
        if arguments[0] is not str:
            raise TypeError(f"mapping keys must be str: {annotation}")
        return cls("mapping", items=(cls.of(arguments[1]),))
    if isinstance(annotation, type):
        python = cast("type[Any]", annotation)
        if python in cls.PRIMITIVES:
            return cls("primitive", python=python)
        if issubclass(python, enum.Enum):
            return cls("enum", python=python)
        if dataclasses.is_dataclass(python):
            return cls("record", python=python)
    if annotation is None:
        return cls("primitive", python=type(None))
    raise TypeError(f"the protocol does not support {annotation!r}")

hints cached staticmethod

hints(record: type[Any]) -> dict[str, object]

A dataclass's resolved field annotations (resolved once per class).

Source code in src/grounded_decision/session/wire.py

@staticmethod
@functools.cache
def hints(record: type[Any]) -> dict[str, object]:
    """A dataclass's resolved field annotations (resolved once per class)."""

    return typing.get_type_hints(record)

plan cached staticmethod

plan(
    record: type[Any],
) -> tuple[tuple[str, bool, WireType], ...]

How to decode a record, worked out once per class: each wire field's name, whether a message must carry it, and its type.

Source code in src/grounded_decision/session/wire.py

@staticmethod
@functools.cache
def plan(record: type[Any]) -> tuple[tuple[str, bool, WireType], ...]:
    """How to decode a record, worked out once per class: each wire field's
    name, whether a message must carry it, and its type."""

    hints = WireType.hints(record)
    steps: list[tuple[str, bool, WireType]] = []
    for item in wire_fields(record):
        kind = WireType.of(hints[item.name])
        steps.append((item.name, is_required(item) or kind.discriminator, kind))
    return tuple(steps)

schema

schema(builder: SchemaBuilder) -> JsonObject

This type in JSON Schema (draft 2020-12); records and enums go to $defs.

Source code in src/grounded_decision/session/wire.py

def schema(self, builder: SchemaBuilder) -> JsonObject:
    """This type in JSON Schema (draft 2020-12); records and enums go to ``$defs``."""

    if self.kind == "primitive":
        assert self.python is not None
        return {"type": self.PRIMITIVES[self.python]}
    if self.kind == "literal":
        if len(self.literals) == 1:
            return {"type": "string", "const": self.literals[0]}
        return {"type": "string", "enum": list(self.literals)}
    if self.kind in ("enum", "record"):
        assert self.python is not None
        return builder.reference(self.python)
    if self.kind == "array":
        return {"type": "array", "items": self.items[0].schema(builder)}
    if self.kind == "mapping":
        return {
            "type": "object",
            "additionalProperties": self.items[0].schema(builder),
        }
    members = [item.schema(builder) for item in self.items]
    if all(set(member) == {"type"} for member in members):
        names: list[JsonValue] = []
        for member in members:
            kind = member["type"]
            if kind not in names:
                names.append(kind)
        if "integer" in names and "number" in names:
            names.remove("integer")
        return {"type": names}
    return {"anyOf": cast("list[JsonValue]", members)}

decode

decode(value: JsonValue, path: str = '') -> Any

A JSON value as this type, checked all the way down.

Raises:

Type	Description
WireError	The value does not fit (wrong type, unknown field, missing field, unknown enumeration value …).

Source code in src/grounded_decision/session/wire.py

def decode(self, value: JsonValue, path: str = "") -> Any:
    """A JSON value as this type, checked all the way down.

    Raises:
        WireError: The value does not fit (wrong type, unknown field,
            missing field, unknown enumeration value …).
    """

    if self.kind == "primitive":
        return self._primitive(value, path)
    if self.kind == "literal":
        if not isinstance(value, str) or value not in self.literals:
            raise WireError(
                path, f"expected one of {list(self.literals)}, got {value!r}"
            )
        return value
    if self.kind == "enum":
        assert self.python is not None
        try:
            return self.python(value)
        except ValueError:
            raise WireError(
                path, f"{value!r} is not a {self.python.__name__}"
            ) from None
    if self.kind == "array":
        if not isinstance(value, list):
            raise WireError(path, f"expected an array, got {type(value).__name__}")
        return tuple(
            self.items[0].decode(item, f"{path}[{index}]")
            for index, item in enumerate(value)
        )
    if self.kind == "mapping":
        if not isinstance(value, dict):
            raise WireError(path, f"expected an object, got {type(value).__name__}")
        return {
            key: self.items[0].decode(item, f"{path}.{key}")
            for key, item in value.items()
        }
    if self.kind == "record":
        assert self.python is not None
        return self._record(self.python, value, path)
    problems: list[str] = []
    for member in self.items:
        try:
            return member.decode(value, path)
        except WireError as error:
            problems.append(error.problem)
    raise WireError(path, "; ".join(problems))

SchemaBuilder dataclass

SchemaBuilder(
    definitions: dict[str, JsonObject] = dict[
        str, JsonObject
    ](),
    _classes: dict[str, type[Any]] = dict[str, type[Any]](),
)

Builds the $defs of every dataclass and enumeration a schema refers to.

Definitions are named by their class name (which must be unique) and added in the order they are first reached, so the schema is stable.

reference

reference(target: type[Any]) -> JsonObject

{"$ref": "#/$defs/Name"}, defining Name the first time.

Source code in src/grounded_decision/session/wire.py

def reference(self, target: type[Any]) -> JsonObject:
    """``{"$ref": "#/$defs/Name"}``, defining ``Name`` the first time."""

    name = target.__name__
    known = self._classes.get(name)
    if known is not None and known is not target:
        raise TypeError(f"two protocol types are called {name}")
    if known is None:
        self._classes[name] = target
        self.definitions[name] = {}
        self.definitions[name] = self._define(target)
    return {"$ref": f"#/$defs/{name}"}

wire_fields

wire_fields(cls: type[Any]) -> tuple[Field[Any], ...]

The fields of a protocol dataclass that travel on the wire, in order.

Source code in src/grounded_decision/session/wire.py

def wire_fields(cls: type[Any]) -> tuple[dataclasses.Field[Any], ...]:
    """The fields of a protocol dataclass that travel on the wire, in order."""

    return tuple(
        item
        for item in dataclasses.fields(cls)
        if item.init and not item.metadata.get(WIRE_EXCLUDE, False)
    )

is_required

is_required(item: Field[Any]) -> bool

Whether a field has no default (a sender must always give it).

Source code in src/grounded_decision/session/wire.py

def is_required(item: dataclasses.Field[Any]) -> bool:
    """Whether a field has no default (a sender must always give it)."""

    return (
        item.default is dataclasses.MISSING
        and item.default_factory is dataclasses.MISSING
    )

encode

encode(value: object) -> JsonValue

A protocol value (dataclass, enum, tuple, mapping, scalar) as a JSON value.

Source code in src/grounded_decision/session/wire.py

def encode(value: object) -> JsonValue:
    """A protocol value (dataclass, enum, tuple, mapping, scalar) as a JSON value."""

    return WireEncoder().encode(value)

decode

decode(record: type[Record], value: JsonValue) -> Record

A JSON value as the protocol dataclass record, validated.

Raises:

Type	Description
WireError	The value does not fit.

Source code in src/grounded_decision/session/wire.py

def decode[Record](record: type[Record], value: JsonValue) -> Record:
    """A JSON value as the protocol dataclass ``record``, validated.

    Raises:
        WireError: The value does not fit.
    """

    return cast("Record", WireType("record", python=record).decode(value))

compatibility

Semantic versioning for the protocol: what a schema change breaks.

:class:SchemaChanges compares the recorded schema of the last released protocol version (protocol.baseline.schema.json) with the schema the code generates now and sorts every difference into breaking (an existing front end or engine could fail) or additive (old peers keep working if, as the protocol requires, readers ignore fields they do not know). :meth:SchemaChanges.problems then says whether :data:~grounded_decision.session.protocol.PROTOCOL_VERSION was bumped enough: a breaking change needs a new major version, an additive one a new minor version. Descriptions never count.

The rules, per $defs entry (docs/reference/protocol.md):

• removing a type, a property, an enumeration value or a union member is breaking; changing a property's type is breaking;
• adding a type, an optional property, an enumeration value or a union member is additive;
• a newly required property is breaking in a request (old front ends do not send it) and additive in a reply (old front ends ignore it); a property that stops being required is the reverse.

Version dataclass

Version(major: int, minor: int, patch: int)

A semantic version.

Attributes:

Name	Type	Description
major	int	Incompatible changes.
minor	int	Compatible additions.
patch	int	Documentation and fixes.

parse classmethod

parse(text: str) -> Version

Raises: ValueError: Not major.minor.patch with non-negative integers.

Source code in src/grounded_decision/session/compatibility.py

@classmethod
def parse(cls, text: str) -> Version:
    """Raises:
    ValueError: Not ``major.minor.patch`` with non-negative integers.
    """

    parts = text.split(".")
    if len(parts) != 3 or not all(part.isdigit() for part in parts):
        raise ValueError(f"a version looks like 1.2.3, not {text!r}")
    return cls(int(parts[0]), int(parts[1]), int(parts[2]))

SchemaChanges dataclass

SchemaChanges(
    breaking: tuple[str, ...], additive: tuple[str, ...]
)

Every difference between two protocol schemas, sorted by what it breaks.

Attributes:

Name	Type	Description
breaking	tuple[str, ...]	Changes an existing peer may not survive.
additive	tuple[str, ...]	Changes old peers can ignore.

problems

problems(old_version: str, new_version: str) -> list[str]

Why new_version is not a correct semver bump for these changes ([] if it is).

Source code in src/grounded_decision/session/compatibility.py

def problems(self, old_version: str, new_version: str) -> list[str]:
    """Why ``new_version`` is not a correct semver bump for these changes ([] if it is)."""

    old, new = Version.parse(old_version), Version.parse(new_version)
    if new.key < old.key:
        return [f"the version went backwards: {old} -> {new}"]
    if self.breaking and new.major <= old.major:
        return [
            f"breaking changes need a new major version (now {new}, baseline {old}): "
            + "; ".join(self.breaking)
        ]
    if self.additive and (new.major, new.minor) <= (old.major, old.minor):
        return [
            f"additions need a new minor version (now {new}, baseline {old}): "
            + "; ".join(self.additive)
        ]
    return []

tokens

Design tokens: the terminal's colours and glyphs, for every other front end.

The palette and glyphs of :mod:~grounded_decision.examples.games.views are the one source (design decision D22). :class:DesignTokens turns them into a JSON document (make protocol writes it to web/packages/protocol/generated/tokens.json): a colour and a glyph for every :class:~grounded_decision.examples.games.views.CellKind and :class:~grounded_decision.examples.games.views.Overlay, in a dark theme (the terminal's own colours) and a light theme derived from it by two rules:

• a foreground (a piece, a mark, a digit) is darkened, by mixing in black, until it has at least :data:MIN_CONTRAST (WCAG's 3:1 for graphical objects) against the light board;
• a dark background (a highlight, a covered tile) is lightened by mixing in :data:LIGHT_MIX of white, so highlights stay visible but soft; mid-tone materials (wood, checker squares, 2048 tiles) are kept.

Concepts and the classes that own them:

:class:Colour An sRGB colour: hex text, luminance, contrast, mixing. :class:Theme One theme's colours. :class:DesignTokens Both themes and the glyphs, as one JSON document.

MIN_CONTRAST module-attribute

MIN_CONTRAST = 3.0

Least contrast of a foreground against the light board (WCAG 2.2, non-text).

LIGHT_MIX module-attribute

LIGHT_MIX = 0.55

Share of white mixed into a dark background for the light theme.

DARK_BACKGROUND module-attribute

DARK_BACKGROUND = 0.12

Backgrounds with less relative luminance than this are lightened.

LIGHT_BOARD module-attribute

LIGHT_BOARD = (246, 244, 239)

The light theme's board colour.

Colour dataclass

Colour(red: int, green: int, blue: int)

An sRGB colour with 8-bit channels.

Attributes:

Name	Type	Description
red	int	0-255.
green	int	0-255.
blue	int	0-255.

luminance property

luminance: float

Relative luminance (WCAG 2.2), 0 for black to 1 for white.

contrast

contrast(other: Colour) -> float

The WCAG contrast ratio, 1 to 21.

Source code in src/grounded_decision/session/tokens.py

def contrast(self, other: Colour) -> float:
    """The WCAG contrast ratio, 1 to 21."""

    high, low = sorted((self.luminance, other.luminance), reverse=True)
    return (high + 0.05) / (low + 0.05)

mix

mix(other: Colour, share: float) -> Colour

This colour with share (0-1) of other mixed in.

Source code in src/grounded_decision/session/tokens.py

def mix(self, other: Colour, share: float) -> Colour:
    """This colour with ``share`` (0-1) of ``other`` mixed in."""

    return Colour.of(
        views.blend(
            (self.red, self.green, self.blue),
            (other.red, other.green, other.blue),
            share,
        )
    )

readable_on

readable_on(
    background: Colour, ratio: float = MIN_CONTRAST
) -> Colour

This colour, darkened by steps of 5% black until it reaches ratio on background.

Source code in src/grounded_decision/session/tokens.py

def readable_on(self, background: Colour, ratio: float = MIN_CONTRAST) -> Colour:
    """This colour, darkened by steps of 5% black until it reaches ``ratio`` on ``background``."""

    black = Colour(0, 0, 0)
    for step in range(21):
        candidate = self.mix(black, step / 20)
        if candidate.contrast(background) >= ratio:
            return candidate
    return black

softened

softened() -> Colour

For the light theme: a dark background lightened, a mid-tone kept.

Source code in src/grounded_decision/session/tokens.py

def softened(self) -> Colour:
    """For the light theme: a dark background lightened, a mid-tone kept."""

    if self.luminance < DARK_BACKGROUND:
        return self.mix(Colour(255, 255, 255), LIGHT_MIX)
    return self

Theme dataclass

Theme(name: str, light: bool)

One theme: colours for every cell kind, overlay, seat and panel element.

Attributes:

Name	Type	Description
name	str	dark or light.
light	bool	Whether foregrounds and backgrounds are adapted to a light board.

DesignTokens dataclass

DesignTokens(
    themes: tuple[Theme, ...] = (
        Theme("dark", light=False),
        Theme("light", light=True),
    ),
)

The design-token document: glyphs and both themes.

Attributes:

Name	Type	Description
themes	tuple[Theme, ...]	The themes, dark first.

registry

The registered games as objects: loading, configuring and seating one.

Every way of starting a match (grounded-decision play, the session layer in :mod:grounded_decision.session that the command line and the web front end share, compare and audit) goes through :class:GameModule, so a size, a world, a rule option or a line-up is accepted or refused with the same words everywhere.

Concepts and the classes that own them:

:class:GameModule One game module of :data:~grounded_decision.examples.games.GAME_MODULES and what its module contract offers (see :mod:~grounded_decision.examples.games.play): its agents, worlds, variants, named sizes, checks, and how to build the game and its seats. :class:GameEntry A registered name and whether its module can be loaded. :class:PlayError A request the game cannot honour (unknown game, agent, size …); the command line exits with status 2.

HUMAN module-attribute

HUMAN = 'you'

The agent name that seats a human (--agents you,grounded).

DEFAULT_MAX_STEPS module-attribute

DEFAULT_MAX_STEPS = 10000000

The move budget of a match when none is given (--max-steps).

VARIANT_OPTIONS module-attribute

VARIANT_OPTIONS: tuple[tuple[str, str], ...] = (
    ("BOARDS", "--board"),
    ("STARTS", "--start"),
    ("LAYOUTS", "--layout"),
)

Module dictionaries of named variants and the option that picks one.

PlayError

Bases: Exception

A problem with the request (unknown game, agent, size …); exit status 2.

HumanProposer

Bases: Protocol

A person's seat as an agent: the terminal prompt or the session's submitted moves.

certify is set when the seat is guarded: whether the gate would pass a move, so the seat can say which moves are safe before the person picks one.

GameEntry dataclass

GameEntry(
    name: str, module: ModuleType | None, problem: str = ""
)

One registered game, and whether it can be loaded.

GameModule dataclass

GameModule(name: str, module: ModuleType)

A registered game module and everything its module contract offers.

Attributes:

Name	Type	Description
name	str	The command-line name ("2048" for game_2048).
module	ModuleType	The loaded module.

module_name property

module_name: str

The module's own name (game_2048), used in some messages.

factories property

factories: dict[str, Callable[..., Any]]

Named agent factories, in display order.

default_mode property

default_mode: str

"watch" or "you": what play GAME does with no flags.

check_names property

check_names: tuple[str, ...]

Every check the game's gate can run (CHECK_DESCRIPTIONS): one panel row each.

note_rows property

note_rows: int | None

Panel rows for notes (PANEL_NOTE_ROWS), or None for the default.

worlds property

worlds: dict[str, str]

The game's world modes (WORLDS: mode -> description), or {}.

size_presets property

size_presets: dict[str, tuple[int, int]]

The game's named sizes (SIZE_PRESETS: name -> (width, height)), or {}.

sizes property

sizes: SizeRule | None

The sizes the board may take (SIZES), or None for a fixed size.

configure_options property

configure_options: tuple[str, ...]

The keyword options of configure beyond the size (world, mines …).

option_rules property

option_rules: dict[str, OptionRule]

What the game takes for each rule option (OPTIONS: keyword -> rule), or {}.

option_defaults property

option_defaults: dict[str, object]

configure's default for each keyword option (None: it depends).

find staticmethod

find(name: str) -> GameEntry

Load a registered game module, or explain why it is not available.

Source code in src/grounded_decision/examples/games/registry.py

@staticmethod
def find(name: str) -> GameEntry:
    """Load a registered game module, or explain why it is not available."""

    if name not in GAME_MODULES:
        return GameEntry(
            name, None, f"unknown game; choose from {', '.join(GAME_MODULES)}"
        )
    try:
        module = cast("ModuleType", load_game_module(name))
    except ModuleNotFoundError as error:
        if error.name and GAME_MODULES[name].startswith(error.name):
            return GameEntry(
                name, None, "not installed yet (its module does not exist)"
            )
        raise
    missing = [
        attr
        for attr in ("GAME", "agents", "DEFAULT_AGENTS")
        if not hasattr(module, attr)
    ]
    if missing:
        return GameEntry(
            name, None, f"its module does not provide {', '.join(missing)} yet"
        )
    return GameEntry(name, module)

load classmethod

load(name: str) -> GameModule

Raises: PlayError: The game is unknown or not available.

Source code in src/grounded_decision/examples/games/registry.py

@classmethod
def load(cls, name: str) -> GameModule:
    """Raises:
    PlayError: The game is unknown or not available.
    """

    entry = cls.find(name)
    if entry.module is None:
        raise PlayError(f"{name}: {entry.problem}")
    return cls(name, entry.module)

of classmethod

of(module: ModuleType) -> GameModule

The registered game of an already loaded module.

Source code in src/grounded_decision/examples/games/registry.py

@classmethod
def of(cls, module: ModuleType) -> GameModule:
    """The registered game of an already loaded module."""

    for name, path in GAME_MODULES.items():
        if path == module.__name__:
            return cls(name, module)
    return cls(module.__name__.rsplit(".", 1)[-1], module)

available classmethod

available() -> tuple[GameModule, ...]

Every registered game whose module loads, in registry order.

Source code in src/grounded_decision/examples/games/registry.py

@classmethod
def available(cls) -> tuple[GameModule, ...]:
    """Every registered game whose module loads, in registry order."""

    games: list[GameModule] = []
    for name in GAME_MODULES:
        entry = cls.find(name)
        if entry.module is not None:
            games.append(cls(name, entry.module))
    return tuple(games)

constant

constant(attribute: str, default: Any = None) -> Any

A module-contract constant (GATED_AGENT, TITLE …), or default.

Source code in src/grounded_decision/examples/games/registry.py

def constant(self, attribute: str, default: Any = None) -> Any:
    """A module-contract constant (``GATED_AGENT``, ``TITLE`` …), or ``default``."""

    return getattr(self.module, attribute, default)

descriptions

descriptions(attribute: str) -> dict[str, str]

A {name: text} constant (CHECK_DESCRIPTIONS, WORLDS …), or {}.

Source code in src/grounded_decision/examples/games/registry.py

def descriptions(self, attribute: str) -> dict[str, str]:
    """A ``{name: text}`` constant (``CHECK_DESCRIPTIONS``, ``WORLDS`` …), or ``{}``."""

    return dict(cast("Mapping[str, str]", self.constant(attribute, {}) or {}))

check_names_for

check_names_for(game: object) -> tuple[str, ...]

The checks a configured game's gate runs: its check_names if it declares them (Tetris: fewer in placement mode than in real time), else :attr:check_names.

Source code in src/grounded_decision/examples/games/registry.py

def check_names_for(self, game: object) -> tuple[str, ...]:
    """The checks a configured game's gate runs: its ``check_names`` if it declares
    them (Tetris: fewer in placement mode than in real time), else :attr:`check_names`."""

    declared = getattr(game, "check_names", None)
    if isinstance(declared, tuple):
        return tuple(str(name) for name in cast("tuple[object, ...]", declared))
    return self.check_names

variants

variants(attribute: str) -> dict[str, str]

BOARDS/STARTS/LAYOUTS (name -> description), or {}.

Source code in src/grounded_decision/examples/games/registry.py

def variants(self, attribute: str) -> dict[str, str]:
    """``BOARDS``/``STARTS``/``LAYOUTS`` (name -> description), or ``{}``."""

    return self.descriptions(attribute)

parse_size

parse_size(text: str) -> tuple[int, int]

--size: WxH or one of the game's named sizes.

Raises:

Type	Description
PlayError	Not WxH with positive integers, nor a named size.

Source code in src/grounded_decision/examples/games/registry.py

def parse_size(self, text: str) -> tuple[int, int]:
    """``--size``: ``WxH`` or one of the game's named sizes.

    Raises:
        PlayError: Not ``WxH`` with positive integers, nor a named size.
    """

    return parse_dimensions(text, self.size_presets)

validate_world

validate_world(world: str) -> WorldSpec

Parse --world for this game.

Raises:

Type	Description
PlayError	The game has no worlds, or the mode is unknown or malformed.

Source code in src/grounded_decision/examples/games/registry.py

def validate_world(self, world: str) -> WorldSpec:
    """Parse ``--world`` for this game.

    Raises:
        PlayError: The game has no worlds, or the mode is unknown or malformed.
    """

    worlds = self.worlds
    if not worlds:
        self._no_worlds()
    try:
        return parse_world(world, worlds)
    except WorldError as error:
        raise PlayError(f"--world for {self.module_name}: {error}") from None

check_world

check_world(
    game: TurnGame[Any, Any], agents: Sequence[object]
) -> WorldSpec | None

The game's world, checked against the agents' regimes; None without worlds.

Raises:

Type	Description
PlayError	A clairvoyant agent in a world that depends on the agent's moves.

Source code in src/grounded_decision/examples/games/registry.py

def check_world(
    self, game: TurnGame[Any, Any], agents: Sequence[object]
) -> WorldSpec | None:
    """The game's world, checked against the agents' regimes; ``None`` without worlds.

    Raises:
        PlayError: A clairvoyant agent in a world that depends on the agent's moves.
    """

    worlds = self.worlds
    if not worlds:
        return None
    spec = parse_world(str(getattr(game, "world", DEFAULT_WORLD)), worlds)
    try:
        check_regimes(spec, agents, independent=self.constant("INDEPENDENT_WORLDS"))
    except WorldError as error:
        raise PlayError(str(error)) from None
    return spec

default_game

default_game() -> TurnGame[Any, Any]

The game at its default size and rules.

Source code in src/grounded_decision/examples/games/registry.py

def default_game(self) -> TurnGame[Any, Any]:
    """The game at its default size and rules."""

    game = self.module.GAME
    return cast(
        "TurnGame[Any, Any]",
        game()
        if isinstance(game, type)
        or (callable(game) and not hasattr(game, "legal_moves"))
        else game,
    )

make_game

make_game(
    size: tuple[int, int] | None = None,
    until_length: int | None = None,
    *,
    players: int | None = None,
    rules: Mapping[str, Any] | None = None,
    world: str | None = None,
) -> TurnGame[Any, Any]

The game at its default size, or configured.

Parameters:

Name	Type	Description	Default
size	tuple[int, int] | None	(width, height) (--size).	None
until_length	int | None	Snake's length goal (--until-length).	None
players	int | None	Number of players (--players), for games whose configure takes a players keyword (Chinese checkers).	None
rules	Mapping[str, Any] | None	Extra rule options passed to configure by keyword (Chinese checkers: stall_plies, clear_finished; Hexxagon: board, start, holes, layout …).	None
world	str | None	--world MODE (or MODE:p), for games with a world (a module WORLDS); passed to configure(world=...).	None

Raises:

Type	Description
PlayError	The size, length goal, number of players, world or a rule option is not supported.

Source code in src/grounded_decision/examples/games/registry.py

def make_game(
    self,
    size: tuple[int, int] | None = None,
    until_length: int | None = None,
    *,
    players: int | None = None,
    rules: Mapping[str, Any] | None = None,
    world: str | None = None,
) -> TurnGame[Any, Any]:
    """The game at its default size, or configured.

    Args:
        size: ``(width, height)`` (``--size``).
        until_length: Snake's length goal (``--until-length``).
        players: Number of players (``--players``), for games whose
            ``configure`` takes a ``players`` keyword (Chinese checkers).
        rules: Extra rule options passed to ``configure`` by keyword
            (Chinese checkers: ``stall_plies``, ``clear_finished``;
            Hexxagon: ``board``, ``start``, ``holes``, ``layout`` …).
        world: ``--world MODE`` (or ``MODE:p``), for games with a world
            (a module ``WORLDS``); passed to ``configure(world=...)``.

    Raises:
        PlayError: The size, length goal, number of players, world or a
            rule option is not supported.
    """

    configure = self.constant("configure")
    if world is not None:
        self.validate_world(world)
        rules = {**(rules or {}), "world": world}
    if rules:
        accepted: set[str] = (
            set(inspect.signature(configure).parameters)
            if callable(configure)
            else set()
        )
        unknown = [name for name in rules if name not in accepted]
        if unknown or not callable(configure):
            flags = ", ".join("--" + name.replace("_", "-") for name in unknown)
            raise PlayError(f"{flags} is not an option of {self.module_name}")
        options = dict(rules)
        if until_length is not None:
            if "target_length" not in accepted:
                raise PlayError(
                    "--until-length is only for games with a length (snake)"
                )
            options["target_length"] = until_length
        if size is None:
            size = self._default_size()
        if players is not None:
            options["players"] = players
        try:
            return cast("TurnGame[Any, Any]", configure(*size, **options))
        except ValueError as error:
            raise PlayError(str(error)) from error
    if players is not None:
        if (
            not callable(configure)
            or "players" not in inspect.signature(configure).parameters
        ):
            raise PlayError(
                "--players is only for games with a choice of player count "
                "(chinese_checkers)"
            )
        if size is None:
            size = self._default_size()
        if until_length is not None:
            raise PlayError(
                "--until-length is only for games with a length (snake)"
            )
        try:
            return cast("TurnGame[Any, Any]", configure(*size, players=players))
        except ValueError as error:
            raise PlayError(str(error)) from error
    if size is None and until_length is None:
        return self.default_game()
    if not callable(configure):
        raise PlayError(f"{self.module_name} has a fixed size")
    if size is None:
        size = self._default_size()
    try:
        if until_length is not None:
            parameters = inspect.signature(configure).parameters
            if "target_length" not in parameters:
                raise PlayError(
                    "--until-length is only for games with a length (snake)"
                )
            return cast(
                "TurnGame[Any, Any]", configure(*size, target_length=until_length)
            )
        return cast("TurnGame[Any, Any]", configure(*size))
    except ValueError as error:
        raise PlayError(str(error)) from error

default_line_up

default_line_up(players: int) -> tuple[str, ...]

The default agent names for players seats.

The module's default_agents(players) if it has one, else DEFAULT_AGENTS (repeated round the table if the game has more seats).

Source code in src/grounded_decision/examples/games/registry.py

def default_line_up(self, players: int) -> tuple[str, ...]:
    """The default agent names for ``players`` seats.

    The module's ``default_agents(players)`` if it has one, else
    ``DEFAULT_AGENTS`` (repeated round the table if the game has more seats).
    """

    custom = self.constant("default_agents")
    if callable(custom):
        return tuple(cast("Sequence[str]", custom(players)))
    defaults = tuple(self.module.DEFAULT_AGENTS)
    if len(defaults) == players or not defaults:
        return defaults
    return tuple(defaults[seat % len(defaults)] for seat in range(players))

seat_index staticmethod

seat_index(
    game: TurnGame[Any, Any], you_play: str | None
) -> int

The seat --you-play names (a player's name or number), else the first.

Raises:

Type	Description
PlayError	No such player.

Source code in src/grounded_decision/examples/games/registry.py

@staticmethod
def seat_index(game: TurnGame[Any, Any], you_play: str | None) -> int:
    """The seat ``--you-play`` names (a player's name or number), else the first.

    Raises:
        PlayError: No such player.
    """

    if you_play is None:
        return 0
    names = [name.lower() for name in game.player_names]
    if you_play.lower() in names:
        return names.index(you_play.lower())
    if you_play.isdigit() and 1 <= int(you_play) <= len(names):
        return int(you_play) - 1
    raise PlayError(f"--you-play must be one of {', '.join(game.player_names)}")

seat_names

seat_names(
    game: TurnGame[Any, Any],
    agents: Sequence[str] | None,
    *,
    you: bool = False,
    you_play: str | None = None,
) -> tuple[str, ...]

One agent name per seat (:data:HUMAN for a person).

agents empty or None means the default line-up; a single name fills every seat, except you, which takes one seat (you_play) beside the default line-up. you puts a person in that seat whatever the line-up.

Raises:

Type	Description
PlayError	The wrong number of names (unknown names are refused by :meth:seat_agents).

Source code in src/grounded_decision/examples/games/registry.py

def seat_names(
    self,
    game: TurnGame[Any, Any],
    agents: Sequence[str] | None,
    *,
    you: bool = False,
    you_play: str | None = None,
) -> tuple[str, ...]:
    """One agent name per seat (:data:`HUMAN` for a person).

    ``agents`` empty or ``None`` means the default line-up; a single
    name fills every seat, except ``you``, which takes one seat
    (``you_play``) beside the default line-up. ``you`` puts a person in
    that seat whatever the line-up.

    Raises:
        PlayError: The wrong number of names (unknown names are refused
            by :meth:`seat_agents`).
    """

    players = len(game.player_names)
    names = list(agents) if agents else list(self.default_line_up(players))
    if names == [HUMAN] and players > 1:
        # `you` takes one seat; the others get the default line-up.
        names = list(self.default_line_up(players))
        names[self.seat_index(game, you_play)] = HUMAN
    elif len(names) == 1 and players > 1 and agents:
        names = names * players
    if len(names) != players:
        seat_word = "seat" if players == 1 else "seats"
        raise PlayError(
            f"{game.name} has {players} {seat_word} ({', '.join(game.player_names)}) "
            f"but --agents named {len(names)} ({', '.join(names)}). --agents gives "
            "one agent per seat, in seat order. To compare agents over the same "
            f"games, use: grounded-decision compare {game.name} --agents "
            + ",".join(names)
        )
    if you and HUMAN not in names:
        names[self.seat_index(game, you_play)] = HUMAN
    return tuple(names)

seat_agents

seat_agents(
    game: TurnGame[Any, Any],
    names: Sequence[str],
    *,
    seed: int,
    guarded: bool = False,
    human: Callable[[], HumanProposer],
) -> list[Agent[Any, Any]]

One fresh agent per seat; seat k gets seed + k.

A :data:HUMAN seat is human(), behind the game's gate when guarded (:meth:guard).

Raises:

Type	Description
PlayError	An unknown agent name, or guarded for a game without a gate.

Source code in src/grounded_decision/examples/games/registry.py

def seat_agents(
    self,
    game: TurnGame[Any, Any],
    names: Sequence[str],
    *,
    seed: int,
    guarded: bool = False,
    human: Callable[[], HumanProposer],
) -> list[Agent[Any, Any]]:
    """One fresh agent per seat; seat ``k`` gets ``seed + k``.

    A :data:`HUMAN` seat is ``human()``, behind the game's gate when
    ``guarded`` (:meth:`guard`).

    Raises:
        PlayError: An unknown agent name, or ``guarded`` for a game
            without a gate.
    """

    factories = self.factories
    agents: list[Agent[Any, Any]] = []
    for seat, name in enumerate(names):
        if name == HUMAN:
            agents.append(self.guard(game, human(), guarded))
            continue
        if name not in factories:
            raise PlayError(
                f"unknown agent {name!r} for {game.name}; choose from "
                f"{', '.join([*factories, HUMAN])}"
            )
        agents.append(make_agent(factories[name], seed + seat))
    return agents

guard

guard(
    game: TurnGame[Any, Any],
    proposer: HumanProposer,
    guarded: bool,
) -> Agent[Any, Any]

A person's seat, behind the game's gate (guarded(proposer)) if asked.

Any move the person picks is then only a proposal: the gate checks it and plays its certified fallback if a check fails. The proposer's certify is set to the gate's checks.

Raises:

Type	Description
PlayError	guarded for a game without a gate.

Source code in src/grounded_decision/examples/games/registry.py

def guard(
    self, game: TurnGame[Any, Any], proposer: HumanProposer, guarded: bool
) -> Agent[Any, Any]:
    """A person's seat, behind the game's gate (``guarded(proposer)``) if asked.

    Any move the person picks is then only a proposal: the gate checks
    it and plays its certified fallback if a check fails. The
    proposer's ``certify`` is set to the gate's checks.

    Raises:
        PlayError: ``guarded`` for a game without a gate.
    """

    if not guarded:
        return cast("Agent[Any, Any]", proposer)
    guard = self.constant("guarded")
    if not callable(guard):
        raise PlayError(f"{game.name} has no gate to guard your moves")
    agent = cast("Agent[Any, Any]", guard(proposer))
    if isinstance(agent, GatedAgent):
        gated = agent
        proposer.certify = lambda state, move: all(
            check.passed for check in gated.checks(state, move)
        )
    return agent

parse_dimensions

parse_dimensions(
    text: str,
    presets: Mapping[str, tuple[int, int]] | None = None,
) -> tuple[int, int]

"20x20" → (20, 20), or a named size from presets.

Raises:

Type	Description
PlayError	Not WxH with positive integers, nor a named size.

Source code in src/grounded_decision/examples/games/registry.py

def parse_dimensions(
    text: str, presets: Mapping[str, tuple[int, int]] | None = None
) -> tuple[int, int]:
    """``"20x20"`` → ``(20, 20)``, or a named size from ``presets``.

    Raises:
        PlayError: Not ``WxH`` with positive integers, nor a named size.
    """

    named_sizes = presets or {}
    if text.lower() in named_sizes:
        return named_sizes[text.lower()]
    try:
        width_text, height_text = text.lower().split("x")
        width, height = int(width_text), int(height_text)
    except ValueError:
        named = f" or be one of {', '.join(named_sizes)}" if named_sizes else ""
        raise PlayError(f"--size must look like 20x20{named}, not {text!r}") from None
    if width < 1 or height < 1:
        raise PlayError("--size needs positive numbers")
    return width, height

make_agent

make_agent(
    factory: Callable[..., Any], seed: int
) -> Agent[Any, Any]

Call an agent factory, passing seed= if it accepts one.

Source code in src/grounded_decision/examples/games/registry.py

def make_agent(factory: Callable[..., Any], seed: int) -> Agent[Any, Any]:
    """Call an agent factory, passing ``seed=`` if it accepts one."""

    try:
        parameters = inspect.signature(factory).parameters
    except (TypeError, ValueError):
        parameters = {}
    if "seed" in parameters:
        return cast("Agent[Any, Any]", factory(seed=seed))
    return cast("Agent[Any, Any]", factory())