feat(gems): add multiplayer gem game with custom content packs
treefmt / nix fmt (pull_request) Successful in 6s
test ebook search / test-ebook-search (pull_request) Successful in 39s
pytest / pytest (pull_request) Successful in 40s
build_systems / build-brain (pull_request) Successful in 1m0s
build_systems / build-bob (pull_request) Successful in 1m2s
build_systems / build-rhapsody-in-green (pull_request) Successful in 1m33s
build_systems / build-jeeves (pull_request) Successful in 2m25s
test ebook search / test-ebook-search (push) Successful in 35s
build_systems / build-brain (push) Successful in 37s
build_systems / build-bob (push) Successful in 37s
build_systems / build-jeeves (push) Successful in 2m17s
treefmt / nix fmt (push) Successful in 5s
pytest / pytest (push) Successful in 29s
build_systems / build-rhapsody-in-green (push) Successful in 49s

- implement FastAPI and HTMX lobby and game interfaces
- support up to four human and AI-controlled players
- add configurable rules, victory conditions, and expansion modules
- support validated JSON cards, patrons, objectives, and outposts
This commit was merged in pull request #43.
This commit is contained in:
2026-07-22 20:47:03 -04:00
parent 247e951a27
commit add7a6a848
41 changed files with 4161 additions and 1 deletions
+54
View File
@@ -0,0 +1,54 @@
# Gems
Gems is a server-rendered, turn-based resource-engine game for one to four human or AI players. It uses FastAPI,
Jinja, HTMX, server-sent events, and SQLite.
The application deliberately contains no playable card deck, patron/governor set, objective set, official artwork,
or copied rulebook text. A room host must upload a content pack they are entitled to use before starting a game.
## Run locally
```shell
uv run gems --host 127.0.0.1 --port 8082
```
The default database and installation key are created under `.gems/`. The following environment variables override
runtime behavior:
- `GEMS_DATABASE_PATH`
- `GEMS_KEY_PATH`
- `GEMS_PUBLIC_ORIGIN`
- `GEMS_SECURE_COOKIES`
- `GEMS_HOST`
- `GEMS_PORT`
## Content packs
The current schema is available from a running server at `/schemas/content-pack-v1.json`. A pack defines exactly
five normal resources, one wild resource, cards, and optional patrons, objectives, and outposts. `patrons` is the
canonical field name; `governors` is accepted as an input alias.
Cards may use only the built-in, bounded effect vocabulary:
- `none`
- `virtual_wild`
- `copy_bonus`
- `copy_and_claim`
- `multi_bonus`
- `claim_free`
- an optional discard-cards alternate cost
Unknown fields, resource references, executable expressions, HTML, artwork URLs, and files larger than 512 KiB are
rejected. The normalized pack is private to its room and becomes immutable when play starts.
## Neutral module mapping
Gems calls the four optional mechanics Objectives, Outposts, Eastern Decks, and Fortifications. Lobby presets combine
these mechanics into the familiar base, objective-race, objective-plus-outpost, eastern-plus-fortification, and
all-module configurations. Component identities and values always come from the uploaded pack.
## Jeeves
The NixOS module runs one Uvicorn worker on `127.0.0.1:8002`, stores state in
`/zfs/media/services/gems`, and publishes it through HAProxy at `https://gems.tmmworkshop.com`. The DNS record must
point to Jeeves before ACME can issue the certificate.
+1
View File
@@ -0,0 +1 @@
"""Gems multiplayer card-engine game."""
+5
View File
@@ -0,0 +1,5 @@
"""AI controllers for Gems."""
from .runner import choose_ai_command
__all__ = ["choose_ai_command"]
+68
View File
@@ -0,0 +1,68 @@
"""Deterministic AI command selection at three difficulty levels."""
from __future__ import annotations
import random
from typing import TYPE_CHECKING
from python.gems.domain.engine import RuleError, apply_command, score
from python.gems.domain.legal_actions import legal_commands
if TYPE_CHECKING:
from python.gems.domain.models import ContentPack, GameCommand, GameSettings, GameState
def choose_ai_command(
state: GameState,
pack: ContentPack,
settings: GameSettings,
seat: int,
difficulty: str,
) -> GameCommand:
"""Choose from legal commands using only public/current-player information."""
actions = legal_commands(state, pack, settings, seat)
if not actions:
error = "AI has no legal command"
raise RuntimeError(error)
rng = random.Random(f"{state.seed}:{state.revision}:{seat}:{difficulty}") # noqa: S311 - deterministic AI
if difficulty == "easy":
return rng.choice(actions)
ranked = sorted(actions, key=lambda item: _heuristic(item, state, pack), reverse=True)
if difficulty == "medium":
return ranked[0]
# Bounded deterministic rollout: examine at most the twelve strongest actions.
best = ranked[0]
best_value = float("-inf")
for candidate in ranked[:12]:
try:
future = apply_command(state, candidate, pack, settings, actor_seat=seat)
except RuleError:
continue
value = score(future.players[seat], pack) * 100 + _heuristic(candidate, state, pack)
value += rng.random() * 0.001
if value > best_value:
best, best_value = candidate, value
return best
def _heuristic(command: GameCommand, state: GameState, pack: ContentPack) -> float:
player = state.players[state.current_seat]
value = 0.0
if command.type == "purchase":
card = pack.card(str(command.payload.get("card_id")))
value += 40 + card.points * 25 + (8 if card.bonus_resource else 0)
value -= sum(command.payload.get("payment", {}).values())
elif command.type == "take_distinct":
value += 10 + len(command.payload.get("resources", []))
elif command.type == "reserve":
value += 6
elif command.type == "take_double":
value += 4
elif command.type == "choose":
value += 20
elif command.type == "decline":
value -= 2
value += score(player, pack) * 0.01
return value
+25
View File
@@ -0,0 +1,25 @@
"""Runtime configuration loaded from environment variables."""
from __future__ import annotations
from pathlib import Path
from pydantic_settings import BaseSettings, SettingsConfigDict
class GemsConfig(BaseSettings):
"""Configure persistence, public URLs, cookies, and the HTTP server."""
model_config = SettingsConfigDict(env_prefix="GEMS_")
database_path: Path = Path(".gems/gems.sqlite3")
key_path: Path = Path(".gems/instance.key")
public_origin: str = "http://127.0.0.1:8082"
secure_cookies: bool = False
host: str = "127.0.0.1"
port: int = 8082
def load_config() -> GemsConfig:
"""Load Gems configuration from defaults and environment variables."""
return GemsConfig()
+55
View File
@@ -0,0 +1,55 @@
"""Content-pack parsing, normalization, and JSON Schema publication."""
from __future__ import annotations
import hashlib
import json
from dataclasses import dataclass
from typing import Any
from pydantic import ValidationError
from .domain.models import ContentPack
MAX_PACK_BYTES = 512 * 1024
@dataclass(frozen=True, slots=True)
class ParsedPack:
"""Validated content plus its canonical representation."""
pack: ContentPack
canonical_json: str
digest: str
class ContentPackError(ValueError):
"""A safe, user-visible content validation error."""
def parse_content_pack(raw: bytes | str) -> ParsedPack:
"""Validate a content pack and return stable canonical JSON."""
data = raw.encode() if isinstance(raw, str) else raw
if len(data) > MAX_PACK_BYTES:
error = "Content pack exceeds the 512 KiB limit"
raise ContentPackError(error)
try:
decoded: Any = json.loads(data)
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
error = f"Invalid JSON: {exc}"
raise ContentPackError(error) from exc
try:
pack = ContentPack.model_validate(decoded)
except ValidationError as exc:
messages = []
for issue in exc.errors(include_url=False):
path = ".".join(str(part) for part in issue["loc"])
messages.append(f"{path or '$'}: {issue['msg']}")
raise ContentPackError("\n".join(messages)) from exc
canonical = json.dumps(pack.model_dump(mode="json", by_alias=False), sort_keys=True, separators=(",", ":"))
return ParsedPack(pack=pack, canonical_json=canonical, digest=hashlib.sha256(canonical.encode()).hexdigest())
def content_pack_schema() -> dict[str, Any]:
"""Return the authoritative version-one JSON Schema."""
return ContentPack.model_json_schema(by_alias=False)
+1
View File
@@ -0,0 +1 @@
"""Domain models and rules for Gems."""
+903
View File
@@ -0,0 +1,903 @@
"""Deterministic, server-authoritative Gems rule engine."""
from __future__ import annotations
import random
from typing import Any
from .models import (
CardDefinition,
CardEffectKind,
ContentPack,
GameCommand,
GameSettings,
GameState,
OutpostDefinition,
OutpostPower,
OwnedCard,
PendingChoice,
PlayerState,
)
from .requirements import requirements_met
MAX_PLAYERS = 4
TWO_PLAYER_COUNT = 2
THREE_PLAYER_COUNT = 3
DISTINCT_TAKE_COUNT = 3
DOUBLE_TAKE_MINIMUM = 4
BLIND_RESERVE_DRAW_COUNT = 2
class RuleError(ValueError):
"""A command was not legal for the current state."""
def deck_key(deck: str, tier: int) -> str:
"""Build the state key for a named deck and tier."""
return f"{deck}:{tier}"
def new_game(
room_code: str,
names: list[str],
pack: ContentPack,
settings: GameSettings,
*,
seed: int,
) -> GameState:
"""Create and deal a deterministic game."""
if not 1 <= len(names) <= MAX_PLAYERS:
error = "Games require one to four seats"
raise RuleError(error)
if settings.first_player_mode == "selected" and settings.first_player_seat >= len(names):
error = "The selected first player is no longer in the room"
raise RuleError(error)
_validate_startable(pack, settings)
rng = random.Random(seed) # noqa: S311 - deterministic seeded shuffle
first_seat = settings.first_player_seat if settings.first_player_mode == "selected" else rng.randrange(len(names))
normal_count = 4 if len(names) <= TWO_PLAYER_COUNT else 5 if len(names) == THREE_PLAYER_COUNT else 7
supply = dict.fromkeys(pack.resource_ids, normal_count)
supply[pack.wild_resource.id] = 5
players = [
PlayerState(
seat=index,
name=name,
tokens=dict.fromkeys((*pack.resource_ids, pack.wild_resource.id), 0),
fortifications_available=settings.fortifications_per_player if settings.modules.fortifications else 0,
)
for index, name in enumerate(names)
]
decks: dict[str, list[str]] = {}
markets: dict[str, list[str]] = {}
for source in ("base", "eastern"):
if source == "eastern" and not settings.modules.eastern_decks:
continue
market_size = settings.base_market_size if source == "base" else settings.eastern_market_size
for tier in (1, 2, 3):
key = deck_key(source, tier)
cards = [card.id for card in pack.cards if card.deck == source and card.tier == tier]
rng.shuffle(cards)
markets[key] = [cards.pop() for _ in range(min(market_size, len(cards)))]
decks[key] = cards
patrons = [patron.id for patron in pack.patrons]
rng.shuffle(patrons)
patrons = patrons[: min(len(patrons), len(players) + 1)]
objectives = [objective.id for objective in pack.objectives]
rng.shuffle(objectives)
objectives = objectives[: settings.objective_count] if settings.modules.objectives else []
return GameState(
room_code=room_code,
seed=seed,
players=players,
supply=supply,
decks=decks,
markets=markets,
available_patrons=[] if settings.modules.objectives else patrons,
available_objectives=objectives,
current_seat=first_seat,
first_seat=first_seat,
)
def _validate_startable(pack: ContentPack, settings: GameSettings) -> None:
for tier in (1, 2, 3):
if not any(card.deck == "base" and card.tier == tier for card in pack.cards):
error = f"The base deck has no tier-{tier} cards"
raise RuleError(error)
if settings.modules.eastern_decks and not any(
card.deck == "eastern" and card.tier == tier for card in pack.cards
):
error = f"The eastern deck has no tier-{tier} cards"
raise RuleError(error)
if settings.modules.objectives and not pack.objectives:
error = "The Objectives module needs objective definitions"
raise RuleError(error)
if settings.modules.outposts and not pack.outposts:
error = "The Outposts module needs outpost definitions"
raise RuleError(error)
def card_map(pack: ContentPack) -> dict[str, CardDefinition]:
"""Index all cards in a content pack by identifier."""
return {card.id: card for card in pack.cards}
def bonuses(player: PlayerState, pack: ContentPack) -> dict[str, int]:
"""Calculate effective bonuses after copies, multiples, and discards."""
cards = card_map(pack)
result = dict.fromkeys(pack.resource_ids, 0)
for owned in player.cards:
definition = cards[owned.card_id]
resource = owned.copied_resource or definition.bonus_resource
if resource is not None:
result[resource] += definition.effect.amount if definition.effect.kind == CardEffectKind.MULTI_BONUS else 1
return result
def score(player: PlayerState, pack: ContentPack) -> int:
"""Calculate a player's score from cards, patrons, and outposts."""
cards = card_map(pack)
patrons = {patron.id: patron for patron in pack.patrons}
outposts = {outpost.id: outpost for outpost in pack.outposts}
value = sum(cards[item.card_id].points for item in player.cards)
value += sum(patrons[item].points for item in player.patrons)
for item in player.outposts:
definition = outposts[item]
if definition.power == OutpostPower.POINTS_PER_OUTPOST:
value += definition.value * len(player.outposts)
return value
def affordable_payments(player: PlayerState, card: CardDefinition, pack: ContentPack, *, double_wild: bool) -> bool:
"""Report whether a player can cover a card's standard cost."""
discount = bonuses(player, pack)
shortage = sum(
max(0, amount - discount.get(resource, 0) - player.tokens.get(resource, 0))
for resource, amount in card.cost.items()
)
wild_value = 2 if double_wild else 1
return shortage <= player.tokens.get(pack.wild_resource.id, 0) * wild_value
def visible_cards(state: GameState) -> set[str]:
"""Return the identifiers of cards currently visible in markets."""
return {card for market in state.markets.values() for card in market}
def apply_command(
state: GameState,
command: GameCommand,
pack: ContentPack,
settings: GameSettings,
*,
actor_seat: int,
) -> GameState:
"""Validate and apply one command, returning a new state snapshot."""
if state.finished:
error = "The game is finished"
raise RuleError(error)
if command.expected_revision != state.revision:
error = "The board changed; refresh and try again"
raise RuleError(error)
if actor_seat != state.current_seat:
error = "It is not your turn"
raise RuleError(error)
if state.pending and state.pending.seat != actor_seat:
error = "Another player must resolve the pending choice"
raise RuleError(error)
result = state.model_copy(deep=True)
if result.pending:
_apply_pending(result, command, pack, settings)
elif command.type == "take_distinct":
_take_distinct(result, command.payload, pack, settings)
elif command.type == "take_double":
_take_double(result, command.payload, pack, settings)
elif command.type == "reserve":
_reserve(result, command.payload, pack, settings)
elif command.type == "purchase":
_purchase(result, command.payload, pack, settings, is_conquest=False)
else:
error = "Choose a normal turn action"
raise RuleError(error)
result.revision += 1
return result
def _take_distinct(state: GameState, payload: dict[str, Any], pack: ContentPack, settings: GameSettings) -> None:
colors = payload.get("resources")
if not isinstance(colors, list) or len(colors) != len(set(colors)):
error = "Choose distinct resources"
raise RuleError(error)
available = [resource for resource in pack.resource_ids if state.supply.get(resource, 0) > 0]
required = DISTINCT_TAKE_COUNT if len(available) >= DISTINCT_TAKE_COUNT else None
if required is not None and len(colors) != required:
error = "Take exactly three different resources when possible"
raise RuleError(error)
if required is None and not 1 <= len(colors) <= len(available):
error = "Choose one or more available resources"
raise RuleError(error)
if not set(colors) <= set(available):
error = "A selected resource is unavailable"
raise RuleError(error)
player = state.players[state.current_seat]
for resource in colors:
state.supply[resource] -= 1
player.tokens[resource] += 1
state.log.append(f"{player.name} took {len(colors)} different resources")
_after_standard_action(state, pack, settings, action="take_distinct")
def _take_double(state: GameState, payload: dict[str, Any], pack: ContentPack, settings: GameSettings) -> None:
resource = payload.get("resource")
if resource not in pack.resource_ids or state.supply.get(resource, 0) < DOUBLE_TAKE_MINIMUM:
error = "That resource cannot be taken twice"
raise RuleError(error)
player = state.players[state.current_seat]
state.supply[resource] -= 2
player.tokens[resource] += 2
state.log.append(f"{player.name} took two {resource} resources")
extra = _owned_outpost(player, pack, OutpostPower.RESOURCE_AFTER_DOUBLE)
if settings.modules.outposts and extra:
options = [item for item in pack.resource_ids if item != resource and state.supply.get(item, 0) > 0]
if options:
state.pending = PendingChoice(
kind="resource", seat=player.seat, options=options, context={"next": "after_action"}
)
return
_after_standard_action(state, pack, settings, action="take_double")
def _reserve(state: GameState, payload: dict[str, Any], pack: ContentPack, settings: GameSettings) -> None:
player = state.players[state.current_seat]
if len(player.reserved) >= settings.reserve_limit:
error = "Your reserve is full"
raise RuleError(error)
card_id = payload.get("card_id")
source = payload.get("deck")
drawn: list[str]
if card_id:
if card_id not in visible_cards(state):
error = "That card is not visible"
raise RuleError(error)
_assert_not_blocked(state, card_id, player.seat)
_remove_visible(state, card_id)
drawn = [card_id]
elif isinstance(source, str) and source in state.decks:
if not state.decks[source]:
error = "That deck is empty"
raise RuleError(error)
draw_count = 1
post = _owned_outpost(player, pack, OutpostPower.BLIND_RESERVE_TWO)
if post and (not source.startswith("eastern:") or settings.interactions.outpost_reserve_applies_eastern):
draw_count = min(BLIND_RESERVE_DRAW_COUNT, len(state.decks[source]))
drawn = [state.decks[source].pop() for _ in range(draw_count)]
else:
error = "Choose a visible card or deck"
raise RuleError(error)
if len(drawn) == BLIND_RESERVE_DRAW_COUNT:
state.pending = PendingChoice(kind="reserve_keep", seat=player.seat, options=drawn, context={"deck": source})
return
_finish_reserve(state, drawn[0], pack, settings)
def _finish_reserve(state: GameState, card_id: str, pack: ContentPack, settings: GameSettings) -> None:
player = state.players[state.current_seat]
player.reserved.append(card_id)
_refill_markets(state, settings)
wild = pack.wild_resource.id
if state.supply[wild] > 0:
state.supply[wild] -= 1
player.tokens[wild] += 1
state.log.append(f"{player.name} reserved a card")
_after_standard_action(state, pack, settings, action="reserve")
def _purchase(
state: GameState,
payload: dict[str, Any],
pack: ContentPack,
settings: GameSettings,
*,
is_conquest: bool,
) -> None:
player = state.players[state.current_seat]
card_id = payload.get("card_id")
if not isinstance(card_id, str):
error = "Choose a card"
raise RuleError(error)
from_reserve = card_id in player.reserved
if not from_reserve:
if card_id not in visible_cards(state):
error = "That card is not available"
raise RuleError(error)
_assert_not_blocked(state, card_id, player.seat)
if is_conquest and state.fortifications.get(card_id, {}).get(player.seat, 0) < settings.fortifications_per_player:
error = "All your fortifications must occupy the conquest card"
raise RuleError(error)
card = pack.card(card_id)
_pay_for_purchase(state, player, card, payload, pack, settings)
if from_reserve:
player.reserved.remove(card_id)
else:
_remove_visible(state, card_id)
_return_fortifications(state, card_id)
owned = OwnedCard(card_id=card_id)
player.cards.append(owned)
player.purchased_card_count += 1
state.turn_purchase_count += 1
state.log.append(f"{player.name} purchased {card.label}")
if card.effect.kind in {CardEffectKind.COPY_BONUS, CardEffectKind.COPY_AND_CLAIM}:
# Do not allow the new copy card to satisfy its own target requirement.
prior = player.cards.pop()
options = sorted(resource for resource, amount in bonuses(player, pack).items() if amount)
player.cards.append(prior)
if not options:
error = "This card requires an existing bonus to copy"
raise RuleError(error)
state.pending = PendingChoice(
kind="copy_bonus",
seat=player.seat,
options=options,
context={"card_id": card_id, "is_conquest": is_conquest},
)
return
if card.effect.kind == CardEffectKind.CLAIM_FREE:
_queue_free_card(state, card, pack, settings, is_conquest=is_conquest)
return
_after_purchase(state, pack, settings, is_conquest=is_conquest)
def _pay_for_purchase(
state: GameState,
player: PlayerState,
card: CardDefinition,
payload: dict[str, Any],
pack: ContentPack,
settings: GameSettings,
) -> None:
if card.alternate_cost:
discarded = payload.get("discard_cards", [])
_pay_alternate(player, card, discarded, pack)
if not settings.interactions.retain_outposts_after_discard:
_reconcile_outposts(player, pack)
else:
payment = payload.get("payment", {})
virtual_cards = payload.get("virtual_wild_cards", [])
_pay_tokens(state, player, card, payment, virtual_cards, pack, settings)
def _pay_tokens( # noqa: C901 - payment validation mirrors the rule sequence
state: GameState,
player: PlayerState,
card: CardDefinition,
payment: object,
virtual_cards: object,
pack: ContentPack,
settings: GameSettings,
) -> None:
if not isinstance(payment, dict) or any(not isinstance(value, int) or value < 0 for value in payment.values()):
error = "Provide a valid payment"
raise RuleError(error)
allowed = {*pack.resource_ids, pack.wild_resource.id}
if not set(payment) <= allowed:
error = "Payment contains an unknown resource"
raise RuleError(error)
if any(payment.get(resource, 0) > player.tokens.get(resource, 0) for resource in allowed):
error = "Payment uses resources you do not have"
raise RuleError(error)
discount = bonuses(player, pack)
if not isinstance(virtual_cards, list) or len(virtual_cards) != len(set(virtual_cards)):
error = "Virtual-wild selections must be unique"
raise RuleError(error)
owned_by_id = {item.card_id: item for item in player.cards}
virtual_value = 0
for card_id in virtual_cards:
owned = owned_by_id.get(card_id)
if owned is None:
error = "You do not own a selected virtual-wild card"
raise RuleError(error)
definition = pack.card(card_id)
if definition.effect.kind != CardEffectKind.VIRTUAL_WILD:
error = "A selected card does not grant virtual wild resources"
raise RuleError(error)
virtual_value += definition.effect.amount
wild = payment.get(pack.wild_resource.id, 0)
double = bool(_owned_outpost(player, pack, OutpostPower.DOUBLE_WILD))
wild_value = 2 if double else 1
virtual_multiplier = 2 if double and settings.interactions.virtual_wild_can_double else 1
remaining_wild_value = wild * wild_value + virtual_value * virtual_multiplier
for resource in pack.resource_ids:
due = max(0, card.cost.get(resource, 0) - discount.get(resource, 0))
normal = payment.get(resource, 0)
if normal > due:
error = "Payment overpays a normal resource"
raise RuleError(error)
remaining_wild_value -= due - normal
if remaining_wild_value < 0:
error = "Payment does not exactly cover the card cost"
raise RuleError(error)
for resource, amount in payment.items():
player.tokens[resource] -= amount
state.supply[resource] += amount
if virtual_cards:
player.cards = [item for item in player.cards if item.card_id not in virtual_cards]
def _pay_alternate(player: PlayerState, card: CardDefinition, selected: object, pack: ContentPack) -> None:
if not isinstance(selected, list) or not card.alternate_cost or len(selected) != card.alternate_cost.count:
error = "Choose the required owned cards to discard"
raise RuleError(error)
if len(selected) != len(set(selected)):
error = "A card can only be discarded once"
raise RuleError(error)
owned_by_id = {item.card_id: item for item in player.cards}
for card_id in selected:
owned = owned_by_id.get(card_id)
if owned is None:
error = "You do not own a selected discard"
raise RuleError(error)
definition = pack.card(card_id)
effective = owned.copied_resource or definition.bonus_resource
if effective != card.alternate_cost.discard_resource:
error = "A selected discard has the wrong color"
raise RuleError(error)
# Copy cards of the effective color must be discarded first.
matching_copies = [
item.card_id for item in player.cards if item.copied_resource == card.alternate_cost.discard_resource
]
if any(item not in selected for item in matching_copies[: min(len(matching_copies), len(selected))]):
error = "Copied cards of this color must be discarded first"
raise RuleError(error)
player.cards = [item for item in player.cards if item.card_id not in selected]
def _apply_pending( # noqa: C901, PLR0911, PLR0912, PLR0915 - explicit pending-choice state machine
state: GameState, command: GameCommand, pack: ContentPack, settings: GameSettings
) -> None:
pending = state.pending
if pending is None or command.type not in {"choose", "decline", "purchase"}:
error = "Resolve the pending choice"
raise RuleError(error)
player = state.players[state.current_seat]
choice = command.payload.get("choice")
if pending.kind == "reserve_keep":
if choice not in pending.options:
error = "Choose one of the drawn cards"
raise RuleError(error)
other = next(item for item in pending.options if item != choice)
state.decks[pending.context["deck"]].insert(0, other)
state.pending = None
_finish_reserve(state, choice, pack, settings)
return
if pending.kind == "copy_bonus":
if choice not in pending.options:
error = "Choose an existing bonus"
raise RuleError(error)
card_id = pending.context["card_id"]
owned = next(item for item in reversed(player.cards) if item.card_id == card_id)
owned.copied_resource = choice
definition = pack.card(card_id)
state.pending = None
if definition.effect.kind == CardEffectKind.COPY_AND_CLAIM:
_queue_free_card(state, definition, pack, settings, is_conquest=bool(pending.context.get("is_conquest")))
else:
_after_purchase(
state,
pack,
settings,
is_conquest=bool(pending.context.get("is_conquest")),
chained=bool(pending.context.get("chained")),
)
return
if pending.kind == "free_card":
if choice not in pending.options:
error = "Choose an eligible free card"
raise RuleError(error)
if settings.interactions.fortifications_block_free_claim:
_assert_not_blocked(state, choice, player.seat)
_remove_visible(state, choice)
_return_fortifications(state, choice)
claimed = pack.card(choice)
player.cards.append(OwnedCard(card_id=choice))
state.turn_chain_free_count += 1
state.log.append(f"{player.name} claimed {claimed.label} without payment")
state.pending = None
if claimed.effect.kind in {CardEffectKind.COPY_BONUS, CardEffectKind.COPY_AND_CLAIM}:
prior = player.cards.pop()
options = sorted(resource for resource, amount in bonuses(player, pack).items() if amount)
player.cards.append(prior)
if options:
state.pending = PendingChoice(
kind="copy_bonus",
seat=player.seat,
options=options,
context={
"card_id": choice,
"is_conquest": pending.context.get("is_conquest", False),
"chained": True,
},
)
return
_after_purchase(state, pack, settings, is_conquest=bool(pending.context.get("is_conquest")), chained=True)
return
if pending.kind == "resource":
if choice not in pending.options or state.supply.get(str(choice), 0) <= 0:
error = "Choose an available resource"
raise RuleError(error)
state.supply[str(choice)] -= 1
player.tokens[str(choice)] += 1
context = pending.context
state.pending = None
remaining = int(context.get("remaining", 1))
if context.get("next") == "after_purchase" and remaining > 1:
options = [item for item in pack.resource_ids if state.supply.get(item, 0) > 0]
if options:
state.pending = PendingChoice(
kind="resource",
seat=player.seat,
options=options,
context={**context, "remaining": remaining - 1},
)
return
if context.get("next") == "after_purchase":
_after_purchase(state, pack, settings, is_conquest=bool(context.get("is_conquest")), skip_resource=True)
else:
_after_standard_action(state, pack, settings, action="take_double")
return
if pending.kind == "discard_tokens":
discard = command.payload.get("tokens", {})
allowed = {*pack.resource_ids, pack.wild_resource.id}
if not isinstance(discard, dict) or not set(discard) <= allowed or sum(discard.values()) != pending.amount:
error = "Discard exactly the required number of resources"
raise RuleError(error)
for resource, amount in discard.items():
if not isinstance(amount, int) or amount <= 0 or amount > player.tokens.get(resource, 0):
error = "Invalid token discard"
raise RuleError(error)
for resource, amount in discard.items():
player.tokens[resource] -= amount
state.supply[resource] += amount
state.pending = None
_end_checks(state, pack, settings)
return
if pending.kind == "patron":
if choice not in pending.options:
error = "Choose an eligible patron"
raise RuleError(error)
player.patrons.append(choice)
state.available_patrons.remove(choice)
state.pending = None
_check_outposts_or_objectives(state, pack, settings)
return
if pending.kind == "outpost":
if choice not in pending.options:
error = "Choose an eligible outpost"
raise RuleError(error)
player.outposts.append(choice)
state.pending = None
_check_objectives(state, pack, settings)
return
if pending.kind == "fortification":
target = command.payload.get("card_id")
mode = command.payload.get("mode")
if target not in visible_cards(state):
error = "Choose a visible card"
raise RuleError(error)
if (
target in state.markets.get(deck_key("eastern", pack.card(target).tier), [])
and not settings.interactions.fortifications_on_eastern
):
error = "Fortifications cannot occupy eastern cards with this setting"
raise RuleError(error)
if mode == "place":
occupants = state.fortifications.get(target, {})
if any(seat != player.seat and count for seat, count in occupants.items()):
error = "An opponent occupies that card"
raise RuleError(error)
source = command.payload.get("from_card")
if source:
if state.fortifications.get(source, {}).get(player.seat, 0) <= 0:
error = "You have no fortification there"
raise RuleError(error)
state.fortifications[source][player.seat] -= 1
elif player.fortifications_available > 0:
player.fortifications_available -= 1
else:
error = "Move one of your placed fortifications"
raise RuleError(error)
state.fortifications.setdefault(target, {})[player.seat] = occupants.get(player.seat, 0) + 1
elif mode == "remove":
occupants = state.fortifications.get(target, {})
opponents = [(seat, count) for seat, count in occupants.items() if seat != player.seat and count == 1]
if len(opponents) != 1:
error = "Choose a card with exactly one opposing fortification"
raise RuleError(error)
seat, _ = opponents[0]
occupants[seat] = 0
state.players[seat].fortifications_available += 1
else:
error = "Choose place or remove"
raise RuleError(error)
remaining = int(pending.context.get("remaining", 1))
state.pending = None
if remaining > 1 and _queue_fortification(state, pack, settings, remaining=remaining - 1):
return
_after_fortification(state, pack, settings)
return
if pending.kind == "conquest":
if command.type == "decline":
state.pending = None
_enforce_token_limit_or_checks(state, pack, settings)
return
if command.type != "purchase":
error = "Purchase the conquest card or decline"
raise RuleError(error)
state.pending = None
_purchase(state, command.payload, pack, settings, is_conquest=True)
return
error = "Unsupported pending choice"
raise RuleError(error)
def _queue_free_card(
state: GameState, card: CardDefinition, pack: ContentPack, settings: GameSettings, *, is_conquest: bool
) -> None:
options = []
for card_id in visible_cards(state):
candidate = pack.card(card_id)
if candidate.tier != card.effect.target_tier:
continue
if settings.interactions.fortifications_block_free_claim:
occupants = state.fortifications.get(card_id, {})
if any(seat != state.current_seat and count for seat, count in occupants.items()):
continue
options.append(card_id)
if options:
state.pending = PendingChoice(
kind="free_card", seat=state.current_seat, options=sorted(options), context={"is_conquest": is_conquest}
)
else:
_after_purchase(state, pack, settings, is_conquest=is_conquest)
def _after_purchase(
state: GameState,
pack: ContentPack,
settings: GameSettings,
*,
is_conquest: bool,
chained: bool = False,
skip_resource: bool = False,
) -> None:
player = state.players[state.current_seat]
_refill_markets(state, settings)
resource_post = _owned_outpost(player, pack, OutpostPower.RESOURCE_AFTER_PURCHASE)
triggers = not is_conquest or settings.interactions.purchase_resource_on_conquest
if settings.modules.outposts and resource_post and triggers and not skip_resource:
options = [item for item in pack.resource_ids if state.supply.get(item, 0) > 0]
if options:
remaining = 1
if chained and not settings.interactions.chained_claim_is_not_purchase:
remaining += state.turn_chain_free_count
state.pending = PendingChoice(
kind="resource",
seat=player.seat,
options=options,
context={"next": "after_purchase", "is_conquest": is_conquest, "remaining": remaining},
)
return
if settings.modules.fortifications:
remaining = 1
if chained and not settings.interactions.one_fortification_decision_per_purchase_chain:
remaining += state.turn_chain_free_count
if _queue_fortification(state, pack, settings, remaining=remaining):
return
_after_fortification(state, pack, settings)
def _after_fortification(state: GameState, pack: ContentPack, settings: GameSettings) -> None:
player = state.players[state.current_seat]
conquest = [
card_id
for card_id, occupants in state.fortifications.items()
if occupants.get(player.seat, 0) >= settings.fortifications_per_player and card_id in visible_cards(state)
]
if settings.modules.fortifications and conquest:
state.turn_chain_free_count = 0
state.pending = PendingChoice(kind="conquest", seat=player.seat, options=conquest)
return
state.turn_chain_free_count = 0
_enforce_token_limit_or_checks(state, pack, settings)
def _queue_fortification(
state: GameState,
pack: ContentPack,
settings: GameSettings,
*,
remaining: int,
) -> bool:
player = state.players[state.current_seat]
options = [
card_id
for card_id in visible_cards(state)
if settings.interactions.fortifications_on_eastern or pack.card(card_id).deck != "eastern"
]
can_act = False
for card_id in options:
occupants = state.fortifications.get(card_id, {})
can_place = not any(seat != player.seat and count for seat, count in occupants.items())
can_remove = any(seat != player.seat and count == 1 for seat, count in occupants.items())
if can_place or can_remove:
can_act = True
break
if not can_act:
return False
state.pending = PendingChoice(
kind="fortification",
seat=player.seat,
options=sorted(options),
context={"remaining": remaining},
)
return True
def _after_standard_action(state: GameState, pack: ContentPack, settings: GameSettings, *, action: str) -> None:
del action
if settings.modules.fortifications and state.turn_purchase_count:
_after_purchase(state, pack, settings, is_conquest=False)
else:
_enforce_token_limit_or_checks(state, pack, settings)
def _enforce_token_limit_or_checks(state: GameState, pack: ContentPack, settings: GameSettings) -> None:
player = state.players[state.current_seat]
excess = sum(player.tokens.values()) - settings.token_limit
if excess > 0:
state.pending = PendingChoice(kind="discard_tokens", seat=player.seat, amount=excess)
return
_end_checks(state, pack, settings)
def _end_checks(state: GameState, pack: ContentPack, settings: GameSettings) -> None:
player = state.players[state.current_seat]
bonus = bonuses(player, pack)
eligible_patrons = [
patron.id
for patron in pack.patrons
if patron.id in state.available_patrons and requirements_met(patron.requirements, bonus)
]
if eligible_patrons:
state.pending = PendingChoice(kind="patron", seat=player.seat, options=eligible_patrons)
return
_check_outposts_or_objectives(state, pack, settings)
def _check_outposts_or_objectives(state: GameState, pack: ContentPack, settings: GameSettings) -> None:
if settings.modules.outposts and settings.interactions.outposts_before_objectives and _queue_outpost(state, pack):
return
_check_objectives(state, pack, settings)
def _queue_outpost(state: GameState, pack: ContentPack) -> bool:
player = state.players[state.current_seat]
bonus = bonuses(player, pack)
choices = [
outpost.id
for outpost in pack.outposts
if outpost.id not in player.outposts and requirements_met(outpost.requirements, bonus)
]
if choices:
state.pending = PendingChoice(kind="outpost", seat=player.seat, options=choices)
return True
return False
def _check_objectives(state: GameState, pack: ContentPack, settings: GameSettings) -> None:
player = state.players[state.current_seat]
if settings.modules.objectives:
bonus = bonuses(player, pack)
choices = [
objective.id
for objective in pack.objectives
if objective.id in state.available_objectives
and score(player, pack) >= objective.minimum_score
and requirements_met(objective.requirements, bonus)
]
if choices:
player.objective_met = choices[0]
if player.seat not in state.objective_qualifiers:
state.objective_qualifiers.append(player.seat)
if (
settings.modules.outposts
and not settings.interactions.outposts_before_objectives
and _queue_outpost(state, pack)
):
return
_finish_turn(state, pack, settings)
def _finish_turn(state: GameState, pack: ContentPack, settings: GameSettings) -> None:
player = state.players[state.current_seat]
has_score = score(player, pack) >= settings.target_score
has_objective = player.objective_met is not None
triggered = {
"score": has_score,
"objective": has_objective,
"either": has_score or has_objective,
"both": has_score and has_objective,
}[settings.victory_condition]
if triggered and state.finish_at_seat is None:
state.finish_at_seat = (state.first_seat - 1) % len(state.players)
if state.finish_at_seat == state.current_seat:
candidates = (
state.objective_qualifiers if settings.victory_condition == "objective" else list(range(len(state.players)))
)
if not candidates:
candidates = list(range(len(state.players)))
best_score = max(score(state.players[seat], pack) for seat in candidates)
candidates = [seat for seat in candidates if score(state.players[seat], pack) == best_score]
fewest = min(state.players[seat].purchased_card_count for seat in candidates)
state.winners = [seat for seat in candidates if state.players[seat].purchased_card_count == fewest]
state.finished = True
state.pending = None
return
state.current_seat = (state.current_seat + 1) % len(state.players)
if state.current_seat == state.first_seat:
state.round_number += 1
state.turn_purchase_count = 0
state.turn_chain_free_count = 0
def _remove_visible(state: GameState, card_id: str) -> None:
for market in state.markets.values():
if card_id in market:
market.remove(card_id)
return
error = "Card is not visible"
raise RuleError(error)
def _refill_markets(state: GameState, settings: GameSettings) -> None:
for key, market in state.markets.items():
desired = settings.eastern_market_size if key.startswith("eastern:") else settings.base_market_size
deck = state.decks[key]
while len(market) < desired and deck:
market.append(deck.pop())
def _assert_not_blocked(state: GameState, card_id: str, seat: int) -> None:
occupants = state.fortifications.get(card_id, {})
if any(owner != seat and count > 0 for owner, count in occupants.items()):
error = "An opponent's fortification protects that card"
raise RuleError(error)
def _return_fortifications(state: GameState, card_id: str) -> None:
for seat, count in state.fortifications.pop(card_id, {}).items():
state.players[seat].fortifications_available += count
def _owned_outpost(player: PlayerState, pack: ContentPack, power: OutpostPower) -> OutpostDefinition | None:
definitions = {item.id: item for item in pack.outposts}
return next((definitions[item] for item in player.outposts if definitions[item].power == power), None)
def _reconcile_outposts(player: PlayerState, pack: ContentPack) -> None:
definitions = {item.id: item for item in pack.outposts}
bonus = bonuses(player, pack)
player.outposts = [item for item in player.outposts if requirements_met(definitions[item].requirements, bonus)]
def public_state(state: GameState, viewer_seat: int | None) -> dict[str, Any]:
"""Serialize state while redacting other players' reserved cards and deck order."""
data = state.model_dump(mode="json")
for player in data["players"]:
if player["seat"] != viewer_seat:
player["reserved"] = [None] * len(player["reserved"])
data["decks"] = {key: len(value) for key, value in state.decks.items()}
if state.pending and state.pending.seat != viewer_seat:
data["pending"]["options"] = []
data["pending"]["context"] = {}
return data
+159
View File
@@ -0,0 +1,159 @@
"""Legal command generation used by bots and server-rendered controls."""
from __future__ import annotations
import itertools
import math
import uuid
from .engine import bonuses, visible_cards
from .models import ContentPack, GameCommand, GameSettings, GameState, OutpostPower
DISTINCT_TAKE_COUNT = 3
DOUBLE_TAKE_MINIMUM = 4
def command(state: GameState, type_: str, payload: dict | None = None) -> GameCommand:
"""Create a command targeting the state's current revision."""
return GameCommand(
command_id=uuid.uuid4().hex,
expected_revision=state.revision,
type=type_, # type: ignore[arg-type]
payload=payload or {},
)
def legal_commands( # noqa: C901, PLR0911, PLR0912 - mirrors pending and normal action branches
state: GameState, pack: ContentPack, settings: GameSettings, seat: int
) -> list[GameCommand]:
"""Enumerate meaningful legal commands without revealing hidden information."""
if state.finished or seat != state.current_seat:
return []
player = state.players[seat]
if state.pending:
pending = state.pending
if pending.seat != seat:
return []
if pending.kind == "discard_tokens":
available = [item for item, amount in player.tokens.items() for _ in range(amount)]
payload: dict[str, int] = {}
for item in available[: pending.amount]:
payload[item] = payload.get(item, 0) + 1
return [command(state, "choose", {"tokens": payload})]
if pending.kind == "fortification":
pending_results = []
for target in pending.options:
occupants = state.fortifications.get(target, {})
if not any(owner != seat and count for owner, count in occupants.items()):
if player.fortifications_available:
pending_results.append(command(state, "choose", {"card_id": target, "mode": "place"}))
for source, source_occupants in state.fortifications.items():
if source_occupants.get(seat, 0):
pending_results.append(
command(state, "choose", {"card_id": target, "from_card": source, "mode": "place"})
)
if any(owner != seat and count == 1 for owner, count in occupants.items()):
pending_results.append(command(state, "choose", {"card_id": target, "mode": "remove"}))
return pending_results
if pending.kind == "conquest":
conquest_results = [command(state, "decline")]
for card_id in pending.options:
payment = default_payment(state, pack, settings, card_id, seat)
if payment is not None:
conquest_results.append(command(state, "purchase", {"card_id": card_id, **payment}))
return conquest_results
return [command(state, "choose", {"choice": item}) for item in pending.options]
results: list[GameCommand] = []
available = [resource for resource in pack.resource_ids if state.supply.get(resource, 0)]
take_size = DISTINCT_TAKE_COUNT if len(available) >= DISTINCT_TAKE_COUNT else 1
sizes = [take_size] if len(available) >= DISTINCT_TAKE_COUNT else list(range(1, len(available) + 1))
for size in sizes:
results.extend(
command(state, "take_distinct", {"resources": list(items)})
for items in itertools.combinations(available, size)
)
results.extend(
command(state, "take_double", {"resource": resource})
for resource in pack.resource_ids
if state.supply.get(resource, 0) >= DOUBLE_TAKE_MINIMUM
)
if len(player.reserved) < settings.reserve_limit:
results.extend(
command(state, "reserve", {"card_id": card_id})
for card_id in visible_cards(state)
if not _blocked(state, card_id, seat)
)
results.extend(command(state, "reserve", {"deck": key}) for key, cards in state.decks.items() if cards)
for card_id in [*visible_cards(state), *player.reserved]:
if card_id not in player.reserved and _blocked(state, card_id, seat):
continue
payment = default_payment(state, pack, settings, card_id, seat)
if payment is not None:
results.append(command(state, "purchase", {"card_id": card_id, **payment}))
return results
def default_payment( # noqa: C901, PLR0912 - ordered payment rules are intentionally explicit
state: GameState,
pack: ContentPack,
settings: GameSettings,
card_id: str,
seat: int,
) -> dict | None:
"""Return one legal colored-first payment, or None when unaffordable."""
player = state.players[seat]
card = pack.card(card_id)
if card.alternate_cost:
matching = []
definitions = {item.id: item for item in pack.cards}
for owned in player.cards:
resource = owned.copied_resource or definitions[owned.card_id].bonus_resource
if resource == card.alternate_cost.discard_resource:
matching.append(owned.card_id)
copies = [
item for item in matching if next(owned for owned in player.cards if owned.card_id == item).copied_resource
]
ordered = [*copies, *(item for item in matching if item not in copies)]
if len(ordered) < card.alternate_cost.count:
return None
return {"discard_cards": ordered[: card.alternate_cost.count]}
discount = bonuses(player, pack)
payment: dict[str, int] = {}
shortage = 0
for resource in pack.resource_ids:
due = max(0, card.cost.get(resource, 0) - discount.get(resource, 0))
amount = min(due, player.tokens.get(resource, 0))
if amount:
payment[resource] = amount
shortage += due - amount
double_wild = any(
outpost.power == OutpostPower.DOUBLE_WILD and outpost.id in player.outposts for outpost in pack.outposts
)
wild_value = 2 if double_wild else 1
virtual_multiplier = 2 if double_wild and settings.interactions.virtual_wild_can_double else 1
virtual_cards: list[str] = []
virtual_total = 0
if shortage > player.tokens.get(pack.wild_resource.id, 0) * wild_value:
for owned in player.cards:
definition = pack.card(owned.card_id)
if definition.effect.kind.value != "virtual_wild":
continue
virtual_cards.append(owned.card_id)
virtual_total += definition.effect.amount * virtual_multiplier
if virtual_total + player.tokens.get(pack.wild_resource.id, 0) * wild_value >= shortage:
break
remaining = max(0, shortage - virtual_total)
wild_needed = math.ceil(remaining / wild_value)
if wild_needed > player.tokens.get(pack.wild_resource.id, 0) or virtual_total + wild_needed * wild_value < shortage:
return None
if wild_needed:
payment[pack.wild_resource.id] = wild_needed
result: dict[str, object] = {"payment": payment}
if virtual_cards:
result["virtual_wild_cards"] = virtual_cards
return result
def _blocked(state: GameState, card_id: str, seat: int) -> bool:
return any(owner != seat and count for owner, count in state.fortifications.get(card_id, {}).items())
+394
View File
@@ -0,0 +1,394 @@
"""Typed content, configuration, commands, and game state."""
from __future__ import annotations
from enum import StrEnum
from typing import Annotated, Any, Literal
from pydantic import BaseModel, ConfigDict, Field, model_validator
class StrictModel(BaseModel):
"""Base model that rejects misspelled fields."""
model_config = ConfigDict(extra="forbid")
Slug = Annotated[str, Field(pattern=r"^[a-z][a-z0-9_-]{0,47}$")]
ShortText = Annotated[str, Field(min_length=1, max_length=80)]
TOKEN_DARK_INK_THRESHOLD = 0.82
STANDARD_RESOURCE_SYMBOLS = {
"onyx": "O",
"sapphire": "S",
"emerald": "E",
"ruby": "R",
"diamond": "D",
"gold": "G",
}
class ResourceDefinition(StrictModel):
"""A normal or wild resource rendered by the UI."""
id: Slug
label: ShortText
symbol: Annotated[str, Field(min_length=1, max_length=4)]
color: Annotated[str, Field(pattern=r"^#[0-9a-fA-F]{6}$")]
@model_validator(mode="after")
def normalize_standard_symbol(self) -> ResourceDefinition:
"""Store the conventional symbol for a standard gem label."""
self.symbol = STANDARD_RESOURCE_SYMBOLS.get(self.label.casefold(), self.symbol)
return self
@property
def ink_color(self) -> str:
"""Return readable token lettering for the configured background."""
red, green, blue = (int(self.color[index : index + 2], 16) for index in (1, 3, 5))
perceived_brightness = (299 * red + 587 * green + 114 * blue) / 255_000
return "#111111" if perceived_brightness >= TOKEN_DARK_INK_THRESHOLD else "#ffffff"
class RequirementKind(StrEnum):
"""Identify how a requirement selects resource colors."""
COLOR = "color"
ANY_COLOR = "any_color"
class Requirement(StrictModel):
"""A fixed-color or same-color bonus requirement."""
id: Slug
kind: RequirementKind
count: Annotated[int, Field(ge=1, le=99)]
resource: Slug | None = None
exclude: list[Slug] = Field(default_factory=list)
distinct_from: list[Slug] = Field(default_factory=list)
@model_validator(mode="after")
def validate_shape(self) -> Requirement:
"""Validate fields that depend on the requirement kind."""
if self.kind == RequirementKind.COLOR and self.resource is None:
error = "color requirements need a resource"
raise ValueError(error)
if self.kind == RequirementKind.ANY_COLOR and self.resource is not None:
error = "any_color requirements cannot name a resource"
raise ValueError(error)
return self
class CardEffectKind(StrEnum):
"""Identify a supported server-side card effect."""
NONE = "none"
VIRTUAL_WILD = "virtual_wild"
COPY_BONUS = "copy_bonus"
COPY_AND_CLAIM = "copy_and_claim"
MULTI_BONUS = "multi_bonus"
CLAIM_FREE = "claim_free"
class CardEffect(StrictModel):
"""A bounded built-in card effect; uploaded code is never evaluated."""
kind: CardEffectKind = CardEffectKind.NONE
amount: Annotated[int, Field(ge=1, le=10)] = 1
target_tier: Annotated[int, Field(ge=1, le=3)] | None = None
class AlternateCost(StrictModel):
"""Purchase a card by discarding owned cards of one effective color."""
discard_resource: Slug
count: Annotated[int, Field(ge=1, le=10)]
class CardDefinition(StrictModel):
"""A card supplied by a user-owned content pack."""
id: Slug
label: ShortText
deck: Literal["base", "eastern"] = "base"
tier: Annotated[int, Field(ge=1, le=3)]
points: Annotated[int, Field(ge=0, le=99)] = 0
bonus_resource: Slug | None = None
cost: dict[Slug, Annotated[int, Field(ge=0, le=99)]] = Field(default_factory=dict)
effect: CardEffect = Field(default_factory=CardEffect)
alternate_cost: AlternateCost | None = None
@model_validator(mode="after")
def validate_effect(self) -> CardDefinition:
"""Validate fields that depend on the selected card effect."""
effect = self.effect
if effect.kind in {CardEffectKind.CLAIM_FREE, CardEffectKind.COPY_AND_CLAIM}:
if effect.target_tier is None:
error = "claim effects require target_tier"
raise ValueError(error)
if effect.target_tier >= self.tier:
error = "free-card effects must target a lower tier"
raise ValueError(error)
elif effect.target_tier is not None:
error = "target_tier is only valid for claim effects"
raise ValueError(error)
if effect.kind == CardEffectKind.MULTI_BONUS and self.bonus_resource is None:
error = "multi_bonus cards require bonus_resource"
raise ValueError(error)
return self
class PatronDefinition(StrictModel):
"""Define a patron claimed by meeting permanent-bonus requirements."""
id: Slug
label: ShortText
points: Annotated[int, Field(ge=0, le=99)]
requirements: list[Requirement]
class ObjectiveDefinition(StrictModel):
"""Define an optional objective-based victory condition."""
id: Slug
label: ShortText
minimum_score: Annotated[int, Field(ge=0, le=999)]
requirements: list[Requirement]
class OutpostPower(StrEnum):
"""Identify a built-in outpost ability."""
RESOURCE_AFTER_PURCHASE = "resource_after_purchase"
RESOURCE_AFTER_DOUBLE = "resource_after_double"
DOUBLE_WILD = "double_wild"
POINTS_PER_OUTPOST = "points_per_outpost"
BLIND_RESERVE_TWO = "blind_reserve_two"
class OutpostDefinition(StrictModel):
"""Define an outpost and the bonuses required to claim it."""
id: Slug
label: ShortText
requirements: list[Requirement]
power: OutpostPower
value: Annotated[int, Field(ge=1, le=10)] = 1
class PackMetadata(StrictModel):
"""Describe the identity and authorship of a content pack."""
id: Slug
name: ShortText
version: Annotated[str, Field(min_length=1, max_length=32)]
author: Annotated[str, Field(max_length=80)] = ""
class ContentPack(StrictModel):
"""Versioned user-provided game content."""
schema_version: Literal[1]
metadata: PackMetadata
resources: Annotated[list[ResourceDefinition], Field(min_length=5, max_length=5)]
wild_resource: ResourceDefinition
cards: list[CardDefinition]
patrons: list[PatronDefinition] = Field(default_factory=list)
objectives: list[ObjectiveDefinition] = Field(default_factory=list)
outposts: list[OutpostDefinition] = Field(default_factory=list)
@model_validator(mode="before")
@classmethod
def normalize_patrons(cls, data: object) -> object:
"""Normalize the legacy governors key to the neutral patrons key."""
if isinstance(data, dict):
if "patrons" in data and "governors" in data:
error = "use patrons or governors, not both"
raise ValueError(error)
if "governors" in data:
data = dict(data)
data["patrons"] = data.pop("governors")
return data
@model_validator(mode="after")
def validate_references(self) -> ContentPack: # noqa: C901, PLR0912 - validates each reference family
"""Validate identifiers and references across the complete pack."""
resources = {item.id for item in self.resources}
if self.wild_resource.id in resources:
error = "wild resource ID must be distinct"
raise ValueError(error)
id_collections = [
[item.id for item in self.resources],
[item.id for item in self.cards],
[item.id for item in self.patrons],
[item.id for item in self.objectives],
[item.id for item in self.outposts],
]
for ids in id_collections:
if len(ids) != len(set(ids)):
error = "IDs must be unique within each content collection"
raise ValueError(error)
req_ids: set[str]
for card in self.cards:
if card.bonus_resource is not None and card.bonus_resource not in resources:
error = f"card {card.id} has an unknown bonus resource"
raise ValueError(error)
if not set(card.cost) <= resources:
error = f"card {card.id} has an unknown cost resource"
raise ValueError(error)
if card.alternate_cost and card.alternate_cost.discard_resource not in resources:
error = f"card {card.id} has an unknown alternate-cost resource"
raise ValueError(error)
requirement_owners: list[PatronDefinition | ObjectiveDefinition | OutpostDefinition] = [
*self.patrons,
*self.objectives,
*self.outposts,
]
for owner in requirement_owners:
req_ids = {requirement.id for requirement in owner.requirements}
if len(req_ids) != len(owner.requirements):
error = f"{owner.id} has duplicate requirement IDs"
raise ValueError(error)
for requirement in owner.requirements:
if requirement.resource and requirement.resource not in resources:
error = f"{owner.id} references an unknown resource"
raise ValueError(error)
if not set(requirement.exclude) <= resources:
error = f"{owner.id} excludes an unknown resource"
raise ValueError(error)
if not set(requirement.distinct_from) <= req_ids:
error = f"{owner.id} references an unknown requirement"
raise ValueError(error)
return self
@property
def resource_ids(self) -> tuple[str, ...]:
"""Return normal resource identifiers in display order."""
return tuple(resource.id for resource in self.resources)
def card(self, card_id: str) -> CardDefinition:
"""Return a card definition by identifier."""
return next(card for card in self.cards if card.id == card_id)
class Modules(StrictModel):
"""Select optional rule modules for a game."""
objectives: bool = False
outposts: bool = False
eastern_decks: bool = False
fortifications: bool = False
class Interactions(StrictModel):
"""Configure interactions between optional rule modules."""
outpost_reserve_applies_eastern: bool = True
virtual_wild_can_double: bool = True
retain_outposts_after_discard: bool = True
outposts_before_objectives: bool = True
purchase_resource_on_conquest: bool = True
chained_claim_is_not_purchase: bool = True
fortifications_on_eastern: bool = True
fortifications_block_free_claim: bool = True
one_fortification_decision_per_purchase_chain: bool = True
class GameSettings(StrictModel):
"""Configure win conditions, limits, markets, and rule modules."""
target_score: Annotated[int, Field(ge=1, le=999)] = 15
victory_condition: Literal["score", "objective", "either", "both"] = "score"
first_player_mode: Literal["random", "selected"] = "random"
first_player_seat: Annotated[int, Field(ge=0, le=3)] = 0
token_limit: Annotated[int, Field(ge=1, le=99)] = 10
reserve_limit: Annotated[int, Field(ge=0, le=20)] = 3
base_market_size: Annotated[int, Field(ge=1, le=10)] = 4
eastern_market_size: Annotated[int, Field(ge=1, le=10)] = 2
objective_count: Annotated[int, Field(ge=1, le=10)] = 3
fortifications_per_player: Annotated[int, Field(ge=1, le=10)] = 3
modules: Modules = Field(default_factory=Modules)
interactions: Interactions = Field(default_factory=Interactions)
class OwnedCard(StrictModel):
"""Track an owned card and any copied resource assignment."""
card_id: Slug
copied_resource: Slug | None = None
class PlayerState(StrictModel):
"""Store the mutable state belonging to one player."""
seat: int
name: ShortText
tokens: dict[str, int]
cards: list[OwnedCard] = Field(default_factory=list)
reserved: list[Slug] = Field(default_factory=list)
patrons: list[Slug] = Field(default_factory=list)
outposts: list[Slug] = Field(default_factory=list)
objective_met: Slug | None = None
fortifications_available: int = 0
purchased_card_count: int = 0
class PendingChoice(StrictModel):
"""Describe a choice that must be resolved before play continues."""
kind: Literal[
"discard_tokens",
"copy_bonus",
"free_card",
"reserve_keep",
"resource",
"fortification",
"conquest",
"patron",
"outpost",
]
seat: int
options: list[str] = Field(default_factory=list)
amount: int = 1
context: dict[str, Any] = Field(default_factory=dict)
class GameState(StrictModel):
"""Store a complete authoritative game snapshot."""
room_code: str
seed: int
revision: int = 0
players: list[PlayerState]
supply: dict[str, int]
decks: dict[str, list[Slug]]
markets: dict[str, list[Slug]]
available_patrons: list[Slug]
available_objectives: list[Slug]
fortifications: dict[Slug, dict[int, int]] = Field(default_factory=dict)
current_seat: int = 0
first_seat: int = 0
round_number: int = 1
pending: PendingChoice | None = None
finish_at_seat: int | None = None
objective_qualifiers: list[int] = Field(default_factory=list)
winners: list[int] = Field(default_factory=list)
finished: bool = False
turn_purchase_count: int = 0
turn_chain_free_count: int = 0
log: list[str] = Field(default_factory=list)
class GameCommand(StrictModel):
"""Represent one revision-bound action submitted to the engine."""
command_id: Annotated[str, Field(min_length=8, max_length=64)]
expected_revision: Annotated[int, Field(ge=0)]
type: Literal[
"take_distinct",
"take_double",
"reserve",
"purchase",
"choose",
"decline",
]
payload: dict[str, Any] = Field(default_factory=dict)
+58
View File
@@ -0,0 +1,58 @@
"""Requirement evaluation shared by patrons, objectives, outposts, and AI."""
from __future__ import annotations
from .models import Requirement, RequirementKind
def requirements_met(requirements: list[Requirement], bonuses: dict[str, int]) -> bool:
"""Return whether one assignment satisfies all fixed and wildcard clauses."""
chosen: dict[str, str] = {}
def visit(index: int) -> bool:
if index == len(requirements):
return True
requirement = requirements[index]
if requirement.kind == RequirementKind.COLOR:
if bonuses.get(requirement.resource or "", 0) < requirement.count:
return False
chosen[requirement.id] = requirement.resource or ""
return visit(index + 1)
forbidden = set(requirement.exclude)
forbidden.update(chosen[item] for item in requirement.distinct_from if item in chosen)
for resource, amount in bonuses.items():
if resource in forbidden or amount < requirement.count:
continue
chosen[requirement.id] = resource
if visit(index + 1):
return True
chosen.pop(requirement.id, None)
return False
fixed = [item for item in requirements if item.kind == RequirementKind.COLOR]
flexible = [item for item in requirements if item.kind == RequirementKind.ANY_COLOR]
return visit_ordered([*fixed, *flexible], bonuses, chosen)
def visit_ordered(requirements: list[Requirement], bonuses: dict[str, int], chosen: dict[str, str]) -> bool:
"""Backtracking evaluator kept separate for straightforward unit testing."""
if not requirements:
return True
requirement, *rest = requirements
if requirement.kind == RequirementKind.COLOR:
resource = requirement.resource or ""
if bonuses.get(resource, 0) < requirement.count:
return False
chosen[requirement.id] = resource
return visit_ordered(rest, bonuses, chosen)
forbidden = set(requirement.exclude)
forbidden.update(chosen[item] for item in requirement.distinct_from if item in chosen)
for resource, amount in bonuses.items():
if resource in forbidden or amount < requirement.count:
continue
chosen[requirement.id] = resource
if visit_ordered(rest, bonuses, chosen):
return True
chosen.pop(requirement.id, None)
return False
+35
View File
@@ -0,0 +1,35 @@
"""In-process revision notifications for SSE clients."""
from __future__ import annotations
import asyncio
import contextlib
from collections import defaultdict
class EventBroker:
"""Fan out room revisions; clients always reload the latest snapshot."""
def __init__(self) -> None:
"""Initialize an empty subscription registry."""
self._queues: dict[str, set[asyncio.Queue[int]]] = defaultdict(set)
def subscribe(self, room_code: str) -> asyncio.Queue[int]:
"""Subscribe a bounded notification queue to a room."""
queue: asyncio.Queue[int] = asyncio.Queue(maxsize=1)
self._queues[room_code].add(queue)
return queue
def unsubscribe(self, room_code: str, queue: asyncio.Queue[int]) -> None:
"""Remove a room notification queue from the registry."""
self._queues[room_code].discard(queue)
if not self._queues[room_code]:
self._queues.pop(room_code, None)
def publish(self, room_code: str, revision: int) -> None:
"""Publish the newest room revision to every subscriber."""
for queue in tuple(self._queues.get(room_code, ())):
if queue.full():
with contextlib.suppress(asyncio.QueueEmpty):
queue.get_nowait()
queue.put_nowait(revision)
+91
View File
@@ -0,0 +1,91 @@
"""FastAPI entry point for Gems."""
from __future__ import annotations
import asyncio
from contextlib import asynccontextmanager, suppress
from typing import TYPE_CHECKING, Annotated
import typer
import uvicorn
from fastapi import FastAPI, Request, Response
from fastapi.staticfiles import StaticFiles
from .config import load_config
from .events import EventBroker
from .persistence import Repository
from .rooms import RoomService
from .routes import router
from .security import CredentialService
from .web import STATIC_DIR
if TYPE_CHECKING:
from collections.abc import AsyncIterator
from starlette.middleware.base import RequestResponseEndpoint
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
"""Initialize and close application-scoped services."""
config = app.state.config
repository = Repository(config.database_path)
app.state.repository = repository
app.state.rooms = RoomService(repository, CredentialService(config.key_path), EventBroker())
repository.cleanup()
cleanup_task = asyncio.create_task(_cleanup_rooms(repository))
try:
yield
finally:
cleanup_task.cancel()
with suppress(asyncio.CancelledError):
await cleanup_task
repository.close()
async def _cleanup_rooms(repository: Repository) -> None:
while True:
await asyncio.sleep(3600)
repository.cleanup()
def create_app() -> FastAPI:
"""Create an isolated application instance."""
app = FastAPI(title="Gems", docs_url=None, redoc_url=None, lifespan=lifespan)
app.state.config = load_config()
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
app.include_router(router)
@app.middleware("http")
async def security_headers(request: Request, call_next: RequestResponseEndpoint) -> Response:
response = await call_next(request)
response.headers.setdefault("Referrer-Policy", "no-referrer")
response.headers.setdefault("X-Content-Type-Options", "nosniff")
response.headers.setdefault("X-Frame-Options", "DENY")
response.headers.setdefault("X-Robots-Tag", "noindex, nofollow")
response.headers.setdefault(
"Content-Security-Policy",
"default-src 'self'; script-src 'self'; style-src 'self'; style-src-attr 'unsafe-inline'; "
"img-src 'self' data:; connect-src 'self'",
)
if request.url.path.startswith("/rooms/"):
response.headers.setdefault("Cache-Control", "no-store")
return response
return app
app = create_app()
def serve(
host: Annotated[str | None, typer.Option()] = None,
port: Annotated[int | None, typer.Option()] = None,
) -> None:
"""Run the Gems ASGI application with Uvicorn."""
config = load_config()
uvicorn.run("python.gems.main:app", host=host or config.host, port=port or config.port, workers=1)
if __name__ == "__main__":
typer.run(serve)
+255
View File
@@ -0,0 +1,255 @@
"""SQLite persistence for rooms, memberships, snapshots, and events."""
from __future__ import annotations
import json
import sqlite3
import threading
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from typing import TYPE_CHECKING
from .domain.models import ContentPack, GameSettings, GameState
if TYPE_CHECKING:
from pathlib import Path
@dataclass(slots=True)
class Member:
"""Represent a human or AI seat persisted for a room."""
room_code: str
seat: int
name: str
controller: str
difficulty: str | None
credential_hash: str | None
is_host: bool
ready: bool
last_seen: str
@dataclass(slots=True)
class Room:
"""Represent persisted room metadata and its current snapshot."""
code: str
status: str
settings: GameSettings
pack: ContentPack | None
pack_digest: str | None
state: GameState | None
revision: int
created_at: str
updated_at: str
class Repository:
"""Small transactional repository designed for one application worker."""
def __init__(self, path: Path) -> None:
"""Open the SQLite database and initialize its schema."""
path.parent.mkdir(parents=True, exist_ok=True)
self._connection = sqlite3.connect(path, check_same_thread=False)
self._connection.row_factory = sqlite3.Row
self._lock = threading.RLock()
with self._connection:
self._connection.execute("PRAGMA journal_mode=WAL")
self._connection.execute("PRAGMA foreign_keys=ON")
self._connection.execute("PRAGMA busy_timeout=5000")
self._connection.executescript(
"""
CREATE TABLE IF NOT EXISTS schema_version(version INTEGER NOT NULL);
INSERT INTO schema_version(version) SELECT 1 WHERE NOT EXISTS (SELECT 1 FROM schema_version);
CREATE TABLE IF NOT EXISTS rooms(
code TEXT PRIMARY KEY, status TEXT NOT NULL, settings_json TEXT NOT NULL,
pack_json TEXT, pack_digest TEXT, state_json TEXT, revision INTEGER NOT NULL,
created_at TEXT NOT NULL, updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS members(
room_code TEXT NOT NULL REFERENCES rooms(code) ON DELETE CASCADE,
seat INTEGER NOT NULL, name TEXT NOT NULL, controller TEXT NOT NULL,
difficulty TEXT, credential_hash TEXT, is_host INTEGER NOT NULL,
ready INTEGER NOT NULL, last_seen TEXT NOT NULL,
PRIMARY KEY(room_code, seat), UNIQUE(room_code, credential_hash)
);
CREATE TABLE IF NOT EXISTS room_events(
room_code TEXT NOT NULL REFERENCES rooms(code) ON DELETE CASCADE,
revision INTEGER NOT NULL, command_id TEXT NOT NULL, actor_seat INTEGER,
event_type TEXT NOT NULL, payload_json TEXT NOT NULL, created_at TEXT NOT NULL,
PRIMARY KEY(room_code, command_id)
);
"""
)
def room_exists(self, code: str) -> bool:
"""Report whether a room code exists."""
return self._connection.execute("SELECT 1 FROM rooms WHERE code=?", (code,)).fetchone() is not None
def create_room(self, room: Room, host: Member) -> None:
"""Persist a new room and its host in one transaction."""
with self._lock, self._connection:
self._connection.execute(
"INSERT INTO rooms VALUES(?,?,?,?,?,?,?,?,?)",
(
room.code,
room.status,
room.settings.model_dump_json(),
None,
None,
None,
room.revision,
room.created_at,
room.updated_at,
),
)
self._insert_member(host)
def _insert_member(self, member: Member) -> None:
self._connection.execute(
"INSERT INTO members VALUES(?,?,?,?,?,?,?,?,?)",
(
member.room_code,
member.seat,
member.name,
member.controller,
member.difficulty,
member.credential_hash,
int(member.is_host),
int(member.ready),
member.last_seen,
),
)
def add_member(self, member: Member) -> None:
"""Add a member and refresh the room's activity timestamp."""
with self._lock, self._connection:
self._insert_member(member)
self.touch(member.room_code)
def update_member(self, member: Member) -> None:
"""Persist all mutable fields for an existing member."""
with self._lock, self._connection:
self._connection.execute(
"""UPDATE members SET name=?,controller=?,difficulty=?,credential_hash=?,is_host=?,ready=?,last_seen=?
WHERE room_code=? AND seat=?""",
(
member.name,
member.controller,
member.difficulty,
member.credential_hash,
int(member.is_host),
int(member.ready),
member.last_seen,
member.room_code,
member.seat,
),
)
def remove_member(self, code: str, seat: int) -> None:
"""Remove a member from a room seat."""
with self._lock, self._connection:
self._connection.execute("DELETE FROM members WHERE room_code=? AND seat=?", (code, seat))
def get_room(self, code: str) -> Room | None:
"""Load a room by code, including its typed JSON fields."""
row = self._connection.execute("SELECT * FROM rooms WHERE code=?", (code.upper(),)).fetchone()
if row is None:
return None
return Room(
code=row["code"],
status=row["status"],
settings=GameSettings.model_validate_json(row["settings_json"]),
pack=ContentPack.model_validate_json(row["pack_json"]) if row["pack_json"] else None,
pack_digest=row["pack_digest"],
state=GameState.model_validate_json(row["state_json"]) if row["state_json"] else None,
revision=row["revision"],
created_at=row["created_at"],
updated_at=row["updated_at"],
)
def members(self, code: str) -> list[Member]:
"""Load all room members ordered by seat."""
rows = self._connection.execute("SELECT * FROM members WHERE room_code=? ORDER BY seat", (code,)).fetchall()
return [
Member(
room_code=row["room_code"],
seat=row["seat"],
name=row["name"],
controller=row["controller"],
difficulty=row["difficulty"],
credential_hash=row["credential_hash"],
is_host=bool(row["is_host"]),
ready=bool(row["ready"]),
last_seen=row["last_seen"],
)
for row in rows
]
def member_for_credential(self, code: str, credential_hash: str) -> Member | None:
"""Find the room member associated with a credential digest."""
return next((member for member in self.members(code) if member.credential_hash == credential_hash), None)
def save_lobby(self, room: Room) -> None:
"""Persist mutable lobby configuration and revision data."""
with self._lock, self._connection:
self._connection.execute(
"""UPDATE rooms SET settings_json=?,pack_json=?,pack_digest=?,revision=?,updated_at=? WHERE code=?""",
(
room.settings.model_dump_json(),
room.pack.model_dump_json() if room.pack else None,
room.pack_digest,
room.revision,
room.updated_at,
room.code,
),
)
def save_state(self, room: Room, command_id: str, actor_seat: int | None, payload: dict) -> bool:
"""Persist a game snapshot and its idempotent command event."""
now = utc_now()
with self._lock, self._connection:
existing = self._connection.execute(
"SELECT 1 FROM room_events WHERE room_code=? AND command_id=?", (room.code, command_id)
).fetchone()
if existing:
return False
self._connection.execute(
"UPDATE rooms SET status=?,state_json=?,revision=?,updated_at=? WHERE code=?",
(room.status, room.state.model_dump_json() if room.state else None, room.revision, now, room.code),
)
self._connection.execute(
"INSERT INTO room_events VALUES(?,?,?,?,?,?,?)",
(room.code, room.revision, command_id, actor_seat, "command", json.dumps(payload), now),
)
return True
def has_command(self, code: str, command_id: str) -> bool:
"""Report whether a command was already persisted for a room."""
row = self._connection.execute(
"SELECT 1 FROM room_events WHERE room_code=? AND command_id=?",
(code, command_id),
).fetchone()
return row is not None
def touch(self, code: str) -> None:
"""Refresh a room's activity timestamp."""
with self._connection:
self._connection.execute("UPDATE rooms SET updated_at=? WHERE code=?", (utc_now(), code))
def cleanup(self, days: int = 30) -> int:
"""Delete rooms inactive for the requested number of days."""
cutoff = (datetime.now(UTC) - timedelta(days=days)).isoformat()
with self._lock, self._connection:
cursor = self._connection.execute("DELETE FROM rooms WHERE updated_at < ?", (cutoff,))
return cursor.rowcount
def close(self) -> None:
"""Close the underlying SQLite connection."""
self._connection.close()
def utc_now() -> str:
"""Return the current UTC time as an ISO 8601 string."""
return datetime.now(UTC).isoformat()
+375
View File
@@ -0,0 +1,375 @@
"""Application service for anonymous rooms and AI turns."""
from __future__ import annotations
import asyncio
import secrets
from collections import defaultdict
from datetime import UTC, datetime
from typing import TYPE_CHECKING
from .ai import choose_ai_command
from .domain.engine import RuleError, apply_command, new_game
from .domain.models import GameCommand, GameSettings
from .persistence import Member, Repository, Room, utc_now
if TYPE_CHECKING:
from .content import ParsedPack
from .events import EventBroker
from .security import CredentialService
ROOM_ALPHABET = "23456789ABCDEFGHJKMNPQRSTUVWXYZ"
MAX_PLAYERS = 4
OFFLINE_AFTER_SECONDS = 30
DISPLAY_NAME_MAX_LENGTH = 32
CONTROL_CHARACTER_LIMIT = 32
class RoomError(ValueError):
"""Safe user-facing room operation error."""
class RoomService:
"""Coordinates persistence, authorization, per-room locking, and bots."""
def __init__(self, repository: Repository, credentials: CredentialService, broker: EventBroker) -> None:
"""Initialize the service with persistence, security, and events."""
self.repository = repository
self.credentials = credentials
self.broker = broker
self._locks: dict[str, asyncio.Lock] = defaultdict(asyncio.Lock)
self._ai_tasks: dict[str, asyncio.Task[None]] = {}
def create(self, host_name: str, credential: str | None = None) -> tuple[Room, str]:
"""Create a lobby and assign its first human member as host."""
name = validate_name(host_name)
code = self._new_code()
credential = credential or self.credentials.issue()
now = utc_now()
room = Room(
code=code,
status="lobby",
settings=GameSettings(),
pack=None,
pack_digest=None,
state=None,
revision=0,
created_at=now,
updated_at=now,
)
host = Member(
room_code=code,
seat=0,
name=name,
controller="human",
difficulty=None,
credential_hash=self.credentials.digest(credential),
is_host=True,
ready=False,
last_seen=now,
)
self.repository.create_room(room, host)
return room, credential
def join(self, code: str, name: str, credential: str | None = None) -> tuple[Room, str]:
"""Join an available lobby or restore an existing membership."""
room = self.require_room(code)
credential = credential or self.credentials.issue()
existing = self.repository.member_for_credential(room.code, self.credentials.digest(credential))
if existing is not None:
return room, credential
if room.status != "lobby":
error = "This room is no longer accepting players"
raise RoomError(error)
members = self.repository.members(room.code)
if len(members) >= MAX_PLAYERS:
error = "This room is full"
raise RoomError(error)
clean_name = validate_name(name)
if any(member.name.casefold() == clean_name.casefold() for member in members):
error = "That display name is already in use"
raise RoomError(error)
seat = next(item for item in range(MAX_PLAYERS) if item not in {member.seat for member in members})
member = Member(
room_code=room.code,
seat=seat,
name=clean_name,
controller="human",
difficulty=None,
credential_hash=self.credentials.digest(credential),
is_host=False,
ready=False,
last_seen=utc_now(),
)
self.repository.add_member(member)
self._bump_lobby(room)
return room, credential
def authenticate(self, code: str, credential: str | None) -> tuple[Room, Member]:
"""Authenticate a room member and refresh their presence timestamp."""
room = self.require_room(code)
if not credential:
error = "Join the room to continue"
raise RoomError(error)
member = self.repository.member_for_credential(room.code, self.credentials.digest(credential))
if member is None:
error = "Join the room to continue"
raise RoomError(error)
member.last_seen = utc_now()
self.repository.update_member(member)
return room, member
def require_room(self, code: str) -> Room:
"""Load a room or raise a safe user-facing error."""
room = self.repository.get_room(code.upper())
if room is None:
error = "Room not found"
raise RoomError(error)
return room
def set_pack(self, room: Room, actor: Member, parsed: ParsedPack) -> Room:
"""Assign a validated content pack to a lobby."""
self._require_host_lobby(room, actor)
room.pack = parsed.pack
room.pack_digest = parsed.digest
self._bump_lobby(room)
return room
def set_settings(self, room: Room, actor: Member, settings: GameSettings) -> Room:
"""Replace the editable rules for a lobby."""
self._require_host_lobby(room, actor)
room.settings = settings
self._bump_lobby(room)
return room
def set_ready(self, room: Room, actor: Member, *, ready: bool) -> None:
"""Update a human lobby member's readiness."""
if room.status != "lobby" or actor.controller != "human":
error = "Readiness can only change in the lobby"
raise RoomError(error)
actor.ready = ready
actor.last_seen = utc_now()
self.repository.update_member(actor)
self._bump_lobby(room)
def add_ai(self, room: Room, actor: Member, difficulty: str) -> None:
"""Add an AI-controlled seat to a lobby."""
self._require_host_lobby(room, actor)
if difficulty not in {"easy", "medium", "hard"}:
error = "Choose easy, medium, or hard AI"
raise RoomError(error)
members = self.repository.members(room.code)
if len(members) >= MAX_PLAYERS:
error = "This room is full"
raise RoomError(error)
seat = next(item for item in range(MAX_PLAYERS) if item not in {member.seat for member in members})
number = 1 + sum(member.controller == "ai" for member in members)
self.repository.add_member(
Member(
room_code=room.code,
seat=seat,
name=f"Bot {number}",
controller="ai",
difficulty=difficulty,
credential_hash=None,
is_host=False,
ready=True,
last_seen=utc_now(),
)
)
self._bump_lobby(room)
def remove_seat(self, room: Room, actor: Member, seat: int) -> None:
"""Remove a non-host seat from a lobby."""
self._require_host_lobby(room, actor)
target = next((member for member in self.repository.members(room.code) if member.seat == seat), None)
if target is None or target.is_host:
error = "That seat cannot be removed"
raise RoomError(error)
self.repository.remove_member(room.code, seat)
self._bump_lobby(room)
def transfer_host(self, room: Room, actor: Member, seat: int) -> None:
"""Transfer lobby ownership to another human player."""
self._require_host_lobby(room, actor)
target = next((member for member in self.repository.members(room.code) if member.seat == seat), None)
if target is None or target.controller != "human" or target.seat == actor.seat:
error = "Choose another human player"
raise RoomError(error)
actor.is_host = False
target.is_host = True
self.repository.update_member(actor)
self.repository.update_member(target)
self._bump_lobby(room)
async def start(self, room: Room, actor: Member) -> Room:
"""Validate the lobby and start its first game."""
self._require_host_lobby(room, actor)
if room.pack is None:
error = "Upload a valid content pack first"
raise RoomError(error)
members = self.repository.members(room.code)
humans = [member for member in members if member.controller == "human"]
if not humans or any(not member.ready for member in humans):
error = "Every human player must be ready"
raise RoomError(error)
ordered = sorted(members, key=lambda member: member.seat)
# Compress lobby seat gaps so engine seats always index the players list.
for new_seat, member in enumerate(ordered):
if member.seat != new_seat:
self.repository.remove_member(room.code, member.seat)
member.seat = new_seat
self.repository.add_member(member)
try:
state = new_game(
room.code, [member.name for member in ordered], room.pack, room.settings, seed=secrets.randbits(63)
)
except RuleError as exc:
raise RoomError(str(exc)) from exc
room.state = state
room.status = "playing"
room.revision = state.revision
self.repository.save_state(room, f"start-{secrets.token_hex(8)}", actor.seat, {"type": "start"})
self.broker.publish(room.code, room.revision)
self.schedule_ai(room.code)
return room
async def play_again(self, room: Room, actor: Member) -> Room:
"""Start a fresh game in a finished room while preserving its table setup."""
if not actor.is_host or room.status != "finished":
error = "Only the host can replay a finished game"
raise RoomError(error)
async with self._locks[room.code]:
room = self.require_room(room.code)
if room.status != "finished" or room.pack is None:
error = "This game is not ready for a replay"
raise RoomError(error)
members = sorted(self.repository.members(room.code), key=lambda member: member.seat)
try:
room.state = new_game(
room.code,
[member.name for member in members],
room.pack,
room.settings,
seed=secrets.randbits(63),
)
except RuleError as exc:
raise RoomError(str(exc)) from exc
room.status = "playing"
room.revision = room.state.revision
self.repository.save_state(
room,
f"replay-{secrets.token_hex(8)}",
actor.seat,
{"type": "play_again"},
)
self.broker.publish(room.code, room.revision)
self.schedule_ai(room.code)
return room
async def submit(self, room: Room, actor: Member, command: GameCommand) -> Room:
"""Apply and persist one idempotent human game command."""
if room.status != "playing" or room.state is None or room.pack is None:
error = "This game is not active"
raise RoomError(error)
async with self._locks[room.code]:
room = self.require_room(room.code)
if self.repository.has_command(room.code, command.command_id):
return room
if room.state is None or room.pack is None:
error = "This game is not active"
raise RoomError(error)
try:
room.state = apply_command(room.state, command, room.pack, room.settings, actor_seat=actor.seat)
except RuleError as exc:
raise RoomError(str(exc)) from exc
room.revision = room.state.revision
room.status = "finished" if room.state.finished else "playing"
room.state.log = room.state.log[-200:]
self.repository.save_state(room, command.command_id, actor.seat, command.model_dump(mode="json"))
self.broker.publish(room.code, room.revision)
self.schedule_ai(room.code)
return room
def replace_with_ai(self, room: Room, actor: Member, seat: int, difficulty: str) -> None:
"""Replace a disconnected human player with an AI controller."""
if not actor.is_host or room.status != "playing":
error = "Only the host can replace a player during a game"
raise RoomError(error)
target = next((item for item in self.repository.members(room.code) if item.seat == seat), None)
if target is None or target.controller != "human" or target.is_host:
error = "That player cannot be replaced"
raise RoomError(error)
last_seen = datetime.fromisoformat(target.last_seen)
if (datetime.now(UTC) - last_seen).total_seconds() < OFFLINE_AFTER_SECONDS:
error = "That player is still connected"
raise RoomError(error)
target.controller = "ai"
target.difficulty = difficulty
target.credential_hash = None
target.ready = True
self.repository.update_member(target)
self.broker.publish(room.code, room.revision)
self.schedule_ai(room.code)
def schedule_ai(self, code: str) -> None:
"""Schedule an AI runner when a room does not already have one."""
task = self._ai_tasks.get(code)
if task is None or task.done():
self._ai_tasks[code] = asyncio.create_task(self._run_ai(code))
async def _run_ai(self, code: str) -> None:
while True:
room = self.require_room(code)
if room.status != "playing" or room.state is None or room.pack is None:
return
member = next(
(item for item in self.repository.members(code) if item.seat == room.state.current_seat), None
)
if member is None or member.controller != "ai":
return
await asyncio.sleep(0.35)
async with self._locks[code]:
room = self.require_room(code)
if room.state is None or room.pack is None or room.state.current_seat != member.seat:
continue
chosen = choose_ai_command(
room.state, room.pack, room.settings, member.seat, member.difficulty or "medium"
)
try:
room.state = apply_command(room.state, chosen, room.pack, room.settings, actor_seat=member.seat)
except RuleError:
return
room.revision = room.state.revision
room.status = "finished" if room.state.finished else "playing"
room.state.log = room.state.log[-200:]
self.repository.save_state(room, chosen.command_id, member.seat, chosen.model_dump(mode="json"))
self.broker.publish(code, room.revision)
def _new_code(self) -> str:
while True:
code = "".join(secrets.choice(ROOM_ALPHABET) for _ in range(8))
if not self.repository.room_exists(code):
return code
def _require_host_lobby(self, room: Room, actor: Member) -> None:
if not actor.is_host or room.status != "lobby":
error = "Only the host can change this lobby"
raise RoomError(error)
def _bump_lobby(self, room: Room) -> None:
room.revision += 1
room.updated_at = utc_now()
self.repository.save_lobby(room)
self.broker.publish(room.code, room.revision)
def validate_name(name: str) -> str:
"""Normalize and validate a player display name."""
clean = " ".join(name.split())
if not 1 <= len(clean) <= DISPLAY_NAME_MAX_LENGTH or any(
ord(character) < CONTROL_CHARACTER_LIMIT for character in clean
):
error = "Display names must be 1-32 plain-text characters"
raise RoomError(error)
return clean
+5
View File
@@ -0,0 +1,5 @@
"""HTTP routes for Gems."""
from .app import router
__all__ = ["router"]
+626
View File
@@ -0,0 +1,626 @@
"""FastAPI pages, HTMX mutations, and SSE stream."""
from __future__ import annotations
import asyncio
import html
from collections.abc import AsyncIterator, Callable, Mapping
from datetime import UTC, datetime
from typing import Annotated, Literal
from fastapi import APIRouter, Form, Request, Response, UploadFile
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse, StreamingResponse
from pydantic import ValidationError
from python.gems.content import ContentPackError, content_pack_schema, parse_content_pack
from python.gems.domain.engine import bonuses, public_state, score
from python.gems.domain.legal_actions import command as build_command
from python.gems.domain.legal_actions import legal_commands
from python.gems.domain.models import GameCommand, GameSettings, Interactions, Modules
from python.gems.persistence import Member, Room
from python.gems.rooms import RoomError, RoomService
from python.gems.web import templates
router = APIRouter()
COOKIE_NAME = "gems_credential"
DISPLAY_NAME_COOKIE = "gems_display_name"
DISPLAY_NAME_MAX_AGE = 365 * 24 * 60 * 60
DISPLAY_NAME_MAX_LENGTH = 32
OFFLINE_AFTER_SECONDS = 30
RoomOperation = Callable[[Room, Member], None]
def service(request: Request) -> RoomService:
"""Return the application-scoped room service."""
return request.app.state.rooms
def _request_origin(request: Request) -> str:
return f"{request.url.scheme}://{request.url.netloc}".rstrip("/")
def _public_origin(request: Request) -> str:
"""Prefer the current address when the untouched local default is configured."""
configured = request.app.state.config.public_origin.rstrip("/")
if configured == "http://127.0.0.1:8082":
return _request_origin(request)
return configured
def _context(request: Request, room: Room, member: Member, *, error: str | None = None) -> dict[str, object]:
members = service(request).repository.members(room.code)
humans = [item for item in members if item.controller == "human"]
all_humans_ready = bool(humans) and all(item.ready for item in humans)
csrf = service(request).credentials.csrf(request.cookies.get(COOKIE_NAME, ""), room.code)
context: dict[str, object] = {
"request": request,
"room": room,
"member": member,
"members": members,
"csrf": csrf,
"error": error,
"public_origin": _public_origin(request),
"all_humans_ready": all_humans_ready,
"can_start": room.pack is not None and all_humans_ready,
"offline_seats": {
item.seat
for item in members
if item.controller == "human"
and (datetime.now(UTC) - datetime.fromisoformat(item.last_seen)).total_seconds() >= OFFLINE_AFTER_SECONDS
},
}
if room.state and room.pack:
context["state"] = public_state(room.state, member.seat)
context["pack"] = room.pack
context["cards"] = {card.id: card for card in room.pack.cards}
context["resources"] = {resource.id: resource for resource in [*room.pack.resources, room.pack.wild_resource]}
context["normal_resource_ids"] = room.pack.resource_ids
context["all_resource_ids"] = (*room.pack.resource_ids, room.pack.wild_resource.id)
context["member_by_seat"] = {item.seat: item for item in members}
context["bonus_counts"] = {player.seat: bonuses(player, room.pack) for player in room.state.players}
context["patrons"] = {item.id: item for item in room.pack.patrons}
context["objectives"] = {item.id: item for item in room.pack.objectives}
context["outposts"] = {item.id: item for item in room.pack.outposts}
commands = (
legal_commands(room.state, room.pack, room.settings, member.seat) if member.controller == "human" else []
)
context["actions"] = [(action_label(item, room), item.model_dump_json()) for item in commands]
context["double_resource_ids"] = {
str(item.payload["resource"])
for item in commands
if item.type == "take_double" and item.payload.get("resource")
}
card_actions: dict[str, dict[str, str]] = {}
deck_reserve_actions: dict[str, str] = {}
for item in commands:
card_id = item.payload.get("card_id")
if item.type in {"purchase", "reserve"} and card_id:
card_actions.setdefault(str(card_id), {})[item.type] = item.model_dump_json()
elif item.type == "reserve" and item.payload.get("deck"):
deck_reserve_actions[str(item.payload["deck"])] = item.model_dump_json()
context["card_actions"] = card_actions
context["deck_reserve_actions"] = deck_reserve_actions
context["scores"] = {player.seat: score(player, room.pack) for player in room.state.players}
return context
def action_label(command: GameCommand, room: Room) -> str:
"""Build a human-readable label for a legal game command."""
payload = command.payload
if command.type == "take_distinct":
label = "Take " + ", ".join(payload["resources"])
elif command.type == "take_double":
label = f"Take two {payload['resource']}"
elif command.type == "reserve":
if payload.get("card_id") and room.pack:
label = f"Reserve {room.pack.card(payload['card_id']).label}"
else:
label = f"Reserve from {payload.get('deck', 'deck')}"
elif command.type == "purchase" and room.pack:
label = f"Purchase {room.pack.card(payload['card_id']).label}"
elif command.type == "choose" and isinstance(payload.get("tokens"), dict) and room.pack:
resources = {item.id: item for item in [*room.pack.resources, room.pack.wild_resource]}
discarded = payload["tokens"]
label = "Discard " + ", ".join(
f"{amount} {resources[str(resource)].label}" for resource, amount in discarded.items()
)
elif command.type == "decline":
label = "Decline"
else:
choice = payload.get("choice") or payload.get("card_id") or "selection"
label = f"Choose {choice}"
return label
def _selected_gem_command(room: Room, seat: int, selected: list[str]) -> GameCommand:
if room.state is None or room.pack is None:
error = "The game has not started"
raise RoomError(error)
if not selected:
error = "Select gem piles before taking gems"
raise RoomError(error)
if len(selected) != len(set(selected)) or not set(selected) <= set(room.pack.resource_ids):
error = "Choose each normal gem pile at most once"
raise RoomError(error)
commands = legal_commands(room.state, room.pack, room.settings, seat)
chosen = next(
(
item
for item in commands
if item.type == "take_double" and len(selected) == 1 and item.payload.get("resource") == selected[0]
),
None,
)
if chosen is None:
chosen = next(
(
item
for item in commands
if item.type == "take_distinct"
and set(item.payload.get("resources", [])) == set(selected)
and len(item.payload.get("resources", [])) == len(selected)
),
None,
)
if chosen is None:
error = "Choose three different available gems, or one pile with at least four gems to take a pair"
raise RoomError(error)
return chosen
def _double_gem_command(room: Room, seat: int, resource: str) -> GameCommand:
if room.state is None or room.pack is None:
error = "The game has not started"
raise RoomError(error)
chosen = next(
(
item
for item in legal_commands(room.state, room.pack, room.settings, seat)
if item.type == "take_double" and item.payload.get("resource") == resource
),
None,
)
if chosen is None:
error = "That gem pair is unavailable; a pile needs at least four gems"
raise RoomError(error)
return chosen
def _discard_token_command(room: Room, seat: int, submitted: Mapping[str, object]) -> GameCommand:
if room.state is None or room.pack is None:
error = "The game has not started"
raise RoomError(error)
pending = room.state.pending
if pending is None or pending.kind != "discard_tokens" or pending.seat != seat:
error = "There are no excess tokens to discard"
raise RoomError(error)
discard: dict[str, int] = {}
for resource in (*room.pack.resource_ids, room.pack.wild_resource.id):
value = submitted.get(resource, "0")
try:
amount = int(value) if isinstance(value, str) else 0
except ValueError as exc:
error = "Token discard amounts must be whole numbers"
raise RoomError(error) from exc
if amount < 0:
error = "Token discard amounts cannot be negative"
raise RoomError(error)
if amount:
discard[resource] = amount
return build_command(room.state, "choose", {"tokens": discard})
def _authenticate(request: Request, code: str) -> tuple[Room, Member]:
return service(request).authenticate(code, request.cookies.get(COOKIE_NAME))
def _saved_display_name(request: Request) -> str:
name = request.cookies.get(DISPLAY_NAME_COOKIE, "").strip()
return name if len(name) <= DISPLAY_NAME_MAX_LENGTH else ""
def _set_identity_cookies(response: Response, request: Request, credential: str, name: str) -> None:
cookie_options = {"httponly": True, "secure": request.app.state.config.secure_cookies, "samesite": "lax"}
response.set_cookie(COOKIE_NAME, credential, **cookie_options)
response.set_cookie(DISPLAY_NAME_COOKIE, name.strip(), max_age=DISPLAY_NAME_MAX_AGE, **cookie_options)
def _check_csrf(request: Request, room_code: str, csrf: str) -> None:
credential = request.cookies.get(COOKIE_NAME, "")
if not service(request).credentials.valid_csrf(credential, room_code, csrf):
error = "Your form expired; reload and try again"
raise RoomError(error)
origin = request.headers.get("origin")
allowed_origins = {
request.app.state.config.public_origin.rstrip("/"),
_request_origin(request),
}
if origin and origin.rstrip("/") not in allowed_origins:
error = "Request origin was rejected"
raise RoomError(error)
@router.get("/", response_class=HTMLResponse)
def home(request: Request) -> Response:
"""Render the room creation and join page."""
return templates.TemplateResponse(request, "home.html", {"display_name": _saved_display_name(request)})
@router.post("/rooms")
def create_room(request: Request, name: Annotated[str, Form()]) -> Response:
"""Create a room and redirect its host to the lobby."""
try:
room, credential = service(request).create(name, request.cookies.get(COOKIE_NAME))
except RoomError as exc:
return templates.TemplateResponse(
request, "home.html", {"error": str(exc), "display_name": name}, status_code=422
)
response = RedirectResponse(f"/rooms/{room.code}", status_code=303)
_set_identity_cookies(response, request, credential, name)
return response
@router.post("/join")
def join_code(code: Annotated[str, Form()]) -> Response:
"""Normalize a room code and redirect to its join page."""
return RedirectResponse(f"/join/{code.strip().upper()}", status_code=303)
@router.get("/join/{code}", response_class=HTMLResponse)
def join_page(request: Request, code: str) -> Response:
"""Render the display-name form for an existing room."""
try:
room = service(request).require_room(code)
except RoomError:
return templates.TemplateResponse(
request,
"join.html",
{"code": code, "error": "Room not found", "display_name": _saved_display_name(request)},
status_code=404,
)
return templates.TemplateResponse(
request, "join.html", {"code": room.code, "display_name": _saved_display_name(request)}
)
@router.post("/join/{code}")
def join_room(request: Request, code: str, name: Annotated[str, Form()]) -> Response:
"""Join a room and persist the member's browser identity."""
try:
room, credential = service(request).join(code, name, request.cookies.get(COOKIE_NAME))
except RoomError as exc:
return templates.TemplateResponse(
request, "join.html", {"code": code, "error": str(exc), "display_name": name}, status_code=422
)
response = RedirectResponse(f"/rooms/{room.code}", status_code=303)
_set_identity_cookies(response, request, credential, name)
return response
@router.get("/rooms/{code}", response_class=HTMLResponse)
def room_page(request: Request, code: str) -> Response:
"""Render the authenticated lobby or game table."""
try:
room, member = _authenticate(request, code)
except RoomError:
return RedirectResponse(f"/join/{code}", status_code=303)
return templates.TemplateResponse(request, "room.html", _context(request, room, member))
def _partial(request: Request, code: str, operation: RoomOperation, *, error_status: int = 422) -> Response:
try:
room, member = _authenticate(request, code)
operation(room, member)
room = service(request).require_room(code)
return templates.TemplateResponse(request, "partials/room_state.html", _context(request, room, member))
except (RoomError, ValidationError, ContentPackError) as exc:
try:
room, member = _authenticate(request, code)
return templates.TemplateResponse(
request,
"partials/room_state.html",
_context(request, room, member, error=str(exc)),
status_code=error_status,
)
except RoomError:
return HTMLResponse(html.escape(str(exc)), status_code=error_status)
def _csrf_partial(request: Request, code: str, csrf: str, operation: RoomOperation) -> Response:
def authorized(room: Room, member: Member) -> None:
_check_csrf(request, room.code, csrf)
operation(room, member)
return _partial(request, code, authorized)
@router.post("/rooms/{code}/pack", response_class=HTMLResponse)
async def upload_pack(request: Request, code: str, csrf: Annotated[str, Form()], pack_file: UploadFile) -> Response:
"""Validate and activate an uploaded JSON content pack."""
raw = await pack_file.read(512 * 1024 + 1)
def operation(room: Room, member: Member) -> None:
_check_csrf(request, room.code, csrf)
service(request).set_pack(room, member, parse_content_pack(raw))
return _partial(request, code, operation)
@router.post("/rooms/{code}/settings/preset", response_class=HTMLResponse)
def apply_preset(
request: Request,
code: str,
csrf: Annotated[str, Form()],
preset: Annotated[str, Form()],
) -> Response:
"""Apply a built-in rules preset to a lobby."""
presets: dict[str, tuple[Modules, Literal["score", "objective", "either", "both"]]] = {
"classic": (Modules(), "score"),
"objectives": (Modules(objectives=True), "objective"),
"objectives_outposts": (Modules(objectives=True, outposts=True), "objective"),
"eastern_fortifications": (Modules(eastern_decks=True, fortifications=True), "score"),
"all": (Modules(objectives=True, outposts=True, eastern_decks=True, fortifications=True), "objective"),
}
def operation(room: Room, member: Member) -> None:
_check_csrf(request, room.code, csrf)
if preset not in presets:
error = "Unknown rules preset"
raise RoomError(error)
modules, victory = presets[preset]
settings = room.settings.model_copy(deep=True)
settings.modules = modules
settings.victory_condition = victory
settings.interactions = Interactions()
service(request).set_settings(room, member, settings)
return _partial(request, code, operation)
@router.get("/rooms/{code}/pack")
def download_pack(request: Request, code: str) -> Response:
"""Download the room's active content pack as JSON."""
room, _ = _authenticate(request, code)
if room.pack is None:
return JSONResponse({"detail": "No pack uploaded"}, status_code=404)
return JSONResponse(
room.pack.model_dump(mode="json"),
headers={"Content-Disposition": f'attachment; filename="{room.pack.metadata.id}.json"'},
)
@router.post("/rooms/{code}/settings", response_class=HTMLResponse)
async def update_settings(request: Request, code: str) -> Response:
"""Validate and save custom lobby rules."""
form = await request.form()
def checked(name: str) -> bool:
return name in form
def value(name: str, default: str) -> str:
submitted = form.get(name)
return submitted if isinstance(submitted, str) else default
def operation(room: Room, member: Member) -> None:
_check_csrf(request, room.code, value("csrf", ""))
settings = GameSettings.model_validate(
{
"target_score": value("target_score", "15"),
"victory_condition": value("victory_condition", "score"),
"first_player_mode": value("first_player_mode", "random"),
"first_player_seat": value("first_player_seat", "0"),
"token_limit": value("token_limit", "10"),
"reserve_limit": value("reserve_limit", "3"),
"base_market_size": value("base_market_size", "4"),
"eastern_market_size": value("eastern_market_size", "2"),
"objective_count": value("objective_count", "3"),
"fortifications_per_player": value("fortifications_per_player", "3"),
"modules": Modules(
objectives=checked("objectives"),
outposts=checked("outposts"),
eastern_decks=checked("eastern_decks"),
fortifications=checked("fortifications"),
),
"interactions": Interactions(**{field: checked(field) for field in Interactions.model_fields}),
}
)
service(request).set_settings(room, member, settings)
return _partial(request, code, operation)
@router.post("/rooms/{code}/ready", response_class=HTMLResponse)
def ready(
request: Request,
code: str,
csrf: Annotated[str, Form()],
ready_value: Annotated[bool, Form(alias="ready")],
) -> Response:
"""Update the authenticated human player's ready state."""
return _csrf_partial(
request, code, csrf, lambda room, member: service(request).set_ready(room, member, ready=ready_value)
)
@router.post("/rooms/{code}/seats/ai", response_class=HTMLResponse)
def add_ai(request: Request, code: str, csrf: Annotated[str, Form()], difficulty: Annotated[str, Form()]) -> Response:
"""Add an AI-controlled player to a lobby."""
return _csrf_partial(request, code, csrf, lambda room, member: service(request).add_ai(room, member, difficulty))
@router.post("/rooms/{code}/seats/{seat}/remove", response_class=HTMLResponse)
def remove_seat(request: Request, code: str, seat: int, csrf: Annotated[str, Form()]) -> Response:
"""Remove a non-host player from a lobby."""
return _csrf_partial(request, code, csrf, lambda room, member: service(request).remove_seat(room, member, seat))
@router.post("/rooms/{code}/seats/{seat}/make-host", response_class=HTMLResponse)
def transfer_host(request: Request, code: str, seat: int, csrf: Annotated[str, Form()]) -> Response:
"""Transfer lobby ownership to another human player."""
return _csrf_partial(request, code, csrf, lambda room, member: service(request).transfer_host(room, member, seat))
@router.post("/rooms/{code}/seats/{seat}/replace-with-ai", response_class=HTMLResponse)
def replace_with_ai(
request: Request,
code: str,
seat: int,
csrf: Annotated[str, Form()],
difficulty: Annotated[str, Form()] = "medium",
) -> Response:
"""Replace a disconnected player with an AI controller."""
return _csrf_partial(
request,
code,
csrf,
lambda room, member: service(request).replace_with_ai(room, member, seat, difficulty),
)
@router.post("/rooms/{code}/start", response_class=HTMLResponse)
async def start_game(request: Request, code: str, csrf: Annotated[str, Form()]) -> Response:
"""Start a game after validating the lobby's readiness."""
try:
room, member = _authenticate(request, code)
_check_csrf(request, room.code, csrf)
await service(request).start(room, member)
room = service(request).require_room(code)
return templates.TemplateResponse(request, "partials/room_state.html", _context(request, room, member))
except RoomError as exc:
room, member = _authenticate(request, code)
return templates.TemplateResponse(
request, "partials/room_state.html", _context(request, room, member, error=str(exc)), status_code=422
)
@router.post("/rooms/{code}/play-again", response_class=HTMLResponse)
async def play_again(request: Request, code: str, csrf: Annotated[str, Form()]) -> Response:
"""Start a fresh game with the finished room's current setup."""
try:
room, member = _authenticate(request, code)
_check_csrf(request, room.code, csrf)
room = await service(request).play_again(room, member)
return templates.TemplateResponse(request, "partials/room_state.html", _context(request, room, member))
except RoomError as exc:
room, member = _authenticate(request, code)
return templates.TemplateResponse(
request, "partials/room_state.html", _context(request, room, member, error=str(exc)), status_code=422
)
@router.post("/rooms/{code}/commands", response_class=HTMLResponse)
async def game_command(
request: Request, code: str, csrf: Annotated[str, Form()], command_json: Annotated[str, Form()]
) -> Response:
"""Validate and submit one serialized game command."""
try:
room, member = _authenticate(request, code)
_check_csrf(request, room.code, csrf)
command = GameCommand.model_validate_json(command_json)
room = await service(request).submit(room, member, command)
return templates.TemplateResponse(request, "partials/room_state.html", _context(request, room, member))
except (RoomError, ValidationError) as exc:
room, member = _authenticate(request, code)
return templates.TemplateResponse(
request, "partials/room_state.html", _context(request, room, member, error=str(exc)), status_code=422
)
@router.post("/rooms/{code}/take-gems", response_class=HTMLResponse)
async def take_gems(request: Request, code: str) -> Response:
"""Translate the visual gem selection into a legal command."""
form = await request.form()
try:
room, member = _authenticate(request, code)
_check_csrf(request, room.code, str(form.get("csrf", "")))
pair_resource = form.get("pair_resource")
if pair_resource is not None:
chosen = _double_gem_command(room, member.seat, str(pair_resource))
else:
selected = [str(item) for item in form.getlist("resources")]
chosen = _selected_gem_command(room, member.seat, selected)
room = await service(request).submit(room, member, chosen)
return templates.TemplateResponse(request, "partials/room_state.html", _context(request, room, member))
except RoomError as exc:
room, member = _authenticate(request, code)
return templates.TemplateResponse(
request, "partials/room_state.html", _context(request, room, member, error=str(exc)), status_code=422
)
@router.post("/rooms/{code}/discard-tokens", response_class=HTMLResponse)
async def discard_tokens(request: Request, code: str) -> Response:
"""Return a player's selected excess tokens to the supply."""
form = await request.form()
try:
room, member = _authenticate(request, code)
_check_csrf(request, room.code, str(form.get("csrf", "")))
submitted = {
resource: form.get(f"token_{resource}", "0")
for resource in ((*room.pack.resource_ids, room.pack.wild_resource.id) if room.pack else ())
}
chosen = _discard_token_command(room, member.seat, submitted)
room = await service(request).submit(room, member, chosen)
return templates.TemplateResponse(request, "partials/room_state.html", _context(request, room, member))
except RoomError as exc:
room, member = _authenticate(request, code)
return templates.TemplateResponse(
request, "partials/room_state.html", _context(request, room, member, error=str(exc)), status_code=422
)
@router.get("/rooms/{code}/events")
async def room_events(request: Request, code: str) -> Response:
"""Stream room revisions to an authenticated browser over SSE."""
try:
room, member = _authenticate(request, code)
except RoomError:
return HTMLResponse("Unauthorized", status_code=401)
queue = service(request).broker.subscribe(room.code)
stream_member = member
async def stream() -> AsyncIterator[str]:
current_member = stream_member
try:
# Comments keep the SSE connection alive without replacing in-progress form input.
yield ": connected\n\n"
while True:
try:
await asyncio.wait_for(queue.get(), timeout=15)
latest = service(request).require_room(code)
yield _sse_fragment(request, latest, current_member)
except TimeoutError:
_, current_member = _authenticate(request, code)
yield ": keepalive\n\n"
if await request.is_disconnected():
break
finally:
service(request).broker.unsubscribe(room.code, queue)
return StreamingResponse(
stream(), media_type="text/event-stream", headers={"Cache-Control": "no-store", "X-Accel-Buffering": "no"}
)
def _sse_fragment(request: Request, room: Room, member: Member) -> str:
rendered = templates.get_template("partials/room_state.html").render(_context(request, room, member))
return "event: room\n" + "\n".join(f"data: {line}" for line in rendered.splitlines()) + "\n\n"
@router.get("/schemas/content-pack-v1.json")
def schema() -> Response:
"""Return the JSON Schema for content-pack version one."""
return JSONResponse(content_pack_schema())
@router.get("/healthz")
def health() -> dict[str, str]:
"""Report that the HTTP process is running."""
return {"status": "ok"}
@router.get("/readyz")
def readyz(request: Request) -> dict[str, str]:
"""Report readiness after checking the database connection."""
request.app.state.repository._connection.execute("SELECT 1").fetchone() # noqa: SLF001
return {"status": "ready"}
+40
View File
@@ -0,0 +1,40 @@
"""Opaque browser credentials and CSRF protection."""
from __future__ import annotations
import hashlib
import hmac
import secrets
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from pathlib import Path
class CredentialService:
"""Hash browser credentials with a stable per-installation key."""
def __init__(self, key_path: Path) -> None:
"""Load or create the installation's credential-signing key."""
key_path.parent.mkdir(parents=True, exist_ok=True)
if not key_path.exists():
key_path.write_bytes(secrets.token_bytes(32))
key_path.chmod(0o600)
self._key = key_path.read_bytes()
@staticmethod
def issue() -> str:
"""Issue a cryptographically random browser credential."""
return secrets.token_urlsafe(32)
def digest(self, credential: str) -> str:
"""Create the persistent keyed digest of a browser credential."""
return hmac.new(self._key, credential.encode(), hashlib.sha256).hexdigest()
def csrf(self, credential: str, room_code: str) -> str:
"""Create a room-scoped CSRF token for a browser credential."""
return hmac.new(self._key, f"csrf:{credential}:{room_code}".encode(), hashlib.sha256).hexdigest()
def valid_csrf(self, credential: str, room_code: str, candidate: str) -> bool:
"""Validate a candidate CSRF token using constant-time comparison."""
return hmac.compare_digest(self.csrf(credential, room_code), candidate)
File diff suppressed because one or more lines are too long
+18
View File
@@ -0,0 +1,18 @@
.preset-row{display:flex;gap:.5rem;flex-wrap:wrap;margin-bottom:1rem}.invite-link{margin:0;min-width:min(430px,55vw)}.danger{color:var(--danger)}.pack-status{border-left:3px solid var(--danger);padding:.65rem .8rem;background:#4a201d55;border-radius:0 9px 9px 0}button:disabled,button:disabled:hover{background:#252b2f;border-color:#3b4348;color:#7f898f;cursor:not-allowed;opacity:.7}
.room-topbar{height:auto;min-height:68px;gap:1.25rem}.room-identity{display:flex;align-items:baseline;gap:.65rem;margin-right:auto}.room-identity .eyebrow{margin:0}.room-identity strong{font-size:1.2rem;letter-spacing:.14em}.room-invite{display:flex;align-items:center;grid-template-columns:auto minmax(260px,430px);gap:.65rem}.room-invite input{padding:.5rem .65rem}
@media(max-width:700px){.room-topbar{flex-wrap:wrap;padding-block:.7rem}.room-invite{order:3;width:100%;grid-template-columns:auto 1fr}.invite-link{min-width:0}.room-identity strong{font-size:1rem}}
.game-layout{grid-template-columns:330px minmax(0,1fr) 280px;grid-template-areas:"players market supply";align-items:start}.board-sidebar{display:contents}.players-stack{grid-area:players;display:grid;gap:.55rem}.market{grid-area:market}.right-supply{grid-area:supply;position:sticky;top:1rem}.sidebar-player{padding:.75rem}.sidebar-player.active{border-color:var(--gold);box-shadow:0 0 0 1px var(--gold),0 12px 28px #0003}.sidebar-player-summary{list-style:none;cursor:pointer}.sidebar-player-summary::-webkit-details-marker{display:none}.sidebar-player-heading{display:flex;justify-content:space-between;align-items:center;gap:.5rem}.sidebar-player-heading>span{color:var(--gold);font-size:.82rem}.count-legend{display:flex;justify-content:space-between;color:var(--muted);font-size:.68rem;margin:.45rem 0 .25rem}.player-color-counts{display:grid;grid-template-columns:repeat(3,1fr);gap:.3rem}.player-color-count{--gem-color:#697780;display:flex;align-items:center;justify-content:space-between;gap:.25rem;padding:.3rem .4rem;border:1px solid color-mix(in srgb,var(--gem-color) 55%,var(--line));border-radius:8px;background:color-mix(in srgb,var(--gem-color) 10%,#0e1418);font-size:.75rem}.player-color-count b{color:color-mix(in srgb,var(--gem-color) 70%,white);text-shadow:0 1px 2px #000}.player-detail-hint{display:block;margin-top:.45rem}.sidebar-player-details{border-top:1px solid var(--line);margin-top:.65rem;padding-top:.65rem}.owned-card-grid{display:grid;gap:.4rem;max-height:18rem;overflow-y:auto}.owned-card-inspector{--gem-color:#697780;border:1px solid color-mix(in srgb,var(--gem-color) 55%,var(--line));border-radius:9px;overflow:hidden;background:color-mix(in srgb,var(--gem-color) 9%,#0e1418)}.owned-card-inspector>summary{display:grid;grid-template-columns:1.8rem 1fr auto;align-items:center;gap:.45rem;padding:.5rem;cursor:pointer}.owned-card-bonus{display:grid;place-items:center;width:1.8rem;height:1.8rem;border-radius:50%;background:var(--gem-color);color:var(--gem-ink,#fff);font-weight:900;text-shadow:0 1px 2px #000}.owned-card-detail{padding:.5rem;border-top:1px solid var(--line);font-size:.78rem}.owned-card-detail p{margin:.25rem 0}.owned-card-detail .resource-cost{display:inline-block;margin:.15rem}
.gem-picker{display:grid;gap:.55rem}.gem-picker .fine-print{margin:.1rem 0 .35rem}.gem-pile,.wild-pile{--gem-color:#697780;display:grid;grid-template-columns:auto 1fr auto;align-items:center;gap:.55rem;margin:0;padding:.55rem;border:1px solid color-mix(in srgb,var(--gem-color) 65%,var(--line));border-radius:12px;background:#0e1418;color:var(--text)}.gem-pile input{width:1.1rem;height:1.1rem;margin:0;accent-color:var(--gem-color)}.gem-choice{display:grid;grid-template-columns:2.6rem 1fr;align-items:center;gap:.55rem;margin:0;cursor:pointer}.gem-pile:has(input:checked){border-color:var(--gem-color);box-shadow:0 0 0 1px var(--gem-color),0 0 14px color-mix(in srgb,var(--gem-color) 35%,transparent);background:color-mix(in srgb,var(--gem-color) 18%,#0e1418)}.gem-pile.empty{opacity:.45}.take-pair{padding:.45rem .55rem;white-space:nowrap}.gem-disc{display:grid;place-items:center;width:2.5rem;height:2.5rem;border-radius:50%;border:3px solid color-mix(in srgb,var(--gem-color) 55%,white);background:radial-gradient(circle at 35% 30%,color-mix(in srgb,var(--gem-color) 45%,white),var(--gem-color) 65%);color:var(--gem-ink,#fff);text-shadow:0 1px 2px #000,0 0 3px #000;font-size:1.15rem;font-weight:900;box-shadow:0 4px 8px #0008,0 0 8px color-mix(in srgb,var(--gem-color) 35%,transparent)}.card-bonus-token{width:2rem;height:2rem;border-width:2px;font-size:.9rem;line-height:1;flex:none}.gem-pile strong,.wild-pile strong{color:color-mix(in srgb,var(--gem-color) 68%,white)}.gem-pile small,.wild-pile small{display:block}.gem-picker.readonly .gem-pile{grid-template-columns:2.6rem 1fr;cursor:default}.wild-pile{grid-template-columns:2.6rem 1fr;margin-top:.7rem}.resource-cost{border-color:color-mix(in srgb,var(--gem-color) 65%,var(--line))!important;box-shadow:inset 3px 0 0 var(--gem-color)}
.gem-disc.light-gem{border-color:#090c0e;background:radial-gradient(circle at 32% 25%,#fff 0 18%,#f8f7f2 48%,#d9dde0 100%);color:#090c0e;font-weight:800;text-shadow:none;-webkit-font-smoothing:antialiased;text-rendering:geometricPrecision;box-shadow:0 4px 9px #0009,0 0 0 1px #596168,inset 0 0 0 2px #fff9}.hand-gem.light-gem>b{border:2px solid #090c0e;background:radial-gradient(circle at 32% 25%,#fff 0 18%,#f8f7f2 48%,#d9dde0 100%);color:#090c0e;font-weight:800;text-shadow:none;-webkit-font-smoothing:antialiased;text-rendering:geometricPrecision;box-shadow:0 2px 5px #0008,0 0 0 1px #596168,inset 0 0 0 1px #fff9}
.card-bonus-token.light-gem{border-color:#090c0e;background:radial-gradient(circle at 32% 25%,#fff 0 16%,#f7f6f1 42%,#d9dde0 100%);color:#090c0e;text-shadow:none;box-shadow:0 3px 7px #000b,0 0 0 1px #596168,inset 0 0 0 2px #fff9}
.tier-label{align-items:center}.compact-button{padding:.35rem .55rem}.game-card{display:flex;flex-direction:column;padding:0;overflow:hidden}.card-face{min-height:170px;padding:.9rem;flex:1}.game-card.actionable{border-color:#46615d}.card-actions{display:flex;gap:.45rem;padding:.75rem;border-top:1px solid var(--line);background:#0e1418}.card-actions form,.card-actions>button{flex:1}.card-actions button{width:100%;padding:.5rem}.board-help{grid-template-columns:170px 1fr}.board-help p{margin:0;color:var(--muted)}
.reserved-cards{margin-top:1rem}.reserved-cards h3{margin-bottom:.5rem}.reserved-card{border:1px solid var(--line);border-radius:10px;margin-top:.4rem;overflow:hidden}.reserved-card-face{display:flex;justify-content:space-between;gap:.5rem;padding:.6rem}.reserved-card.actionable{border-color:var(--teal)}.reserved-card-face span{color:var(--muted);font-size:.8rem}
@media(max-width:900px){.game-layout{grid-template-columns:1fr;grid-template-areas:"players" "market" "supply"}.right-supply{position:static}.gem-picker{grid-template-columns:repeat(2,minmax(0,1fr))}.gem-picker .fine-print,.gem-picker>.take-checked{grid-column:1/-1}}
@media(max-width:540px){.gem-picker{grid-template-columns:1fr}.gem-picker .fine-print,.gem-picker>.take-checked{grid-column:auto}.board-help{display:flex}}
.player-gem-dock{position:fixed;left:50%;bottom:.55rem;transform:translateX(-50%);z-index:15;display:flex;align-items:center;gap:.65rem;max-width:calc(100vw - 1rem);padding:.4rem .65rem;border:1px solid var(--gold);border-radius:14px;background:#151c21ee;backdrop-filter:blur(16px);box-shadow:0 12px 35px #0009}.player-gem-dock>strong{white-space:nowrap;color:var(--gold)}.hand-total{display:grid;place-items:center;min-width:3.8rem;padding:.25rem .45rem;border:1px solid var(--gold);border-radius:10px;background:#2a2417}.hand-total strong{color:var(--gold);font-size:.9rem}.hand-total small{font-size:.6rem;text-transform:uppercase;letter-spacing:.08em}.hand-legend{white-space:nowrap;color:var(--muted);font-size:.72rem}.player-gem-hand{display:flex;gap:.4rem}.hand-gem{--gem-color:#697780;display:grid;grid-template-columns:1.8rem auto;align-items:center;gap:.35rem;padding:.25rem .5rem .25rem .25rem;border:1px solid color-mix(in srgb,var(--gem-color) 65%,var(--line));border-radius:999px;background:color-mix(in srgb,var(--gem-color) 13%,#0e1418)}.hand-gem>b{display:grid;place-items:center;width:1.8rem;height:1.8rem;border-radius:50%;background:var(--gem-color);color:var(--gem-ink,#fff);text-shadow:0 1px 2px #000}.hand-gem>span{display:flex;align-items:center;gap:.2rem}.hand-gem>span strong{font-size:.95rem}.hand-gem i{color:var(--muted);font-style:normal}.action-dock{bottom:4.7rem}
.bottom-reserved{display:flex;align-items:center;gap:.35rem;padding-left:.6rem;border-left:1px solid var(--line)}.bottom-reserved-card{display:flex;align-items:center;gap:.35rem;padding:.25rem;border:1px solid var(--line);border-radius:9px;background:#0e1418}.bottom-reserved-card>span{max-width:8rem}.bottom-reserved-card strong,.bottom-reserved-card small{display:block;overflow:hidden;text-overflow:ellipsis;white-space:nowrap}.bottom-reserved-card strong{font-size:.72rem}.bottom-reserved-card small{color:var(--muted);font-size:.62rem}.bottom-reserved-card button{padding:.35rem .45rem;font-size:.7rem}
.action-dock>div:first-child small{display:block;margin-top:.25rem}.discard-picker{display:flex;align-items:center;gap:.65rem;min-width:0}.discard-colors{display:flex;gap:.4rem;overflow-x:auto;padding:.15rem;min-width:0}.discard-color{display:grid;grid-template-columns:2rem minmax(4.5rem,auto) 3.5rem;align-items:center;gap:.4rem;flex:none;margin:0;padding:.35rem;border:1px solid color-mix(in srgb,var(--gem-color) 65%,var(--line));border-radius:10px;background:color-mix(in srgb,var(--gem-color) 10%,#0e1418)}.discard-color .gem-disc{width:2rem;height:2rem;font-size:.9rem}.discard-color>span:nth-child(2){font-size:.75rem;font-weight:700;color:color-mix(in srgb,var(--gem-color) 70%,white)}.discard-color small{display:block;font-size:.6rem}.discard-color input{width:3.5rem;margin:0;padding:.4rem}.discard-picker>button{flex:none;white-space:nowrap}
@media(max-width:700px){.player-gem-dock{width:calc(100vw - 1rem);overflow-x:auto;justify-content:flex-start}.player-gem-hand{flex:none}.player-gem-dock>strong{font-size:.8rem}.hand-gem{grid-template-columns:1.5rem auto}.hand-gem b{width:1.5rem;height:1.5rem}}
@media(min-width:901px){.board-main{max-width:1900px;padding:clamp(.55rem,1.2vh,1rem) 1rem 7rem;scroll-padding-bottom:7rem}.board-main .status-strip{padding:.5rem 0;margin-bottom:.55rem}.board-main .game-layout{grid-template-columns:300px minmax(0,1fr) 270px;gap:.55rem}.board-main .players-stack{gap:.35rem}.board-main .sidebar-player{padding:.5rem}.board-main .count-legend{margin:.25rem 0 .15rem}.board-main .player-color-counts{gap:.2rem}.board-main .player-color-count{padding:.2rem .3rem}.board-main .player-detail-hint{margin-top:.25rem}.board-main .market{gap:.45rem}.board-main .market-row{padding:.55rem}.board-main .tier-label{margin-bottom:.4rem}.board-main .card-row{gap:.45rem}.board-main .card-face{min-height:108px;padding:.6rem}.board-main .game-card h3{font-size:1rem;line-height:1.15;margin:.55rem 0 .25rem}.board-main .game-card p{margin:.25rem 0}.board-main .card-actions{padding:.4rem;gap:.3rem}.board-main .card-actions button{padding:.4rem .25rem}.board-main .right-supply{padding:.65rem;top:.5rem}.board-main .right-supply h2{font-size:1.1rem;margin-bottom:.35rem}.board-main .right-supply .section-heading{margin-bottom:.35rem}.board-main .right-supply .gem-picker{gap:.3rem}.board-main .right-supply .gem-pile,.board-main .right-supply .wild-pile{padding:.3rem;gap:.35rem}.board-main .right-supply .gem-choice{grid-template-columns:2rem 1fr;gap:.35rem}.board-main .right-supply .gem-disc{width:2rem;height:2rem;font-size:.9rem}.board-main .right-supply .take-pair{padding:.3rem}.board-main .right-supply .fine-print{font-size:.7rem}.history-inline{position:relative}.history-inline summary{cursor:pointer;color:var(--teal)}.history-inline>div{position:absolute;right:0;top:1.8rem;width:min(360px,70vw);max-height:50vh;overflow:auto;z-index:20;padding:.7rem;border:1px solid var(--line);border-radius:10px;background:var(--panel);box-shadow:0 16px 40px #0008}.history-inline p{margin:.3rem 0;font-size:.8rem}.patrons-section{padding-bottom:.5rem;margin-bottom:.55rem;border-bottom:1px solid var(--line)}.patron-list{display:grid;grid-template-columns:repeat(2,minmax(0,1fr));gap:.3rem}.patron-card{display:grid;grid-template-columns:auto 1fr;align-items:start;gap:.3rem;padding:.35rem;border:1px solid var(--line);border-radius:8px;background:#0e1418;font-size:.72rem}.patron-card>strong{color:var(--gold)}.patron-card b,.patron-card small{display:block}.requirement-chip{display:inline-block;margin:.15rem .1rem 0;padding:.1rem .25rem;border:1px solid var(--line);border-radius:999px}.claimed-patrons{display:grid;gap:.25rem;margin-top:.5rem}.claimed-patrons span{font-size:.78rem;color:var(--gold)}}
+13
View File
@@ -0,0 +1,13 @@
Zero-Clause BSD
=============
Permission to use, copy, modify, and/or distribute this software for
any purpose with or without fee is hereby granted.
THE SOFTWARE IS PROVIDED “AS IS” AND THE AUTHOR DISCLAIMS ALL
WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES
OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE
FOR ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN
AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+14
View File
@@ -0,0 +1,14 @@
BSD Zero Clause License
Copyright (c) 2023, Alexander Petros
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
File diff suppressed because one or more lines are too long
+1
View File
@@ -0,0 +1 @@
(function(){var g;htmx.defineExtension("sse",{init:function(e){g=e;if(htmx.createEventSource==undefined){htmx.createEventSource=t}},getSelectors:function(){return["[sse-connect]","[data-sse-connect]","[sse-swap]","[data-sse-swap]"]},onEvent:function(e,t){var r=t.target||t.detail.elt;switch(e){case"htmx:beforeCleanupElement":var n=g.getInternalData(r);var s=n.sseEventSource;if(s){g.triggerEvent(r,"htmx:sseClose",{source:s,type:"nodeReplaced"});n.sseEventSource.close()}return;case"htmx:afterProcessNode":i(r)}}});function t(e){return new EventSource(e,{withCredentials:true})}function a(n){if(g.getAttributeValue(n,"sse-swap")){var s=g.getClosestMatch(n,v);if(s==null){return null}var e=g.getInternalData(s);var a=e.sseEventSource;var t=g.getAttributeValue(n,"sse-swap");var r=t.split(",");for(var i=0;i<r.length;i++){const u=r[i].trim();const c=function(e){if(l(s)){return}if(!g.bodyContains(n)){a.removeEventListener(u,c);return}if(!g.triggerEvent(n,"htmx:sseBeforeMessage",e)){return}f(n,e.data);g.triggerEvent(n,"htmx:sseMessage",e)};g.getInternalData(n).sseEventListener=c;a.addEventListener(u,c)}}if(g.getAttributeValue(n,"hx-trigger")){var s=g.getClosestMatch(n,v);if(s==null){return null}var e=g.getInternalData(s);var a=e.sseEventSource;var o=g.getTriggerSpecs(n);o.forEach(function(t){if(t.trigger.slice(0,4)!=="sse:"){return}var r=function(e){if(l(s)){return}if(!g.bodyContains(n)){a.removeEventListener(t.trigger.slice(4),r)}htmx.trigger(n,t.trigger,e);htmx.trigger(n,"htmx:sseMessage",e)};g.getInternalData(n).sseEventListener=r;a.addEventListener(t.trigger.slice(4),r)})}}function i(e,t){if(e==null){return null}if(g.getAttributeValue(e,"sse-connect")){var r=g.getAttributeValue(e,"sse-connect");if(r==null){return}n(e,r,t)}a(e)}function n(r,e,n){var s=htmx.createEventSource(e);s.onerror=function(e){g.triggerErrorEvent(r,"htmx:sseError",{error:e,source:s});if(l(r)){return}if(s.readyState===EventSource.CLOSED){n=n||0;n=Math.max(Math.min(n*2,128),1);var t=n*500;window.setTimeout(function(){i(r,n)},t)}};s.onopen=function(e){g.triggerEvent(r,"htmx:sseOpen",{source:s});if(n&&n>0){const t=r.querySelectorAll("[sse-swap], [data-sse-swap], [hx-trigger], [data-hx-trigger]");for(let e=0;e<t.length;e++){a(t[e])}n=0}};g.getInternalData(r).sseEventSource=s;var t=g.getAttributeValue(r,"sse-close");if(t){s.addEventListener(t,function(){g.triggerEvent(r,"htmx:sseClose",{source:s,type:"message"});s.close()})}}function l(e){if(!g.bodyContains(e)){var t=g.getInternalData(e).sseEventSource;if(t!=undefined){g.triggerEvent(e,"htmx:sseClose",{source:t,type:"nodeMissing"});t.close();return true}}return false}function f(t,r){g.withExtensions(t,function(e){r=e.transformResponse(r,null,t)});var e=g.getSwapSpecification(t);var n=g.getTarget(t);g.swap(n,r,e)}function v(e){return g.getInternalData(e).sseEventSource!=null}})();
+18
View File
@@ -0,0 +1,18 @@
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<meta name="theme-color" content="#101418">
<meta name="htmx-config" content='{"responseHandling":[{"code":"204","swap":false},{"code":"[23]..","swap":true},{"code":"422","swap":true,"error":false},{"code":"[45]..","swap":false,"error":true}]}'>
<title>{% block title %}Gems{% endblock %}</title>
<link rel="stylesheet" href="/static/gems.css">
<link rel="stylesheet" href="/static/overrides.css">
<script src="/static/vendor/htmx.min.js" defer></script>
<script src="/static/vendor/sse.min.js" defer></script>
</head>
<body>
{% block topbar %}<header class="topbar"><a href="/" class="brand"><span class="brand-mark"></span> Gems</a></header>{% endblock %}
<main class="{% block main_class %}{% endblock %}">{% block content %}{% endblock %}</main>
</body>
</html>
+22
View File
@@ -0,0 +1,22 @@
{% extends "base.html" %}
{% block content %}
<section class="hero">
<p class="eyebrow">A private strategy table</p>
<h1>Build an engine.<br>Race for prestige.</h1>
<p class="lede">Bring your own content pack, invite up to three friends or bots, and shape the rules before the first turn.</p>
{% if error %}<p class="alert" role="alert">{{ error }}</p>{% endif %}
<div class="home-grid">
<form class="panel" method="post" action="/rooms">
<h2>Create a table</h2>
<label>Display name<input name="name" maxlength="32" value="{{ display_name }}" required autocomplete="nickname"></label>
<button class="primary">Create room</button>
</form>
<form class="panel" method="post" action="/join">
<h2>Join a table</h2>
<label>Invite code<input name="code" minlength="8" maxlength="8" required autocapitalize="characters"></label>
<button>Continue</button>
</form>
</div>
<p class="fine-print">No cards, artwork, or playable content are bundled with Gems.</p>
</section>
{% endblock %}
+14
View File
@@ -0,0 +1,14 @@
{% extends "base.html" %}
{% block content %}
<section class="narrow">
<p class="eyebrow">Invitation {{ code }}</p>
<div class="panel">
<h1>Take a seat</h1>
{% if error %}<p class="alert" role="alert">{{ error }}</p>{% endif %}
<form method="post" action="/join/{{ code }}">
<label>Display name<input name="name" maxlength="32" value="{{ display_name }}" required autofocus autocomplete="nickname"></label>
<button class="primary">Join room</button>
</form>
</div>
</section>
{% endblock %}
@@ -0,0 +1,211 @@
{% if error %}<p class="alert floating" role="alert">{{ error }}</p>{% endif %}
{% if room.status == "lobby" %}
<div class="lobby-grid">
<section class="panel">
<div class="section-heading"><h2>Seats</h2><span>{{ members|length }}/4</span></div>
<ol class="seat-list">
{% for seat in members %}
<li>
<span class="seat-number">{{ seat.seat + 1 }}</span>
<span><strong>{{ seat.name }}</strong><small>{{ seat.difficulty|capitalize if seat.controller == 'ai' else ('Ready' if seat.ready else 'Not ready') }}{% if seat.is_host %} · Host{% endif %}</small></span>
{% if member.is_host and not seat.is_host %}
<form hx-post="/rooms/{{ room.code }}/seats/{{ seat.seat }}/make-host" hx-target="#room-state" hx-swap="innerHTML">
<input type="hidden" name="csrf" value="{{ csrf }}"><button class="quiet">Make host</button>
</form>
<form hx-post="/rooms/{{ room.code }}/seats/{{ seat.seat }}/remove" hx-target="#room-state" hx-swap="innerHTML">
<input type="hidden" name="csrf" value="{{ csrf }}"><button class="icon-button" aria-label="Remove {{ seat.name }}">×</button>
</form>
{% endif %}
</li>
{% endfor %}
</ol>
{% if member.is_host and members|length < 4 %}
<form class="inline-form" hx-post="/rooms/{{ room.code }}/seats/ai" hx-target="#room-state" hx-swap="innerHTML">
<input type="hidden" name="csrf" value="{{ csrf }}">
<select name="difficulty" aria-label="AI difficulty"><option value="easy">Easy AI</option><option value="medium" selected>Medium AI</option><option value="hard">Hard AI</option></select>
<button>Add bot</button>
</form>
{% endif %}
{% if member.controller == 'human' %}
<form hx-post="/rooms/{{ room.code }}/ready" hx-target="#room-state" hx-swap="innerHTML">
<input type="hidden" name="csrf" value="{{ csrf }}"><input type="hidden" name="ready" value="{{ 'false' if member.ready else 'true' }}">
<button class="{{ 'quiet' if member.ready else 'primary' }}">{{ 'Not ready' if member.ready else 'Ready up' }}</button>
</form>
{% endif %}
</section>
<section class="panel">
<div class="section-heading"><h2>Content pack</h2>{% if room.pack %}<span class="success">Loaded and validated</span>{% else %}<span class="danger">Not loaded</span>{% endif %}</div>
{% if room.pack %}<p><strong>{{ room.pack.metadata.name }}</strong> <span class="muted">v{{ room.pack.metadata.version }}</span></p>{% else %}<p class="pack-status" role="status">No content pack is loaded. The game cannot start until the host uploads a valid JSON pack.</p>{% endif %}
{% if member.is_host %}
<form hx-post="/rooms/{{ room.code }}/pack" hx-target="#room-state" hx-swap="innerHTML" hx-encoding="multipart/form-data">
<input type="hidden" name="csrf" value="{{ csrf }}"><label>JSON file<input type="file" name="pack_file" accept="application/json,.json" required></label>
<button>Validate and use pack</button>
</form>
{% endif %}
<p class="fine-print"><a href="/schemas/content-pack-v1.json">JSON Schema</a>{% if room.pack %} · <a href="/rooms/{{ room.code }}/pack">Download active pack</a>{% endif %}</p>
</section>
<section class="panel rules-panel">
<div class="section-heading"><h2>Table rules</h2><span>Frozen on start</span></div>
{% if member.is_host %}<div class="preset-row">
{% for key, label in [('classic','Classic'),('objectives','Objective race'),('objectives_outposts','Objective + outpost'),('eastern_fortifications','Eastern + fortification'),('all','All modules')] %}
<form hx-post="/rooms/{{ room.code }}/settings/preset" hx-target="#room-state" hx-swap="innerHTML"><input type="hidden" name="csrf" value="{{ csrf }}"><input type="hidden" name="preset" value="{{ key }}"><button class="quiet">{{ label }}</button></form>
{% endfor %}
</div>{% endif %}
<form hx-post="/rooms/{{ room.code }}/settings" hx-target="#room-state" hx-swap="innerHTML">
<input type="hidden" name="csrf" value="{{ csrf }}">
<fieldset {% if not member.is_host %}disabled{% endif %}>
<div class="settings-grid">
<label>Target score<input type="number" name="target_score" min="1" max="999" value="{{ room.settings.target_score }}"></label>
<label>Victory<select name="victory_condition"><option value="score" {% if room.settings.victory_condition == 'score' %}selected{% endif %}>Score</option><option value="objective" {% if room.settings.victory_condition == 'objective' %}selected{% endif %}>Objective</option><option value="either" {% if room.settings.victory_condition == 'either' %}selected{% endif %}>Either</option><option value="both" {% if room.settings.victory_condition == 'both' %}selected{% endif %}>Both</option></select></label>
<label>First player<select name="first_player_mode"><option value="random" {% if room.settings.first_player_mode == 'random' %}selected{% endif %}>Random</option><option value="selected" {% if room.settings.first_player_mode == 'selected' %}selected{% endif %}>Choose player</option></select></label>
<label>Chosen player<select name="first_player_seat" aria-label="Chosen first player">{% for seat in members %}<option value="{{ loop.index0 }}" {% if room.settings.first_player_seat == loop.index0 %}selected{% endif %}>{{ seat.name }}</option>{% endfor %}</select><small>Used when “Choose player” is selected.</small></label>
<label>Token limit<input type="number" name="token_limit" min="1" max="99" value="{{ room.settings.token_limit }}"></label>
<label>Reserve limit<input type="number" name="reserve_limit" min="0" max="20" value="{{ room.settings.reserve_limit }}"></label>
<label>Base market/tier<input type="number" name="base_market_size" min="1" max="10" value="{{ room.settings.base_market_size }}"></label>
<label>Eastern market/tier<input type="number" name="eastern_market_size" min="1" max="10" value="{{ room.settings.eastern_market_size }}"></label>
<label>Objectives shown<input type="number" name="objective_count" min="1" max="10" value="{{ room.settings.objective_count }}"></label>
<label>Fortifications/player<input type="number" name="fortifications_per_player" min="1" max="10" value="{{ room.settings.fortifications_per_player }}"></label>
</div>
<h3>Modules</h3>
<div class="toggle-grid">
{% for key, label in [('objectives','Objectives'),('outposts','Outposts'),('eastern_decks','Eastern decks'),('fortifications','Fortifications')] %}
<label class="toggle"><input type="checkbox" name="{{ key }}" {% if room.settings.modules[key] %}checked{% endif %}><span>{{ label }}</span></label>
{% endfor %}
</div>
<details><summary>Advanced module interactions</summary><div class="toggle-grid advanced">
{% for key, value in room.settings.interactions %}<label class="toggle"><input type="checkbox" name="{{ key }}" {% if value %}checked{% endif %}><span>{{ key|replace('_',' ')|capitalize }}</span></label>{% endfor %}
</div></details>
</fieldset>
{% if member.is_host %}<button>Save rules</button>{% endif %}
</form>
</section>
</div>
{% if member.is_host %}
<form class="start-bar" hx-post="/rooms/{{ room.code }}/start" hx-target="#room-state" hx-swap="innerHTML"><input type="hidden" name="csrf" value="{{ csrf }}"><span>{{ 'Pack loaded' if room.pack else 'Pack required' }} · {{ 'Players ready' if all_humans_ready else 'Waiting for every human to ready up' }} · Rules set</span><button class="primary"{% if not can_start %} disabled{% endif %}>Start game</button></form>
{% endif %}
{% else %}
<section class="status-strip">
<span>Round {{ state.round_number }}</span><strong>{% if room.status == 'finished' %}Game complete · {% for seat in state.winners %}{{ state.players[seat].name }}{% if not loop.last %}, {% endif %}{% endfor %}{% else %}{{ state.players[state.current_seat].name }}s turn{% endif %}</strong><span>Target {{ room.settings.target_score }}</span><details class="history-inline"><summary>History</summary><div>{% for entry in state.log[-12:]|reverse %}<p>{{ entry }}</p>{% endfor %}</div></details>{% if room.status == 'finished' %}{% if member.is_host %}<form hx-post="/rooms/{{ room.code }}/play-again" hx-target="#room-state" hx-swap="innerHTML"><input type="hidden" name="csrf" value="{{ csrf }}"><button class="primary">Play again</button></form>{% else %}<span class="muted">Waiting for the host</span>{% endif %}{% endif %}
</section>
<div class="game-layout">
<aside class="board-sidebar">
<section class="players-stack" aria-label="Players">
{% for player in state.players %}
<details class="sidebar-player panel compact{% if player.seat == state.current_seat %} active{% endif %}">
<summary class="sidebar-player-summary">
<div class="sidebar-player-heading"><strong>{{ player.name }}</strong><span>{% if member_by_seat[player.seat].controller == 'ai' %}Bot · {% endif %}{{ scores[player.seat] }}◆</span></div>
<div class="count-legend"><span>Color</span><span>Gems / Cards</span></div>
<div class="player-color-counts">
{% for resource_id in normal_resource_ids %}{% set resource = resources[resource_id] %}
<span class="player-color-count" style="--gem-color:{{ resource.color }}" title="{{ resource.label }}: {{ player.tokens[resource_id] }} gems, {{ bonus_counts[player.seat].get(resource_id, 0) }} cards"><b>{{ resource.symbol }}</b><span>{{ player.tokens[resource_id] }} / {{ bonus_counts[player.seat].get(resource_id, 0) }}</span></span>
{% endfor %}
{% set wild = pack.wild_resource %}<span class="player-color-count" style="--gem-color:{{ wild.color }}" title="{{ wild.label }}: {{ player.tokens[wild.id] }} gems"><b>{{ wild.symbol }}</b><span>{{ player.tokens[wild.id] }} / 0</span></span>
</div>
<small class="player-detail-hint">Click for card details · {{ player.cards|length }} owned · {{ player.reserved|length }} reserved</small>
</summary>
<div class="sidebar-player-details">
{% if player.cards %}<div class="owned-card-grid">
{% for owned in player.cards %}{% set card = cards[owned.card_id] %}{% set bonus_id = owned.copied_resource or card.bonus_resource %}
<details class="owned-card-inspector" {% if bonus_id %}style="--gem-color:{{ resources[bonus_id].color }};--gem-ink:{{ resources[bonus_id].ink_color }}"{% endif %}>
<summary><span class="owned-card-bonus">{{ resources[bonus_id].symbol if bonus_id else '◇' }}</span><strong>{{ card.label }}</strong><span>{{ card.points }}◆</span></summary>
<div class="owned-card-detail"><p>Cost {% for resource_id, value in card.cost.items() %}<span class="resource-cost" style="--gem-color:{{ resources[resource_id].color }}">{{ resources[resource_id].symbol }} {{ value }}</span>{% else %}<span>Free</span>{% endfor %}</p><p>Effect: {{ card.effect.kind|replace('_',' ') }}{% if owned.copied_resource %} · copied {{ resources[owned.copied_resource].label }}{% endif %}</p></div>
</details>
{% endfor %}
</div>{% else %}<p class="muted">No owned cards yet.</p>{% endif %}
{% if player.patrons %}<div class="claimed-patrons"><strong>Claimed patrons</strong>{% for patron_id in player.patrons %}<span>{{ patrons[patron_id].label }} · {{ patrons[patron_id].points }}◆</span>{% endfor %}</div>{% endif %}
{% if player.seat == member.seat and player.reserved %}
<div class="reserved-cards"><h3>Your reserved cards</h3>
{% for card_id in player.reserved %}{% set card = cards[card_id] %}{% set purchase = card_actions.get(card_id, {}).get('purchase') %}
<div class="reserved-card{% if purchase %} actionable{% endif %}"><div class="reserved-card-face"><strong>{{ card.label }}</strong><span>{{ card.points }}◆ · {% for resource_id, value in card.cost.items() %}{{ resources[resource_id].symbol }} {{ value }} {% endfor %}</span></div><div class="card-actions">{% if purchase %}<form hx-post="/rooms/{{ room.code }}/commands" hx-target="#room-state" hx-swap="innerHTML"><input type="hidden" name="csrf" value="{{ csrf }}"><input type="hidden" name="command_json" value="{{ purchase|e }}"><button class="primary">Purchase</button></form>{% else %}<button class="primary" disabled>Purchase</button>{% endif %}<button disabled>Reserved</button></div></div>
{% endfor %}
</div>
{% endif %}
{% if member.is_host and player.seat in offline_seats and player.seat != member.seat and room.status == 'playing' %}<form hx-post="/rooms/{{ room.code }}/seats/{{ player.seat }}/replace-with-ai" hx-target="#room-state" hx-swap="innerHTML"><input type="hidden" name="csrf" value="{{ csrf }}"><input type="hidden" name="difficulty" value="medium"><button>Replace with AI</button></form>{% endif %}
</div>
</details>
{% endfor %}
</section>
<section class="supply right-supply panel compact">
<div class="patrons-section">
<div class="section-heading"><h2>Patrons</h2><span>{{ state.available_patrons|length }} available</span></div>
<div class="patron-list">
{% for patron_id in state.available_patrons %}{% set patron = patrons[patron_id] %}
<article class="patron-card"><strong>{{ patron.points }}◆</strong><span><b>{{ patron.label }}</b><small>{% for requirement in patron.requirements %}{% if requirement.resource %}<span class="requirement-chip resource-cost" style="--gem-color:{{ resources[requirement.resource].color }}">{{ resources[requirement.resource].symbol }} {{ requirement.count }}</span>{% else %}<span class="requirement-chip">Any {{ requirement.count }}</span>{% endif %}{% endfor %}</small></span></article>
{% else %}<p class="muted">No unclaimed patrons.</p>
{% endfor %}
</div>
</div>
<div class="section-heading"><h2>Gem piles</h2><span>Supply</span></div>
{% if member.seat == state.current_seat and not state.pending and room.status == 'playing' %}
<form class="gem-picker" hx-post="/rooms/{{ room.code }}/take-gems" hx-target="#room-state" hx-swap="innerHTML">
<input type="hidden" name="csrf" value="{{ csrf }}">
<p class="fine-print">Check 3 different gems, or use Take 2 beside one pile.</p>
{% for resource_id in normal_resource_ids %}{% set resource = resources[resource_id] %}{% set amount = state.supply[resource_id] %}
<div class="gem-pile{% if not amount %} empty{% endif %}" style="--gem-color:{{ resource.color }};--gem-ink:{{ resource.ink_color }}">
<input id="gem-{{ resource_id }}" type="checkbox" name="resources" value="{{ resource_id }}" {% if not amount %}disabled{% endif %}>
<label class="gem-choice" for="gem-{{ resource_id }}"><span class="gem-disc{% if resource.ink_color == '#111111' %} light-gem{% endif %}">{{ resource.symbol }}</span><span><strong>{{ resource.label }}</strong><small>{{ amount }} available</small></span></label>
<button class="take-pair" type="submit" name="pair_resource" value="{{ resource_id }}" {% if resource_id not in double_resource_ids %}disabled{% endif %}>Take 2</button>
</div>
{% endfor %}
<button class="primary take-checked">Take checked gems</button>
</form>
{% else %}
<div class="gem-picker readonly">
{% for resource_id in normal_resource_ids %}{% set resource = resources[resource_id] %}
<div class="gem-pile" style="--gem-color:{{ resource.color }};--gem-ink:{{ resource.ink_color }}"><span class="gem-disc{% if resource.ink_color == '#111111' %} light-gem{% endif %}">{{ resource.symbol }}</span><span><strong>{{ resource.label }}</strong><small>{{ state.supply[resource_id] }} available</small></span></div>
{% endfor %}
</div>
{% endif %}
{% set wild = pack.wild_resource %}
<div class="wild-pile" style="--gem-color:{{ wild.color }};--gem-ink:{{ wild.ink_color }}"><span class="gem-disc{% if wild.ink_color == '#111111' %} light-gem{% endif %}">{{ wild.symbol }}</span><span><strong>{{ wild.label }}</strong><small>{{ state.supply[wild.id] }} available · gained by reserving</small></span></div>
</section>
</aside>
<section class="market">
{% for key, market in state.markets.items() %}
<div class="market-row">
<div class="tier-label"><strong>{{ key|replace(':',' · Tier ') }}</strong><span>{{ state.decks[key] }} left</span>{% if deck_reserve_actions.get(key) %}<form hx-post="/rooms/{{ room.code }}/commands" hx-target="#room-state" hx-swap="innerHTML"><input type="hidden" name="csrf" value="{{ csrf }}"><input type="hidden" name="command_json" value="{{ deck_reserve_actions[key]|e }}"><button class="quiet compact-button">Reserve top card</button></form>{% else %}<button class="quiet compact-button" disabled>Reserve top card</button>{% endif %}</div>
<div class="card-row">
{% for card_id in market %}{% set card = cards[card_id] %}{% set available = card_actions.get(card_id, {}) %}
<article class="game-card{% if available %} actionable{% endif %}">
<div class="card-face"><div class="card-top"><span>{{ card.points }}◆</span>{% if card.bonus_resource %}<span class="gem-disc card-bonus-token{% if resources[card.bonus_resource].ink_color == '#111111' %} light-gem{% endif %}" style="--gem-color:{{ resources[card.bonus_resource].color }};--gem-ink:{{ resources[card.bonus_resource].ink_color }}" role="img" aria-label="{{ resources[card.bonus_resource].label }} bonus" title="{{ resources[card.bonus_resource].label }} bonus">{{ resources[card.bonus_resource].symbol }}</span>{% else %}<span></span>{% endif %}</div><h3>{{ card.label }}</h3><p>Tier {{ card.tier }} · {{ card.effect.kind|replace('_',' ') }}</p><div class="cost">{% for resource_id, value in card.cost.items() %}<span class="resource-cost" style="--gem-color:{{ resources[resource_id].color }}">{{ resources[resource_id].symbol }} {{ value }}</span>{% endfor %}</div>{% if state.fortifications.get(card_id) %}<small>Fortified</small>{% endif %}</div>
<div class="card-actions">
{% if available.get('purchase') %}<form hx-post="/rooms/{{ room.code }}/commands" hx-target="#room-state" hx-swap="innerHTML"><input type="hidden" name="csrf" value="{{ csrf }}"><input type="hidden" name="command_json" value="{{ available['purchase']|e }}"><button class="primary">Purchase</button></form>{% else %}<button class="primary" disabled>Purchase</button>{% endif %}
{% if available.get('reserve') %}<form hx-post="/rooms/{{ room.code }}/commands" hx-target="#room-state" hx-swap="innerHTML"><input type="hidden" name="csrf" value="{{ csrf }}"><input type="hidden" name="command_json" value="{{ available['reserve']|e }}"><button>Reserve</button></form>{% else %}<button disabled>Reserve</button>{% endif %}
</div>
</article>
{% endfor %}
</div>
</div>
{% endfor %}
</section>
</div>
{% set viewing_player = state.players[member.seat] %}
<section class="player-gem-dock" aria-label="{{ viewing_player.name }} gems and cards">
<strong>{{ viewing_player.name }}</strong><span class="hand-total"><strong>{{ viewing_player.tokens.values()|sum }} / {{ room.settings.token_limit }}</strong><small>Gems</small></span><span class="hand-legend">Gems / Cards</span>
<div class="player-gem-hand">
{% for resource_id in normal_resource_ids %}{% set resource = resources[resource_id] %}<span class="hand-gem{% if resource.ink_color == '#111111' %} light-gem{% endif %}" style="--gem-color:{{ resource.color }};--gem-ink:{{ resource.ink_color }}" title="{{ resource.label }}: {{ viewing_player.tokens[resource_id] }} gems, {{ bonus_counts[viewing_player.seat].get(resource_id, 0) }} cards"><b>{{ resource.symbol }}</b><span><strong>{{ viewing_player.tokens[resource_id] }}</strong><i>/</i><strong>{{ bonus_counts[viewing_player.seat].get(resource_id, 0) }}</strong></span></span>{% endfor %}
{% set wild = pack.wild_resource %}<span class="hand-gem wild{% if wild.ink_color == '#111111' %} light-gem{% endif %}" style="--gem-color:{{ wild.color }};--gem-ink:{{ wild.ink_color }}" title="{{ wild.label }}: {{ viewing_player.tokens[wild.id] }} gems, 0 cards"><b>{{ wild.symbol }}</b><span><strong>{{ viewing_player.tokens[wild.id] }}</strong><i>/</i><strong>0</strong></span></span>
</div>
{% if viewing_player.reserved %}<div class="bottom-reserved"><span class="hand-legend">Reserved</span>{% for card_id in viewing_player.reserved %}{% set card = cards[card_id] %}{% set purchase = card_actions.get(card_id, {}).get('purchase') %}<div class="bottom-reserved-card"><span><strong>{{ card.label }}</strong><small>{{ card.points }}◆ · {% for resource_id, value in card.cost.items() %}{{ resources[resource_id].symbol }}{{ value }} {% endfor %}</small></span>{% if purchase %}<form hx-post="/rooms/{{ room.code }}/commands" hx-target="#room-state" hx-swap="innerHTML"><input type="hidden" name="csrf" value="{{ csrf }}"><input type="hidden" name="command_json" value="{{ purchase|e }}"><button class="primary">Purchase</button></form>{% else %}<button class="primary" disabled>Purchase</button>{% endif %}</div>{% endfor %}</div>{% endif %}
</section>
{% if room.status != 'finished' and member.seat == state.current_seat and state.pending %}
<section class="action-dock"><div><p class="eyebrow">Your move</p><strong>Resolve: {{ state.pending.kind|replace('_',' ') }}</strong>{% if state.pending.kind == 'discard_tokens' %}<small>Return exactly {{ state.pending.amount }} excess gem{{ '' if state.pending.amount == 1 else 's' }}.</small>{% endif %}</div>
{% if state.pending.kind == 'discard_tokens' %}
<form class="discard-picker" hx-post="/rooms/{{ room.code }}/discard-tokens" hx-target="#room-state" hx-swap="innerHTML">
<input type="hidden" name="csrf" value="{{ csrf }}">
<div class="discard-colors">{% for resource_id in all_resource_ids %}{% set resource = resources[resource_id] %}<label class="discard-color" style="--gem-color:{{ resource.color }};--gem-ink:{{ resource.ink_color }}"><span class="gem-disc{% if resource.ink_color == '#111111' %} light-gem{% endif %}">{{ resource.symbol }}</span><span>{{ resource.label }}<small>{{ viewing_player.tokens[resource_id] }} held</small></span><input type="number" name="token_{{ resource_id }}" min="0" max="{{ viewing_player.tokens[resource_id] }}" value="0" aria-label="{{ resource.label }} gems to return"></label>{% endfor %}</div>
<button class="primary">Return {{ state.pending.amount }} gem{{ '' if state.pending.amount == 1 else 's' }}</button>
</form>
{% else %}
<div class="actions">{% for label, command_json in actions %}<form hx-post="/rooms/{{ room.code }}/commands" hx-target="#room-state" hx-swap="innerHTML"><input type="hidden" name="csrf" value="{{ csrf }}"><input type="hidden" name="command_json" value="{{ command_json|e }}"><button>{{ label }}</button></form>{% else %}<span class="muted">No legal action available.</span>{% endfor %}</div>
{% endif %}
</section>
{% endif %}
{% endif %}
+17
View File
@@ -0,0 +1,17 @@
{% extends "base.html" %}
{% block title %}Room {{ room.code }} · Gems{% endblock %}
{% block main_class %}board-main{% endblock %}
{% block topbar %}
<header class="topbar room-topbar">
<a href="/" class="brand"><span class="brand-mark"></span> Gems</a>
<div class="room-identity"><span class="eyebrow">Room</span><strong>{{ room.code }}</strong></div>
<label class="invite-link room-invite">Invite link<input readonly value="{{ public_origin }}/join/{{ room.code }}" aria-label="Invite link"></label>
</header>
{% endblock %}
{% block content %}
<div class="room-shell">
<div id="room-state" hx-ext="sse" sse-connect="/rooms/{{ room.code }}/events" sse-swap="room">
{% include "partials/room_state.html" %}
</div>
</div>
{% endblock %}
+11
View File
@@ -0,0 +1,11 @@
"""Template and static resource configuration."""
from pathlib import Path
from fastapi.templating import Jinja2Templates
PACKAGE_DIR = Path(__file__).resolve().parent
TEMPLATE_DIR = PACKAGE_DIR / "templates"
STATIC_DIR = PACKAGE_DIR / "static"
templates = Jinja2Templates(directory=TEMPLATE_DIR)