Events Guide¶

Events enable decoupled communication between systems in MAID. Instead of systems calling each other directly, they emit and subscribe to events. This guide covers creating events, subscribing to them, and best practices for event-driven architecture.

Event Basics¶

What Are Events?¶

Events are messages that describe something that happened in the game. They carry data about the occurrence and can be subscribed to by any interested system.

from dataclasses import dataclass
from uuid import UUID

from maid_engine.core.events import Event


@dataclass
class DamageDealtEvent(Event):
    """Emitted when damage is dealt to an entity."""

    source_id: UUID | None   # Who dealt the damage
    target_id: UUID          # Who received the damage
    damage: int              # Amount of damage
    damage_type: str         # Type: physical, fire, ice, etc.

The Event Base Class¶

All events inherit from Event, which provides:

@dataclass
class Event:
    event_type: str      # Set automatically to class name
    timestamp: datetime  # Set automatically to current time
    cancelled: bool      # Can be set to stop propagation

    def cancel(self) -> None:
        """Cancel this event."""
        self.cancelled = True

Creating Events¶

Simple Events¶

Events are dataclasses with fields for their data:

from dataclasses import dataclass
from uuid import UUID

from maid_engine.core.events import Event


@dataclass
class PlayerJoinedEvent(Event):
    """A player joined the game."""

    player_id: UUID
    player_name: str


@dataclass
class RoomEnterEvent(Event):
    """An entity entered a room."""

    entity_id: UUID
    room_id: UUID
    from_room_id: UUID | None = None

Events with Default Values¶

Use default values for optional fields:

@dataclass
class ItemDroppedEvent(Event):
    """An item was dropped."""

    entity_id: UUID
    item_id: UUID
    room_id: UUID
    quantity: int = 1
    intentional: bool = True  # vs dropped on death

Events with Complex Data¶

Events can contain structured data:

from dataclasses import dataclass, field

@dataclass
class CombatRoundEvent(Event):
    """A round of combat completed."""

    combatant_ids: list[UUID]
    room_id: UUID
    round_number: int
    actions_taken: dict[str, str] = field(default_factory=dict)

Emitting Events¶

Basic Emission¶

Emit events through the event bus:

# In a system
async def update(self, delta: float) -> None:
    event = DamageDealtEvent(
        source_id=attacker.id,
        target_id=target.id,
        damage=25,
        damage_type="physical",
    )
    await self.events.emit(event)

Emission from Commands¶

async def attack_command(ctx: CommandContext) -> bool:
    # ... calculate attack ...

    await ctx.world.events.emit(DamageDealtEvent(
        source_id=ctx.player_id,
        target_id=target_id,
        damage=damage,
        damage_type="physical",
    ))
    return True

Sync Emission¶

For synchronous contexts, queue events for later processing:

# Queue for later (doesn't await)
world.events.emit_sync(MyEvent(...))

# Process queued events later
count = await world.events.process_pending()

Subscribing to Events¶

Basic Subscription¶

Subscribe handlers to event types:

async def on_damage_dealt(event: DamageDealtEvent) -> None:
    print(f"Damage: {event.damage} to {event.target_id}")


# Subscribe
world.events.subscribe(DamageDealtEvent, on_damage_dealt)

In Systems¶

Subscribe in the system's startup method:

class LoggingSystem(System):
    async def startup(self) -> None:
        self.events.subscribe(DamageDealtEvent, self._on_damage)
        self.events.subscribe(EntityDeathEvent, self._on_death)

    async def _on_damage(self, event: DamageDealtEvent) -> None:
        # Log damage...
        pass

    async def _on_death(self, event: EntityDeathEvent) -> None:
        # Log death...
        pass

Sync Handlers¶

Synchronous handlers are automatically wrapped:

def sync_handler(event: DamageDealtEvent) -> None:
    # Sync code works fine
    print(f"Damage dealt: {event.damage}")


world.events.subscribe(DamageDealtEvent, sync_handler)

Event Priority¶

Handlers run in priority order. Higher priority runs first.

from maid_engine.core.events import EventPriority


# Runs first - can modify or cancel event
world.events.subscribe(
    DamageDealtEvent,
    damage_reduction_handler,
    priority=EventPriority.HIGHEST,
)

# Runs in the middle
world.events.subscribe(
    DamageDealtEvent,
    normal_handler,
    priority=EventPriority.NORMAL,
)

# Runs last - for logging/cleanup
world.events.subscribe(
    DamageDealtEvent,
    logging_handler,
    priority=EventPriority.LOWEST,
)

Priority levels:

Priority	Use Case
`HIGHEST`	Validation, blocking, modification
`HIGH`	Pre-processing
`NORMAL`	Standard handlers
`LOW`	Post-processing
`LOWEST`	Logging, cleanup

Canceling Events¶

Handlers can cancel events to prevent further processing:

