"""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