Skip to content

Combat System Tutorial - Part 4: Commands

In this part, you will create the player-facing commands that allow users to initiate and interact with the combat system.

Overview

We will create the following commands:

  • attack: Initiate combat against a target
  • kill: Alias for attack (common in MUDs)
  • consider: Preview potential damage against a target

Step 1: Create the Attack Command

The attack command is the primary way players engage in combat:

# src/maid_combat_system/commands/combat_commands.py
"""Combat commands for player interaction."""

from __future__ import annotations

from typing import TYPE_CHECKING

from maid_engine.commands.registry import AccessLevel, CommandContext
from maid_stdlib.components import DescriptionComponent, HealthComponent, PositionComponent

from ..components import AttackComponent, DefenseComponent
from ..events import AttackRequestEvent

if TYPE_CHECKING:
    from maid_engine.commands.registry import LayeredCommandRegistry


async def attack_command(ctx: CommandContext) -> bool:
    """Attack a target in the same room.

    Usage:
        attack <target>
        attack goblin
        attack 2.goblin  (attack the second goblin)

    Args:
        ctx: The command context containing session, player, args, etc.

    Returns:
        True if the command was handled.
    """
    # Check for target argument
    if not ctx.args:
        await ctx.session.send("Attack whom?")
        await ctx.session.send("Usage: attack <target>")
        return True

    target_name = ctx.args[0].lower()

    # Get the player entity
    player = ctx.world.entities.get(ctx.player_id)
    if not player:
        await ctx.session.send("Error: Could not find your character.")
        return False

    # Check if player can attack
    if not player.has(AttackComponent):
        await ctx.session.send("You are unable to attack.")
        return True

    # Get player's position
    player_pos = player.try_get(PositionComponent)
    if not player_pos:
        await ctx.session.send("You need to be somewhere to attack!")
        return True

    # Find the target
    target = await find_target_in_room(ctx, player_pos.room_id, target_name)
    if not target:
        await ctx.session.send(f"You don't see '{ctx.args[0]}' here.")
        return True

    # Check if target is self
    if target.id == ctx.player_id:
        await ctx.session.send("You can't attack yourself!")
        return True

    # Check if target has health (can be attacked)
    if not target.has(HealthComponent):
        await ctx.session.send("You cannot attack that.")
        return True

    # Check if target is already dead
    target_health = target.get(HealthComponent)
    if not target_health.is_alive:
        target_desc = target.get(DescriptionComponent)
        target_name_display = str(target_desc.name) if target_desc else "It"
        await ctx.session.send(f"{target_name_display} is already dead.")
        return True

    # Get names for messages
    player_desc = player.try_get(DescriptionComponent)
    target_desc = target.try_get(DescriptionComponent)
    player_name = str(player_desc.name) if player_desc else "You"
    target_name_display = str(target_desc.name) if target_desc else "the target"

    # Send attack initiation message
    await ctx.session.send(f"You attack {target_name_display}!")

    # Emit attack request event
    await ctx.world.events.emit(AttackRequestEvent(
        attacker_id=ctx.player_id,
        target_id=target.id,
        attack_type="melee",
    ))

    return True


