from collections.abc import Callable
from dataclasses import dataclass
from yasuki_core.engine import ops
from yasuki_core.engine.players import PlayerId
from yasuki_core.engine.rules.events import (
CardDiscarded,
CounterGained,
Destroyed,
EnteredPlay,
GameEvent,
TurnStarted,
)
from yasuki_core.engine.rules.decisions import ChooseCards
from yasuki_core.engine.rules.modifiers import Duration, Modifier, Stat
from yasuki_core.engine.rules.state import GameState
from yasuki_core.engine.rules.work import ResumeCascade
from yasuki_core.engine.table import DeckKey, ZoneKey, ZoneRole
from yasuki_core.game_pieces.cards import L5RCard
from yasuki_core.game_pieces.constants import Side
from yasuki_core.game_pieces.counters import Counter, SINCERITY, WEALTH
from yasuki_core.game_pieces.dynasty import DynastyHolding
# A sanity bound on the fixpoint walk: a converging cascade drains in a handful of events, so far
# more than this means a trigger re-emits an event that re-fires it — a card-logic bug, raised loudly.
_MAX_CASCADE = 1000
# The keyword whose cards accrue and receive seeded Sincerity tokens.
SINCERITY_KEYWORD = "Sincerity"
[docs]
@dataclass(frozen=True, slots=True)
class AdjustCounter:
"""Effect: add ``delta`` to a counter on a card (floored at zero by the card). A grant is a
positive delta, a removal negative. The rules-side twin of the sandbox ``AdjustCounter`` intent,
applied through :func:`apply_effect` rather than ``apply_intent``."""
card_id: str
counter: Counter
delta: int
[docs]
@dataclass(frozen=True, slots=True)
class DrawCard:
"""Effect: ``seat`` draws a card from its fate deck."""
seat: PlayerId
[docs]
@dataclass(frozen=True, slots=True)
class Destroy:
"""Effect: destroy a card, sending it to its owner's discard by side."""
card_id: str
[docs]
@dataclass(frozen=True, slots=True)
class GrantModifier:
"""Effect: record a continuous stat modifier — the ``source`` card grants ``target`` a change of
``amount`` to ``stat`` for ``duration``. The single created-effect entry point; a card's
counters and attachments grant their bonuses without one (they are derived on read)."""
source_id: str
target_id: str
stat: Stat
amount: int
duration: Duration
[docs]
@dataclass(frozen=True, slots=True)
class Bow:
"""Effect: bow a card."""
card_id: str
[docs]
@dataclass(frozen=True, slots=True)
class Straighten:
"""Effect: straighten (unbow) a card."""
card_id: str
[docs]
@dataclass(frozen=True, slots=True)
class BanishTopFate:
"""Effect: banish the top card of ``seat``'s Fate deck; a no-op if the deck is empty."""
seat: PlayerId
[docs]
@dataclass(frozen=True, slots=True)
class GainGold:
"""Effect: add ``amount`` gold to ``seat``'s pool — gold produced outside a payment (a card that
produces gold on entry), transient and cleared at the end of the phase."""
seat: PlayerId
amount: int
[docs]
@dataclass(frozen=True, slots=True)
class IgnoreHonorRequirements:
"""Effect: grant ``seat`` the standing waiver of every Personality's Honor Requirement when
recruiting."""
seat: PlayerId
[docs]
@dataclass(frozen=True, slots=True)
class Choose:
"""Effect: pause the cascade so ``seat`` picks between ``minimum`` and ``maximum`` of
``candidates``; the chosen ids feed the registered ``resolver``, whose effects apply on resume.
The one interruption point in the effect vocabulary — every other effect commits at once, so a
trigger returns a Choose as its sole effect.
Attributes
----------
seat : PlayerId
The seat that chooses.
candidates : tuple of str
The card ids the seat may pick among.
minimum : int
The fewest cards the seat may pick — zero when the choice is optional.
maximum : int
The most cards the seat may pick.
resolver : str
The registered choice resolver naming what the chosen ids do.
source_id : str
The card whose trigger raised the choice, passed to the resolver.
"""
seat: PlayerId
candidates: tuple[str, ...]
minimum: int
maximum: int
resolver: str
source_id: str
Effect = (
AdjustCounter
| DrawCard
| Destroy
| GrantModifier
| Bow
| Straighten
| BanishTopFate
| GainGold
| IgnoreHonorRequirements
| Choose
)
[docs]
@dataclass(frozen=True, slots=True)
class TriggerContext:
"""What a trigger reads: the live game, the card whose trigger is firing, and the event."""
game: GameState
card: L5RCard
event: GameEvent
Trigger = Callable[[TriggerContext], list[Effect]]
# event type -> printed_id -> triggers. Populated by the @on decorators below, on import; kept
# grouped by printed_id so collection is a lookup, not a rebuild per event.
_TRIGGERS: dict[type, dict[str, list[Trigger]]] = {}
[docs]
def on(event_type: type, printed_id: str) -> Callable[[Trigger], Trigger]:
"""Register the decorated function as ``printed_id``'s trigger for ``event_type``."""
def register(trigger: Trigger) -> Trigger:
_TRIGGERS.setdefault(event_type, {}).setdefault(printed_id, []).append(trigger)
return trigger
return register
# A choice resolver turns the ids a Choose collected into the effects the choice produces. Keyed by a
# string so a paused ChooseCards names its resolver, keeping the pending decision replay-stable (a
# stored closure would not rebuild to an equal object).
Resolver = Callable[[GameState, str, tuple[str, ...]], list[Effect]]
CHOICE_RESOLVERS: dict[str, Resolver] = {}
[docs]
def choice_resolver(key: str) -> Callable[[Resolver], Resolver]:
"""Register the decorated function as the choice resolver named ``key``."""
def register(resolver: Resolver) -> Resolver:
CHOICE_RESOLVERS[key] = resolver
return resolver
return register
[docs]
def at_cap(card: L5RCard, counter: Counter, cap: int) -> bool:
"""Whether ``card`` already holds ``cap`` or more of ``counter`` — a shared trigger guard."""
return card.counters.get(counter.key, 0) >= cap
[docs]
def caused_by(ctx: TriggerContext, seat: PlayerId) -> bool:
"""Whether ``seat``'s own action caused the event — the "if the action was yours" guard. Reads
the event's ``by_seat``; only meaningful for events that carry one."""
return ctx.event.by_seat is seat
[docs]
def once_per_turn(game: GameState, card: L5RCard, tag: str) -> bool:
"""Claim a once-per-turn use for ``card``'s ``tag``: True the first time this turn, then False.
Turn-scoped, so it resets each turn without clearing ``GameState.once_per``."""
return game.use_once(f"{card.id}:{tag}:t{game.turn}")
[docs]
def apply_effect(game: GameState, effect: Effect) -> list[GameEvent]:
"""Commit one effect and return the events it raises, for the fixpoint walk to drain. This is the
single mutation boundary; triggers themselves never mutate."""
match effect:
case AdjustCounter(card_id=card_id, counter=counter, delta=delta):
card = game.table.cards_by_id.get(card_id)
if card is None:
return []
before = card.counters.get(counter.key, 0)
card.adjust_counter(counter.key, delta)
gained = card.counters.get(counter.key, 0) - before
if gained > 0:
return [CounterGained(card_id, counter, gained)]
case DrawCard(seat=seat):
ops.draw_to_hand(game.table, seat)
case Destroy(card_id=card_id):
card = game.table.cards_by_id.get(card_id)
if card is None or card.owner is None:
return []
role = ZoneRole.DYNASTY_DISCARD if card.side is Side.DYNASTY else ZoneRole.FATE_DISCARD
ops.move_card(game.table, card, ZoneKey(card.owner, role))
return [Destroyed(card_id)]
case GrantModifier(
source_id=source_id, target_id=target_id, stat=stat, amount=amount, duration=duration
):
game.modifiers.append(Modifier(source_id, target_id, stat, amount, duration))
case Bow(card_id=card_id):
card = game.table.cards_by_id.get(card_id)
if card is not None:
card.bow()
case Straighten(card_id=card_id):
card = game.table.cards_by_id.get(card_id)
if card is not None:
card.unbow()
case BanishTopFate(seat=seat):
deck = game.table.decks[DeckKey(seat, Side.FATE)]
if deck.cards:
ops.move_card(game.table, deck.cards[-1], ZoneKey(seat, ZoneRole.FATE_BANISH))
case GainGold(seat=seat, amount=amount):
game.add_gold(seat, amount)
case IgnoreHonorRequirements(seat=seat):
ops.set_ignore_honor_requirements(game.table, seat, True)
case Choose():
raise RuntimeError("a Choose pauses the trigger cascade; it is never applied directly")
return []
def _collect(game: GameState, event: GameEvent) -> list[tuple[L5RCard, Trigger]]:
by_id = _TRIGGERS.get(type(event))
if not by_id:
return []
return [
(card, trigger)
for card in game.table.battlefield.cards
for trigger in by_id.get(card.printed_id, ())
]
def _canonical_order(pair: tuple[L5RCard, Trigger]) -> tuple[str, str]:
card = pair[0]
return (card.owner.name if card.owner else "", card.id)
def _choice_request(choice: Choose) -> ChooseCards:
return ChooseCards(
seat=choice.seat,
candidates=choice.candidates,
minimum=choice.minimum,
maximum=choice.maximum,
resolver=choice.resolver,
source_id=choice.source_id,
)
def _advance(
game: GameState,
effects: tuple[Effect, ...],
firing: list[tuple[L5RCard, Trigger]],
event: GameEvent | None,
queue: list[GameEvent],
) -> None:
"""Run the effect-and-trigger cascade to a fixpoint from an arbitrary resume point.
One resumable worklist machine, in three repeating steps: apply the ``effects`` in hand (each
committing at once, its derived events joining ``queue``); then fire the next trigger still
``firing`` for ``event``, whose effects become the next ``effects`` in hand; then pop the next
event off ``queue`` and collect its triggers. A ``Choose`` among the effects pauses the machine:
it records the decision and stashes the exact remainder — the effects after it, the triggers not
yet fired, the event, and the queue — as a :class:`ResumeCascade`, so :func:`resume_cascade`
continues from precisely here once the choice is answered."""
resolved = 0
firing = list(firing)
while True:
for index, effect in enumerate(effects):
if isinstance(effect, Choose):
game.pending = _choice_request(effect)
_stash(game, tuple(effects[index + 1 :]), firing, event, queue)
return
queue.extend(apply_effect(game, effect))
effects = ()
if firing:
card, trigger = firing.pop(0)
effects = tuple(trigger(TriggerContext(game, card, event)))
continue
if not queue:
return
resolved += 1
if resolved > _MAX_CASCADE:
raise RuntimeError(f"trigger cascade did not converge after {_MAX_CASCADE} events")
event = queue.pop(0)
firing = _collect(game, event)
firing.sort(key=_canonical_order)
def _stash(
game: GameState,
effects: tuple[Effect, ...],
firing: list[tuple[L5RCard, Trigger]],
event: GameEvent | None,
queue: list[GameEvent],
) -> None:
remaining = tuple((card.id, trigger) for card, trigger in firing)
game.stack.append(ResumeCascade(effects, remaining, event, tuple(queue)))
[docs]
def resume_cascade(game: GameState, item: ResumeCascade, produced: list[Effect]) -> None:
"""Continue a cascade a choice paused, with ``produced`` — the effects the answered choice
produced — spliced in where the ``Choose`` stood, ahead of the effects, triggers, and events the
pause stashed. Triggers whose card has since left play are dropped."""
firing = [
(game.table.cards_by_id[card_id], trigger)
for card_id, trigger in item.firing
if card_id in game.table.cards_by_id
]
_advance(game, tuple(produced) + item.effects, firing, item.event, list(item.queue))
[docs]
def fire(game: GameState, event: GameEvent) -> None:
"""Resolve ``event`` and the cascade it triggers, running the worklist to a fixpoint."""
_advance(game, (), [], None, [event])
[docs]
def resolve_effects(game: GameState, effects: list[Effect]) -> None:
"""Apply ``effects`` — an ability's or a choice resolver's output — and run the derived-event
cascade the same way :func:`fire` does, so a triggered reaction to those effects still resolves."""
_advance(game, tuple(effects), [], None, [])
# Per-card triggers, registered on import of this module (as effects.py holds its gold handlers).
@on(TurnStarted, "rice_farm")
def _rice_farm(ctx: TriggerContext) -> list[Effect]:
"""After your turn begins, give this Holding a +1GP Wealth token (max four)."""
if ctx.card.owner is not ctx.event.seat or at_cap(ctx.card, WEALTH, 4):
return []
return [AdjustCounter(ctx.card.id, WEALTH, 1)]
@on(CardDiscarded, "caravansary")
def _caravansary(ctx: TriggerContext) -> list[Effect]:
"""If your action discarded a Fate card, give this Holding a +1GP Wealth token (max three)."""
if not caused_by(ctx, ctx.card.owner) or ctx.event.side is not Side.FATE:
return []
if at_cap(ctx.card, WEALTH, 3):
return []
return [AdjustCounter(ctx.card.id, WEALTH, 1)]
@on(CounterGained, "shosuro_aoki_yoritomo_kayoko_experienced")
def _shosuro_aoki(ctx: TriggerContext) -> list[Effect]:
"""After your Holding gains any Wealth tokens, once per turn, draw a card."""
if ctx.event.counter is not WEALTH:
return []
gainer = ctx.game.table.cards_by_id.get(ctx.event.card_id)
if not isinstance(gainer, DynastyHolding) or gainer.owner is not ctx.card.owner:
return []
if not once_per_turn(ctx.game, ctx.card, "aoki_draw"):
return []
return [DrawCard(ctx.card.owner)]
@on(EnteredPlay, "rural_market")
def _rural_market_enters_play(ctx: TriggerContext) -> list[Effect]:
"""After this Holding enters play, give it a +1GP Wealth token."""
if ctx.event.card_id != ctx.card.id:
return []
return [AdjustCounter(ctx.card.id, WEALTH, 1)]
@on(Destroyed, "rural_market")
def _rural_market_farm_destroyed(ctx: TriggerContext) -> list[Effect]:
"""After your Farm is destroyed, give this Holding a +1GP Wealth token."""
destroyed = ctx.game.table.cards_by_id.get(ctx.event.card_id)
if destroyed is None or destroyed.owner is not ctx.card.owner:
return []
if "Farm" not in destroyed.keywords:
return []
return [AdjustCounter(ctx.card.id, WEALTH, 1)]
@on(EnteredPlay, "wheat_farm")
def _wheat_farm(ctx: TriggerContext) -> list[Effect]:
"""After this Holding enters play, let its controller give zero to two other Farms they control a
+1GP Wealth token."""
if ctx.event.card_id != ctx.card.id:
return []
others = tuple(
card.id
for card in ctx.game.table.battlefield.cards
if card.owner is ctx.card.owner
and card is not ctx.card
and isinstance(card, DynastyHolding)
and "Farm" in card.keywords
)
if not others:
return []
return [Choose(ctx.card.owner, others, 0, min(2, len(others)), "wheat_farm", ctx.card.id)]
@choice_resolver("wheat_farm")
def _wheat_farm_grant(game: GameState, source_id: str, chosen: tuple[str, ...]) -> list[Effect]:
return [AdjustCounter(card_id, WEALTH, 1) for card_id in chosen]
@on(EnteredPlay, "pawnbroker")
def _pawnbroker(ctx: TriggerContext) -> list[Effect]:
"""After this Holding enters play, turn each Sincerity token it accrued into a +1GP Wealth
token."""
if ctx.event.card_id != ctx.card.id:
return []
sincerity = ctx.card.counters.get(SINCERITY.key, 0)
if sincerity == 0:
return []
return [AdjustCounter(ctx.card.id, WEALTH, sincerity)]
@on(EnteredPlay, "sapphire_mine")
def _sapphire_mine(ctx: TriggerContext) -> list[Effect]:
"""Sincerity: after this Holding enters play, if it accrued two or more Sincerity tokens, give it
a +1GP Wealth token."""
if ctx.event.card_id != ctx.card.id:
return []
if ctx.card.counters.get(SINCERITY.key, 0) < 2:
return []
return [AdjustCounter(ctx.card.id, WEALTH, 1)]
@on(EnteredPlay, "mishime_sensei")
def _mishime_sensei_enters_play(ctx: TriggerContext) -> list[Effect]:
"""Mishime Sensei: grant its controller the ignore-Honor-Requirements waiver as it enters
play."""
if ctx.event.card_id != ctx.card.id or ctx.card.owner is None:
return []
return [IgnoreHonorRequirements(ctx.card.owner)]
@on(EnteredPlay, "the_kurai_district_court")
def _kurai_district_court(ctx: TriggerContext) -> list[Effect]:
"""After this Holding enters play, produce one Gold for each Sincerity token it accrued."""
if ctx.event.card_id != ctx.card.id:
return []
sincerity = ctx.card.counters.get(SINCERITY.key, 0)
if sincerity == 0:
return []
return [GainGold(ctx.card.owner, sincerity)]
[docs]
def sincerity_seed_targets(game: GameState, seat: PlayerId) -> list[str]:
"""The seat's face-up Sincerity cards still in a Province with no Sincerity tokens — the legal
recipients of a seeded Sincerity token."""
return [
card.id
for key, zone in game.table.zones.items()
if key.owner is seat and key.role is ZoneRole.PROVINCE
for card in zone.cards
if card.face_up
and SINCERITY_KEYWORD in card.keywords
and card.counters.get(SINCERITY.key, 0) == 0
]
[docs]
def province_holdings(game: GameState, seat: PlayerId) -> list[str]:
"""The seat's face-up Holdings still in a Province — the recruitable targets of a targeted
recruit ability."""
return [
card.id
for key, zone in game.table.zones.items()
if key.owner is seat and key.role is ZoneRole.PROVINCE
for card in zone.cards
if card.face_up and isinstance(card, DynastyHolding)
]
@choice_resolver("modest_farm_straighten")
def _modest_farm_straighten(
game: GameState, source_id: str, chosen: tuple[str, ...]
) -> list[Effect]:
# source_id is the recruited target; chosen holds Modest Farm's id when its controller sacrifices
# it to straighten the target.
if not chosen:
return []
return [Destroy(chosen[0]), Straighten(source_id)]
@on(EnteredPlay, "training_court")
def _training_court(ctx: TriggerContext) -> list[Effect]:
"""Political Tireless Response: after Training Court enters play, seed a Sincerity token onto one
of its controller's token-less Sincerity cards still in a Province."""
if ctx.event.card_id != ctx.card.id:
return []
targets = tuple(sincerity_seed_targets(ctx.game, ctx.card.owner))
if not targets:
return []
return [Choose(ctx.card.owner, targets, 1, 1, "sincerity_seed", ctx.card.id)]
@choice_resolver("sincerity_seed")
def _sincerity_seed(game: GameState, source_id: str, chosen: tuple[str, ...]) -> list[Effect]:
return [AdjustCounter(card_id, SINCERITY, 1) for card_id in chosen]