Add 30-second idle timeout for combat encounters

Encounters track last_action_at (updated on attack and defend). If 30
seconds pass with no actions, combat fizzles out with a message to both
players and combat mode is popped. start_encounter initializes the
timestamp so fresh encounters don't immediately timeout.
This commit is contained in:
Jared Miller 2026-02-08 11:39:04 -05:00
parent e368ed1843
commit 1b3684dc65
Signed by: shmup
GPG key ID: 22B5C6D66A38B06C
4 changed files with 143 additions and 2 deletions

View file

@ -20,6 +20,9 @@ class CombatState(Enum):
# Telegraph phase duration in seconds (3 game ticks at 100ms/tick) # Telegraph phase duration in seconds (3 game ticks at 100ms/tick)
TELEGRAPH_DURATION = 0.3 TELEGRAPH_DURATION = 0.3
# Seconds of no action before combat fizzles out
IDLE_TIMEOUT = 30.0
@dataclass @dataclass
class ResolveResult: class ResolveResult:
@ -42,6 +45,7 @@ class CombatEncounter:
current_move: CombatMove | None = None current_move: CombatMove | None = None
move_started_at: float = 0.0 move_started_at: float = 0.0
pending_defense: CombatMove | None = None pending_defense: CombatMove | None = None
last_action_at: float = 0.0
def attack(self, move: CombatMove) -> None: def attack(self, move: CombatMove) -> None:
"""Initiate or switch an attack move. """Initiate or switch an attack move.
@ -65,6 +69,7 @@ class CombatEncounter:
self.current_move = move self.current_move = move
self.attacker.stamina -= move.stamina_cost self.attacker.stamina -= move.stamina_cost
self.last_action_at = time.monotonic()
if self.state == CombatState.IDLE: if self.state == CombatState.IDLE:
self.state = CombatState.TELEGRAPH self.state = CombatState.TELEGRAPH
@ -77,6 +82,7 @@ class CombatEncounter:
move: The defense move to attempt move: The defense move to attempt
""" """
self.pending_defense = move self.pending_defense = move
self.last_action_at = time.monotonic()
def tick(self, now: float) -> None: def tick(self, now: float) -> None:
"""Advance the state machine based on current time. """Advance the state machine based on current time.

View file

@ -2,7 +2,7 @@
import time import time
from mudlib.combat.encounter import CombatEncounter, CombatState from mudlib.combat.encounter import IDLE_TIMEOUT, CombatEncounter, CombatState
from mudlib.entity import Entity from mudlib.entity import Entity
# Global list of active combat encounters # Global list of active combat encounters
@ -32,7 +32,11 @@ def start_encounter(attacker: Entity, defender: Entity) -> CombatEncounter:
raise ValueError(msg) raise ValueError(msg)
# Create and register the encounter # Create and register the encounter
encounter = CombatEncounter(attacker=attacker, defender=defender) encounter = CombatEncounter(
attacker=attacker,
defender=defender,
last_action_at=time.monotonic(),
)
active_encounters.append(encounter) active_encounters.append(encounter)
return encounter return encounter
@ -71,6 +75,20 @@ async def process_combat() -> None:
now = time.monotonic() now = time.monotonic()
for encounter in active_encounters[:]: # Copy list to allow modification for encounter in active_encounters[:]: # Copy list to allow modification
# Check for idle timeout
if now - encounter.last_action_at > IDLE_TIMEOUT:
await encounter.attacker.send("Combat has fizzled out.\r\n")
await encounter.defender.send("Combat has fizzled out.\r\n")
from mudlib.player import Player
for entity in (encounter.attacker, encounter.defender):
if isinstance(entity, Player) and entity.mode == "combat":
entity.mode_stack.pop()
end_encounter(encounter)
continue
# Tick the state machine # Tick the state machine
encounter.tick(now) encounter.tick(now)

View file