async def consider_command(ctx: CommandContext) -> bool:
    """Evaluate a potential target's combat statistics.

    Usage:
        consider <target>
        consider goblin

    Args:
        ctx: The command context

    Returns:
        True if the command was handled.
    """
    if not ctx.args:
        await ctx.session.send("Consider whom?")
        await ctx.session.send("Usage: consider <target>")
        return True

    target_name = ctx.args[0].lower()

    # Get the player entity
    player = ctx.world.entities.get(ctx.player_id)
    if not player:
        await ctx.session.send("Error: Could not find your character.")
        return False

    player_pos = player.try_get(PositionComponent)
    if not player_pos:
        await ctx.session.send("You need to be somewhere!")
        return True

    # Find the target
    target = await find_target_in_room(ctx, player_pos.room_id, target_name)
    if not target:
        await ctx.session.send(f"You don't see '{ctx.args[0]}' here.")
        return True

    # Get target info
    target_desc = target.try_get(DescriptionComponent)
    target_name_display = str(target_desc.name) if target_desc else "the target"

    # Check if target can be fought
    if not target.has(HealthComponent):
        await ctx.session.send(f"{target_name_display} cannot be fought.")
        return True

    # Build the assessment
    lines = [f"You consider {target_name_display}:"]

    # Health assessment
    target_health = target.get(HealthComponent)
    if not target_health.is_alive:
        lines.append("  It is dead.")
    else:
        health_pct = target_health.percentage
        if health_pct >= 90:
            lines.append("  It is in excellent condition.")
        elif health_pct >= 70:
            lines.append("  It has some scratches.")
        elif health_pct >= 50:
            lines.append("  It is wounded.")
        elif health_pct >= 25:
            lines.append("  It is badly wounded.")
        else:
            lines.append("  It is near death.")

    # Attack assessment (if target has attack)
    target_attack = target.try_get(AttackComponent)
    if target_attack:
        attack_power = target_attack.attack_power
        if attack_power >= 30:
            lines.append("  It looks extremely dangerous!")
        elif attack_power >= 20:
            lines.append("  It looks quite dangerous.")
        elif attack_power >= 10:
            lines.append("  It looks moderately threatening.")
        else:
            lines.append("  It doesn't look very dangerous.")

    # Defense assessment (if target has defense)
    target_defense = target.try_get(DefenseComponent)
    if target_defense:
        defense = target_defense.defense
        if defense >= 20:
            lines.append("  It appears heavily armored.")
        elif defense >= 10:
            lines.append("  It appears well-protected.")
        elif defense >= 5:
            lines.append("  It has some protection.")
        else:
            lines.append("  It appears lightly armored.")

    # Damage preview (if player has attack)
    from ..systems import CombatSystem
    combat_system = ctx.world.systems.get(CombatSystem)
    if combat_system:
        preview = combat_system.get_damage_preview(ctx.player_id, target.id)
        if preview:
            lines.append(f"  Estimated damage: {preview['min_damage']}-{preview['max_damage']}")
            lines.append(f"  Hit chance: {preview['hit_chance']}%")

    # Send all lines
    for line in lines:
        await ctx.session.send(line)

    return True


async def flee_command(ctx: CommandContext) -> bool:
    """Attempt to flee from combat.

    This is a placeholder for future combat state management.

    Usage:
        flee

    Args:
        ctx: The command context

    Returns:
        True if the command was handled.
    """
    await ctx.session.send("You attempt to flee!")
    await ctx.session.send("(Combat state management not yet implemented)")
    return True


async def find_target_in_room(
    ctx: CommandContext,
    room_id,
    target_name: str,
):
    """Find an entity by name in a room.

    Supports targeting syntax:
    - "goblin" - first goblin
    - "2.goblin" - second goblin
    - "goblin 2" - second goblin (alternate syntax)

    Args:
        ctx: Command context with world reference
        room_id: UUID of the room to search
        target_name: Name or keyword to search for

    Returns:
        The matching entity, or None if not found.
    """
    # Parse index prefix (e.g., "2.goblin")
    index = 1
    name = target_name

    if "." in target_name:
        parts = target_name.split(".", 1)
        if parts[0].isdigit():
            index = int(parts[0])
            name = parts[1]

    # Find all matching entities in the room
    matches = []
    for entity in ctx.world.entities.with_components(PositionComponent, DescriptionComponent):
        pos = entity.get(PositionComponent)
        if pos.room_id != room_id:
            continue

        desc = entity.get(DescriptionComponent)
        entity_name = str(desc.name).lower()

        # Check if name matches
        if name in entity_name:
            matches.append(entity)
        # Check keywords
        elif desc.matches_keyword(name):
            matches.append(entity)

    # Return the indexed match
    if index <= len(matches):
        return matches[index - 1]

    return None


