Replace defense lock with sleep-based commitment blocking

Defense moves now asyncio.sleep for timing_window_ms instead of using
a cooldown field. Input queues naturally since the per-player loop is
sequential. Outside combat shows "parry the air!" flavor text.
This commit is contained in:
Jared Miller 2026-02-08 12:23:53 -05:00
parent 2de1ebd59e
commit 3c5c1490e6
Signed by: shmup
GPG key ID: 22B5C6D66A38B06C
2 changed files with 21 additions and 51 deletions

View file

@ -1,5 +1,6 @@
"""Combat command handlers.""" """Combat command handlers."""
import asyncio
import time import time
from collections import defaultdict from collections import defaultdict
from pathlib import Path from pathlib import Path
@ -90,20 +91,11 @@ async def do_defend(player: Player, _args: str, move: CombatMove) -> None:
_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
""" """
now = time.monotonic()
# Commitment check: locked from previous defense
if player.defense_locked_until > now:
await player.send("You're still recovering!\r\n")
return
# Check stamina # Check stamina
if player.stamina < move.stamina_cost: if player.stamina < move.stamina_cost:
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
# Apply lock and stamina cost
player.defense_locked_until = now + (move.timing_window_ms / 1000.0)
player.stamina -= move.stamina_cost player.stamina -= move.stamina_cost
# If in combat, queue the defense on the encounter # If in combat, queue the defense on the encounter
@ -118,10 +110,16 @@ async def do_defend(player: Player, _args: str, move: CombatMove) -> None:
player, player,
player.x, player.x,
player.y, player.y,
f"{player.name} {move.name}s!\r\n", f"{player.name} {move.command}s!\r\n",
) )
# Commitment: block for the timing window (inputs queue naturally)
await asyncio.sleep(move.timing_window_ms / 1000.0)
if encounter is not None:
await player.send(f"You {move.name}!\r\n") await player.send(f"You {move.name}!\r\n")
else:
await player.send(f"You {move.command} the air!\r\n")
def _make_direct_handler(move: CombatMove, handler_fn): def _make_direct_handler(move: CombatMove, handler_fn):

View file

@ -169,12 +169,13 @@ async def test_attack_sends_telegraph_to_defender(player, target, punch_right):
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_defense_works_outside_combat(player, dodge_left): async def test_defense_works_outside_combat(player, dodge_left):
"""Test do_defend works outside combat (costs stamina, applies lock).""" """Test do_defend works outside combat (costs stamina, 'the air' message)."""
initial_stamina = player.stamina initial_stamina = player.stamina
await combat_commands.do_defend(player, "", dodge_left) await combat_commands.do_defend(player, "", dodge_left)
assert player.stamina == initial_stamina - dodge_left.stamina_cost assert player.stamina == initial_stamina - dodge_left.stamina_cost
assert player.defense_locked_until > 0 messages = [call[0][0] for call in player.writer.write.call_args_list]
assert any("dodge the air" in msg.lower() for msg in messages)
@pytest.mark.asyncio @pytest.mark.asyncio
@ -334,52 +335,21 @@ async def test_switch_attack_sends_new_telegraph(
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_defense_rejected_while_locked(player, dodge_left): async def test_defense_blocks_for_timing_window(player, dodge_left):
"""Test defense rejected during recovery lock.""" """Test defense sleeps for timing_window_ms (commitment via blocking)."""
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() before = time.monotonic()
await combat_commands.do_defend(player, "", dodge_left) await combat_commands.do_defend(player, "", dodge_left)
elapsed = time.monotonic() - before
expected_lock = dodge_left.timing_window_ms / 1000.0 expected = dodge_left.timing_window_ms / 1000.0
# Lock should be roughly now + timing_window assert elapsed >= expected - 0.05
assert player.defense_locked_until >= before + expected_lock - 0.1
assert player.defense_locked_until <= before + expected_lock + 0.5
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_defense_in_combat_queues_on_encounter( async def test_defense_in_combat_queues_on_encounter(
player, target, punch_right, dodge_left player, target, punch_right, dodge_left
): ):
"""Test defense in combat queues on encounter AND applies lock.""" """Test defense in combat queues on encounter and shows move name."""
await combat_commands.do_attack(player, "Vegeta", punch_right) await combat_commands.do_attack(player, "Vegeta", punch_right)
await combat_commands.do_defend(target, "", dodge_left) await combat_commands.do_defend(target, "", dodge_left)
@ -387,7 +357,9 @@ async def test_defense_in_combat_queues_on_encounter(
encounter = get_encounter(target) encounter = get_encounter(target)
assert encounter is not None assert encounter is not None
assert encounter.pending_defense is dodge_left assert encounter.pending_defense is dodge_left
assert target.defense_locked_until > 0 # In combat: full move name (not "the air")
messages = [call[0][0] for call in target.writer.write.call_args_list]
assert any("dodge left" in msg.lower() for msg in messages)
@pytest.mark.asyncio @pytest.mark.asyncio