Add forced action queue for double knockouts (Issue #10)

Changed forced_action from single item to FIFO queue to support
scenarios where multiple forced actions are needed simultaneously:

- forced_actions: list[ForcedAction] replaces forced_action: ForcedAction | None
- Added queue management methods:
  - has_forced_action() - check if queue has pending actions
  - get_current_forced_action() - get first action without removing
  - add_forced_action(action) - add to end of queue
  - pop_forced_action() - remove and return first action
  - clear_forced_actions() - clear all pending actions
- Updated engine, turn_manager, rules_validator, and visibility filter
- Added 8 new tests for forced action queue including double knockout scenario

This fixes the bug where simultaneous knockouts (e.g., mutual poison damage)
would lose one player's select_active action due to overwriting.

795 tests passing.
This commit is contained in:
Cal Corum 2026-01-26 11:33:47 -06:00
parent 8e084d250a
commit 9432499018
13 changed files with 378 additions and 119 deletions

View File

@ -3,7 +3,7 @@
**Date:** 2026-01-25
**Updated:** 2026-01-26
**Reviewed By:** Multi-agent system review
**Status:** 789 tests passing, 94% coverage
**Status:** 795 tests passing, 94% coverage
---
@ -165,12 +165,18 @@ This ensures Pokemon with multiple abilities (e.g., one limited to 1/turn, anoth
---
### 10. Double Knockout - Only One Forced Action
**File:** `app/core/turn_manager.py:478-486`
### 10. ~~Double Knockout - Only One Forced Action~~ FIXED
**File:** `app/core/models/game_state.py`
**Status:** FIXED in current session
If both players' active Pokemon are KO'd simultaneously, only one `forced_action` can be set (last writer wins).
Changed `forced_action: ForcedAction | None` to `forced_actions: list[ForcedAction]` (FIFO queue):
- `has_forced_action()` - Check if queue has pending actions
- `get_current_forced_action()` - Get first action without removing
- `add_forced_action(action)` - Add to end of queue
- `pop_forced_action()` - Remove and return first action
- `clear_forced_actions()` - Clear all pending actions
**Fix:** Support a queue of forced actions or handle simultaneous KOs specially.
When both players' active Pokemon are KO'd, both forced actions are queued and processed in order.
---
@ -306,10 +312,9 @@ Also added `GameEndReason.TURN_LIMIT` enum value to distinguish from `TIMEOUT` (
9. ~~Fix per-ability usage tracking~~ DONE (#9)
### Phase 3: Polish (Before Production)
10. Add CardDefinition field validation
11. Handle double knockouts
12. Improve effect error handling
13. Add missing effect handlers
10. ~~Handle double knockouts~~ DONE (#10)
11. Improve effect error handling
12. Add missing effect handlers
---
@ -331,7 +336,7 @@ Also added `GameEndReason.TURN_LIMIT` enum value to distinguish from `TIMEOUT` (
| `win_conditions.py` | 99% | Near complete |
| `engine.py` | 81% | Gaps in error paths |
| `rng.py` | 93% | Good |
| **TOTAL** | **94%** | **789 tests** |
| **TOTAL** | **94%** | **795 tests** |
### New Tests Added
- `tests/core/test_evolution_stack.py` - 28 tests for evolution stack, devolve, knockout with attachments, find_card_instance
@ -339,6 +344,7 @@ Also added `GameEndReason.TURN_LIMIT` enum value to distinguish from `TIMEOUT` (
- `tests/core/test_engine.py::TestTurnLimitCheck` - 5 tests for turn limit checking
- `tests/core/test_turn_manager.py::TestPrizeCardModeKnockout` - 4 tests for prize card mode in knockouts
- `tests/core/test_models/test_card.py::TestCardInstanceTurnState::test_can_use_ability_independent_tracking` - 1 test for per-ability usage tracking
- `tests/core/test_models/test_game_state.py::TestForcedActionQueue` - 8 tests for forced action queue management
---
@ -352,6 +358,17 @@ Also added `GameEndReason.TURN_LIMIT` enum value to distinguish from `TIMEOUT` (
## Change Log
### 2026-01-26 - Forced Action Queue for Double Knockouts (Issue #10)
Changed `forced_action` from single item to FIFO queue:
- `forced_actions: list[ForcedAction]` replaces `forced_action: ForcedAction | None`
- Added queue management methods: `has_forced_action()`, `get_current_forced_action()`,
`add_forced_action()`, `pop_forced_action()`, `clear_forced_actions()`
- Updated engine, turn_manager, rules_validator, and visibility filter
- Added 8 new tests for forced action queue including double knockout scenario
**Total: 795 tests passing**
### 2026-01-26 - Per-Ability Usage Tracking (Issue #9)
Fixed issue #9 - ability usage now tracked per-ability instead of globally:

View File

@ -875,8 +875,8 @@ class GameEngine:
player.active.add(new_active)
# Clear forced action
game.forced_action = None
# Pop the completed forced action from the queue
game.pop_forced_action()
return ActionResult(
success=True,
@ -915,15 +915,16 @@ class GameEngine:
# Handle multi-prize selection (e.g., for EX/VMAX knockouts)
win_result = None
if game.forced_action and game.forced_action.action_type == "select_prize":
remaining = game.forced_action.params.get("count", 1) - 1
current_forced = game.get_current_forced_action()
if current_forced and current_forced.action_type == "select_prize":
remaining = current_forced.params.get("count", 1) - 1
if remaining > 0 and len(player.prizes) > 0:
# More prizes to take
game.forced_action.params["count"] = remaining
game.forced_action.reason = f"Select {remaining} more prize card(s)"
# More prizes to take - update the current action
current_forced.params["count"] = remaining
current_forced.reason = f"Select {remaining} more prize card(s)"
else:
# Done taking prizes
game.forced_action = None
# Done taking prizes - pop this action from queue
game.pop_forced_action()
# Check for win by taking all prizes
if len(player.prizes) == 0:

View File

@ -374,7 +374,8 @@ class GameState(BaseModel):
stadium_in_play: The current Stadium card in play, if any.
turn_order: List of player IDs in turn order.
first_turn_completed: Whether the very first turn of the game is done.
forced_action: A ForcedAction that must be completed before game proceeds.
forced_actions: Queue of ForcedAction items that must be completed before game proceeds.
Actions are processed in FIFO order (first added = first to resolve).
action_log: Log of actions taken (for replays/debugging).
"""
@ -401,8 +402,9 @@ class GameState(BaseModel):
# First turn tracking
first_turn_completed: bool = False
# Forced action (e.g., select new active after KO)
forced_action: ForcedAction | None = None
# Forced actions queue (e.g., select new active after KO, select prizes)
# Actions are processed in FIFO order - first added is first to resolve
forced_actions: list[ForcedAction] = Field(default_factory=list)
# Optional action log for replays
action_log: list[dict[str, Any]] = Field(default_factory=list)
@ -460,6 +462,48 @@ class GameState(BaseModel):
"""Check if the game has ended."""
return self.winner_id is not None
# =========================================================================
# Forced Action Queue Management
# =========================================================================
def has_forced_action(self) -> bool:
"""Check if there are any pending forced actions."""
return len(self.forced_actions) > 0
def get_current_forced_action(self) -> ForcedAction | None:
"""Get the current (first) forced action without removing it.
Returns:
The first forced action in the queue, or None if queue is empty.
"""
if self.forced_actions:
return self.forced_actions[0]
return None
def add_forced_action(self, action: ForcedAction) -> None:
"""Add a forced action to the queue.
Actions are processed in FIFO order, so this adds to the end.
Args:
action: The ForcedAction to add.
"""
self.forced_actions.append(action)
def pop_forced_action(self) -> ForcedAction | None:
"""Remove and return the current (first) forced action.
Returns:
The first forced action that was removed, or None if queue was empty.
"""
if self.forced_actions:
return self.forced_actions.pop(0)
return None
def clear_forced_actions(self) -> None:
"""Clear all pending forced actions."""
self.forced_actions.clear()
def get_player_count(self) -> int:
"""Return the number of players in this game."""
return len(self.players)

View File

@ -83,7 +83,7 @@ def validate_action(game: GameState, player_id: str, action: Action) -> Validati
return ValidationResult(valid=False, reason="Game is over")
# 2. Check for forced action
if game.forced_action is not None:
if game.has_forced_action():
return _check_forced_action(game, player_id, action)
# 3. Check if it's the player's turn (resign always allowed)
@ -138,9 +138,10 @@ def _check_forced_action(game: GameState, player_id: str, action: Action) -> Val
"""Check if the action satisfies a forced action requirement.
When a forced action is pending, only the specified player can act,
and only with the specified action type.
and only with the specified action type. Only the first action in the
queue is checked - subsequent actions are handled after completion.
"""
forced = game.forced_action
forced = game.get_current_forced_action()
if forced.player_id != player_id:
return ValidationResult(

View File

@ -520,10 +520,12 @@ class TurnManager:
if player.has_benched_pokemon() and not player.has_active_pokemon():
from app.core.models.game_state import ForcedAction
game.forced_action = ForcedAction(
player_id=player_id,
action_type="select_active",
reason="Your active Pokemon was knocked out. Select a new active Pokemon.",
game.add_forced_action(
ForcedAction(
player_id=player_id,
action_type="select_active",
reason="Your active Pokemon was knocked out. Select a new active Pokemon.",
)
)
return None
@ -640,14 +642,15 @@ class TurnManager:
prize_card = opponent.prizes.cards.pop(idx)
opponent.hand.add(prize_card)
else:
# Player chooses - set up forced action
# Note: If there's already a forced action, we may need to queue
# For now, we overwrite (select_prize takes priority)
game.forced_action = ForcedAction(
player_id=opponent_id,
action_type="select_prize",
reason=f"Select {prizes_to_take} prize card(s)",
params={"count": prizes_to_take},
# Player chooses - add to forced action queue
# Multiple forced actions can be queued (e.g., double knockout)
game.add_forced_action(
ForcedAction(
player_id=opponent_id,
action_type="select_prize",
reason=f"Select {prizes_to_take} prize card(s)",
params={"count": prizes_to_take},
)
)
# Check for win by all prizes taken

View File

@ -270,14 +270,15 @@ def get_visible_state(game: GameState, viewer_id: str) -> VisibleGameState:
# Filter each player's state
visible_players = {pid: _filter_player_state(game, pid, viewer_id) for pid in game.players}
# Extract forced action info (if any)
# Extract forced action info (if any) - only shows the current/first action in queue
forced_action_player = None
forced_action_type = None
forced_action_reason = None
if game.forced_action:
forced_action_player = game.forced_action.player_id
forced_action_type = game.forced_action.action_type
forced_action_reason = game.forced_action.reason
current_forced = game.get_current_forced_action()
if current_forced:
forced_action_player = current_forced.player_id
forced_action_type = current_forced.action_type
forced_action_reason = current_forced.reason
return VisibleGameState(
game_id=game.game_id,
@ -363,14 +364,15 @@ def get_spectator_state(game: GameState) -> VisibleGameState:
vstar_power_used=player.vstar_power_used,
)
# Extract forced action info
# Extract forced action info - only shows the current/first action in queue
forced_action_player = None
forced_action_type = None
forced_action_reason = None
if game.forced_action:
forced_action_player = game.forced_action.player_id
forced_action_type = game.forced_action.action_type
forced_action_reason = game.forced_action.reason
current_forced = game.get_current_forced_action()
if current_forced:
forced_action_player = current_forced.player_id
forced_action_type = current_forced.action_type
forced_action_reason = current_forced.reason
return VisibleGameState(
game_id=game.game_id,

View File

@ -919,11 +919,13 @@ def game_with_forced_action(extended_card_registry, card_instance_factory) -> Ga
turn_number=3,
phase=TurnPhase.ATTACK,
first_turn_completed=True,
forced_action=ForcedAction(
player_id="player2",
action_type="select_active",
reason="Active Pokemon was knocked out",
),
forced_actions=[
ForcedAction(
player_id="player2",
action_type="select_active",
reason="Active Pokemon was knocked out",
)
],
)
return game

View File

@ -194,11 +194,13 @@ class TestForcedActionEdgeCases:
turn_number=3,
phase=TurnPhase.ATTACK,
first_turn_completed=True,
forced_action=ForcedAction(
player_id="player2", # Player2 must act
action_type="select_active",
reason="Active Pokemon was knocked out",
),
forced_actions=[
ForcedAction(
player_id="player2", # Player2 must act
action_type="select_active",
reason="Active Pokemon was knocked out",
)
],
)
# Player1 tries to act (should fail)
@ -231,11 +233,13 @@ class TestForcedActionEdgeCases:
turn_number=3,
phase=TurnPhase.ATTACK,
first_turn_completed=True,
forced_action=ForcedAction(
player_id="player2",
action_type="select_active", # Must select active
reason="Active Pokemon was knocked out",
),
forced_actions=[
ForcedAction(
player_id="player2",
action_type="select_active", # Must select active
reason="Active Pokemon was knocked out",
)
],
)
# Player2 tries to pass instead of selecting active
@ -269,11 +273,13 @@ class TestForcedActionEdgeCases:
turn_number=3,
phase=TurnPhase.ATTACK,
first_turn_completed=True,
forced_action=ForcedAction(
player_id="player2",
action_type="invalid_action_type_xyz", # Not a valid forced action
reason="Corrupted state",
),
forced_actions=[
ForcedAction(
player_id="player2",
action_type="invalid_action_type_xyz", # Not a valid forced action
reason="Corrupted state",
)
],
)
# Create a mock action with matching type to bypass initial check
@ -284,7 +290,7 @@ class TestForcedActionEdgeCases:
# isn't in the validators dict
# Actually, let's test when forced action type doesn't have a validator
game.forced_action.action_type = "attack" # Not in forced action validators
game.forced_actions[0].action_type = "attack" # Not in forced action validators
action = AttackAction(attack_index=0)
result = validate_action(game, "player2", action)
@ -729,11 +735,13 @@ class TestForcedActionPlayerNotFound:
turn_number=3,
phase=TurnPhase.ATTACK,
first_turn_completed=True,
forced_action=ForcedAction(
player_id="ghost_player", # Doesn't exist!
action_type="select_active",
reason="Ghost player needs to act",
),
forced_actions=[
ForcedAction(
player_id="ghost_player", # Doesn't exist!
action_type="select_active",
reason="Ghost player needs to act",
)
],
)
# The ghost player tries to act

View File

@ -1080,11 +1080,13 @@ class TestSelectPrizeAction:
initial_prize_count = len(player.prizes)
# Set up forced action (as if a knockout happened)
game.forced_action = ForcedAction(
player_id="player1",
action_type="select_prize",
reason="Select a prize card",
params={"count": 1},
game.add_forced_action(
ForcedAction(
player_id="player1",
action_type="select_prize",
reason="Select a prize card",
params={"count": 1},
)
)
result = await engine.execute_action(game, "player1", SelectPrizeAction(prize_index=0))
@ -1136,11 +1138,13 @@ class TestSelectPrizeAction:
engine.start_turn(game)
# Set up forced action for prize selection
game.forced_action = ForcedAction(
player_id="player1",
action_type="select_prize",
reason="Select your last prize card",
params={"count": 1},
game.add_forced_action(
ForcedAction(
player_id="player1",
action_type="select_prize",
reason="Select your last prize card",
params={"count": 1},
)
)
result = await engine.execute_action(game, "player1", SelectPrizeAction(prize_index=0))
@ -2119,10 +2123,12 @@ class TestSelectActiveAction:
p2.active.add(CardInstance(instance_id="p2-active", definition_id=basic_pokemon_def.id))
# Set forced action
game.forced_action = ForcedAction(
player_id="player1",
action_type="select_active",
reason="Active Pokemon was knocked out",
game.add_forced_action(
ForcedAction(
player_id="player1",
action_type="select_active",
reason="Active Pokemon was knocked out",
)
)
game.phase = TurnPhase.MAIN
@ -2156,7 +2162,7 @@ class TestSelectActiveAction:
assert "p1-bench-1" not in p1.bench
# Forced action should be cleared
assert game_with_forced_action.forced_action is None
assert not game_with_forced_action.has_forced_action()
@pytest.mark.asyncio
async def test_select_active_not_on_bench(

View File

@ -962,6 +962,168 @@ class TestGameStateWinConditions:
assert game.is_game_over()
class TestForcedActionQueue:
"""Tests for the forced action queue system.
The forced action queue allows multiple forced actions to be queued
and processed in FIFO order. This is essential for scenarios like
double knockouts where both players need to select new active Pokemon.
"""
def test_has_forced_action_empty(self) -> None:
"""
Verify has_forced_action returns False when queue is empty.
"""
game = GameState(game_id="test")
assert not game.has_forced_action()
def test_has_forced_action_with_items(self) -> None:
"""
Verify has_forced_action returns True when queue has items.
"""
from app.core.models.game_state import ForcedAction
game = GameState(game_id="test")
game.add_forced_action(
ForcedAction(
player_id="player1",
action_type="select_active",
reason="Test",
)
)
assert game.has_forced_action()
def test_add_forced_action_fifo_order(self) -> None:
"""
Verify forced actions are processed in FIFO order.
First action added should be first action returned.
"""
from app.core.models.game_state import ForcedAction
game = GameState(game_id="test")
# Add actions in order
game.add_forced_action(
ForcedAction(player_id="player1", action_type="select_active", reason="First")
)
game.add_forced_action(
ForcedAction(player_id="player2", action_type="select_active", reason="Second")
)
game.add_forced_action(
ForcedAction(player_id="player1", action_type="select_prize", reason="Third")
)
# First should be player1's first action
first = game.get_current_forced_action()
assert first.player_id == "player1"
assert first.reason == "First"
def test_pop_forced_action_removes_first(self) -> None:
"""
Verify pop_forced_action removes and returns the first item.
"""
from app.core.models.game_state import ForcedAction
game = GameState(game_id="test")
game.add_forced_action(
ForcedAction(player_id="player1", action_type="select_active", reason="First")
)
game.add_forced_action(
ForcedAction(player_id="player2", action_type="select_active", reason="Second")
)
# Pop first
popped = game.pop_forced_action()
assert popped.reason == "First"
# Second should now be current
current = game.get_current_forced_action()
assert current.reason == "Second"
# Pop second
popped = game.pop_forced_action()
assert popped.reason == "Second"
# Queue should now be empty
assert not game.has_forced_action()
assert game.get_current_forced_action() is None
assert game.pop_forced_action() is None
def test_clear_forced_actions(self) -> None:
"""
Verify clear_forced_actions removes all items.
"""
from app.core.models.game_state import ForcedAction
game = GameState(game_id="test")
game.add_forced_action(
ForcedAction(player_id="player1", action_type="select_active", reason="First")
)
game.add_forced_action(
ForcedAction(player_id="player2", action_type="select_active", reason="Second")
)
assert game.has_forced_action()
game.clear_forced_actions()
assert not game.has_forced_action()
assert len(game.forced_actions) == 0
def test_double_knockout_scenario(self) -> None:
"""
Test the double knockout scenario where both players need actions.
When both players' active Pokemon are knocked out simultaneously
(e.g., from mutual damage or poison), both need to select new actives.
The queue ensures both actions are tracked and processed in order.
"""
from app.core.models.game_state import ForcedAction
game = GameState(game_id="test")
# Simulate double KO - both players need to select new active
# Player 1's active was KO'd first (or turn order determines)
game.add_forced_action(
ForcedAction(
player_id="player1",
action_type="select_active",
reason="Player1's active was knocked out",
)
)
# Player 2's active was also KO'd
game.add_forced_action(
ForcedAction(
player_id="player2",
action_type="select_active",
reason="Player2's active was knocked out",
)
)
# Two actions should be queued
assert len(game.forced_actions) == 2
# Player 1 acts first
current = game.get_current_forced_action()
assert current.player_id == "player1"
# Player 1 completes their action
game.pop_forced_action()
# Now player 2 should act
current = game.get_current_forced_action()
assert current.player_id == "player2"
# Player 2 completes their action
game.pop_forced_action()
# Queue should be empty
assert not game.has_forced_action()
class TestGameStateCardSearch:
"""Tests for GameState card search operations."""

View File

@ -1678,10 +1678,12 @@ class TestSelectPrizeValidation:
player.prizes.add(card_instance_factory("pikachu_base_001", instance_id=f"prize_{i}"))
# Set up forced action
game_in_main_phase.forced_action = ForcedAction(
player_id="player1",
action_type="select_prize",
reason="Knocked out opponent's Pokemon",
game_in_main_phase.add_forced_action(
ForcedAction(
player_id="player1",
action_type="select_prize",
reason="Knocked out opponent's Pokemon",
)
)
action = SelectPrizeAction(prize_index=0)
@ -1699,10 +1701,12 @@ class TestSelectPrizeValidation:
game_in_main_phase.phase = TurnPhase.END
# Set up forced action
game_in_main_phase.forced_action = ForcedAction(
player_id="player1",
action_type="select_prize",
reason="Test",
game_in_main_phase.add_forced_action(
ForcedAction(
player_id="player1",
action_type="select_prize",
reason="Test",
)
)
action = SelectPrizeAction(prize_index=0)
@ -1725,10 +1729,12 @@ class TestSelectPrizeValidation:
for i in range(2):
player.prizes.add(card_instance_factory("pikachu_base_001", instance_id=f"prize_{i}"))
game_in_main_phase.forced_action = ForcedAction(
player_id="player1",
action_type="select_prize",
reason="Test",
game_in_main_phase.add_forced_action(
ForcedAction(
player_id="player1",
action_type="select_prize",
reason="Test",
)
)
action = SelectPrizeAction(prize_index=5) # Out of range!
@ -1748,10 +1754,12 @@ class TestSelectPrizeValidation:
player = game_in_main_phase.players["player1"]
player.prizes.clear()
game_in_main_phase.forced_action = ForcedAction(
player_id="player1",
action_type="select_prize",
reason="Test",
game_in_main_phase.add_forced_action(
ForcedAction(
player_id="player1",
action_type="select_prize",
reason="Test",
)
)
action = SelectPrizeAction(prize_index=0)

View File

@ -1082,9 +1082,10 @@ class TestKnockoutProcessing:
turn_manager.process_knockout(two_player_game, active_id, "player2")
assert two_player_game.forced_action is not None
assert two_player_game.forced_action.player_id == "player1"
assert two_player_game.forced_action.action_type == "select_active"
current_forced = two_player_game.get_current_forced_action()
assert current_forced is not None
assert current_forced.player_id == "player1"
assert current_forced.action_type == "select_active"
def test_process_bench_knockout(
self,
@ -1356,9 +1357,10 @@ class TestStatusKnockoutIntegration:
turn_manager.end_turn(two_player_game, seeded_rng)
# Should have forced action to select new active
assert two_player_game.forced_action is not None
assert two_player_game.forced_action.player_id == player.player_id
assert two_player_game.forced_action.action_type == "select_active"
current_forced = two_player_game.get_current_forced_action()
assert current_forced is not None
assert current_forced.player_id == player.player_id
assert current_forced.action_type == "select_active"
# =============================================================================
@ -1478,10 +1480,11 @@ class TestPrizeCardModeKnockout:
turn_manager.process_knockout(prize_card_game, bench_id, "player2", seeded_rng)
# Should have forced action for prize selection
assert prize_card_game.forced_action is not None
assert prize_card_game.forced_action.action_type == "select_prize"
assert prize_card_game.forced_action.player_id == "player2"
assert prize_card_game.forced_action.params.get("count") == 1
current_forced = prize_card_game.get_current_forced_action()
assert current_forced is not None
assert current_forced.action_type == "select_prize"
assert current_forced.player_id == "player2"
assert current_forced.params.get("count") == 1
def test_knockout_ex_awards_two_prizes(
self,

View File

@ -618,10 +618,12 @@ class TestGameStateInformation:
Both players need to know when a forced action is pending.
"""
full_game.forced_action = ForcedAction(
player_id="player1",
action_type="select_active",
reason="Your active Pokemon was knocked out.",
full_game.add_forced_action(
ForcedAction(
player_id="player1",
action_type="select_active",
reason="Your active Pokemon was knocked out.",
)
)
visible = get_visible_state(full_game, "player2")
@ -740,7 +742,7 @@ class TestEdgeCases:
"""
Test that missing forced action is handled correctly.
"""
full_game.forced_action = None
full_game.clear_forced_actions()
visible = get_visible_state(full_game, "player1")