def register_commands(registry: LayeredCommandRegistry, pack_name: str) -> None:
    """Register all combat commands.

    Args:
        registry: The command registry to register with
        pack_name: Name of this content pack
    """
    # Attack command
    registry.register(
        name="attack",
        handler=attack_command,
        pack_name=pack_name,
        aliases=["kill", "k", "hit"],
        category="combat",
        description="Attack a target in the same room",
        usage="attack <target>",
        access_level=AccessLevel.PLAYER,
        metadata={
            "requires_target": True,
            "initiates_combat": True,
        },
    )

    # Consider command
    registry.register(
        name="consider",
        handler=consider_command,
        pack_name=pack_name,
        aliases=["con", "assess"],
        category="combat",
        description="Evaluate a potential target",
        usage="consider <target>",
        access_level=AccessLevel.PLAYER,
        metadata={
            "requires_target": True,
        },
    )

    # Flee command
    registry.register(
        name="flee",
        handler=flee_command,
        pack_name=pack_name,
        aliases=["escape", "run"],
        category="combat",
        description="Attempt to flee from combat",
        usage="flee",
        access_level=AccessLevel.PLAYER,
        metadata={
            "combat_only": True,
        },
    )

Update the commands init file:

# src/maid_combat_system/commands/__init__.py
"""Combat commands.

Commands for player combat interaction:
- attack: Initiate combat against a target
- consider: Evaluate a target's combat stats
- flee: Attempt to escape from combat
"""

from .combat_commands import (
    attack_command,
    consider_command,
    flee_command,
    register_commands,
)

__all__ = [
    "attack_command",
    "consider_command",
    "flee_command",
    "register_commands",
]

Step 2: Create Combat Message Handlers

To provide feedback to players during combat, create event handlers that send messages:

# src/maid_combat_system/systems/combat_messages.py
"""Combat message handlers for player feedback."""

from __future__ import annotations

from typing import TYPE_CHECKING, ClassVar

from maid_engine.core.ecs import System
from maid_stdlib.components import DescriptionComponent

from ..events import CombatMissEvent, DamageDealtEvent, EntityDeathEvent

if TYPE_CHECKING:
    from maid_engine.core.world import World


