Add defense commitment lock and defense-everywhere support

Defenses now work outside combat mode with stamina cost, recovery lock
(based on timing_window_ms), and broadcast to nearby players. Lock
prevents spamming defenses — you commit to the move. Stamina deduction
moved from encounter.defend() to do_defend command layer. Defense
commands registered with mode="*" instead of "combat".
This commit is contained in:
Jared Miller 2026-02-08 11:37:09 -05:00
parent cf423fb22b
commit e368ed1843
Signed by: shmup
GPG key ID: 22B5C6D66A38B06C
6 changed files with 138 additions and 35 deletions

View file

@ -1,5 +1,6 @@
"""Combat command handlers.""" """Combat command handlers."""
import time
from collections import defaultdict from collections import defaultdict
from pathlib import Path from pathlib import Path
@ -81,15 +82,19 @@ async def do_attack(player: Player, target_args: str, move: CombatMove) -> None:
async def do_defend(player: Player, _args: str, move: CombatMove) -> None: async def do_defend(player: Player, _args: str, move: CombatMove) -> None:
"""Core defense logic with a resolved move. """Core defense logic with a resolved move.
Works both in and outside combat. Applies a recovery lock
(based on timing_window_ms) so defenses have commitment.
Args: Args:
player: The defending player player: The defending player
_args: Unused (defense moves don't take a target) _args: Unused (defense moves don't take a target)
move: The resolved combat move move: The resolved combat move
""" """
# Check if in combat now = time.monotonic()
encounter = get_encounter(player)
if encounter is None: # Commitment check: locked from previous defense
await player.send("You're not in combat.\r\n") if player.defense_locked_until > now:
await player.send("You're still recovering!\r\n")
return return
# Check stamina # Check stamina
@ -97,9 +102,26 @@ async def do_defend(player: Player, _args: str, move: CombatMove) -> None:
await player.send("You don't have enough stamina for that move.\r\n") await player.send("You don't have enough stamina for that move.\r\n")
return return
# Queue the defense # Apply lock and stamina cost
player.defense_locked_until = now + (move.timing_window_ms / 1000.0)
player.stamina -= move.stamina_cost
# If in combat, queue the defense on the encounter
encounter = get_encounter(player)
if encounter is not None:
encounter.defend(move) encounter.defend(move)
await player.send(f"You attempt to {move.name}!\r\n")
# Broadcast to nearby players
from mudlib.commands.movement import send_nearby_message
await send_nearby_message(
player,
player.x,
player.y,
f"{player.name} {move.name}s!\r\n",
)
await player.send(f"You {move.name}!\r\n")
def _make_direct_handler(move: CombatMove, handler_fn): def _make_direct_handler(move: CombatMove, handler_fn):
@ -174,7 +196,7 @@ def register_combat_commands(content_dir: Path) -> None:
# Register simple moves (roundhouse, sweep, duck, jump) # Register simple moves (roundhouse, sweep, duck, jump)
for move in simple_moves: for move in simple_moves:
handler_fn = do_attack if move.move_type == "attack" else do_defend handler_fn = do_attack if move.move_type == "attack" else do_defend
mode = "*" if move.move_type == "attack" else "combat" mode = "*"
action = "Attack" if move.move_type == "attack" else "Defend" action = "Attack" if move.move_type == "attack" else "Defend"
register( register(
CommandDefinition( CommandDefinition(

View file

@ -69,16 +69,15 @@ class CombatEncounter:
self.state = CombatState.TELEGRAPH self.state = CombatState.TELEGRAPH
def defend(self, move: CombatMove) -> None: def defend(self, move: CombatMove) -> None:
"""Queue a defense move. """Queue a defense move on the encounter.
Stamina cost and lock are handled by the command layer (do_defend).
Args: Args:
move: The defense move to attempt move: The defense move to attempt
""" """
self.pending_defense = move self.pending_defense = move
# Apply stamina cost
self.defender.stamina -= move.stamina_cost
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

@ -14,6 +14,7 @@ class Entity:
pl: float = 100.0 # power level (health and damage multiplier) pl: float = 100.0 # power level (health and damage multiplier)
stamina: float = 100.0 # current stamina stamina: float = 100.0 # current stamina
max_stamina: float = 100.0 # stamina ceiling max_stamina: float = 100.0 # stamina ceiling
defense_locked_until: float = 0.0 # monotonic time when defense recovery ends
async def send(self, message: str) -> None: async def send(self, message: str) -> None:
"""Send a message to this entity. Base implementation is a no-op.""" """Send a message to this entity. Base implementation is a no-op."""

View file

@ -1,10 +1,12 @@
"""Tests for combat commands.""" """Tests for combat commands."""
import time
from pathlib import Path from pathlib import Path
from unittest.mock import AsyncMock, MagicMock from unittest.mock import AsyncMock, MagicMock
import pytest import pytest
import mudlib.commands.movement as movement_mod
from mudlib.combat import commands as combat_commands from mudlib.combat import commands as combat_commands
from mudlib.combat.engine import active_encounters, get_encounter from mudlib.combat.engine import active_encounters, get_encounter
from mudlib.combat.moves import load_moves from mudlib.combat.moves import load_moves
@ -21,6 +23,18 @@ def clear_state():
players.clear() players.clear()
@pytest.fixture(autouse=True)
def mock_world():
"""Inject a mock world for send_nearby_message."""
fake_world = MagicMock()
fake_world.width = 256
fake_world.height = 256
old = movement_mod.world
movement_mod.world = fake_world
yield fake_world
movement_mod.world = old
@pytest.fixture @pytest.fixture
def mock_writer(): def mock_writer():
writer = MagicMock() writer = MagicMock()
@ -154,25 +168,23 @@ async def test_attack_sends_telegraph_to_defender(player, target, punch_right):
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_defense_only_in_combat(player, dodge_left): async def test_defense_works_outside_combat(player, dodge_left):
"""Test do_defend only works in combat mode.""" """Test do_defend works outside combat (costs stamina, applies lock)."""
initial_stamina = player.stamina
await combat_commands.do_defend(player, "", dodge_left) await combat_commands.do_defend(player, "", dodge_left)
player.writer.write.assert_called() assert player.stamina == initial_stamina - dodge_left.stamina_cost
message = player.writer.write.call_args[0][0] assert player.defense_locked_until > 0
assert "not in combat" in message.lower()
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_defense_records_pending_defense(player, target, punch_right, dodge_left): async def test_defense_records_pending_defense(player, target, punch_right, dodge_left):
"""Test do_defend records the defense move.""" """Test do_defend queues defense on encounter and costs stamina."""
# Start combat # Start combat
await combat_commands.do_attack(player, "Vegeta", punch_right) await combat_commands.do_attack(player, "Vegeta", punch_right)
player.writer.write.reset_mock() player.writer.write.reset_mock()
# Switch to defender's perspective initial_stamina = target.stamina
target.writer = player.writer
target.mode_stack = ["combat"]
# Defend # Defend
await combat_commands.do_defend(target, "", dodge_left) await combat_commands.do_defend(target, "", dodge_left)
@ -181,23 +193,19 @@ async def test_defense_records_pending_defense(player, target, punch_right, dodg
assert encounter is not None assert encounter is not None
assert encounter.pending_defense is not None assert encounter.pending_defense is not None
assert encounter.pending_defense.name == "dodge left" assert encounter.pending_defense.name == "dodge left"
assert target.stamina == initial_stamina - dodge_left.stamina_cost
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_defense_insufficient_stamina(player, target, punch_right, dodge_left): async def test_defense_insufficient_stamina(player, dodge_left):
"""Test do_defend with insufficient stamina gives error.""" """Test do_defend with insufficient stamina gives error."""
# Start combat player.stamina = 1.0 # Not enough for dodge (costs 3)
await combat_commands.do_attack(player, "Vegeta", punch_right)
target.writer = player.writer
target.mode_stack = ["combat"]
target.stamina = 1.0 # Not enough for dodge (costs 3)
player.writer.write.reset_mock() await combat_commands.do_defend(player, "", dodge_left)
await combat_commands.do_defend(target, "", dodge_left)
target.writer.write.assert_called() player.writer.write.assert_called()
message = target.writer.write.call_args[0][0] messages = [call[0][0] for call in player.writer.write.call_args_list]
assert "stamina" in message.lower() assert any("stamina" in msg.lower() for msg in messages)
# --- variant handler tests --- # --- variant handler tests ---
@ -320,3 +328,75 @@ async def test_switch_attack_sends_new_telegraph(
target_msgs = [call[0][0] for call in target.writer.write.call_args_list] target_msgs = [call[0][0] for call in target.writer.write.call_args_list]
# Defender should get a new telegraph # Defender should get a new telegraph
assert len(target_msgs) > 0 assert len(target_msgs) > 0
# --- defense commitment tests ---
@pytest.mark.asyncio
async def test_defense_rejected_while_locked(player, dodge_left):
"""Test defense rejected during recovery lock."""
await combat_commands.do_defend(player, "", dodge_left)
player.writer.write.reset_mock()
# Immediately try again — should be locked
await combat_commands.do_defend(player, "", dodge_left)
messages = [call[0][0] for call in player.writer.write.call_args_list]
assert any("still recovering" in msg.lower() for msg in messages)
@pytest.mark.asyncio
async def test_defense_works_after_lock_expires(player, dodge_left):
"""Test defense works after recovery lock expires."""
await combat_commands.do_defend(player, "", dodge_left)
# Fast-forward past the lock
player.defense_locked_until = time.monotonic() - 1.0
player.writer.write.reset_mock()
await combat_commands.do_defend(player, "", dodge_left)
messages = [call[0][0] for call in player.writer.write.call_args_list]
# Should succeed, not say "still recovering"
assert not any("still recovering" in msg.lower() for msg in messages)
assert any("dodge left" in msg.lower() for msg in messages)
@pytest.mark.asyncio
async def test_defense_lock_uses_timing_window(player, dodge_left, moves):
"""Test defense lock duration matches move's timing_window_ms."""
before = time.monotonic()
await combat_commands.do_defend(player, "", dodge_left)
expected_lock = dodge_left.timing_window_ms / 1000.0
# Lock should be roughly now + timing_window
assert player.defense_locked_until >= before + expected_lock - 0.1
assert player.defense_locked_until <= before + expected_lock + 0.5
@pytest.mark.asyncio
async def test_defense_in_combat_queues_on_encounter(
player, target, punch_right, dodge_left
):
"""Test defense in combat queues on encounter AND applies lock."""
await combat_commands.do_attack(player, "Vegeta", punch_right)
await combat_commands.do_defend(target, "", dodge_left)
encounter = get_encounter(target)
assert encounter is not None
assert encounter.pending_defense is dodge_left
assert target.defense_locked_until > 0
@pytest.mark.asyncio
async def test_defense_broadcast_to_nearby(player, target, dodge_left):
"""Test defense broadcasts to nearby players."""
target.writer.write.reset_mock()
await combat_commands.do_defend(player, "", dodge_left)
# Target is at same coords, should get broadcast
target_msgs = [call[0][0] for call in target.writer.write.call_args_list]
assert any(player.name in msg for msg in target_msgs)

View file

@ -100,15 +100,15 @@ def test_defend_records_pending_defense(attacker, defender, punch, dodge):
assert encounter.pending_defense is dodge assert encounter.pending_defense is dodge
def test_defend_applies_stamina_cost(attacker, defender, punch, dodge): def test_defend_does_not_deduct_stamina(attacker, defender, punch, dodge):
"""Test defending costs stamina.""" """Test encounter.defend() does not deduct stamina (command layer does)."""
encounter = CombatEncounter(attacker=attacker, defender=defender) encounter = CombatEncounter(attacker=attacker, defender=defender)
encounter.attack(punch) encounter.attack(punch)
initial_stamina = defender.stamina initial_stamina = defender.stamina
encounter.defend(dodge) encounter.defend(dodge)
assert defender.stamina == initial_stamina - dodge.stamina_cost assert defender.stamina == initial_stamina
def test_tick_telegraph_to_window(attacker, defender, punch): def test_tick_telegraph_to_window(attacker, defender, punch):

View file

@ -27,6 +27,7 @@ def test_entity_has_combat_stats():
assert entity.pl == 100.0 assert entity.pl == 100.0
assert entity.stamina == 100.0 assert entity.stamina == 100.0
assert entity.max_stamina == 100.0 assert entity.max_stamina == 100.0
assert entity.defense_locked_until == 0.0
def test_entity_combat_stats_can_be_customized(): def test_entity_combat_stats_can_be_customized():