async def invulnerability_handler(event: DamageDealtEvent) -> None:
    target = world.entities.get(event.target_id)
    if target and has_invulnerability(target):
        event.cancel()  # No more handlers will run


# Register with high priority to run first
world.events.subscribe(
    DamageDealtEvent,
    invulnerability_handler,
    priority=EventPriority.HIGHEST,
)

Check if canceled before processing:

async def damage_handler(event: DamageDealtEvent) -> None:
    if event.cancelled:
        return  # Optional - event bus already checks this

    # Apply damage...

One-Time Subscriptions¶

Subscribe to handle an event only once:

# Handler automatically unsubscribes after first invocation
world.events.subscribe(
    PlayerJoinedEvent,
    first_player_handler,
    once=True,
)

Unsubscribing¶

By Handler ID¶

# Subscribe returns a handler ID
handler_id = world.events.subscribe(DamageDealtEvent, my_handler)

# Later, unsubscribe
world.events.unsubscribe(handler_id)

All Handlers for Event Type¶

# Remove all handlers for an event type
count = world.events.unsubscribe_all(DamageDealtEvent)
print(f"Removed {count} handlers")

Built-in Events¶

Engine Events (maid-engine)¶

Event	Description
`TickEvent`	Emitted at start of each game tick
`EntityCreatedEvent`	Entity was created
`EntityDestroyedEvent`	Entity was destroyed
`ComponentAddedEvent`	Component added to entity
`ComponentRemovedEvent`	Component removed from entity
`PlayerConnectedEvent`	Player connected to server
`PlayerDisconnectedEvent`	Player disconnected
`PlayerCommandEvent`	After command is processed
`RoomEnterEvent`	Entity entered a room
`RoomLeaveEvent`	Entity left a room
`CustomEvent`	Generic event for ad-hoc use

Stdlib Events (maid-stdlib)¶

Event	Description
`CombatStartEvent`	Combat began
`CombatEndEvent`	Combat ended
`DamageDealtEvent`	Damage was dealt
`EntityDeathEvent`	Entity died
`ItemPickedUpEvent`	Item was picked up
`ItemDroppedEvent`	Item was dropped
`MessageEvent`	Communication message

Quest Events (maid-classic-rpg)¶

Event	Description
`QuestAvailableEvent`	Quest became available to a character
`QuestAcceptedEvent`	Character accepted a quest
`QuestObjectiveProgressEvent`	Progress made on a quest objective
`QuestObjectiveCompleteEvent`	A quest objective was completed
`QuestCompleteEvent`	All objectives complete, ready to turn in
`QuestTurnedInEvent`	Quest turned in and rewards granted
`QuestFailedEvent`	Quest failed
`QuestAbandonedEvent`	Character abandoned a quest
`QuestGeneratedEvent`	A procedurally generated quest was created
`QuestOfferedEvent`	A generated quest was offered to a character
`QuestChainEvent`	Follow-up quest chain trigger detected

NPC Autonomy Events (maid-classic-rpg)¶

Event	Description
`NPCActivityChangedEvent`	NPC's scheduled activity changed
`NPCGoalCreatedEvent`	New goal added to an NPC
`NPCGoalCompletedEvent`	NPC completed a goal
`NPCGoalFailedEvent`	NPC goal failed
`NPCActionEvent`	Autonomy selected an action for an NPC
`ActionCompletedEvent`	NPC action reached completion
`ActionFailedEvent`	NPC action failed
`NPCSocialInteractionEvent`	Two NPCs completed a social interaction
`NPCMoodChangedEvent`	NPC mood changed significantly
`NPCBarkEvent`	Ambient bark selected for an NPC
`StorySignalEvent`	Story signal detection produced a signal
`NarrativeReadyEvent`	Queued narrative text is ready
`TickBudgetExceededEvent`	Autonomy tick budget was exceeded
`ArchetypeCircuitBreakerEvent`	Repeated archetype failures triggered protection

Event Patterns¶

Request-Response Pattern¶

Use events for requests that need responses:

@dataclass
class HealRequestEvent(Event):
    """Request to heal an entity."""
    target_id: UUID
    amount: int
    source_id: UUID | None = None


@dataclass
class HealCompletedEvent(Event):
    """Heal was applied."""
    target_id: UUID
    actual_amount: int
    source_id: UUID | None = None


# Healing system handles requests
async def handle_heal_request(event: HealRequestEvent) -> None:
    entity = world.entities.get(event.target_id)
    if not entity:
        return

    health = entity.try_get(HealthComponent)
    if not health:
        return

    actual = health.heal(event.amount)

    # Emit completion event
    await world.events.emit(HealCompletedEvent(
        target_id=event.target_id,
        actual_amount=actual,
        source_id=event.source_id,
    ))

Event Chains¶

Events can trigger other events:

# DamageDealtEvent -> EntityDeathEvent -> LootDroppedEvent