class CombatMessageSystem(System):
    """System that sends combat messages to players.

    This system listens for combat events and sends appropriate
    messages to the involved players.
    """

    priority: ClassVar[int] = 100  # Run after combat processing

    # Message templates
    MESSAGES = {
        "hit": "You hit {target} for {damage} damage!",
        "hit_critical": "CRITICAL! You hit {target} for {damage} damage!",
        "hit_by": "{attacker} hits you for {damage} damage!",
        "hit_by_critical": "CRITICAL! {attacker} hits you for {damage} damage!",
        "miss": "You miss {target}!",
        "miss_evade": "{target} evades your attack!",
        "missed_by": "{attacker} misses you!",
        "evaded": "You evade {attacker}'s attack!",
        "death": "You have been slain by {killer}!",
        "kill": "You have slain {target}!",
        "witness_death": "{target} has been slain by {killer}!",
    }

    def __init__(self, world: World, session_manager=None):
        """Initialize the combat message system.

        Args:
            world: The game world
            session_manager: Optional session manager for sending messages
        """
        super().__init__(world)
        self._session_manager = session_manager

    async def startup(self) -> None:
        """Subscribe to combat events."""
        self.events.subscribe(DamageDealtEvent, self._on_damage_dealt)
        self.events.subscribe(CombatMissEvent, self._on_combat_miss)
        self.events.subscribe(EntityDeathEvent, self._on_entity_death)

    async def update(self, delta: float) -> None:
        """No per-tick processing needed."""
        pass

    async def _on_damage_dealt(self, event: DamageDealtEvent) -> None:
        """Handle damage dealt event by sending messages."""
        # Get entity names
        attacker_name = self._get_entity_name(event.source_id) if event.source_id else "Something"
        target_name = self._get_entity_name(event.target_id)

        # Determine message templates based on critical
        if event.was_critical:
            hit_msg = self.MESSAGES["hit_critical"]
            hit_by_msg = self.MESSAGES["hit_by_critical"]
        else:
            hit_msg = self.MESSAGES["hit"]
            hit_by_msg = self.MESSAGES["hit_by"]

        # Send message to attacker
        if event.source_id:
            attacker_msg = hit_msg.format(
                target=target_name,
                damage=event.damage,
            )
            await self._send_to_entity(event.source_id, attacker_msg)

        # Send message to target
        target_msg = hit_by_msg.format(
            attacker=attacker_name,
            damage=event.damage,
        )
        await self._send_to_entity(event.target_id, target_msg)

    async def _on_combat_miss(self, event: CombatMissEvent) -> None:
        """Handle combat miss event by sending messages."""
        attacker_name = self._get_entity_name(event.attacker_id)
        target_name = self._get_entity_name(event.target_id)

        if event.reason == "evaded":
            # Target evaded
            attacker_msg = self.MESSAGES["miss_evade"].format(target=target_name)
            target_msg = self.MESSAGES["evaded"].format(attacker=attacker_name)
        else:
            # Attacker missed
            attacker_msg = self.MESSAGES["miss"].format(target=target_name)
            target_msg = self.MESSAGES["missed_by"].format(attacker=attacker_name)

        await self._send_to_entity(event.attacker_id, attacker_msg)
        await self._send_to_entity(event.target_id, target_msg)

    async def _on_entity_death(self, event: EntityDeathEvent) -> None:
        """Handle entity death event by sending messages."""
        target_name = self._get_entity_name(event.entity_id)
        killer_name = self._get_entity_name(event.killer_id) if event.killer_id else "something"

        # Message to the killed entity
        death_msg = self.MESSAGES["death"].format(killer=killer_name)
        await self._send_to_entity(event.entity_id, death_msg)

        # Message to the killer
        if event.killer_id:
            kill_msg = self.MESSAGES["kill"].format(target=target_name)
            await self._send_to_entity(event.killer_id, kill_msg)

    def _get_entity_name(self, entity_id) -> str:
        """Get the display name for an entity.

        Args:
            entity_id: UUID of the entity

        Returns:
            The entity's name or a default string.
        """
        if not entity_id:
            return "something"

        entity = self.entities.get(entity_id)
        if not entity:
            return "something"

        desc = entity.try_get(DescriptionComponent)
        if desc:
            return str(desc.name)

        return "something"

    async def _send_to_entity(self, entity_id, message: str) -> None:
        """Send a message to an entity's session.

        Args:
            entity_id: UUID of the entity to message
            message: The message to send

        Note:
            This requires a session manager to be configured.
            In a full implementation, this would look up the
            entity's active session and send the message.
        """
        # This is a placeholder - in a full implementation,
        # you would look up the player's session and send the message
        if self._session_manager:
            session = self._session_manager.get_session_for_entity(entity_id)
            if session:
                await session.send(message)

Step 3: Testing Commands

Create tests for the combat commands:

# tests/test_commands.py
"""Tests for combat commands."""

import pytest
from uuid import uuid4

from maid_engine.commands.registry import CommandContext
from maid_stdlib.components import DescriptionComponent, HealthComponent, PositionComponent

from maid_combat_system.commands import attack_command, consider_command
from maid_combat_system.components import AttackComponent, DefenseComponent
from maid_combat_system.events import AttackRequestEvent


@pytest.fixture
def command_context(world, player_entity, mock_session):
    """Create a command context for testing."""
    return CommandContext(
        session=mock_session,
        player_id=player_entity.id,
        command="attack",
        args=[],
        raw_input="attack",
        world=world,
    )


@pytest.fixture
def attackable_player(player_entity):
    """Add attack component to player."""
    player_entity.add(AttackComponent(attack_power=10))
    return player_entity