@ -393,3 +393,29 @@ def test_resolve_uses_final_move(attacker, defender, punch, sweep):
expected_damage = attacker.pl * sweep.damage_pct * 1.5 expected_damage = attacker.pl * sweep.damage_pct * 1.5
assert defender.pl == initial_pl - expected_damage assert defender.pl == initial_pl - expected_damage
assert "sweep" in result.attacker_msg.lower() assert "sweep" in result.attacker_msg.lower()
# --- last_action_at tracking tests ---
def test_last_action_at_updates_on_attack(attacker, defender, punch):
"""Test last_action_at is set when attack() is called."""
encounter = CombatEncounter(attacker=attacker, defender=defender)
assert encounter.last_action_at == 0.0
before = time.monotonic()
encounter.attack(punch)
assert encounter.last_action_at >= before
def test_last_action_at_updates_on_defend(attacker, defender, punch, dodge):
"""Test last_action_at is set when defend() is called."""
encounter = CombatEncounter(attacker=attacker, defender=defender)
encounter.attack(punch)
first_action = encounter.last_action_at
time.sleep(0.01)
encounter.defend(dodge)
assert encounter.last_action_at > first_action

View file

@ -328,3 +328,94 @@ async def test_process_combat_sends_messages_on_resolve(punch):
assert any("punch right" in msg.lower() for msg in attacker_msgs) assert any("punch right" in msg.lower() for msg in attacker_msgs)
assert any("punch right" in msg.lower() for msg in defender_msgs) assert any("punch right" in msg.lower() for msg in defender_msgs)
# --- Idle timeout tests ---
@pytest.mark.asyncio
async def test_idle_timeout_ends_encounter(punch):
"""Test encounter times out after 30s of no actions."""
w = _mock_writer
attacker = Player(name="Goku", x=0, y=0, writer=w())
defender = Player(name="Vegeta", x=0, y=0, writer=w())
attacker.mode_stack.append("combat")
defender.mode_stack.append("combat")
encounter = start_encounter(attacker, defender)
# Force last_action_at to 31 seconds ago
encounter.last_action_at = time.monotonic() - 31
await process_combat()
assert get_encounter(attacker) is None
assert get_encounter(defender) is None
@pytest.mark.asyncio
async def test_idle_timeout_sends_message(punch):
"""Test timeout sends fizzle message to both players."""
atk_writer = _mock_writer()
def_writer = _mock_writer()
attacker = Player(name="Goku", x=0, y=0, writer=atk_writer)
defender = Player(name="Vegeta", x=0, y=0, writer=def_writer)
attacker.mode_stack.append("combat")
defender.mode_stack.append("combat")
encounter = start_encounter(attacker, defender)
encounter.last_action_at = time.monotonic() - 31
await process_combat()
atk_msgs = [c[0][0] for c in atk_writer.write.call_args_list]
def_msgs = [c[0][0] for c in def_writer.write.call_args_list]
assert any("fizzled" in msg.lower() for msg in atk_msgs)
assert any("fizzled" in msg.lower() for msg in def_msgs)
@pytest.mark.asyncio
async def test_idle_timeout_pops_combat_mode(punch):
"""Test timeout pops combat mode from both players."""
w = _mock_writer
attacker = Player(name="Goku", x=0, y=0, writer=w())
defender = Player(name="Vegeta", x=0, y=0, writer=w())
attacker.mode_stack.append("combat")
defender.mode_stack.append("combat")
encounter = start_encounter(attacker, defender)
encounter.last_action_at = time.monotonic() - 31
await process_combat()
assert attacker.mode_stack == ["normal"]
assert defender.mode_stack == ["normal"]
@pytest.mark.asyncio
async def test_recent_action_prevents_timeout(punch):
"""Test recent action prevents idle timeout."""
w = _mock_writer
attacker = Player(name="Goku", x=0, y=0, writer=w())
defender = Player(name="Vegeta", x=0, y=0, writer=w())
attacker.mode_stack.append("combat")
defender.mode_stack.append("combat")
encounter = start_encounter(attacker, defender)
# last_action_at was set to now by start_encounter
await process_combat()
# Should still be active
assert get_encounter(attacker) is encounter
@pytest.mark.asyncio
async def test_start_encounter_sets_last_action_at():
"""Test start_encounter initializes last_action_at."""
attacker = Entity(name="Goku", x=0, y=0)
defender = Entity(name="Vegeta", x=0, y=0)
before = time.monotonic()
encounter = start_encounter(attacker, defender)
assert encounter.last_action_at >= before