async def check_death(event: DamageDealtEvent) -> None:
    target = world.entities.get(event.target_id)
    if not target:
        return

    health = target.try_get(HealthComponent)
    if health and not health.is_alive:
        await world.events.emit(EntityDeathEvent(
            entity_id=event.target_id,
            killer_id=event.source_id,
        ))


async def drop_loot(event: EntityDeathEvent) -> None:
    # Generate and drop loot
    for item in generate_loot(event.entity_id):
        await world.events.emit(LootDroppedEvent(
            entity_id=event.entity_id,
            item_id=item.id,
        ))

Broadcasting Pattern¶

Send messages to multiple targets. Session lookup goes through the SessionManager — there is no global get_session_for_entity helper. A SessionAware system is handed the manager by the engine (via set_sessions) once the network layer starts, and SessionManager.send_to_room already does the per-player get_by_player + send_line loop for you:

from typing import TYPE_CHECKING
from uuid import UUID

from maid_engine.core.ecs import System
from maid_engine.net.protocols import SessionAware

if TYPE_CHECKING:
    from maid_engine.core.world import World
    from maid_engine.net.session import SessionManager


@dataclass
class RoomMessageEvent(Event):
    """Message to all entities in a room."""
    room_id: UUID
    message: str
    exclude_ids: list[UUID] = field(default_factory=list)


class RoomBroadcaster(System, SessionAware):
    """Broadcasts RoomMessageEvent to the players in a room."""

    def __init__(self, world: "World") -> None:
        # System.__init__ requires the world; it wires up self.world,
        # self.entities, and self.events for you.
        super().__init__(world)
        self._sessions: "SessionManager | None" = None
        self._subscription: UUID | None = None

    def set_sessions(self, sessions: "SessionManager") -> None:
        # Engine injects the SessionManager after the network layer starts.
        self._sessions = sessions

    async def startup(self) -> None:
        # Subscribe when the system starts; keep the id so we can clean up.
        self._subscription = self.events.subscribe(
            RoomMessageEvent, self.on_room_message
        )

    async def shutdown(self) -> None:
        if self._subscription is not None:
            self.events.unsubscribe(self._subscription)
            self._subscription = None

    async def update(self, delta: float) -> None:
        # Event-driven system: no per-tick work.
        pass

    async def on_room_message(self, event: RoomMessageEvent) -> None:
        if self._sessions is None:
            return
        # entities_in_room() yields Entity objects; send_to_room wants their ids.
        # For a player entity, entity.id IS the player id get_by_player expects.
        room_entity_ids = {e.id for e in self.world.entities_in_room(event.room_id)}
        await self._sessions.send_to_room(
            room_entity_ids,
            event.message,
            exclude=set(event.exclude_ids),
        )

Best Practices¶

Keep Events Immutable¶

Don't modify event data in handlers:

# Bad - modifying event data
async def bad_handler(event: DamageDealtEvent) -> None:
    event.damage = event.damage * 2  # Don't do this!

# Good - emit a new event or modify the entity
async def good_handler(event: DamageDealtEvent) -> None:
    modified_damage = event.damage * 2
    # Apply modified damage to entity directly

Use Specific Event Types¶

Create specific events rather than generic ones:

# Good - specific events
@dataclass
class PlayerLeveledUpEvent(Event):
    player_id: UUID
    new_level: int

# Avoid - generic events are harder to work with
@dataclass
class GenericPlayerEvent(Event):
    player_id: UUID
    event_name: str  # "leveled_up", "died", etc.
    data: dict

Document Event Flow¶

Comment the expected event flow:

# Event Flow:
# 1. PlayerAttacksEvent (from command)
# 2. -> DamageCalculatedEvent (from combat system)
# 3. -> DamageDealtEvent (from combat system)
# 4. -> EntityDeathEvent (if target dies)
# 5. -> ExperienceGainedEvent (if target died)
# 6. -> LootDroppedEvent (if target died)

Handle Errors Gracefully¶

The event bus catches handler exceptions, but log them:

async def robust_handler(event: DamageDealtEvent) -> None:
    try:
        # Handle event...
        pass
    except Exception as e:
        logger.error(f"Error handling DamageDealtEvent: {e}")
        # Don't re-raise - other handlers should still run

Clean Up Subscriptions¶

Unsubscribe when systems shut down:

class MySystem(System):
    def __init__(self, world: World) -> None:
        super().__init__(world)
        self._handler_ids: list[UUID] = []

    async def startup(self) -> None:
        handler_id = self.events.subscribe(MyEvent, self._handler)
        self._handler_ids.append(handler_id)

    async def shutdown(self) -> None:
        for handler_id in self._handler_ids:
            self.events.unsubscribe(handler_id)
        self._handler_ids.clear()

Next Steps¶

Persistence Guide - Storing event-related data
Testing Guide - Testing event handlers
Reference: Events - Complete event reference