class TestAttackCommand:
    """Tests for the attack command."""

    @pytest.mark.asyncio
    async def test_attack_no_args_shows_usage(self, command_context):
        """Test attack without target shows usage."""
        result = await attack_command(command_context)

        assert result is True
        assert "attack whom" in command_context.session.messages[0].lower()

    @pytest.mark.asyncio
    async def test_attack_target_not_found(
        self, command_context, attackable_player
    ):
        """Test attack on non-existent target."""
        command_context.args = ["dragon"]

        result = await attack_command(command_context)

        assert result is True
        assert "don't see" in command_context.session.messages[0].lower()

    @pytest.mark.asyncio
    async def test_attack_valid_target(
        self,
        command_context,
        attackable_player,
        enemy_entity
    ):
        """Test attack on valid target emits event."""
        command_context.args = ["goblin"]
        events_received = []

        async def capture_event(event: AttackRequestEvent):
            events_received.append(event)

        command_context.world.events.subscribe(AttackRequestEvent, capture_event)

        result = await attack_command(command_context)

        assert result is True
        assert "attack" in command_context.session.messages[0].lower()
        assert len(events_received) == 1
        assert events_received[0].attacker_id == attackable_player.id
        assert events_received[0].target_id == enemy_entity.id

    @pytest.mark.asyncio
    async def test_attack_self_prevented(
        self, command_context, attackable_player
    ):
        """Test that attacking yourself is prevented."""
        # Get player's name from description
        desc = attackable_player.get(DescriptionComponent)
        command_context.args = [desc.name.lower()]

        result = await attack_command(command_context)

        assert result is True
        assert "yourself" in command_context.session.messages[0].lower()

    @pytest.mark.asyncio
    async def test_attack_dead_target(
        self,
        command_context,
        attackable_player,
        enemy_entity
    ):
        """Test attack on dead target fails."""
        # Kill the enemy
        health = enemy_entity.get(HealthComponent)
        health.current = 0

        command_context.args = ["goblin"]

        result = await attack_command(command_context)

        assert result is True
        assert "dead" in command_context.session.messages[0].lower()

    @pytest.mark.asyncio
    async def test_attack_different_room(
        self,
        command_context,
        attackable_player,
        enemy_entity
    ):
        """Test attack on target in different room fails."""
        # Move enemy to different room
        enemy_pos = enemy_entity.get(PositionComponent)
        enemy_pos.room_id = uuid4()

        command_context.args = ["goblin"]

        result = await attack_command(command_context)

        assert result is True
        assert "don't see" in command_context.session.messages[0].lower()

    @pytest.mark.asyncio
    async def test_attack_without_attack_component(
        self, command_context, player_entity, enemy_entity
    ):
        """Test player without AttackComponent cannot attack."""
        # Player doesn't have AttackComponent (not added)
        command_context.args = ["goblin"]

        result = await attack_command(command_context)

        assert result is True
        assert "unable to attack" in command_context.session.messages[0].lower()


class TestConsiderCommand:
    """Tests for the consider command."""

    @pytest.mark.asyncio
    async def test_consider_no_args_shows_usage(self, command_context):
        """Test consider without target shows usage."""
        command_context.command = "consider"
        command_context.raw_input = "consider"

        result = await consider_command(command_context)

        assert result is True
        assert "consider whom" in command_context.session.messages[0].lower()

    @pytest.mark.asyncio
    async def test_consider_shows_health_status(
        self,
        command_context,
        attackable_player,
        enemy_entity
    ):
        """Test consider shows target health status."""
        command_context.args = ["goblin"]

        result = await consider_command(command_context)

        assert result is True
        messages = " ".join(command_context.session.messages).lower()
        assert "goblin" in messages
        assert "condition" in messages or "wounded" in messages

    @pytest.mark.asyncio
    async def test_consider_shows_danger_level(
        self,
        command_context,
        attackable_player,
        enemy_entity
    ):
        """Test consider shows danger assessment."""
        # Give enemy high attack
        enemy_entity.add(AttackComponent(attack_power=30))
        command_context.args = ["goblin"]

        result = await consider_command(command_context)

        assert result is True
        messages = " ".join(command_context.session.messages).lower()
        assert "dangerous" in messages

    @pytest.mark.asyncio
    async def test_consider_shows_armor(
        self,
        command_context,
        attackable_player,
        enemy_entity
    ):
        """Test consider shows armor level."""
        enemy_entity.add(DefenseComponent(defense=20))
        command_context.args = ["goblin"]

        result = await consider_command(command_context)

        assert result is True
        messages = " ".join(command_context.session.messages).lower()
        assert "armor" in messages or "protected" in messages


class TestTargetSelection:
    """Tests for target selection syntax."""

    @pytest.mark.asyncio
    async def test_indexed_target_syntax(
        self,
        world,
        room_id,
        command_context,
        attackable_player
    ):
        """Test '2.goblin' syntax for second goblin."""
        # Create two goblins
        goblin1 = world.entities.create()
        goblin1.add(PositionComponent(room_id=room_id))
        goblin1.add(HealthComponent(current=30, maximum=30))
        goblin1.add(DescriptionComponent(name="Goblin", short_desc="First goblin"))

        goblin2 = world.entities.create()
        goblin2.add(PositionComponent(room_id=room_id))
        goblin2.add(HealthComponent(current=40, maximum=40))
        goblin2.add(DescriptionComponent(name="Goblin", short_desc="Second goblin"))

        events_received = []

        async def capture_event(event: AttackRequestEvent):
            events_received.append(event)

        world.events.subscribe(AttackRequestEvent, capture_event)

        # Attack second goblin
        command_context.args = ["2.goblin"]

        result = await attack_command(command_context)

        assert result is True
        assert len(events_received) == 1
        assert events_received[0].target_id == goblin2.id

Run the command tests:

uv run pytest tests/test_commands.py -v

Step 4: Command Best Practices

Input Validation

Always validate user input thoroughly:

async def attack_command(ctx: CommandContext) -> bool:
    # 1. Check required arguments
    if not ctx.args:
        await ctx.session.send("Attack whom?")
        return True

    # 2. Validate player state
    player = ctx.world.entities.get(ctx.player_id)
    if not player:
        return False  # Internal error

    # 3. Check player capabilities
    if not player.has(AttackComponent):
        await ctx.session.send("You are unable to attack.")
        return True

    # 4. Validate target exists
    target = find_target(...)
    if not target:
        await ctx.session.send("You don't see that here.")
        return True

    # 5. Validate target state
    if not target.has(HealthComponent):
        await ctx.session.send("You cannot attack that.")
        return True

Clear Feedback

Provide clear, informative messages:

# Good: Specific feedback
await ctx.session.send(f"You don't see '{target_name}' here.")
await ctx.session.send("The goblin is already dead.")
await ctx.session.send("You are too exhausted to attack.")

# Bad: Vague feedback
await ctx.session.send("Invalid target.")
await ctx.session.send("Cannot attack.")
await ctx.session.send("Error.")

Event-Based Actions

Use events for actions instead of direct calls:

# Good: Emit event for system to handle
await ctx.world.events.emit(AttackRequestEvent(
    attacker_id=ctx.player_id,
    target_id=target.id,
))

# Avoid: Direct system calls
combat_system = ctx.world.systems.get(CombatSystem)
await combat_system.process_attack(ctx.player_id, target.id)

Summary

In this part, you:

  1. Created the attack command for initiating combat
  2. Created the consider command for target assessment
  3. Implemented target finding with index syntax support
  4. Created a combat message system for feedback
  5. Added comprehensive tests for commands

Next Steps

In Part 5: Testing, you will learn:

  • Integration testing strategies
  • End-to-end combat scenario testing
  • Using the testing framework effectively

Continue to Part 5: Testing