maid-classic-rpg¶

The maid-classic-rpg package provides a complete classic MUD gameplay experience, building on top of maid-engine and maid-stdlib.

Installation¶

pip install maid-classic-rpg

Module Overview¶

Systems (`maid_classic_rpg.systems`)¶

Game systems implementing classic MUD mechanics:

System Module	Description
`combat`	Turn-based tactical combat
`magic`	Spellcasting and magical effects
`crafting`	Item creation and recipes
`economy`	Shops, trading, banking
`quests`	Quest objectives and rewards
`quests.generation`	Automated quest generation integration
`social`	Guilds, factions, achievements
`pvp`	Arenas, duels, rankings
`world`	Time, weather, ecosystems
`npc`	NPC behavior and AI dialogue
`npc.autonomy`	NPC autonomy (needs, goals, schedules, social, barks)
`skills`	Character skills and progression

Commands (`maid_classic_rpg.commands`)¶

Game-specific commands:

Category	Commands
Combat	`attack`, `defend`, `flee`, `cast`, `use`
Economy	`buy`, `sell`, `trade`, `bank`, `auction`
Social	`guild`, `faction`, `party`, `tell`, `shout`
Dialogue	`talk`, `ask`, `greet`, `conversations`
Crafting	`craft`, `gather`, `recipe`
Quests	`quest`, `objectives`, `abandon`
PvP	`duel`, `arena`, `rankings`
Debug	`debug_brain` — inspect NPC autonomy state (admin only, no `@` prefix)

Models (`maid_classic_rpg.models`)¶

Data models for game entities:

Module	Description
`combat`	Combat-related data structures
`magic`	Spell definitions and effects
`economy`	Shop, trade, and currency models
`crafting`	Recipes and materials
`social`	Guild and faction structures
`npc`	NPC configuration and dialogue
`skills`	Skill definitions and trees
`world`	Time, weather, terrain

Events (`maid_classic_rpg.events`)¶

Game-specific events:

Module	Description
`combat`	Combat-related events
`economy`	Economy/trade events
`quests`	Quest progression events
`autonomy`	NPC autonomy events (need changes, goal updates, schedule transitions, social interactions)

Components (`maid_classic_rpg.components`)¶

Game-specific ECS components:

Component	Description
`CharacterStatsComponent`	D&D-style ability scores (STR, DEX, etc.)
`CharacterInfoComponent`	Character class, race, and background
`SkillsComponent`	Character skills and levels
`AbilitiesComponent`	Known abilities
`WeaponComponent` / `ArmorComponent`	Equipped weapon and armor properties

Systems¶

Combat System¶

Turn-based tactical combat with positioning and abilities.

from maid_classic_rpg.systems.combat import MeleeCombatSystem

# Combat system handles:
# - Initiative and turn order
# - Attack resolution
# - Damage calculation
# - Status effects
# - Death and respawn

combat ¶

Combat systems for melee and ranged combat.

BaseCombatSystem ¶

Bases: System, ABC

Abstract base class for combat systems.

Provides structure for combat flow: engagement, turn processing, damage calculation, and disengagement.

start_combat `abstractmethod` `async` ¶

start_combat(attacker_id: UUID, defender_id: UUID) -> bool

Initiate combat between two entities.

end_combat `abstractmethod` `async` ¶

end_combat(entity_id: UUID, reason: str = 'normal') -> bool

End combat for an entity.

process_attack `abstractmethod` `async` ¶

process_attack(
    attacker: Entity,
    defender: Entity,
    weapon_id: UUID | None = None,
) -> tuple[bool, int]

Process an attack action.

calculate_damage `abstractmethod` `async` ¶

calculate_damage(
    attacker: Entity, defender: Entity, base_damage: int
) -> int

Calculate final damage after modifiers.

is_in_combat ¶

is_in_combat(entity_id: UUID) -> bool

Check if an entity is currently in combat.

BodyDamageResult ¶

BodyDamageResult(
    actual_damage: int = 0,
    part_destroyed: bool = False,
    effects: list[StatusEffectType] | None = None,
    vital_hit: bool = False,
)

Result of applying damage to a body part.

is_fatal `property` ¶

is_fatal: bool

Check if this damage was fatal (vital part destroyed).

BodySystem ¶

BodySystem(world: World)

Bases: System

System for managing body parts and location-based damage.

Handles: - Applying damage to specific body parts - Tracking body part health - Determining effects when parts are damaged/destroyed - Body part regeneration

update `async` ¶

update(delta: float) -> None

Update tick - handle regeneration and effect ticks.

apply_damage_to_location ¶

apply_damage_to_location(
    body: BodyComponent,
    location: str,
    damage: int,
    armor_value: int = 0,
) -> BodyDamageResult

Apply damage to a specific body location.

Parameters:

Name	Type	Description	Default
`body`	`BodyComponent`	The BodyComponent to damage	required
`location`	`str`	Name of the body part	required
`damage`	`int`	Amount of damage to apply	required
`armor_value`	`int`	Additional armor to apply (on top of part's armor)	`0`

Returns:

Type	Description
`BodyDamageResult`	BodyDamageResult with damage details

heal_location ¶

heal_location(
    body: BodyComponent, location: str, amount: int
) -> int

Heal a specific body location.

Parameters:

Name	Type	Description	Default
`body`	`BodyComponent`	The BodyComponent to heal	required
`location`	`str`	Name of the body part	required
`amount`	`int`	Amount of healing	required

Returns:

Type	Description
`int`	Actual amount healed

restore_part ¶

restore_part(body: BodyComponent, location: str) -> bool

Restore a destroyed body part to minimal health.

Parameters:

Name	Type	Description	Default
`body`	`BodyComponent`	The BodyComponent	required
`location`	`str`	Name of the body part to restore	required

Returns:

Type	Description
`bool`	True if part was restored

get_total_health ¶

get_total_health(body: BodyComponent) -> tuple[int, int]

Get total health across all body parts.

Parameters:

Name	Type	Description	Default
`body`	`BodyComponent`	The BodyComponent	required

Returns:

Type	Description
`tuple[int, int]`	Tuple of (current_health, max_health)

get_health_percentage ¶

get_health_percentage(body: BodyComponent) -> float

Get overall health percentage.

Parameters:

Name	Type	Description	Default
`body`	`BodyComponent`	The BodyComponent	required

Returns:

Type	Description
`float`	Health percentage (0-100)

is_vital_destroyed ¶

is_vital_destroyed(body: BodyComponent) -> bool

Check if any vital body part is destroyed.

Parameters:

Name	Type	Description	Default
`body`	`BodyComponent`	The BodyComponent	required

Returns:

Type	Description
`bool`	True if a vital part is destroyed (entity should be dead)

get_status_summary ¶

get_status_summary(body: BodyComponent) -> dict[str, str]

Get a status summary of all body parts.

Parameters:

Name	Type	Description	Default
`body`	`BodyComponent`	The BodyComponent	required

Returns:

Type	Description
`dict[str, str]`	Dict mapping part name to status string

add_part_effect ¶

add_part_effect(
    body: BodyComponent, location: str, effect: str
) -> bool

Add a status effect to a body part.

Parameters:

Name	Type	Description	Default
`body`	`BodyComponent`	The BodyComponent	required
`location`	`str`	Name of the body part	required
`effect`	`str`	Effect to add	required

Returns:

Type	Description
`bool`	True if effect was added

remove_part_effect ¶

remove_part_effect(
    body: BodyComponent, location: str, effect: str
) -> bool

Remove a status effect from a body part.

Parameters:

Name	Type	Description	Default
`body`	`BodyComponent`	The BodyComponent	required
`location`	`str`	Name of the body part	required
`effect`	`str`	Effect to remove	required

Returns:

Type	Description
`bool`	True if effect was removed

clear_part_effects ¶

clear_part_effects(
    body: BodyComponent, location: str
) -> int

Clear all status effects from a body part.

Parameters:

Name	Type	Description	Default
`body`	`BodyComponent`	The BodyComponent	required
`location`	`str`	Name of the body part	required

Returns:

Type	Description
`int`	Number of effects cleared

DamageSystem ¶

DamageSystem(world: World)

Bases: System

System for applying damage from combat to characters.

Handles: - Applying attack results to character vitals - Location-based damage to body parts - Status effect application - Death determination

startup `async` ¶

startup() -> None

Subscribe to entity destruction events for cache cleanup.

update `async` ¶

update(delta: float) -> None

Update tick - check for deaths from damage-over-time effects.

DOT damage is applied by StatusEffectManager via apply_direct_damage(). This method checks for entities that died from DOT since last tick.

shutdown `async` ¶

shutdown() -> None

Clean up body component cache and unsubscribe from events.

remove_body ¶

remove_body(character_id: UUID) -> None

Remove a character's body component from the cache.

Call when an entity is destroyed to prevent unbounded cache growth.

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	The character UUID to remove.	required

get_or_create_body ¶

get_or_create_body(character: Character) -> BodyComponent

Get or create a body component for a character.

Parameters:

Name	Type	Description	Default
`character`	`Character`	The character	required

Returns:

Type	Description
`BodyComponent`	BodyComponent for the character

apply_attack_result `async` ¶

apply_attack_result(
    result: AttackResult, target: Character
) -> bool

Apply an attack result to a target character.

Parameters:

Name	Type	Description	Default
`result`	`AttackResult`	The attack result to apply	required
`target`	`Character`	The target character	required

Returns:

Type	Description
`bool`	True if target was killed

apply_direct_damage ¶

apply_direct_damage(
    target: Character,
    damage: int,
    _damage_type: DamageType = TRUE,
    location: str | None = None,
    source_id: UUID | None = None,
) -> int

Apply direct damage to a target (no attack roll).

Parameters:

Name	Type	Description	Default
`target`	`Character`	The target character	required
`damage`	`int`	Amount of damage	required
`_damage_type`	`DamageType`	Type of damage (for future resistance calculations)	`TRUE`
`location`	`str \| None`	Optional body location	`None`
`source_id`	`UUID \| None`	Optional source entity ID for event emission	`None`

Returns:

Type	Description
`int`	Actual damage dealt

heal_character ¶

heal_character(
    target: Character,
    amount: int,
    location: str | None = None,
) -> int

Heal a character.

Parameters:

Name	Type	Description	Default
`target`	`Character`	The character to heal	required
`amount`	`int`	Amount to heal	required
`location`	`str \| None`	Optional specific body part to heal	`None`

Returns:

Type	Description
`int`	Actual amount healed

is_alive ¶

is_alive(character: Character) -> bool

Check if a character is alive.

Parameters:

Name	Type	Description	Default
`character`	`Character`	The character to check	required

Returns:

Type	Description
`bool`	True if character is alive

get_health_status ¶

get_health_status(character: Character) -> str

Get a description of character's health status.

Parameters:

Name	Type	Description	Default
`character`	`Character`	The character	required

Returns:

Type	Description
`str`	Health status string

get_body_status ¶

get_body_status(character: Character) -> dict[str, str]

Get detailed body part status for a character.

Parameters:

Name	Type	Description	Default
`character`	`Character`	The character	required

Returns:

Type	Description
`dict[str, str]`	Dict mapping body part to status

StatusEffectManager ¶

StatusEffectManager(world: World)

Bases: System

System for managing status effects on entities.

Handles: - Adding/removing status effects - Effect duration tracking - Effect tick processing (damage over time, etc.) - Effect stacking rules

shutdown `async` ¶

shutdown() -> None

Clean up effect tracking to prevent memory leak.

update `async` ¶

update(delta: float) -> None

Process effect ticks and expirations.

add_effect ¶

add_effect(
    target: Character,
    effect_type: StatusEffectType,
    source_id: UUID | None = None,
    duration: float | None = None,
    intensity: float = 1.0,
    stacks: int = 1,
) -> StatusEffect

Add a status effect to a character.

Parameters:

Name	Type	Description	Default
`target`	`Character`	Character to affect	required
`effect_type`	`StatusEffectType`	Type of effect	required
`source_id`	`UUID \| None`	Entity that caused the effect	`None`
`duration`	`float \| None`	Override default duration	`None`
`intensity`	`float`	Effect intensity multiplier	`1.0`
`stacks`	`int`	Number of stacks to apply	`1`

Returns:

Type	Description
`StatusEffect`	The created or updated StatusEffect

remove_effect ¶

remove_effect(
    target: Character, effect_type: StatusEffectType
) -> bool

Remove a status effect from a character.

Parameters:

Name	Type	Description	Default
`target`	`Character`	Character to affect	required
`effect_type`	`StatusEffectType`	Type of effect to remove	required

Returns:

Type	Description
`bool`	True if effect was found and removed

remove_all_effects ¶

remove_all_effects(target: Character) -> int

Remove all effects from a character.

Parameters:

Name	Type	Description	Default
`target`	`Character`	Character to clear	required

Returns:

Type	Description
`int`	Number of effects removed

get_effects ¶

get_effects(target: Character) -> list[StatusEffect]

Get all active effects on a character.

Parameters:

Name	Type	Description	Default
`target`	`Character`	Character to check	required

Returns:

Type	Description
`list[StatusEffect]`	List of active effects (copy)

has_effect ¶

has_effect(
    target: Character, effect_type: StatusEffectType
) -> bool

Check if character has a specific effect.

Parameters:

Name	Type	Description	Default
`target`	`Character`	Character to check	required
`effect_type`	`StatusEffectType`	Effect type to look for	required

Returns:

Type	Description
`bool`	True if effect is active

get_effect_stacks ¶

get_effect_stacks(
    target: Character, effect_type: StatusEffectType
) -> int

Get the number of stacks of an effect.

Parameters:

Name	Type	Description	Default
`target`	`Character`	Character to check	required
`effect_type`	`StatusEffectType`	Effect type	required

Returns:

Type	Description
`int`	Number of stacks (0 if not present)

get_effect_modifier ¶

get_effect_modifier(
    target: Character, effect_type: StatusEffectType
) -> float

Get the modifier value from an active effect.

Parameters:

Name	Type	Description	Default
`target`	`Character`	Character to check	required
`effect_type`	`StatusEffectType`	Effect type	required

Returns:

Type	Description
`float`	Effect intensity * stacks (0.0 if not present)

MeleeCombatSystem ¶

MeleeCombatSystem(world: World)

Bases: System

System for resolving melee combat attacks.

Handles attack resolution including: - Accuracy calculation with all modifiers - Hit location selection - Damage calculation with modifiers - Critical hits - Dodge/block/parry mechanics

update `async` ¶

update(delta: float) -> None

Update tick - melee combat is resolved immediately, not per-tick.

resolve_attack `async` ¶

resolve_attack(
    attacker: Character,
    target: Character,
    weapon: Item | None,
    attacker_position: CombatPosition | None = None,
    target_position: CombatPosition | None = None,
    attacker_zone: CombatZone = MELEE,
    target_zone: CombatZone = MELEE,
    attacker_size: CreatureSize = MEDIUM,
    target_size: CreatureSize = MEDIUM,
    weapon_size: CreatureSize | None = None,
    _other_attackers: list[CombatPosition] | None = None,
    aimed_location: str | None = None,
    attacker_grid_state: CombatantGridState | None = None,
    target_grid_state: CombatantGridState | None = None,
) -> AttackResult

Resolve a melee attack.

Parameters:

Name	Type	Description	Default
`attacker`	`Character`	Character making the attack	required
`target`	`Character`	Character being attacked	required
`weapon`	`Item \| None`	Weapon being used (None for unarmed)	required
`attacker_position`	`CombatPosition \| None`	Tactical position of attacker (legacy 3x3 grid)	`None`
`target_position`	`CombatPosition \| None`	Tactical position of target (legacy 3x3 grid)	`None`
`attacker_zone`	`CombatZone`	Combat zone of attacker	`MELEE`
`target_zone`	`CombatZone`	Combat zone of target	`MELEE`
`attacker_size`	`CreatureSize`	Size of the attacker creature	`MEDIUM`
`target_size`	`CreatureSize`	Size of the target creature	`MEDIUM`
`weapon_size`	`CreatureSize \| None`	Absolute size of the weapon (if different from wielder)	`None`
`_other_attackers`	`list[CombatPosition] \| None`	Positions of other attackers (for flanking, legacy)	`None`
`aimed_location`	`str \| None`	Specific body part to target (optional)	`None`
`attacker_grid_state`	`CombatantGridState \| None`	New variable-grid position/facing state	`None`
`target_grid_state`	`CombatantGridState \| None`	New variable-grid position/facing state for target	`None`

Returns:

Type	Description
`AttackResult`	AttackResult with full attack outcome

RangedCombatSystem ¶

RangedCombatSystem(world: World)

Bases: System

System for resolving ranged combat attacks.

Handles attacks across room boundaries including: - Line of sight checks - Range/distance modifiers - Ranged weapon mechanics - Cover system - Ammunition tracking - Reload mechanics - Moving target penalties

Note: Facing-based bonuses (backstab, flanking) do NOT apply to ranged attacks. Those bonuses only apply to melee combat.

update `async` ¶

update(delta: float) -> None

Update tick - process reload timers.

shutdown `async` ¶

shutdown() -> None

Clean up reload timers and ammo tracking to prevent memory leak.

resolve_attack `async` ¶

resolve_attack(
    attacker: Character,
    target: Character,
    weapon: Item,
    distance_rooms: int = 0,
    attacker_zone: CombatZone = RANGED_NEAR,
    target_zone: CombatZone = MELEE,
    line_of_sight: LineOfSight | None = None,
    target_in_cover: bool = False,
    cover_level: float = 0.0,
    target_is_moving: bool = False,
    aimed_location: str | None = None,
) -> AttackResult

Resolve a ranged attack.

Parameters:

Name	Type	Description	Default
`attacker`	`Character`	Character making the attack	required
`target`	`Character`	Character being attacked	required
`weapon`	`Item`	Ranged weapon being used	required
`distance_rooms`	`int`	Number of rooms between attacker and target	`0`
`attacker_zone`	`CombatZone`	Combat zone of attacker	`RANGED_NEAR`
`target_zone`	`CombatZone`	Combat zone of target	`MELEE`
`line_of_sight`	`LineOfSight \| None`	Line of sight info between positions	`None`
`target_in_cover`	`bool`	Whether target has cover	`False`
`cover_level`	`float`	Cover percentage (0.0-1.0, 1.0 = full cover)	`0.0`
`target_is_moving`	`bool`	Whether target moved this tick	`False`
`aimed_location`	`str \| None`	Specific body part to target	`None`

Returns:

Type	Description
`AttackResult`	AttackResult with full attack outcome

get_reload_remaining ¶

get_reload_remaining(attacker_id: UUID) -> float

Get remaining reload time for an attacker.

is_reloading ¶

is_reloading(attacker_id: UUID) -> bool

Check if attacker is currently reloading.

force_reload ¶

force_reload(
    attacker_id: UUID, weapon_type: RangedWeaponType
) -> None

Manually trigger a reload (for reload command).

cancel_reload ¶

cancel_reload(attacker_id: UUID) -> bool

Cancel an ongoing reload.

Magic System¶

Spellcasting with mana costs, cooldowns, and magical effects.

from maid_classic_rpg.systems.magic import SpellSystem

# Magic system handles:
# - Spell learning and memorization
# - Mana management
# - Spell effects and duration
# - Magical item enchantments

magic ¶

Magic systems for spellcasting and effects.

ActiveBuff `dataclass` ¶

ActiveBuff(
    id: UUID,
    name: str,
    source_id: UUID | None,
    target_id: UUID,
    expires_at: float,
    effect_type: str,
    stat_modifiers: dict[str, int] = dict(),
    damage_reduction: float = 0.0,
    damage_increase: float = 0.0,
    status_immunity: list[StatusEffectType] = list(),
    tick_effect: str | None = None,
    tick_interval: float = 1.0,
    last_tick: float = 0.0,
    tick_damage: int = 0,
    tick_heal: int = 0,
    tick_damage_type: DamageType | None = None,
    stacks: int = 1,
    max_stacks: int = 1,
    is_buff: bool = True,
    can_dispel: bool = True,
    icon: str = "",
)

A temporary buff/debuff on a character.

BuffAppliedEvent `dataclass` ¶

BuffAppliedEvent(
    target_id: UUID,
    buff_name: str,
    source_id: UUID | None = None,
    duration: float = 0.0,
)

Bases: Event

Emitted when a buff is applied.

BuffDispelledEvent `dataclass` ¶

BuffDispelledEvent(
    target_id: UUID,
    buff_name: str,
    dispeller_id: UUID | None = None,
)

Bases: Event

Emitted when a buff is dispelled.

BuffExpiredEvent `dataclass` ¶

BuffExpiredEvent(target_id: UUID, buff_name: str)

Bases: Event

Emitted when a buff expires.

SpellEffectSystem ¶

SpellEffectSystem(world: World)

Bases: System

Manages active spell effects, buffs, and debuffs.

update `async` ¶

update(delta: float) -> None

Process buff expirations and tick effects.

register_handler ¶

register_handler(
    effect_type: str, handler: EffectHandler
) -> None

Register a custom effect handler.

apply_effect `async` ¶

apply_effect(
    caster: Character,
    target: Character,
    effect_type: str,
    params: dict[str, Any],
) -> dict[str, Any]

Apply a spell effect.

Parameters:

Name	Type	Description	Default
`caster`	`Character`	Character casting the spell	required
`target`	`Character`	Target of the effect	required
`effect_type`	`str`	Type of effect to apply	required
`params`	`dict[str, Any]`	Effect parameters	required

Returns:

Type	Description
`dict[str, Any]`	Result dictionary with effect outcome

add_buff ¶

add_buff(target_id: UUID, buff: ActiveBuff) -> bool

Add a buff to a character.

Returns True if buff was added, False if stacking failed.

remove_buff ¶

remove_buff(target_id: UUID, buff_name: str) -> bool

Remove a buff by name. Returns True if removed.

get_buffs ¶

get_buffs(target_id: UUID) -> list[ActiveBuff]

Get all active buffs on a target.

get_buff ¶

get_buff(
    target_id: UUID, buff_name: str
) -> ActiveBuff | None

Get a specific buff by name.

has_buff ¶

has_buff(target_id: UUID, buff_name: str) -> bool

Check if target has a specific buff.

get_stat_modifier ¶

get_stat_modifier(target_id: UUID, stat: str) -> int

Get total stat modifier from all buffs.

get_damage_reduction ¶

get_damage_reduction(target_id: UUID) -> float

Get total damage reduction from all buffs (0.0-1.0).

get_damage_increase ¶

get_damage_increase(target_id: UUID) -> float

Get total damage increase from all buffs (capped at 5.0 / 500%).

has_status_immunity ¶

has_status_immunity(
    target_id: UUID, status: StatusEffectType
) -> bool

Check if target is immune to a status effect.

dispel `async` ¶

dispel(
    target_id: UUID,
    dispeller_id: UUID | None = None,
    count: int = 1,
    buffs_only: bool = False,
    debuffs_only: bool = False,
) -> int

Dispel buffs/debuffs from a target.

Parameters:

Name	Type	Description	Default
`target_id`	`UUID`	Character to dispel	required
`dispeller_id`	`UUID \| None`	Who is dispelling	`None`
`count`	`int`	Max effects to remove	`1`
`buffs_only`	`bool`	Only remove positive effects	`False`
`debuffs_only`	`bool`	Only remove negative effects	`False`

Returns:

Type	Description
`int`	Number of effects dispelled

clear_buffs ¶

clear_buffs(target_id: UUID) -> int

Remove all buffs from a target. Returns count removed.

SpellCastCompleteEvent `dataclass` ¶

SpellCastCompleteEvent(
    caster_id: UUID,
    spell_name: str,
    target_id: UUID | None = None,
    success: bool = True,
    damage_dealt: int = 0,
    healing_done: int = 0,
)

Bases: Event

Emitted when a spell finishes casting.

SpellCastInterruptedEvent `dataclass` ¶

SpellCastInterruptedEvent(
    caster_id: UUID, spell_name: str, reason: str = ""
)

Bases: Event

Emitted when a spell cast is interrupted.

SpellCastStartEvent `dataclass` ¶

SpellCastStartEvent(
    caster_id: UUID,
    spell_name: str,
    target_id: UUID | None = None,
    cast_time: float = 0.0,
)

Bases: Event

Emitted when a spell cast begins.

SpellLearnedEvent `dataclass` ¶

SpellLearnedEvent(character_id: UUID, spell_name: str)

Bases: Event

Emitted when a character learns a spell.

CastingState `dataclass` ¶

CastingState(
    caster_id: UUID,
    spell_name: str,
    target_id: UUID | None,
    started_at: float,
    completes_at: float,
    room_id: UUID | None,
    mana_cost: int = 0,
    cooldown: float = 0.0,
)

Tracks an in-progress spell cast.

SpellCastResult `dataclass` ¶

SpellCastResult(
    success: bool = False,
    message: str = "",
    damage_dealt: int = 0,
    healing_done: int = 0,
    effects_applied: list[StatusEffectType] = list(),
    targets_hit: list[UUID] = list(),
    mana_spent: int = 0,
)

Result of attempting to cast a spell.

SpellSystem ¶

SpellSystem(world: World)

Bases: System, StorageAware

Handles spell casting, effects, and cooldowns.

set_storage ¶

set_storage(store: DocumentStore) -> None

Set the document store for persistence.

startup `async` ¶

startup() -> None

Load spell data from storage.

shutdown `async` ¶

shutdown() -> None

Save spell data and clean up.

update `async` ¶

update(delta: float) -> None

Process casting completions and clean up expired cooldowns.

start_cast `async` ¶

start_cast(
    caster: Character,
    spell_name: str,
    target: Character | None = None,
) -> tuple[bool, str]

Begin casting a spell.

Parameters:

Name	Type	Description	Default
`caster`	`Character`	Character casting the spell	required
`spell_name`	`str`	Internal name of the spell	required
`target`	`Character \| None`	Optional target character	`None`

Returns:

Type	Description
`tuple[bool, str]`	Tuple of (success, message)

interrupt_cast `async` ¶

interrupt_cast(
    caster_id: UUID, reason: str = "interrupted"
) -> bool

Interrupt an in-progress cast.

Refunds mana spent when cast was initiated.

Parameters:

Name	Type	Description	Default
`caster_id`	`UUID`	ID of caster to interrupt	required
`reason`	`str`	Reason for interruption	`'interrupted'`

Returns:

Type	Description
`bool`	True if a cast was interrupted

is_casting ¶

is_casting(caster_id: UUID) -> bool

Check if a character is currently casting.

get_casting_state ¶

get_casting_state(caster_id: UUID) -> CastingState | None

Get the current casting state for a character.

learn_spell ¶

learn_spell(character_id: UUID, spell_name: str) -> bool

Teach a spell to a character.

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	Character to teach	required
`spell_name`	`str`	Spell internal name	required

Returns:

Type	Description
`bool`	True if spell was learned (not already known)

forget_spell ¶

forget_spell(character_id: UUID, spell_name: str) -> bool

Remove a spell from a character's known spells.

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	Character to forget spell	required
`spell_name`	`str`	Spell internal name	required

Returns:

Type	Description
`bool`	True if spell was removed

get_known_spells ¶

get_known_spells(character_id: UUID) -> list[LearnedSpell]

Get all spells known by a character.

get_known_spell ¶

get_known_spell(
    character_id: UUID, spell_name: str
) -> LearnedSpell | None

Get a specific learned spell.

get_cooldown_remaining ¶

get_cooldown_remaining(
    character_id: UUID, spell_name: str
) -> float

Get remaining cooldown time in game seconds.

register_magic_commands ¶

register_magic_commands(
    spell_system: SpellSystem, skill_system: SkillSystem
) -> None

Register all magic and ability commands.

Crafting System¶

Item creation with recipes, materials, and skill requirements.

from maid_classic_rpg.systems.crafting import CraftingSystem

# Crafting system handles:
# - Recipe discovery and learning
# - Material gathering
# - Crafting skill progression
# - Quality and variation

crafting ¶

Crafting system components.

CraftingResult `dataclass` ¶

CraftingResult(
    success: bool,
    error: str = "",
    item: Item | None = None,
    quality: MaterialQuality = COMMON,
    skill_gained: float = 0.0,
    materials_lost: list[UUID] = list(),
)

Result of a crafting attempt.

CraftingSystem ¶

CraftingSystem(world: World)

Bases: System

Manages all crafting operations.

recipe_manager `property` ¶

recipe_manager: RecipeManager

Get the recipe manager.

station_registry `property` ¶

station_registry: StationRegistry

Get the station registry.

set_item_manager ¶

set_item_manager(item_manager: ItemManager) -> None

Set the item manager for creating crafted items.

This allows deferred wiring when the ItemManager is created after the CraftingSystem (e.g., during content pack loading).

startup `async` ¶

startup() -> None

Load recipes and initialize system.

shutdown `async` ¶

shutdown() -> None

Shutdown crafting system.

update `async` ¶

update(delta: float) -> None

Process ongoing crafts with cast times.

attempt_craft `async` ¶

attempt_craft(
    character: Character,
    recipe: Recipe,
    materials: list[MaterialItem],
    station: CraftingStation | None = None,
    item_manager: ItemManager | None = None,
) -> CraftingResult

Attempt to craft an item from a recipe.

Parameters:

Name	Type	Description	Default
`character`	`Character`	Character doing the crafting	required
`recipe`	`Recipe`	Recipe to craft	required
`materials`	`list[MaterialItem]`	Materials to use	required
`station`	`CraftingStation \| None`	Crafting station (if required)	`None`
`item_manager`	`ItemManager \| None`	Optional explicit item manager for creating output. If not provided, uses the system's configured ItemManager or falls back to world.items.	`None`

Returns:

Type	Description
`CraftingResult`	CraftingResult with success/failure and created item

learn_recipe ¶

learn_recipe(character: Character, recipe: Recipe) -> bool

Teach a recipe to a character.

Parameters:

Name	Type	Description	Default
`character`	`Character`	Character to teach	required
`recipe`	`Recipe`	Recipe to learn	required

Returns:

Type	Description
`bool`	True if successfully learned

character_knows_recipe ¶

character_knows_recipe(
    character: Character, recipe: Recipe
) -> bool

Check if character knows a recipe.

Parameters:

Name	Type	Description	Default
`character`	`Character`	Character to check	required
`recipe`	`Recipe`	Recipe to check	required

Returns:

Type	Description
`bool`	True if character knows the recipe

find_materials_for_recipe ¶

find_materials_for_recipe(
    recipe: Recipe, inventory: list[MaterialItem]
) -> tuple[list[MaterialItem], list[str]]

Find matching materials for a recipe from an inventory.

Parameters:

Name	Type	Description	Default
`recipe`	`Recipe`	Recipe to match	required
`inventory`	`list[MaterialItem]`	Available materials	required

Returns:

Type	Description
`tuple[list[MaterialItem], list[str]]`	Tuple of (matched materials, missing material types)

EnchantingSystem ¶

EnchantingSystem()

Handles item enchantment.

enchant_item `async` ¶

enchant_item(
    character: Character,
    item: Item,
    enchantment_type: EnchantmentType,
    reagent_consumed: bool = True,
    station: CraftingStation | None = None,
) -> EnchantResult

Apply an enchantment to an item.

Parameters:

Name	Type	Description	Default
`character`	`Character`	Character enchanting	required
`item`	`Item`	Item to enchant	required
`enchantment_type`	`EnchantmentType`	Type of enchantment	required
`reagent_consumed`	`bool`	Whether reagent cost was paid	`True`
`station`	`CraftingStation \| None`	Enchanting station if available	`None`

Returns:

Type	Description
`EnchantResult`	EnchantResult with success/failure

Enchantment ¶

Bases: MAIDBaseModel

An enchantment applied to an item.

EnchantmentType ¶

Bases: str, Enum

Types of enchantments.

EnchantResult `dataclass` ¶

EnchantResult(
    success: bool,
    error: str = "",
    enchantment: Enchantment | None = None,
    item_damaged: bool = False,
)

Result of an enchanting attempt.

GemType ¶

Bases: str, Enum

Types of socketable gems.

RepairResult `dataclass` ¶

RepairResult(
    success: bool,
    error: str = "",
    durability_restored: int = 0,
    new_durability: int = 0,
)

Result of a repair attempt.

RepairSystem ¶

RepairSystem()

Handles item repair and durability.

repair_item `async` ¶

repair_item(
    character: Character,
    item: Item,
    use_materials: bool = False,
    station: CraftingStation | None = None,
) -> RepairResult

Repair a damaged item.

Parameters:

Name	Type	Description	Default
`character`	`Character`	Character doing the repair	required
`item`	`Item`	Item to repair	required
`use_materials`	`bool`	Whether materials are being used for bonus	`False`
`station`	`CraftingStation \| None`	Crafting station if available	`None`

Returns:

Type	Description
`RepairResult`	RepairResult with success/failure

damage_item ¶

damage_item(item: Item, amount: int = 1) -> bool

Apply durability damage to an item.

Parameters:

Name	Type	Description	Default
`item`	`Item`	Item to damage	required
`amount`	`int`	Durability points to remove	`1`

Returns:

Type	Description
`bool`	True if item is now broken (0 durability)

SocketedGem ¶

Bases: MAIDBaseModel

A gem socketed into an item.

SocketingSystem ¶

Handles gem socketing into items.

socket_gem `async` ¶

socket_gem(
    character: Character, item: Item, gem: Item
) -> SocketResult

Socket a gem into an item.

Parameters:

Name	Type	Description	Default
`character`	`Character`	Character doing the socketing	required
`item`	`Item`	Item to socket into	required
`gem`	`Item`	Gem item to socket	required

Returns:

Type	Description
`SocketResult`	SocketResult with success/failure

remove_gem `async` ¶

remove_gem(
    character: Character, item: Item, socket_index: int
) -> SocketResult

Remove a gem from an item (destroys the gem).

Parameters:

Name	Type	Description	Default
`character`	`Character`	Character doing the removal	required
`item`	`Item`	Item to remove gem from	required
`socket_index`	`int`	Index of socket to empty	required

Returns:

Type	Description
`SocketResult`	SocketResult indicating success/failure

SocketResult `dataclass` ¶

SocketResult(
    success: bool,
    error: str = "",
    gem: SocketedGem | None = None,
)

Result of a gem socketing attempt.

GatheringResult `dataclass` ¶

GatheringResult(
    success: bool,
    error: str = "",
    materials: list[MaterialItem] = list(),
    skill_gained: float = 0.0,
)

Result of a gathering attempt.

GatheringSystem ¶

GatheringSystem(world: World)

Bases: System, StorageAware

Manages resource gathering in the world.

set_storage ¶

set_storage(store: DocumentStore) -> None

Set the document store for persistence.

startup `async` ¶

startup() -> None

Load gathering nodes from storage on startup.

shutdown `async` ¶

shutdown() -> None

Save gathering nodes to storage on shutdown.

update `async` ¶

update(delta: float) -> None

Regenerate resources over time.

register_node ¶

register_node(node: ResourceNode) -> None

Register a resource node.

Parameters:

Name	Type	Description	Default
`node`	`ResourceNode`	Node to register	required

unregister_node ¶

unregister_node(node_id: UUID) -> ResourceNode | None

Unregister a resource node.

Parameters:

Name	Type	Description	Default
`node_id`	`UUID`	ID of node to remove	required

Returns:

Type	Description
`ResourceNode \| None`	Removed node, or None if not found

get_node ¶

get_node(node_id: UUID) -> ResourceNode | None

Get a node by ID.

Parameters:

Name	Type	Description	Default
`node_id`	`UUID`	Node ID	required

Returns:

Type	Description
`ResourceNode \| None`	Node if found

get_nodes_in_room ¶

get_nodes_in_room(
    room_id: UUID, visible_only: bool = True
) -> list[ResourceNode]

Get all resource nodes in a room.

Parameters:

Name	Type	Description	Default
`room_id`	`UUID`	Room ID	required
`visible_only`	`bool`	If True, exclude hidden nodes	`True`

Returns:

Type	Description
`list[ResourceNode]`	List of nodes in the room

attempt_gather `async` ¶

attempt_gather(
    character: Character, node: ResourceNode
) -> GatheringResult

Attempt to gather from a resource node.

Parameters:

Name	Type	Description	Default
`character`	`Character`	Character gathering	required
`node`	`ResourceNode`	Node to gather from	required

Returns:

Type	Description
`GatheringResult`	GatheringResult with success/failure and materials

attempt_skinning `async` ¶

attempt_skinning(
    character: Character, corpse: Item
) -> GatheringResult

Attempt to skin a corpse for leather/materials.

Parameters:

Name	Type	Description	Default
`character`	`Character`	Character skinning	required
`corpse`	`Item`	Corpse item to skin	required

Returns:

Type	Description
`GatheringResult`	GatheringResult with success/failure and materials

all_nodes ¶

all_nodes() -> list[ResourceNode]

Get all registered nodes.

Returns:

Type	Description
`list[ResourceNode]`	List of all nodes

clear ¶

clear() -> None

Clear all registered nodes.

MaterialQuality ¶

Bases: IntEnum

Material quality tiers with numeric values for calculations.

RecipeManager ¶

RecipeManager(
    data_dir: Path | None = None,
    store: DocumentStore | None = None,
)

Manages recipe loading, storage, and queries.

Parameters:

Name	Type	Description	Default
`data_dir`	`Path \| None`	Directory containing recipe YAML files	`None`
`store`	`DocumentStore \| None`	DocumentStore instance for persistence	`None`

load `async` ¶

load() -> int

Load all recipes from storage and YAML files.

Returns:

Type	Description
`int`	Number of recipes loaded

save_recipe `async` ¶

save_recipe(recipe: Recipe) -> None

Save a recipe to persistent storage.

Parameters:

Name	Type	Description	Default
`recipe`	`Recipe`	Recipe to save	required

register ¶

register(recipe: Recipe) -> None

Register a recipe.

Parameters:

Name	Type	Description	Default
`recipe`	`Recipe`	Recipe to register	required

unregister ¶

unregister(recipe_id: UUID) -> Recipe | None

Unregister a recipe.

Parameters:

Name	Type	Description	Default
`recipe_id`	`UUID`	ID of recipe to remove	required

Returns:

Type	Description
`Recipe \| None`	Removed recipe, or None if not found

get ¶

get(recipe_id: UUID) -> Recipe | None

Get a recipe by ID.

Parameters:

Name	Type	Description	Default
`recipe_id`	`UUID`	Recipe ID	required

Returns:

Type	Description
`Recipe \| None`	Recipe if found

find_by_name ¶

find_by_name(name: str) -> Recipe | None

Find a recipe by name (case-insensitive).

Parameters:

Name	Type	Description	Default
`name`	`str`	Recipe name to search for	required

Returns:

Type	Description
`Recipe \| None`	Matching recipe, or None

get_recipes_for_skill ¶

get_recipes_for_skill(
    skill: str, max_level: int = 100
) -> list[Recipe]

Get all recipes for a skill up to a level.

Parameters:

Name	Type	Description	Default
`skill`	`str`	Skill name	required
`max_level`	`int`	Maximum skill level filter	`100`

Returns:

Type	Description
`list[Recipe]`	List of matching recipes

get_learnable_recipes ¶

get_learnable_recipes(
    character: Character, skill: str
) -> list[Recipe]

Get recipes a character can learn but doesn't know.

Parameters:

Name	Type	Description	Default
`character`	`Character`	Character to check	required
`skill`	`str`	Skill to check recipes for	required

Returns:

Type	Description
`list[Recipe]`	List of learnable recipes

get_known_recipes ¶

get_known_recipes(character: Character) -> list[Recipe]

Get all recipes known by a character.

Parameters:

Name	Type	Description	Default
`character`	`Character`	Character to check	required

Returns:

Type	Description
`list[Recipe]`	List of known recipes

get_auto_learned_recipes ¶

get_auto_learned_recipes(
    skill: str, skill_level: int
) -> list[Recipe]

Get recipes that are automatically learned.

Parameters:

Name	Type	Description	Default
`skill`	`str`	Skill name	required
`skill_level`	`int`	Character's skill level	required

Returns:

Type	Description
`list[Recipe]`	List of auto-learned recipes

get_recipe_by_output ¶

get_recipe_by_output(
    output_template_id: UUID,
) -> Recipe | None

Get recipe that produces a specific item template.

Parameters:

Name	Type	Description	Default
`output_template_id`	`UUID`	Item template ID	required

Returns:

Type	Description
`Recipe \| None`	Recipe if found

all_recipes ¶

all_recipes() -> list[Recipe]

Get all registered recipes.

Returns:

Type	Description
`list[Recipe]`	List of all recipes

clear ¶

clear() -> None

Clear all registered recipes.

StationRegistry ¶

StationRegistry()

Manages crafting stations in the world.

register ¶

register(station: CraftingStation) -> None

Register a crafting station.

Parameters:

Name	Type	Description	Default
`station`	`CraftingStation`	Station to register	required

unregister ¶

unregister(station_id: UUID) -> CraftingStation | None

Unregister a crafting station.

Parameters:

Name	Type	Description	Default
`station_id`	`UUID`	ID of station to unregister	required

Returns:

Type	Description
`CraftingStation \| None`	The removed station, or None if not found

get ¶

get(station_id: UUID) -> CraftingStation | None

Get a station by ID.

Parameters:

Name	Type	Description	Default
`station_id`	`UUID`	Station ID	required

Returns:

Type	Description
`CraftingStation \| None`	Station if found, None otherwise

get_stations_in_room ¶

get_stations_in_room(
    room_id: UUID, station_type: StationType | None = None
) -> list[CraftingStation]

Get crafting stations in a room.

Parameters:

Name	Type	Description	Default
`room_id`	`UUID`	Room ID to search	required
`station_type`	`StationType \| None`	Optional filter by station type	`None`

Returns:

Type	Description
`list[CraftingStation]`	List of matching stations

get_stations_by_type ¶

get_stations_by_type(
    station_type: StationType,
) -> list[CraftingStation]

Get all stations of a specific type.

Parameters:

Name	Type	Description	Default
`station_type`	`StationType`	Type to search for	required

Returns:

Type	Description
`list[CraftingStation]`	List of matching stations

find_nearest_station ¶

find_nearest_station(
    room_id: UUID,
    station_type: StationType,
    world: World,
    max_distance: int = 10,
) -> tuple[CraftingStation | None, int]

Find nearest station of type using BFS.

Parameters:

Name	Type	Description	Default
`room_id`	`UUID`	Starting room ID	required
`station_type`	`StationType`	Type of station to find	required
`world`	`World`	World instance for room graph traversal	required
`max_distance`	`int`	Maximum rooms to search	`10`

Returns:

Type	Description
`tuple[CraftingStation \| None, int]`	Tuple of (station or None, distance). Distance is -1 if not found.

all_stations ¶

all_stations() -> list[CraftingStation]

Get all registered stations.

Returns:

Type	Description
`list[CraftingStation]`	List of all stations

clear ¶

clear() -> None

Clear all registered stations.

apply_quality_to_item ¶

apply_quality_to_item(
    item: Item, quality: MaterialQuality
) -> None

Apply quality modifiers to a crafted item's stats.

Modifies: - Weapon damage (dice bonus) - Armor value - Item value (gold) - Stores quality in metadata

Parameters:

Name	Type	Description	Default
`item`	`Item`	The crafted item to modify	required
`quality`	`MaterialQuality`	Quality tier to apply	required

calculate_output_quality ¶

calculate_output_quality(
    crafter_skill: int,
    recipe_difficulty: int,
    materials: list[MaterialItem],
    quality_variance: float = 0.5,
    skill_bonus_divisor: float = 20.0,
) -> MaterialQuality

Calculate the quality of a crafted item.

Formula: - Base quality from average material quality - Skill bonus: (skill - difficulty) / 20 quality levels - Random variance: ± 0.5 quality levels (configurable) - Clamped to valid quality range (1-5)

Parameters:

Name	Type	Description	Default
`crafter_skill`	`int`	Crafter's skill level (1-100)	required
`recipe_difficulty`	`int`	Recipe difficulty (1-100)	required
`materials`	`list[MaterialItem]`	List of materials used	required
`quality_variance`	`float`	Random variance range (default 0.5)	`0.5`
`skill_bonus_divisor`	`float`	Points above difficulty per quality level	`20.0`

Returns:

Type	Description
`MaterialQuality`	Final quality tier for the crafted item

Economy System¶

Shops, trading, banking, and auction house.

from maid_classic_rpg.systems.economy import ShopSystem

# Economy system handles:
# - NPC shops with inventory
# - Player-to-player trading
# - Banking and storage
# - Auction house

economy ¶

Economy systems for shops, trading, banking, and auctions.

AuctionHouse ¶

AuctionHouse(world: World)

Bases: System, StorageAware

Manages the auction house system.

Handles listing, searching, bidding, and purchasing.

active_count `property` ¶

active_count: int

Return number of active listings.

set_storage ¶

set_storage(store: DocumentStore) -> None

Receive storage from the engine via StorageAware protocol.

startup `async` ¶

startup() -> None

Initialize auction house and load data from storage.

shutdown `async` ¶

shutdown() -> None

Persist all auction data before shutdown.

update `async` ¶

update(delta: float) -> None

Process expired auctions and deliver pending gold periodically.

list_item `async` ¶

list_item(
    seller: Character,
    item: Item,
    starting_price: int,
    buyout_price: int | None = None,
    duration_hours: int = 24,
) -> AuctionResult

List an item for auction.

Parameters:

Name	Type	Description	Default
`seller`	`Character`	Character listing the item	required
`item`	`Item`	Item to list	required
`starting_price`	`int`	Starting bid price	required
`buyout_price`	`int \| None`	Optional instant purchase price	`None`
`duration_hours`	`int`	Auction duration	`24`

Returns:

Type	Description
`AuctionResult`	AuctionResult with success/failure and listing

place_bid `async` ¶

place_bid(
    bidder: Character, listing_id: UUID, bid_amount: int
) -> BidResult

Place a bid on an auction, wrapped in critical_write("economy.auction.bid").

buyout `async` ¶

buyout(
    buyer: Character,
    listing_id: UUID,
    item_manager: ItemManager | None = None,
) -> BuyoutResult

Instantly purchase at buyout price, wrapped in critical_write("economy.auction.buyout").

cancel_listing `async` ¶

cancel_listing(
    seller: Character,
    listing_id: UUID,
    item_manager: ItemManager | None = None,
) -> CancelResult

Cancel an auction listing.

Parameters:

Name	Type	Description	Default
`seller`	`Character`	Character cancelling	required
`listing_id`	`UUID`	Listing to cancel	required
`item_manager`	`ItemManager \| None`	Item manager for getting item	`None`

Returns:

Type	Description
`CancelResult`	CancelResult with success/failure

search `async` ¶

search(
    query: str | None = None,
    item_type: str | None = None,
    min_price: int | None = None,
    max_price: int | None = None,
    seller_name: str | None = None,
    sort_by: str = "expires_at",
    limit: int = 50,
    offset: int = 0,
) -> list[AuctionListing]

Search active auction listings.

Parameters:

Name	Type	Description	Default
`query`	`str \| None`	Search term for item name	`None`
`item_type`	`str \| None`	Filter by item type	`None`
`min_price`	`int \| None`	Minimum current bid	`None`
`max_price`	`int \| None`	Maximum current bid	`None`
`seller_name`	`str \| None`	Filter by seller name	`None`
`sort_by`	`str`	Sort field (expires_at, price_asc, price_desc, bid_count)	`'expires_at'`
`limit`	`int`	Maximum results	`50`
`offset`	`int`	Result offset for pagination	`0`

Returns:

Type	Description
`list[AuctionListing]`	List of matching listings

get_listing ¶

get_listing(listing_id: UUID) -> AuctionListing | None

Get a listing by ID.

get_listing_by_short_id ¶

get_listing_by_short_id(
    short_id: str,
) -> AuctionListing | None

Get a listing by short ID prefix.

Parameters:

Name	Type	Description	Default
`short_id`	`str`	First 7 characters of listing ID	required

Returns:

Type	Description
`AuctionListing \| None`	Matching listing, or None

get_seller_listings ¶

get_seller_listings(
    seller_id: UUID, active_only: bool = False
) -> list[AuctionListing]

Get all listings by a seller.

Parameters:

Name	Type	Description	Default
`seller_id`	`UUID`	Seller character ID	required
`active_only`	`bool`	If True, only return active listings	`False`

Returns:

Type	Description
`list[AuctionListing]`	List of listings

get_pending_gold ¶

get_pending_gold(character_id: UUID) -> int

Get pending gold for a character (from sales/refunds).

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	Character ID	required

Returns:

Type	Description
`int`	Pending gold amount

collect_pending_gold `async` ¶

collect_pending_gold(character: Character) -> int

Collect the character's durable owed gold in a single durable commit.

Single-authority rule (issue #139): the durable owed-balance lives on the character's PendingPayoutComponent (round-tripped through Character.pending_payout_gold). The offline-safe _pending_gold store is only an accrual projection; it is folded into the character owed-balance here (and on login) before anything is credited, so the two never diverge. After folding, this method credits the carried gold and drains the owed-balance to zero, then makes BOTH durable in ONE characters upsert via CharacterManager.update. The owed-decrement IS the idempotency record: a re-run after a successful commit sees owed already 0 and grants nothing, so a lost projection write or a retried collect can never double-credit.

Parameters:

Name	Type	Description	Default
`character`	`Character`	Character collecting (must have a live ECS entity for the component dual-write; offline callers still work via the DTO accessor fallback).	required

Returns:

Type	Description
`int`	Amount of gold credited (0 if nothing was owed).

persist_pending_payouts `async` ¶

persist_pending_payouts() -> None

Durably persist the pending-payouts singleton on demand.

Public entry point for the collect handler: after an item delivery is acknowledged (cleared from the in-flight buffer in memory) the cleared owed-state must be made durable BEFORE the operation is treated as complete, otherwise a crash before the lazy periodic save would reload a stale pending item and re-deliver it. The owned-set guard only backstops the residual window between the inventory flush and this call.

commit_character `async` ¶

commit_character(character: Character) -> None

Durably commit the character (inventory + owed-balance + markers).

Public entry point for the collect handler's item delivery: the player character entity is transient so critical_write's dirty-flush SKIPS it; the only durable character primitive is CharacterManager.update -> snapshot components into the DTO -> upsert the characters collection. The handler stamps the inventory add and its per-item auction_item:<id> delivered-marker in the live components, then calls this to land BOTH in ONE durable DTO upsert.

Raises:

Type	Description
`RuntimeError`	If the durable upsert fails (A-R3-1), so the caller can roll back the in-memory add/marker and leave the item owed.

prune_stale_delivered_markers `async` ¶

prune_stale_delivered_markers(character: Character) -> None

Durably prune bounded-retention auction_item delivered-markers.

A-R4-2: each collected item leaves a permanent auction_item:<id> marker (A-R3-2) that suppresses re-delivery after disposal. Pruned here once the marker is provably no longer needed -- the item is owed in NEITHER store (so it can never be re-presented as pending) AND the marker is older than :data:AUCTION_ITEM_MARKER_RETENTION (so we are well outside any crash-recovery window in which a not-yet-synced store might still resurrect it). This keeps the marker set bounded by the items collected within the retention window instead of growing forever.

The prune is made durable in its own CharacterManager.update upsert so a crash cannot reload the pruned markers and regrow the set. A commit failure is swallowed (best-effort cleanup): the markers simply survive to be retried on the next collect -- never a correctness problem, since a stale marker only ever over-suppresses a re-delivery of an item that is no longer owed anyway.

I-R7-1: the "owed in NEITHER store" test reads the IN-MEMORY owed-set, which can diverge from the durable store. acknowledge_delivered clears the in-flight buffer in memory BEFORE persist_pending_payouts makes that clear durable; if the persist failed, the store still owes the item while memory does not. Pruning the marker then would let a restart (which reloads the still-owed item) re-deliver an item the player already received and possibly disposed of. So the prune is WITHHELD entirely unless the durable store is provably in sync with memory (:meth:_is_store_synced); when it is not, the in-memory owed-set cannot be trusted and the markers simply survive to a later collect once the store has caught up.

get_pending_items ¶

get_pending_items(character_id: UUID) -> list[UUID]

Get pending items for a character (from wins).

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	Character ID	required

Returns:

Type	Description
`list[UUID]`	List of pending item IDs

collect_pending_items ¶

collect_pending_items(character_id: UUID) -> list[UUID]

Claim pending items for a character for delivery.

Delivery is acknowledgement-based: the items returned here are moved into an in-flight buffer rather than being discarded outright. They are only considered delivered once the caller has successfully added them to the recipient's authoritative inventory and signalled this by calling :meth:acknowledge_delivered, or explicitly returns them via :meth:requeue_pending_item. Anything left unacknowledged in the in-flight buffer is treated as a crash mid-delivery and recovered to the pending queue on the next collect, so a lookup exception or any other crash after the claim but before the inventory add cannot silently lose the items.

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	Character ID	required

Returns:

Type	Description
`list[UUID]`	List of item IDs claimed for delivery

delivery_lock ¶

delivery_lock(character_id: UUID) -> Lock

Return the per-character lock guarding collect-and-deliver.

Callers hold this lock across the whole claim+deliver critical section so two concurrent collects for the same character are serialised: the second waits for the first to finish and then finds nothing pending and nothing in-flight, preventing double delivery and phantom pending items.

This delegates to the process-global :func:get_character_commit_lock (issue #139, I-R6-1) so the auction's collect-and-deliver section serialises not only against other auction collects but against EVERY other commit section mutating the SAME live character -- e.g. a quest turn-in or an achievement claim crediting gold and flushing via CharacterManager.update. Those subsystems already hold the same shared lock for their grant->commit->rollback window, so sharing it here closes the cross-subsystem interleave that the auction's previous instance-local lock did not cover. The lock is acquired exactly ONCE per collect (at the top of the handler / the auto-deliver loop) and the store-level _payout_store_lock is always taken strictly inside it, so the per-character lock stays outermost and cannot deadlock.

acknowledge_delivered ¶

acknowledge_delivered(
    character_id: UUID, item_id: UUID
) -> None

Acknowledge that a claimed item was successfully delivered.

This is the authoritative "this item landed in the recipient's inventory" signal. The caller MUST invoke it immediately after a successful inventory add (with no intervening await) so the delivered/undelivered decision is made at the exact delivery point rather than re-derived later from mutable inventory state. The item is dropped from the in-flight buffer; once gone it can never be re-owed even if the player later drops, sells, or consumes it.

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	Owner the item was delivered to	required
`item_id`	`UUID`	Item that was successfully delivered	required

requeue_pending_item ¶

requeue_pending_item(
    character_id: UUID, item_id: UUID
) -> None

Return an undeliverable item to the pending queue.

Used when a claimed item cannot be added to the recipient's inventory (e.g. it is full) so the item is preserved rather than lost. The item is cleared from the in-flight buffer (it has been explicitly handled) and re-queued idempotently so it can never be both delivered and pending.

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	Owner the item should remain pending for	required
`item_id`	`UUID`	Item that could not be delivered	required

auto_deliver_pending_gold `async` ¶

auto_deliver_pending_gold() -> int

Deliver pending gold to all online players via the durable path.

Called periodically from update() and on player connection. Routes every delivery through :meth:collect_pending_gold, which folds the accrual projection into the durable owed-balance and commits the credit in a single characters upsert (the single-authority rule, #139), so auto-delivery can never diverge from an explicit auction collect.

Returns:

Type	Description
`int`	Total gold delivered across all players.

clear ¶

clear() -> None

Clear all auction data.

BankSystem ¶

BankSystem(world: World)

Bases: System, StorageAware

System for bank operations.

Banking uses Character.bank_gold for storage. This system handles deposit/withdraw transactions. Bank NPC registrations are persisted via the DocumentStore.

get_transactions ¶

get_transactions(
    character_id: UUID | None = None,
) -> list[BankTransaction]

Get transaction history, optionally filtered by character.

Parameters:

Name	Type	Description	Default
`character_id`	`UUID \| None`	If provided, only return transactions for this character.	`None`

Returns:

Type	Description
`list[BankTransaction]`	List of BankTransaction records.

set_storage ¶

set_storage(store: DocumentStore) -> None

Set the document store for persistence.

startup `async` ¶

startup() -> None

Load banks from storage on startup.

shutdown `async` ¶

shutdown() -> None

Save banks to storage on shutdown.

register_bank ¶

register_bank(bank: BankNPC) -> None

Register a bank NPC.

Parameters:

Name	Type	Description	Default
`bank`	`BankNPC`	The bank NPC to register.	required

Raises:

Type	Description
`ValueError`	If the bank's room_id does not correspond to an existing entity in the world.

unregister_bank ¶

unregister_bank(bank_character_id: UUID) -> BankNPC | None

Unregister a bank NPC.

get_bank_in_room ¶

get_bank_in_room(room_id: UUID) -> BankNPC | None

Get bank NPC in a room.

get_bank ¶

get_bank(bank_character_id: UUID) -> BankNPC | None

Get bank NPC by character ID.

all_banks ¶

all_banks() -> list[BankNPC]

Get all registered banks.

update `async` ¶

update(delta: float) -> None

Bank system tick (interest calculation if enabled).

check_balance `async` ¶

check_balance(
    character_id: UUID, character_manager: CharacterManager
) -> tuple[bool, int]

Check bank balance.

Returns:

Type	Description
`tuple[bool, int]`	Tuple of (success, balance)

deposit `async` ¶

deposit(
    character_id: UUID,
    amount: int,
    bank: BankNPC | None,
    character_manager: CharacterManager,
) -> tuple[bool, str, int]

Deposit gold into bank, wrapped in critical_write("economy.bank.deposit").

withdraw `async` ¶

withdraw(
    character_id: UUID,
    amount: int,
    bank: BankNPC | None,
    character_manager: CharacterManager,
) -> tuple[bool, str, int]

Withdraw gold from bank, wrapped in critical_write("economy.bank.withdraw").

transfer `async` ¶

transfer(
    from_character_id: UUID,
    to_character_id: UUID,
    amount: int,
    character_manager: CharacterManager,
) -> tuple[bool, str]

Transfer gold between bank accounts, wrapped in critical_write("economy.bank.transfer").

ShopManager ¶

ShopManager(world: World | None = None)

Manages shop data and operations.

Handles shop registration, inventory, and pricing calculations.

register_shop ¶

register_shop(shop: Shop) -> None

Register a shop.

Validates room_id exists in world (if world is set) and warns on overwriting an existing shop in the same room.

register_merchant ¶

register_merchant(merchant: MerchantNPC) -> None

Register a merchant NPC linked to a shop.

get_merchant_for_shop ¶

get_merchant_for_shop(shop_id: UUID) -> MerchantNPC | None

Get the MerchantNPC associated with a shop.

unregister_shop ¶

unregister_shop(shop_id: UUID) -> Shop | None

Unregister a shop. Returns the shop if found.

get_shop ¶

get_shop(shop_id: UUID) -> Shop | None

Get shop by ID.

get_shop_in_room ¶

get_shop_in_room(room_id: UUID) -> Shop | None

Get shop operating in a room.

get_shop_by_merchant ¶

get_shop_by_merchant(merchant_id: UUID) -> Shop | None

Get shop by merchant entity ID.

all_shops ¶

all_shops() -> list[Shop]

Get all registered shops.

ShopSystem ¶

ShopSystem(world: World)

Bases: System, StorageAware

System for processing shop transactions and restocking.

Handles: - Buy/sell transactions - Price calculations with modifiers - Inventory restocking over time - Shop gold management

manager `property` ¶

manager: ShopManager

Get the shop manager.

set_storage ¶

set_storage(store: DocumentStore) -> None

Set the document store for persistence.

startup `async` ¶

startup() -> None

Load shops from storage on startup.

shutdown `async` ¶

shutdown() -> None

Save dirty shops to storage on shutdown.

update `async` ¶

update(delta: float) -> None

Process shop restocking and persist dirty shops.

list_inventory `async` ¶

list_inventory(
    shop: Shop,
    item_manager: ItemManager,
    charisma_mod: int = 0,
    reputation: int = 0,
) -> list[tuple[str, int, int]]

Get shop inventory for display.

Parameters:

Name	Type	Description	Default
`shop`	`Shop`	Shop to list	required
`item_manager`	`ItemManager`	Item manager	required
`charisma_mod`	`int`	Player's charisma modifier for price calculation	`0`
`reputation`	`int`	Player's reputation for price calculation	`0`

Returns:

Type	Description
`list[tuple[str, int, int]]`	List of (item_name, price, quantity)

buy_item `async` ¶

buy_item(
    shop: Shop,
    buyer_id: UUID,
    item_template_id: UUID,
    character_manager: CharacterManager,
    item_manager: ItemManager,
    charisma_mod: int = 0,
    reputation: int = 0,
) -> tuple[bool, str, Item | None]

Process a purchase, wrapped in critical_write("economy.shop.buy").

sell_item `async` ¶

sell_item(
    shop: Shop,
    seller_id: UUID,
    item_id: UUID,
    character_manager: CharacterManager,
    item_manager: ItemManager,
    charisma_mod: int = 0,
    reputation: int = 0,
) -> tuple[bool, str, int]

Process a sale, wrapped in critical_write("economy.shop.sell").

get_value `async` ¶

get_value(
    shop: Shop,
    item: Item,
    charisma_mod: int = 0,
    reputation: int = 0,
) -> tuple[bool, int]

Get what a shop would pay for an item.

Returns:

Type	Description
`tuple[bool, int]`	Tuple of (would_buy, price)

get_buy_price `async` ¶

get_buy_price(
    shop: Shop,
    entry: ShopInventoryEntry,
    charisma_mod: int = 0,
    reputation: int = 0,
) -> int

Get the buy price for a shop item entry.

TradeSystem ¶

TradeSystem(world: World)

Bases: System

System for managing player-to-player trades.

Trade flow: 1. Player A initiates trade with Player B 2. Player B accepts (PENDING -> ACTIVE) 3. Both players add/remove items and gold 4. Both players lock their offers (ACTIVE -> LOCKED) 5. Both players confirm (LOCKED -> CONFIRMED -> COMPLETED)

Either player can cancel at any time before completion.

startup `async` ¶

startup() -> None

Initialize trade system.

shutdown `async` ¶

shutdown() -> None

Shutdown trade system.

update `async` ¶

update(_delta: float) -> None

Check for expired trades.

get_player_session ¶

get_player_session(player_id: UUID) -> TradeSession | None

Get the active trade session for a player.

initiate_trade `async` ¶

initiate_trade(
    initiator_id: UUID, target_id: UUID
) -> tuple[bool, str, TradeSession | None]

Start a trade with another player.

Returns:

Type	Description
`tuple[bool, str, TradeSession \| None]`	Tuple of (success, message, session_or_none)

accept_trade `async` ¶

accept_trade(player_id: UUID) -> tuple[bool, str]

Accept a pending trade invitation.

decline_trade `async` ¶

decline_trade(player_id: UUID) -> tuple[bool, str]

Decline a pending trade.

add_item `async` ¶

add_item(
    player_id: UUID,
    item_id: UUID,
    character_manager: CharacterManager,
) -> tuple[bool, str]

Add an item to trade offer.

remove_item `async` ¶

remove_item(
    player_id: UUID, item_id: UUID
) -> tuple[bool, str]

Remove an item from trade offer.

set_gold `async` ¶

set_gold(
    player_id: UUID,
    amount: int,
    character_manager: CharacterManager,
) -> tuple[bool, str]

Set gold amount in trade offer.

lock_offer `async` ¶

lock_offer(player_id: UUID) -> tuple[bool, str]

Lock your trade offer.

unlock_offer `async` ¶

unlock_offer(player_id: UUID) -> tuple[bool, str]

Unlock your trade offer.

confirm_trade `async` ¶

confirm_trade(
    player_id: UUID,
    character_manager: CharacterManager,
    item_manager: ItemManager,
) -> tuple[bool, str]

Confirm and execute trade.

cancel_trade `async` ¶

cancel_trade(
    session_id: UUID,
    cancelled_by: UUID | None = None,
    reason: str = "cancelled",
) -> None

Cancel a trade session.

get_trade_status ¶

get_trade_status(
    player_id: UUID,
) -> dict[str, object] | None

Get current trade status for a player.

Returns:

Type	Description
`dict[str, object] \| None`	Dict with trade details or None if not in trade.

Quest System¶

Quest tracking with objectives, rewards, and progression.

from maid_classic_rpg.systems.quests.generation.system import QuestGenerationSystem

# Quest system handles:
# - Quest discovery and acceptance
# - Objective tracking
# - Reward distribution
# - Quest chains and prerequisites

quests ¶

Quest system for MAID.

Provides quest definitions, tracking, and rewards.

QuestManager ¶

QuestManager(
    quest_data_dir: Path | None = None,
    quest_log_dir: Path | None = None,
    event_bus: EventBus | None = None,
)

Manages quest definitions and character quest logs.

Responsibilities: - Load quest definitions from data files - Store and retrieve quest logs per character - Check quest prerequisites - Manage quest state transitions

Example

manager = QuestManager() await manager.load()

Get available quests for a character¶

available = await manager.get_available_quests(character)

Accept a quest¶

await manager.accept_quest(character, "starter_quest_1")

Parameters:

Name	Type	Description	Default
`quest_data_dir`	`Path \| None`	Directory containing quest definition files	`None`
`quest_log_dir`	`Path \| None`	Directory for storing character quest logs (fallback)	`None`

set_storage ¶

set_storage(store: DocumentStore) -> None

Set the document store for persistence.

load `async` ¶

load() -> None

Load quest definitions and quest logs from storage.

get_quest ¶

get_quest(quest_id: str) -> Quest | None

Get a quest definition by ID.

get_quest_log ¶

get_quest_log(character_id: UUID) -> QuestLog

Get or create a quest log for a character.

shutdown `async` ¶

shutdown() -> None

Await all pending saves to prevent data loss on shutdown.

get_all_quest_logs ¶

get_all_quest_logs() -> list[tuple[UUID, QuestLog]]

Get all character quest logs.

Returns:

Type	Description
`list[tuple[UUID, QuestLog]]`	List of (character_id, quest_log) tuples.

save_quest_log `async` ¶

save_quest_log(log: QuestLog) -> None

Save a character's quest log to storage.

Public wrapper for _save_quest_log.

check_prerequisites ¶

check_prerequisites(
    quest: Quest, character: Character, quest_log: QuestLog
) -> tuple[bool, str]

Check if a character meets quest prerequisites.

Parameters:

Name	Type	Description	Default
`quest`	`Quest`	Quest to check	required
`character`	`Character`	Character to check	required
`quest_log`	`QuestLog`	Character's quest log	required

Returns:

Type	Description
`tuple[bool, str]`	Tuple of (meets_prereqs, reason_if_not)

get_available_quests `async` ¶

get_available_quests(
    character: Character, npc_id: UUID | None = None
) -> list[Quest]

Get quests available to a character.

Parameters:

Name	Type	Description	Default
`character`	`Character`	Character to check	required
`npc_id`	`UUID \| None`	Optional NPC to filter by (only quests this NPC gives)	`None`

Returns:

Type	Description
`list[Quest]`	List of available quests

accept_quest `async` ¶

accept_quest(
    character: Character, quest_id: str
) -> tuple[bool, str]

Accept a quest for a character.

Parameters:

Name	Type	Description	Default
`character`	`Character`	Character accepting the quest	required
`quest_id`	`str`	Quest to accept	required

Returns:

Type	Description
`tuple[bool, str]`	Tuple of (success, message)

abandon_quest `async` ¶

abandon_quest(
    character: Character, quest_id: str
) -> tuple[bool, str]

Abandon an active quest.

Parameters:

Name	Type	Description	Default
`character`	`Character`	Character abandoning the quest	required
`quest_id`	`str`	Quest to abandon	required

Returns:

Type	Description
`tuple[bool, str]`	Tuple of (success, message)

expire_quest `async` ¶

expire_quest(character_id: UUID, quest_id: str) -> bool

Expire an active quest for a character.

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	Character with the active quest.	required
`quest_id`	`str`	Quest identifier to expire.	required

Returns:

Type	Description
`bool`	True if quest was expired.

register_generated_quest ¶

register_generated_quest(quest: Quest) -> None

Register a generated quest at runtime.

Parameters:

Name	Type	Description	Default
`quest`	`Quest`	Generated quest model.	required

Raises:

Type	Description
`ValueError`	If quest_id already exists.

invalidate_quest ¶

invalidate_quest(quest_id: str) -> None

Invalidate a generated quest and remove pending offers.

get_generated_quests ¶

get_generated_quests() -> list[Quest]

Get all currently registered generated quests.

get_npc_quests `async` ¶

get_npc_quests(
    character: Character, npc_id: UUID
) -> dict[str, list[Quest]]

Get quests related to an NPC for a character.

Returns dict with keys: - available: Quests NPC can offer - in_progress: Quests in progress related to NPC - ready_to_turn_in: Completed quests to turn in to NPC

get_quests_by_category `async` ¶

get_quests_by_category(
    character: Character, category: str
) -> list[tuple[Quest, QuestProgress]]

Get active quests by category.

get_all_quests ¶

get_all_quests() -> list[Quest]

Get all registered quest definitions.

DeliverObjectiveHandler ¶

DeliverObjectiveHandler(quest_manager: QuestManager)

Handler for DELIVER objective type.

Deliver objectives require giving a specific item to a specific NPC. This is checked when interacting with the target NPC.

check_delivery `async` ¶

check_delivery(
    character_id: UUID,
    npc_id: UUID,
    item_id: UUID,
    item_template_id: UUID | None,
) -> list[str]

Check if giving this item completes any deliver objectives.

Returns:

Type	Description
`list[str]`	List of quest IDs where delivery was successful

EscortObjectiveHandler ¶

EscortObjectiveHandler(quest_manager: QuestManager)

Handler for ESCORT objective type.

Escort objectives require an NPC to reach a destination room while keeping them alive.

check_escort_arrival `async` ¶

check_escort_arrival(
    npc_id: UUID, room_id: UUID
) -> list[tuple[UUID, str]]

Check if NPC arriving at room completes any escort objectives.

Returns:

Type	Description
`list[tuple[UUID, str]]`	List of (character_id, quest_id) for completed escorts

ObjectiveTracker ¶

ObjectiveTracker(world: World, quest_manager: QuestManager)

Bases: System, StorageAware

System for tracking quest objective progress.

Subscribes to game events and updates quest progress accordingly. Handles all objective types: kill, collect, deliver, visit, talk, etc.

Inherits from StorageAware to receive storage injection and pass it to QuestManager.

quest_manager `property` ¶

quest_manager: QuestManager

Get the quest manager for quest operations.

get_reward_distributor ¶

get_reward_distributor(
    item_manager: Any = None,
) -> RewardDistributor

Get or create the reward distributor for quest turn-ins.

Parameters:

Name	Type	Description	Default
`item_manager`	`Any`	Optional item manager for item rewards.	`None`

Returns:

Type	Description
`RewardDistributor`	RewardDistributor instance.

set_storage ¶

set_storage(store: DocumentStore) -> None

Receive storage from the engine and pass it to QuestManager.

startup `async` ¶

startup() -> None

Subscribe to relevant events.

shutdown `async` ¶

shutdown() -> None

Unsubscribe from events.

update `async` ¶

update(_delta: float) -> None

Check timed objectives.

RewardDistributor ¶

RewardDistributor(
    world: World,
    quest_manager: QuestManager,
    item_manager: ItemManager | None = None,
)

Handles quest reward distribution.

Responsibilities: - Validate quest can be turned in - Grant XP, gold, items, reputation - Handle reward choices - Emit reward events - Update quest state to TURNED_IN

set_item_manager ¶

set_item_manager(item_manager: ItemManager) -> None

Set or update the item manager reference.

This allows deferred wiring when the ItemManager is created after the RewardDistributor (e.g., during content pack loading).

turn_in_quest `async` ¶

turn_in_quest(
    character: Character,
    quest_id: str,
    reward_choice_index: int | None = None,
) -> tuple[bool, str, dict[str, Any] | None]

Turn in a completed quest and grant rewards.

Wrapped in critical_write("quest.turn_in") so reward grants (xp/gold/items/reputation) and quest-state transitions are flushed together. On timeout/overload, returns a retry-oriented failure tuple and the caller should leave the quest in COMPLETE state for retry.

reconcile_quest_completions `async` ¶

reconcile_quest_completions(character: Character) -> None

Repair the quest-log projection from authoritative completion markers.

Lazy on-access recovery for the cross-store durability invariant: the durable completion markers are the authority and the quest-log projection is reconciled to them. For every quest:<id>:<op> marker whose completion the projection does not reflect (a crash committed the marker + gold but lost the projection save), the projection is repaired (marked TURNED_IN, completion count raised via max) and any reward items the marker recorded are re-minted best-effort. Markers are DEDUP-only; the completion count authority remains quest_log.completion_count (raise-only).

Pruning is conservative: a marker is removed ONLY after the durable projection is re-confirmed to reflect it AND it is older than :data:_MARKER_RETENTION, so a fresh marker is never pruned out of its recovery window.

format_rewards ¶

format_rewards(rewards: QuestReward) -> str

Format rewards for display to player.

Guilds, factions, achievements, and social features.

from maid_classic_rpg.systems.social import GuildSystem

# Social system handles:
# - Guild creation and management
# - Faction reputation
# - Achievement tracking
# - Friend lists and ignore

social ¶

Social systems for player communication and groups.

AchievementSystem ¶

AchievementSystem(world: World)

Bases: System, StorageAware

System for tracking achievements.

Listens to game events and updates achievement progress accordingly.

set_storage ¶

set_storage(store: DocumentStore) -> None

Set the document store for persistence.

update `async` ¶

update(delta: float) -> None

No per-tick processing needed — achievements are event-driven via update_criterion().

startup `async` ¶

startup() -> None

Subscribe to relevant game events and load data.

shutdown `async` ¶

shutdown() -> None

Save data, clean up and unsubscribe from events.

register_achievement ¶

register_achievement(achievement: Achievement) -> None

Register an achievement.

unregister_achievement ¶

unregister_achievement(achievement_id: UUID) -> bool

Unregister an achievement.

get_achievement ¶

get_achievement(achievement_id: UUID) -> Achievement | None

Get an achievement by ID.

get_achievement_by_name ¶

get_achievement_by_name(name: str) -> Achievement | None

Get an achievement by name.

get_achievements_by_category ¶

get_achievements_by_category(
    category: AchievementCategory,
) -> list[Achievement]

Get all achievements in a category.

get_all_achievements ¶

get_all_achievements() -> list[Achievement]

Get all registered achievements.

get_progress ¶

get_progress(
    character_id: UUID, achievement_id: UUID
) -> CharacterAchievementProgress

Get a character's progress on an achievement.

get_all_progress ¶

get_all_progress(
    character_id: UUID,
) -> dict[UUID, CharacterAchievementProgress]

Get all achievement progress for a character.

get_earned_achievements ¶

get_earned_achievements(
    character_id: UUID,
) -> list[Achievement]

Get all achievements a character has earned.

get_total_points ¶

get_total_points(character_id: UUID) -> int

Get total achievement points for a character.

has_achievement ¶

has_achievement(
    character_id: UUID, achievement_id: UUID
) -> bool

Check if a character has earned an achievement.

update_criterion `async` ¶

update_criterion(
    character_id: UUID,
    criterion_type: str,
    target_id: UUID | None = None,
    target_type: str = "",
    amount: int = 1,
) -> list[Achievement]

Update progress on all matching criteria.

Returns:

Type	Description
`list[Achievement]`	List of newly completed achievements

grant_achievement `async` ¶

grant_achievement(
    character_id: UUID, achievement_id: UUID
) -> bool

Manually grant an achievement to a character.

Returns:

Type	Description
`bool`	True if achievement was newly granted

claim_rewards `async` ¶

claim_rewards(
    character_id: UUID, achievement_id: UUID
) -> list[AchievementReward]

Claim rewards for a completed achievement.

Returns:

Type	Description
`list[AchievementReward]`	List of rewards granted

clear_character_progress ¶

clear_character_progress(character_id: UUID) -> None

Clear all achievement progress for a character (for cleanup).

FactionSystem ¶

FactionSystem(world: World)

Bases: System, StorageAware

System for tracking faction reputation.

Manages faction definitions and per-character reputation values.

set_storage ¶

set_storage(store: DocumentStore) -> None

Set the document store for persistence.

update `async` ¶

update(delta: float) -> None

Slowly decay reputation toward neutral each tick.

Decay is ~1 point per 10 real-time minutes. We accumulate elapsed time and apply integer decrements when the accumulator exceeds the threshold.

startup `async` ¶

startup() -> None

Load factions and reputations from storage on startup.

shutdown `async` ¶

shutdown() -> None

Save factions and reputations to storage on shutdown.

register_faction ¶

register_faction(faction: Faction) -> None

Register a faction.

unregister_faction ¶

unregister_faction(faction_id: UUID) -> bool

Unregister a faction.

get_faction ¶

get_faction(faction_id: UUID) -> Faction | None

Get a faction by ID.

get_faction_by_name ¶

get_faction_by_name(name: str) -> Faction | None

Get a faction by name.

get_all_factions ¶

get_all_factions() -> list[Faction]

Get all registered factions.

get_reputation ¶

get_reputation(
    character_id: UUID, faction_id: UUID
) -> CharacterReputation

Get a character's reputation with a faction.

get_reputation_level ¶

get_reputation_level(
    character_id: UUID, faction_id: UUID
) -> ReputationLevel

Get a character's reputation level with a faction.

get_all_reputations ¶

get_all_reputations(
    character_id: UUID,
) -> dict[UUID, CharacterReputation]

Get all reputations for a character.

modify_reputation `async` ¶

modify_reputation(
    character_id: UUID,
    faction_id: UUID,
    amount: int,
    propagate: bool = True,
) -> tuple[int, ReputationLevel | None]

Modify a character's reputation with a faction.

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	Character whose reputation to modify	required
`faction_id`	`UUID`	Faction to modify reputation with	required
`amount`	`int`	Amount to add (negative to subtract)	required
`propagate`	`bool`	If True, also modify related factions	`True`

Returns:

Type	Description
`tuple[int, ReputationLevel \| None]`	Tuple of (new_value, new_level if changed)

set_reputation ¶

set_reputation(
    character_id: UUID, faction_id: UUID, value: int
) -> ReputationLevel

Set a character's reputation to a specific value.

Returns:

Type	Description
`ReputationLevel`	The new reputation level

get_unlocked_rewards ¶

get_unlocked_rewards(
    character_id: UUID, faction_id: UUID
) -> list[FactionReward]

Get rewards a character has unlocked with a faction.

get_npc_reaction ¶

get_npc_reaction(
    character_id: UUID, npc_faction_id: UUID
) -> str

Get how an NPC should react based on reputation.

Returns:

Type	Description
`str`	Reaction type: "attack", "refuse", "neutral", "friendly", "eager"

get_price_modifier ¶

get_price_modifier(
    character_id: UUID, faction_id: UUID
) -> float

Get shop price modifier based on reputation.

Returns:

Type	Description
`float`	Multiplier for prices (1.0 = normal, 0.8 = 20% discount)

can_access_vendor ¶

can_access_vendor(
    character_id: UUID,
    faction_id: UUID,
    required_level: ReputationLevel = NEUTRAL,
) -> bool

Check if a character can access a faction vendor.

clear_character_reputations ¶

clear_character_reputations(character_id: UUID) -> None

Clear all reputations for a character (for cleanup).

GroupSystem ¶

GroupSystem(
    world: World, sessions: SessionManager | None = None
)

Bases: System

Manages player groups/parties.

Features: - Group creation and management - Follow leader movement - XP sharing - Loot distribution modes - Group chat (gtell)

set_sessions ¶

set_sessions(sessions: SessionManager) -> None

Set the session manager after initialization.

startup `async` ¶

startup() -> None

Subscribe to movement events for follow.

shutdown `async` ¶

shutdown() -> None

Unsubscribe from events.

update `async` ¶

update(_delta: float) -> None

Remove stale members and dissolve empty groups.

create_group ¶

create_group(
    leader_id: UUID, leader_name: str
) -> tuple[Group | None, str]

Create a new group with player as leader.

disband_group ¶

disband_group(player_id: UUID) -> tuple[bool, str]

Disband a group. Only leader can disband.

invite_player `async` ¶

invite_player(
    inviter_id: UUID,
    inviter_name: str,
    invitee_id: UUID,
    invitee_name: str,
) -> tuple[bool, str]

Invite a player to group.

accept_invite `async` ¶

accept_invite(
    player_id: UUID, player_name: str
) -> tuple[bool, str]

Accept a group invitation.

decline_invite ¶

decline_invite(player_id: UUID) -> tuple[bool, str]

Decline a group invitation.

leave_group `async` ¶

leave_group(player_id: UUID) -> tuple[bool, str]

Leave current group.

kick_member `async` ¶

kick_member(
    kicker_id: UUID, target_id: UUID
) -> tuple[bool, str]

Kick a member from the group.

promote_member ¶

promote_member(
    promoter_id: UUID, target_id: UUID
) -> tuple[bool, str]

Promote a member to assistant.

demote_member ¶

demote_member(
    demoter_id: UUID, target_id: UUID
) -> tuple[bool, str]

Demote an assistant to member.

transfer_leadership ¶

transfer_leadership(
    current_leader_id: UUID, new_leader_id: UUID
) -> tuple[bool, str]

Transfer group leadership to another member.

set_follow ¶

set_follow(
    follower_id: UUID, leader_id: UUID
) -> tuple[bool, str]

Set a player to follow another.

stop_following ¶

stop_following(player_id: UUID) -> tuple[bool, str]

Stop following.

group_tell `async` ¶

group_tell(
    sender_id: UUID, sender_name: str, message: str
) -> tuple[bool, str]

Send message to group.

set_loot_mode ¶

set_loot_mode(
    player_id: UUID, mode: LootMode
) -> tuple[bool, str]

Set group loot mode. Leader only.

get_looter ¶

get_looter(group_id: UUID) -> UUID | None

Get the next player who should loot (for round-robin).

set_xp_share ¶

set_xp_share(
    player_id: UUID, enabled: bool
) -> tuple[bool, str]

Enable or disable XP sharing. Leader only.

calculate_xp_share ¶

calculate_xp_share(
    group_id: UUID, base_xp: int, _source_level: int
) -> dict[UUID, int]

Calculate XP distribution for group members.

get_player_group ¶

get_player_group(player_id: UUID) -> Group | None

Get the group a player is in.

get_group ¶

get_group(group_id: UUID) -> Group | None

Get group by ID.

is_in_group ¶

is_in_group(player_id: UUID) -> bool

Check if player is in a group.

get_pending_invite ¶

get_pending_invite(
    player_id: UUID,
) -> tuple[UUID, UUID] | None

Get pending invite for a player (group_id, inviter_id).

GuildBankError ¶

Bases: GuildError

Error specific to guild bank operations.

GuildError ¶

Bases: Exception

Base exception for guild operations.

GuildFullError ¶

Bases: GuildError

Guild is at maximum capacity.

GuildNotFoundError ¶

Bases: GuildError

Guild does not exist.

GuildPermissionError ¶

Bases: GuildError

Character lacks permission for operation.

GuildSystem ¶

GuildSystem(world: World)

Bases: System, StorageAware

System for managing player guilds.

Handles guild creation, membership, ranks, and progression.

set_storage ¶

set_storage(store: DocumentStore) -> None

Set the document store for persistence.

set_session_provider ¶

set_session_provider(
    provider: Callable[[], list[Any]],
) -> None

Set the function used to retrieve online sessions for chat delivery.

update `async` ¶

update(_delta: float) -> None

Persist dirty guilds and prune stale invite entries.

startup `async` ¶

startup() -> None

Load guilds from storage on startup.

shutdown `async` ¶

shutdown() -> None

Save guilds to storage on shutdown.

get_guild ¶

get_guild(guild_id: UUID) -> Guild | None

Get a guild by ID.

get_guild_by_name ¶

get_guild_by_name(name: str) -> Guild | None

Get a guild by name (case-insensitive).

get_character_guild ¶

get_character_guild(character_id: UUID) -> Guild | None

Get the guild a character belongs to.

is_in_guild ¶

is_in_guild(character_id: UUID) -> bool

Check if a character is in any guild.

get_all_guilds ¶

get_all_guilds() -> list[Guild]

Get all guilds.

create_guild `async` ¶

create_guild(
    name: str, tag: str, leader_id: UUID, leader_name: str
) -> Guild

Create a new guild.

Parameters:

Name	Type	Description	Default
`name`	`str`	Guild name (must be unique)	required
`tag`	`str`	Short guild tag (2-5 chars)	required
`leader_id`	`UUID`	Character ID of the founder	required
`leader_name`	`str`	Character name of the founder	required

Returns:

Type	Description
`Guild`	The created Guild

Raises:

Type	Description
`GuildError`	If character already in guild or name taken

disband_guild `async` ¶

disband_guild(guild_id: UUID, requester_id: UUID) -> None

Disband a guild.

Parameters:

Name	Type	Description	Default
`guild_id`	`UUID`	Guild to disband	required
`requester_id`	`UUID`	Character requesting disbandment	required

Raises:

Type	Description
`GuildNotFoundError`	If guild doesn't exist
`GuildPermissionError`	If requester lacks permission

invite_member `async` ¶

invite_member(
    guild_id: UUID, inviter_id: UUID, invitee_id: UUID
) -> None

Invite a player to the guild.

Parameters:

Name	Type	Description	Default
`guild_id`	`UUID`	Guild doing the inviting	required
`inviter_id`	`UUID`	Character sending the invite	required
`invitee_id`	`UUID`	Character being invited	required

Raises:

Type	Description
`GuildNotFoundError`	If guild doesn't exist
`GuildPermissionError`	If inviter lacks permission
`GuildError`	If invitee already in a guild or guild full

get_pending_invites ¶

get_pending_invites(character_id: UUID) -> list[Guild]

Get all pending guild invites for a character.

accept_invite `async` ¶

accept_invite(
    character_id: UUID, character_name: str, guild_id: UUID
) -> Guild

Accept a guild invitation.

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	Character accepting	required
`character_name`	`str`	Character's name	required
`guild_id`	`UUID`	Guild to join	required

Returns:

Type	Description
`Guild`	The joined Guild

Raises:

Type	Description
`GuildError`	If no pending invite or already in guild

decline_invite `async` ¶

decline_invite(character_id: UUID, guild_id: UUID) -> None

Decline a guild invitation.

leave_guild `async` ¶

leave_guild(character_id: UUID) -> None

Leave current guild.

Raises:

Type	Description
`GuildError`	If not in a guild or is leader

kick_member `async` ¶

kick_member(
    guild_id: UUID, kicker_id: UUID, target_id: UUID
) -> None

Kick a member from the guild.

Raises:

Type	Description
`GuildPermissionError`	If kicker lacks permission or outranked

promote_member `async` ¶

promote_member(
    guild_id: UUID, promoter_id: UUID, target_id: UUID
) -> str

Promote a member to the next rank.

Returns:

Type	Description
`str`	New rank name

demote_member `async` ¶

demote_member(
    guild_id: UUID, demoter_id: UUID, target_id: UUID
) -> str

Demote a member to the previous rank.

Returns:

Type	Description
`str`	New rank name

transfer_leadership `async` ¶

transfer_leadership(
    guild_id: UUID,
    current_leader_id: UUID,
    new_leader_id: UUID,
) -> None

Transfer guild leadership to another member.

add_guild_experience `async` ¶

add_guild_experience(guild_id: UUID, amount: int) -> bool

Add experience to a guild.

Returns:

Type	Description
`bool`	True if guild leveled up

get_bank_tabs ¶

get_bank_tabs(
    guild_id: UUID, character_id: UUID
) -> list[GuildBankTab]

Get accessible bank tabs for a character.

get_tab_contents ¶

get_tab_contents(
    guild_id: UUID, character_id: UUID, tab_index: int
) -> tuple[GuildBankTab, list[UUID]]

Get contents of a specific bank tab.

Returns:

Type	Description
`tuple[GuildBankTab, list[UUID]]`	Tuple of (tab, item_ids)

deposit_item `async` ¶

deposit_item(
    guild_id: UUID,
    character_id: UUID,
    character_name: str,
    item_id: UUID,
    item_name: str,
    tab_index: int = 0,
    character: Character | None = None,
    item_weight: float = 0.0,
) -> None

Deposit an item into the guild bank.

Parameters:

Name	Type	Description	Default
`guild_id`	`UUID`	Guild to deposit into.	required
`character_id`	`UUID`	Character depositing.	required
`character_name`	`str`	Character's name.	required
`item_id`	`UUID`	Item to deposit.	required
`item_name`	`str`	Item's name.	required
`tab_index`	`int`	Bank tab index.	`0`
`character`	`Character \| None`	Character object for atomic inventory removal.	`None`
`item_weight`	`float`	Weight of the item for proper weight tracking.	`0.0`

Raises:

Type	Description
`GuildBankError`	If character is not provided.

withdraw_item `async` ¶

withdraw_item(
    guild_id: UUID,
    character_id: UUID,
    character_name: str,
    item_id: UUID,
    item_name: str,
    tab_index: int = 0,
    character: Character | None = None,
    item_weight: float = 0.0,
) -> None

Withdraw an item from the guild bank.

Raises:

Type	Description
`GuildBankError`	If character is not provided.

deposit_gold `async` ¶

deposit_gold(
    guild_id: UUID,
    character_id: UUID,
    character_name: str,
    amount: int,
    character: Character | None = None,
) -> None

Deposit gold into the guild bank.

Parameters:

Name	Type	Description	Default
`guild_id`	`UUID`	Guild to deposit into.	required
`character_id`	`UUID`	Character depositing.	required
`character_name`	`str`	Character's name.	required
`amount`	`int`	Amount to deposit.	required
`character`	`Character \| None`	Character object for atomic gold deduction.	`None`

Raises:

Type	Description
`GuildBankError`	If character is not provided.

withdraw_gold `async` ¶

withdraw_gold(
    guild_id: UUID,
    character_id: UUID,
    character_name: str,
    amount: int,
    character: Character | None = None,
) -> None

Withdraw gold from the guild bank.

Raises:

Type	Description
`GuildBankError`	If character is not provided.

get_bank_logs ¶

get_bank_logs(
    guild_id: UUID, character_id: UUID, limit: int = 50
) -> list[GuildBankLog]

Get recent bank transaction logs.

get_gold_balance ¶

get_gold_balance(guild_id: UUID, character_id: UUID) -> int

Get the guild's gold balance.

send_guild_message `async` ¶

send_guild_message(
    guild_id: UUID,
    sender_id: UUID,
    sender_name: str,
    message: str,
    is_announcement: bool = False,
) -> None

Send a message to the guild chat.

set_motd ¶

set_motd(
    guild_id: UUID, character_id: UUID, motd: str
) -> None

Set the guild message of the day.

get_motd ¶

get_motd(guild_id: UUID) -> str

Get the guild message of the day.

register_social_commands ¶

register_social_commands(
    channels: ChannelSystem,
    groups: GroupSystem,
    friends: FriendsSystem,
    mail: MailSystem,
) -> None

Register all social commands.

register_guild_commands ¶

register_guild_commands(guilds: GuildSystem) -> None

Register all guild-related player commands.

Parameters:

Name	Type	Description	Default
`guilds`	`GuildSystem`	The GuildSystem instance to delegate operations to.	required

PvP System¶

Player versus player combat with arenas and rankings.

from maid_classic_rpg.systems.pvp import PvPSystem

# PvP system handles:
# - Duel challenges
# - Arena matchmaking
# - Ranking and leaderboards
# - PvP zones and rules

pvp ¶

PvP systems for player-vs-player combat.

ArenaSystem ¶

ArenaSystem(world: World)

Bases: System

System for managing arena matches and queues.

Features: - Queue system with rating-based matchmaking - Multiple match types (1v1, team, FFA) - Dedicated arena rooms - Rating changes based on match outcome

Arena Lifecycle: 1. QUEUED - Players/teams wait for opponents 2. PREPARING - Match found, players teleport to arena 3. IN_PROGRESS - Match active 4. COMPLETED - Match ends, ratings updated

startup `async` ¶

startup() -> None

Initialize arena system.

shutdown `async` ¶

shutdown() -> None

Unsubscribe from events and clean up.

update `async` ¶

update(delta: float) -> None

Update tick - process queues and match timers.

register_arena_room ¶

register_arena_room(
    room_id: UUID, match_types: list[ArenaMatchType]
) -> None

Register a room as an arena for specific match types.

Parameters:

Name	Type	Description	Default
`room_id`	`UUID`	Room to register	required
`match_types`	`list[ArenaMatchType]`	Types of matches this room supports	required

join_queue `async` ¶

join_queue(
    character_ids: list[UUID], match_type: ArenaMatchType
) -> tuple[bool, str]

Join the arena queue.

Parameters:

Name	Type	Description	Default
`character_ids`	`list[UUID]`	Characters joining as a team/solo	required
`match_type`	`ArenaMatchType`	Type of match to queue for	required

Returns:

Type	Description
`tuple[bool, str]`	Tuple of (success, message)

leave_queue `async` ¶

leave_queue(character_id: UUID) -> tuple[bool, str]

Leave the arena queue.

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	Character leaving queue	required

Returns:

Type	Description
`tuple[bool, str]`	Tuple of (success, message)

forfeit_match `async` ¶

forfeit_match(character_id: UUID) -> tuple[bool, str]

Forfeit an active arena match.

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	Character forfeiting	required

Returns:

Type	Description
`tuple[bool, str]`	Tuple of (success, message)

get_queue_position ¶

get_queue_position(
    character_id: UUID,
) -> tuple[int, int] | None

Get queue position for a character.

Returns:

Type	Description
`tuple[int, int] \| None`	Tuple of (position, total_in_queue) or None if not queued

get_active_match ¶

get_active_match(character_id: UUID) -> ArenaMatch | None

Get active match for a character.

is_in_arena ¶

is_in_arena(character_id: UUID) -> bool

Check if character is in an arena match.

get_queue_status ¶

get_queue_status() -> dict[str, int]

Get current queue sizes for all match types.

DuelSystem ¶

DuelSystem(world: World)

Bases: System

System for managing duels between players.

Duels are consensual PvP fights with special rules: - Both players must agree to fight - Combat stops at a configurable HP threshold (default 1 HP) - No permanent death or item loss - Optional gold betting - Spectators can place bets

Duel Lifecycle: 1. PENDING - Challenger issues challenge, target has time to respond 2. ACCEPTED - Challenge accepted, countdown begins 3. IN_PROGRESS - Duel is active, combat proceeds 4. COMPLETED - Duel ends when HP threshold reached or forfeit

startup `async` ¶

startup() -> None

Subscribe to relevant events.

shutdown `async` ¶

shutdown() -> None

Clean up on shutdown.

update `async` ¶

update(delta: float) -> None

Update tick - handle timeouts and countdowns.

challenge `async` ¶

challenge(
    challenger_id: UUID,
    target_id: UUID,
    room_id: UUID,
    bet_amount: int = 0,
    stop_at_hp: int | None = None,
) -> tuple[bool, str, UUID | None]

Issue a duel challenge.

Parameters:

Name	Type	Description	Default
`challenger_id`	`UUID`	Character issuing challenge	required
`target_id`	`UUID`	Character being challenged	required
`room_id`	`UUID`	Room where duel will take place	required
`bet_amount`	`int`	Gold to wager (0 for no bet)	`0`
`stop_at_hp`	`int \| None`	HP threshold to stop fight	`None`

Returns:

Type	Description
`tuple[bool, str, UUID \| None]`	Tuple of (success, message, duel_id)

accept `async` ¶

accept(character_id: UUID) -> tuple[bool, str]

Accept a pending duel challenge.

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	Character accepting (must be target)	required

Returns:

Type	Description
`tuple[bool, str]`	Tuple of (success, message)

decline `async` ¶

decline(character_id: UUID) -> tuple[bool, str]

Decline a pending duel challenge.

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	Character declining (must be target)	required

Returns:

Type	Description
`tuple[bool, str]`	Tuple of (success, message)

forfeit `async` ¶

forfeit(character_id: UUID) -> tuple[bool, str]

Forfeit an active duel.

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	Character forfeiting	required

Returns:

Type	Description
`tuple[bool, str]`	Tuple of (success, message)

place_bet `async` ¶

place_bet(
    duel_id: UUID,
    bettor_id: UUID,
    backed_id: UUID,
    amount: int,
) -> tuple[bool, str]

Place a bet on an active duel.

Parameters:

Name	Type	Description	Default
`duel_id`	`UUID`	Duel to bet on	required
`bettor_id`	`UUID`	Character placing bet	required
`backed_id`	`UUID`	Character being bet on	required
`amount`	`int`	Gold to wager	required

Returns:

Type	Description
`tuple[bool, str]`	Tuple of (success, message)

get_active_duel ¶

get_active_duel(character_id: UUID) -> DuelChallenge | None

Get active duel for a character.

get_pending_challenges ¶

get_pending_challenges(
    character_id: UUID,
) -> list[DuelChallenge]

Get all pending challenges for a character (as target).

is_in_duel ¶

is_in_duel(character_id: UUID) -> bool

Check if character is in an active duel.

get_opponent ¶

get_opponent(character_id: UUID) -> UUID | None

Get duel opponent for a character.

get_countdown ¶

get_countdown(duel_id: UUID) -> float

Get remaining countdown time for a duel.

ArenaMatchEndEvent `dataclass` ¶

ArenaMatchEndEvent(
    match_id: UUID | None = None,
    match_type: ArenaMatchType | None = None,
    winning_team: str | None = None,
    individual_winner: UUID | None = None,
    rating_changes: dict[str, int] = dict(),
)

Bases: Event

Emitted when an arena match ends.

ArenaMatchFoundEvent `dataclass` ¶

ArenaMatchFoundEvent(
    match_id: UUID | None = None,
    match_type: ArenaMatchType | None = None,
    team_a: list[UUID] = list(),
    team_b: list[UUID] = list(),
)

Bases: Event

Emitted when a match is found for queued players.

ArenaMatchStateChangedEvent `dataclass` ¶

ArenaMatchStateChangedEvent(
    match_id: UUID | None = None,
    old_state: ArenaMatchState | None = None,
    new_state: ArenaMatchState | None = None,
)

Bases: Event

Emitted when an arena match state changes.

ArenaQueueJoinEvent `dataclass` ¶

ArenaQueueJoinEvent(
    character_ids: list[UUID] = list(),
    match_type: ArenaMatchType | None = None,
)

Bases: Event

Emitted when a player/team joins arena queue.

ArenaQueueLeaveEvent `dataclass` ¶

ArenaQueueLeaveEvent(
    character_ids: list[UUID] = list(),
    match_type: ArenaMatchType | None = None,
)

Bases: Event

Emitted when a player/team leaves arena queue.

DuelBetPlacedEvent `dataclass` ¶

DuelBetPlacedEvent(
    duel_id: UUID | None = None,
    bettor_id: UUID | None = None,
    backed_id: UUID | None = None,
    amount: int = 0,
)

Bases: Event

Emitted when a spectator places a bet on a duel.

DuelChallengeEvent `dataclass` ¶

DuelChallengeEvent(
    duel_id: UUID | None = None,
    challenger_id: UUID | None = None,
    target_id: UUID | None = None,
    room_id: UUID | None = None,
    bet_amount: int = 0,
)

Bases: Event

Emitted when a duel challenge is issued.

DuelEndEvent `dataclass` ¶

DuelEndEvent(
    duel_id: UUID | None = None,
    winner_id: UUID | None = None,
    loser_id: UUID | None = None,
    reason: str = "",
)

Bases: Event

Emitted when a duel ends.

DuelResponseEvent `dataclass` ¶

DuelResponseEvent(
    duel_id: UUID | None = None,
    challenger_id: UUID | None = None,
    target_id: UUID | None = None,
    accepted: bool = False,
)

Bases: Event

Emitted when a duel challenge is accepted or declined.

DuelStartEvent `dataclass` ¶

DuelStartEvent(
    duel_id: UUID | None = None,
    challenger_id: UUID | None = None,
    target_id: UUID | None = None,
    room_id: UUID | None = None,
)

Bases: Event

Emitted when a duel begins.

PvPDeathEvent `dataclass` ¶

PvPDeathEvent(
    victim_id: UUID | None = None,
    killer_id: UUID | None = None,
    room_id: UUID | None = None,
    loot_dropped: list[UUID] = list(),
)

Bases: Event

Emitted when a player dies in PvP.

PvPFlagChangedEvent `dataclass` ¶

PvPFlagChangedEvent(
    character_id: UUID | None = None,
    old_flag: PvPFlag | None = None,
    new_flag: PvPFlag | None = None,
)

Bases: Event

Emitted when a character's PvP flag changes.

PvPFlaggedEvent `dataclass` ¶

PvPFlaggedEvent(
    character_id: UUID | None = None,
    target_id: UUID | None = None,
    duration_seconds: int = 0,
)

Bases: Event

Emitted when a character becomes temporarily flagged.

PvPKillEvent `dataclass` ¶

PvPKillEvent(
    killer_id: UUID | None = None,
    victim_id: UUID | None = None,
    room_id: UUID | None = None,
    kill_streak: int = 0,
)

Bases: Event

Emitted when a player kills another player.

RankingChangedEvent `dataclass` ¶

RankingChangedEvent(
    character_id: UUID | None = None,
    old_rating: int = 0,
    new_rating: int = 0,
    old_rank: int = 0,
    new_rank: int = 0,
)

Bases: Event

Emitted when a player's ranking changes.

SeasonEndEvent `dataclass` ¶

SeasonEndEvent(
    season_id: UUID | None = None,
    season_name: str = "",
    final_rankings: list[tuple[UUID, int, int]] = list(),
)

Bases: Event

Emitted when a PvP season ends.

TitleEarnedEvent `dataclass` ¶

TitleEarnedEvent(
    character_id: UUID | None = None,
    title_id: str = "",
    title_name: str = "",
)

Bases: Event

Emitted when a player earns a PvP title.

ArenaMatch ¶

Bases: MAIDBaseModel

An arena match instance.

all_participants `property` ¶

all_participants: list[UUID]

Get all participants in the match.

ArenaMatchState ¶

Bases: str, Enum

Arena match lifecycle states.

ArenaMatchType ¶

Bases: str, Enum

Types of arena matches.

ArenaQueueEntry ¶

Bases: MAIDBaseModel

An entry in the arena queue.

team_size `property` ¶

team_size: int

Get number of players in this entry.

DuelBet ¶

Bases: MAIDBaseModel

A bet placed on a duel by a spectator.

DuelChallenge ¶

Bases: MAIDBaseModel

A duel challenge between two players.

is_expired ¶

is_expired() -> bool

Check if challenge has expired.

DuelState ¶

Bases: str, Enum

Duel lifecycle states.

PvPCharacterState ¶

Bases: MAIDBaseModel

PvP state for a single character.

is_pvp_enabled `property` ¶

is_pvp_enabled: bool

Check if character can engage in PvP.

is_flagged `property` ¶

is_flagged: bool

Check if character is temporarily flagged.

kill_death_ratio `property` ¶

kill_death_ratio: float

Calculate K/D ratio.

PvPFlag ¶

Bases: str, Enum

PvP flag states.

PvPLootMode ¶

Bases: str, Enum

Loot rules for PvP death.

PvPSeason ¶

Bases: MAIDBaseModel

A competitive PvP season.

is_current ¶

is_current() -> bool

Check if this season is currently active.

PvPTitle ¶

Bases: BaseModel

A title earned through PvP achievements.

RankingEntry ¶

Bases: MAIDBaseModel

A single ranking entry on the leaderboard.

win_rate `property` ¶

win_rate: float

Calculate win percentage.

PvPSystem ¶

PvPSystem(world: World)

Bases: System

System for managing PvP state, rules, and combat modifications.

Responsibilities: - Track PvP flag state per character - Validate PvP combat is allowed - Apply PvP-specific combat modifiers - Handle PvP death consequences - Manage temporary flagging

The PvP system hooks into combat events to enforce rules and track outcomes for the ranking system.

startup `async` ¶

startup() -> None

Subscribe to combat events.

shutdown `async` ¶

shutdown() -> None

Clean up subscriptions.

update `async` ¶

update(delta: float) -> None

Update tick - check for expired flags and protections.

get_state ¶

get_state(character_id: UUID) -> PvPCharacterState

Get or create PvP state for a character.

enable_pvp `async` ¶

enable_pvp(character_id: UUID) -> bool

Enable PvP for a character.

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	Character to enable PvP for	required

Returns:

Type	Description
`bool`	True if PvP was enabled, False if already enabled or blocked

disable_pvp `async` ¶

disable_pvp(character_id: UUID) -> bool

Disable PvP for a character.

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	Character to disable PvP for	required

Returns:

Type	Description
`bool`	True if PvP was disabled, False if flagged or in combat

can_attack ¶

can_attack(
    attacker_id: UUID, target_id: UUID, room_id: UUID
) -> tuple[bool, str]

Check if PvP attack is allowed.

Parameters:

Name	Type	Description	Default
`attacker_id`	`UUID`	Character attempting attack	required
`target_id`	`UUID`	Target character	required
`room_id`	`UUID`	Room where combat would occur	required

Returns:

Type	Description
`tuple[bool, str]`	Tuple of (allowed, reason_if_denied)

get_damage_modifier ¶

get_damage_modifier(
    attacker_id: UUID, target_id: UUID
) -> float

Get damage modifier for PvP combat.

Returns a multiplier to apply to damage dealt. This allows balancing PvP separately from PvE.

Parameters:

Name	Type	Description	Default
`attacker_id`	`UUID`	Character dealing damage	required
`target_id`	`UUID`	Character receiving damage	required

Returns:

Type	Description
`float`	Damage multiplier (e.g., 0.7 for 70% damage)

is_pvp_combat ¶

is_pvp_combat(attacker_id: UUID, target_id: UUID) -> bool

Check if combat between two entities is PvP.

is_in_pvp_combat ¶

is_in_pvp_combat(character_id: UUID) -> bool

Check if character is currently in PvP combat.

record_kill `async` ¶

record_kill(
    killer_id: UUID, victim_id: UUID, room_id: UUID
) -> None

Record a PvP kill and update statistics.

Parameters:

Name	Type	Description	Default
`killer_id`	`UUID`	Character who got the kill	required
`victim_id`	`UUID`	Character who died	required
`room_id`	`UUID`	Room where kill occurred	required

handle_pvp_death `async` ¶

handle_pvp_death(
    victim_id: UUID, killer_id: UUID | None, room_id: UUID
) -> list[UUID]

Handle PvP death consequences.

Parameters:

Name	Type	Description	Default
`victim_id`	`UUID`	Character who died	required
`killer_id`	`UUID \| None`	Character who killed them (if any)	required
`room_id`	`UUID`	Room where death occurred	required

Returns:

Type	Description
`list[UUID]`	List of item IDs that were dropped as loot

get_protection_remaining ¶

get_protection_remaining(character_id: UUID) -> float

Get remaining protection time in seconds.

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	Character to check	required

Returns:

Type	Description
`float`	Seconds remaining, or 0 if not protected

get_flag_remaining ¶

get_flag_remaining(character_id: UUID) -> float

Get remaining flag time in seconds.

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	Character to check	required

Returns:

Type	Description
`float`	Seconds remaining, or 0 if not flagged

RankingSystem ¶

RankingSystem(world: World)

Bases: System, StorageAware

System for managing PvP rankings and leaderboards.

Features: - Arena rating leaderboards - Kill/death rankings - Seasonal resets - Title rewards

Rankings are updated in response to PvP events and periodically persisted to the database.

set_storage ¶

set_storage(store: DocumentStore) -> None

Receive storage from the engine via StorageAware protocol.

startup `async` ¶

startup() -> None

Subscribe to PvP events and load data.

shutdown `async` ¶

shutdown() -> None

Unsubscribe from events and persist rankings before shutdown.

update `async` ¶

update(delta: float) -> None

Periodic persistence and season checks.

get_ranking ¶

get_ranking(character_id: UUID) -> RankingEntry | None

Get ranking entry for a character.

get_leaderboard ¶

get_leaderboard(
    offset: int = 0,
    limit: int = 10,
    sort_by: str = "rating",
) -> list[RankingEntry]

Get leaderboard entries.

Parameters:

Name	Type	Description	Default
`offset`	`int`	Starting position	`0`
`limit`	`int`	Maximum entries to return	`10`
`sort_by`	`str`	Field to sort by ('rating', 'wins', 'kills')	`'rating'`

Returns:

Type	Description
`list[RankingEntry]`	List of ranking entries

get_rank ¶

get_rank(character_id: UUID) -> int

Get a character's current rank position.

get_earned_titles ¶

get_earned_titles(character_id: UUID) -> list[PvPTitle]

Get all titles earned by a character.

get_available_titles ¶

get_available_titles(character_id: UUID) -> list[PvPTitle]

Get titles a character qualifies for but hasn't earned.

claim_title `async` ¶

claim_title(
    character_id: UUID, title_id: str
) -> tuple[bool, str]

Claim an earned title.

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	Character claiming title	required
`title_id`	`str`	Title to claim	required

Returns:

Type	Description
`tuple[bool, str]`	Tuple of (success, message)

get_current_season ¶

get_current_season() -> PvPSeason | None

Get the current PvP season.

start_season `async` ¶

start_season(
    name: str,
    duration_days: int = 90,
    reward_tiers: dict[str, Any] | None = None,
) -> PvPSeason

Start a new PvP season.

Parameters:

Name	Type	Description	Default
`name`	`str`	Season name	required
`duration_days`	`int`	Length of season	`90`
`reward_tiers`	`dict[str, Any] \| None`	Rewards by rank tier	`None`

Returns:

Type	Description
`PvPSeason`	The new season

World System¶

Time cycles, weather, and dynamic world events.

from maid_classic_rpg.systems.world import GameTimeSystem

# World system handles:
# - Day/night cycles
# - Weather patterns
# - Seasonal changes
# - World events

world ¶

World dynamics systems for MAID.

This package contains systems for managing dynamic world elements: - Time system: Game clock, day/night cycles, seasons - Weather system: Regional weather patterns with gameplay effects - World events: Random encounters, world bosses, invasions - Area resets: Periodic respawning of monsters and items - Ecosystem: Predator/prey relationships, population dynamics - Ambient messages: Biome flavor text delivery to players - Wilderness spawn: Bridge from biome definitions to real ECS entities

EcosystemSystem ¶

EcosystemSystem(
    world: World, config: EcosystemConfig | None = None
)

Bases: System

System for simulating dynamic ecosystems.

Manages predator/prey relationships, resource consumption, reproduction, migration, and population dynamics.

startup `async` ¶

startup() -> None

Initialize ecosystem system.

shutdown `async` ¶

shutdown() -> None

Shut down ecosystem system.

update `async` ¶

update(delta: float) -> None

Update ecosystem simulation.

register_species ¶

register_species(species: SpeciesDefinition) -> None

Register a species definition.

unregister_species ¶

unregister_species(species_id: UUID) -> bool

Unregister a species definition.

add_resource_node ¶

add_resource_node(resource: ResourceNode) -> None

Add a resource node to the ecosystem.

Resolves the area that contains the resource's room so that resources and populations are indexed under the same key. Falls back to using the room_id directly when no area mapping is available.

remove_resource_node ¶

remove_resource_node(
    room_id: UUID, resource_type: ResourceType
) -> bool

Remove a resource node from the ecosystem.

initialize_population ¶

initialize_population(
    area_id: UUID, species_id: UUID, initial_count: int
) -> PopulationState | None

Initialize a population for a species in an area.

get_population_info ¶

get_population_info(area_id: UUID) -> dict[str, Any]

Get population information for an area.

get_ecosystem_status ¶

get_ecosystem_status() -> dict[str, Any]

Get overall ecosystem status.

get_resource_info ¶

get_resource_info(area_id: UUID) -> dict[str, Any]

Get resource information for an area.

WorldEventSystem ¶

WorldEventSystem(world: World)

Bases: System

System for managing world events.

Handles event scheduling, triggering, progression, and rewards.

startup `async` ¶

startup() -> None

Initialize the event system.

shutdown `async` ¶

shutdown() -> None

Shut down the event system.

update `async` ¶

update(delta: float) -> None

Update event system.

register_definition ¶

register_definition(
    definition: WorldEventDefinition,
) -> None

Register an event definition.

get_active_events ¶

get_active_events() -> list[WorldEventInstance]

Get all currently active events.

get_event ¶

get_event(instance_id: UUID) -> WorldEventInstance | None

Get a specific event instance.

get_definition ¶

get_definition(
    definition_id: UUID,
) -> WorldEventDefinition | None

Get an event definition by ID.

trigger_event_by_name `async` ¶

trigger_event_by_name(name: str) -> UUID | None

Trigger an event by its name (admin command).

cancel_event `async` ¶

cancel_event(instance_id: UUID) -> bool

Cancel an active event (admin command).

record_kill ¶

record_kill(
    instance_id: UUID, template_id: str, killer_id: UUID
) -> None

Record a kill for event progress tracking.

schedule_event ¶

schedule_event(
    definition_id: UUID, game_time_minutes: int
) -> bool

Schedule an event to trigger at a specific game time.

get_event_info ¶

get_event_info(instance_id: UUID) -> dict[str, Any] | None

Get detailed information about an active event.

AreaResetSystem ¶

AreaResetSystem(world: World)

Bases: System

System for managing area resets.

Handles periodic respawning of monsters, items, and restoration of room states.

startup `async` ¶

startup() -> None

Initialize reset system.

shutdown `async` ¶

shutdown() -> None

Shut down reset system.

update `async` ¶

update(delta: float) -> None

Check and execute area resets.

register_config ¶

register_config(config: AreaResetConfig) -> None

Register a reset configuration for an area.

unregister_config ¶

unregister_config(area_id: UUID) -> bool

Unregister a reset configuration.

force_reset ¶

force_reset(area_id: UUID) -> bool

Force an immediate area reset (admin command).

get_reset_info ¶

get_reset_info(area_id: UUID) -> dict[str, Any] | None

Get reset information for an area.

get_all_configs ¶

get_all_configs() -> list[AreaResetConfig]

Get all registered reset configurations.

set_enabled ¶

set_enabled(area_id: UUID, enabled: bool) -> bool

Enable or disable resets for an area.

GameTimeSystem ¶

GameTimeSystem(
    world: World, config: TimeConfig | None = None
)

Bases: System

System managing game time progression.

The time system advances game time based on real elapsed time, scaled by a configurable ratio. It emits events for significant time changes that other systems can react to.

Example

system = GameTimeSystem(world)

Get current time¶

current = system.current_time print(f"It is {current.format_time()}")

Check time of day¶

if current.is_night: print("Darkness surrounds you.")

current_time `property` ¶

current_time: GameTime

Get the current game time.

config `property` ¶

config: TimeConfig

Get the time configuration.

time_of_day `property` ¶

time_of_day: TimeOfDay

Get current time of day.

season `property` ¶

season: Season

Get current season.

is_night `property` ¶

is_night: bool

Check if it's night.

set_storage ¶

set_storage(store: DocumentStore) -> None

Set the document store for persistence.

startup `async` ¶

startup() -> None

Initialize the time system, loading saved time if available.

shutdown `async` ¶

shutdown() -> None

Shut down the time system, saving current time.

update `async` ¶

update(delta: float) -> None

Advance game time based on elapsed real time.

Parameters:

Name	Type	Description	Default
`delta`	`float`	Real seconds since last tick	required

set_time ¶

set_time(game_time: GameTime) -> None

Set the current game time directly.

Use for admin commands or loading saved state.

get_visibility_modifier ¶

get_visibility_modifier() -> float

Get visibility modifier based on time of day.

Returns:

Type	Description
`float`	Multiplier for visibility (1.0 = full, 0.0 = none)

WeatherSystem ¶

WeatherSystem(world: World)

Bases: System

System for managing weather across regions.

The weather system tracks weather per region and handles: - Weather state progression - Weather change events - Periodic weather effects (damage, messages) - Weather effect queries for other systems

startup `async` ¶

startup() -> None

Initialize weather system.

shutdown `async` ¶

shutdown() -> None

Shut down weather system.

update `async` ¶

update(delta: float) -> None

Update weather states.

get_weather ¶

get_weather(
    region_id: UUID | None = None,
) -> RegionalWeather

Get weather for a region (or global if not specified).

get_weather_effects ¶

get_weather_effects(
    region_id: UUID | None = None,
) -> WeatherEffect

Get current weather effects for a region.

get_weather_description ¶

get_weather_description(
    region_id: UUID | None = None,
) -> str

Get a text description of current weather.

set_weather ¶

set_weather(
    region_id: UUID | None, weather_type: WeatherType
) -> None

Manually set weather (admin command).

register_region ¶

register_region(config: RegionWeatherConfig) -> None

Register a region with custom weather configuration.

WildernessSpawnAdapter ¶

WildernessSpawnAdapter(world: World)

Converts biome encounter/resource strings into real ECS entities.

Register this adapter's :meth:on_room_generated method as a callback on the WildernessManager so that whenever a wilderness room is created, monsters and resource nodes are spawned automatically.

Example::

adapter = WildernessSpawnAdapter(world)
wilderness_manager.register_room_callback(adapter.on_room_generated)

template_count `property` ¶

template_count: int

Number of loaded monster templates.

spawned_entity_count `property` ¶

spawned_entity_count: int

Total number of entities spawned across all rooms.

load_templates ¶

load_templates() -> None

Load monster templates from YAML data files.

get_template ¶

get_template(name: str) -> MonsterTemplate | None

Look up a monster template by encounter type string.

teardown ¶

teardown() -> None

Clean up all tracked spawned entities and internal state.

Iterates every coordinate that still has spawned entities and destroys them via the existing _cleanup_coord helper, then clears all internal tracking maps.

on_room_generated ¶

on_room_generated(
    room_id: UUID,
    coord: tuple[int, int],
    biome_name: str,
    biome: BiomeDefinition,
    spawned_content: SpawnedContent,
) -> None

Callback for WildernessManager room generation.

Spawns monster entities for encounters and resource nodes for resources listed in the SpawnedContent.

Parameters:

Name	Type	Description	Default
`room_id`	`UUID`	UUID of the generated room.	required
`coord`	`tuple[int, int]`	(x, y) coordinate of the room.	required
`biome_name`	`str`	Name of the biome.	required
`biome`	`BiomeDefinition`	The BiomeDefinition used.	required
`spawned_content`	`SpawnedContent`	Resources and encounters rolled by the manager.	required

on_room_cleaned_up ¶

on_room_cleaned_up(room_id: UUID) -> None

Clean up spawned entities when a wilderness room is destroyed.

Destroys all ECS entities that were spawned for the given room and unregisters them from the BehaviorSystem.

Parameters:

Name	Type	Description	Default
`room_id`	`UUID`	UUID of the room being cleaned up.	required

on_room_destroyed ¶

on_room_destroyed(
    room_id: UUID, coord: tuple[int, int]
) -> None

Cleanup callback matching WildernessManager's RoomCleanupCallback.

Called by WildernessManager before a room is destroyed during stale-room cleanup.

Parameters:

Name	Type	Description	Default
`room_id`	`UUID`	UUID of the room being destroyed.	required
`coord`	`tuple[int, int]`	(x, y) coordinate of the room.	required

NPC System¶

NPC behavior, schedules, and AI-powered dialogue.

from maid_classic_rpg.systems.npc import BehaviorSystem
from maid_classic_rpg.systems.npc.dialogue import NPCDialogueSystem

# NPC system handles:
# - NPC spawning and despawning
# - Behavior patterns
# - Schedules and routines
# - AI-powered conversations

npc ¶

NPC behavior, spawning, dialogue, and autonomy systems.

All public names are lazy-imported to avoid circular dependencies between this package and maid_classic_rpg.events.

AutonomySystem ¶

AutonomySystem(world: World)

Bases: BaseAutonomySystem

Selects and executes NPC autonomous actions each tick.

Extends BaseAutonomySystem with classic-RPG goal generation, utility scoring, LLM integration, and game-specific events.

startup `async` ¶

startup() -> None

Restore persisted state and wire goal/LLM subsystems.

shutdown `async` ¶

shutdown() -> None

Stop the off-tick LLM queue.

BarkLibrary ¶

BarkLibrary(
    templates: dict[str, list[BarkTemplate]] | None = None,
)

Loads and selects bark templates.

set_templates ¶

set_templates(
    templates: dict[str, list[BarkTemplate]],
) -> None

Replace bark templates.

select_bark ¶

select_bark(
    needs: NeedsComponent,
    mood: float,
    archetype_id: str,
    npc_id: UUID | None = None,
    now_minutes: float | None = None,
) -> str | None

Select bark text based on needs, mood, and cooldown.

BarkSystem ¶

BarkSystem(
    world: World,
    library: BarkLibrary | None = None,
    knowledge_manager: Any = None,
)

Bases: System

Emits ambient NPC barks in rooms containing players.

set_knowledge_manager ¶

set_knowledge_manager(km: Any) -> None

Set or replace the knowledge manager for knowledge-based barks.

BehaviorSystem ¶

BehaviorSystem(world: World)

Bases: BaseBehaviorSystem

System for executing NPC behaviors.

Extends BaseBehaviorSystem with classic-rpg-specific aggro logic, CombatComponent handling, and NPCAggroEvent/DamageDealtEvent subscriptions.

Processes: - Aggression checks (aggressive NPCs attack on sight) - Patrol/wander movement - Flee behavior when low health - Territory defense - Pack/ally assistance

startup `async` ¶

startup() -> None

Initialize behavior system.

shutdown `async` ¶

shutdown() -> None

Shutdown behavior system.

register_npc ¶

register_npc(
    entity_id: UUID,
    behavior: BehaviorConfig,
    territory_center: UUID | None = None,
) -> NPCState

Register an NPC for behavior processing.

Parameters:

Name	Type	Description	Default
`entity_id`	`UUID`	NPC entity ID	required
`behavior`	`BehaviorConfig`	Behavior configuration	required
`territory_center`	`UUID \| None`	Optional territory center room	`None`

Returns:

Type	Description
`NPCState`	The created NPCState

check_aggro `async` ¶

check_aggro(
    npc_id: UUID,
    npc_room_id: UUID,
    behavior: BehaviorType,
    aggro_range: int = 1,
) -> list[UUID]

Check for entities that should trigger aggro.

Parameters:

Name	Type	Description	Default
`npc_id`	`UUID`	NPC checking for targets	required
`npc_room_id`	`UUID`	NPC's current room	required
`behavior`	`BehaviorType`	NPC's behavior type	required
`aggro_range`	`int`	Rooms to check (BFS hops from NPC room)	`1`

Returns:

Type	Description
`list[UUID]`	List of entity IDs to aggro on

call_for_help `async` ¶

call_for_help(
    npc_id: UUID,
    npc_room_id: UUID,
    target_id: UUID,
    _call_radius: int = 1,
) -> list[UUID]

Call nearby allies for help.

Parameters:

Name	Type	Description	Default
`npc_id`	`UUID`	NPC calling for help	required
`npc_room_id`	`UUID`	NPC's room	required
`target_id`	`UUID`	Entity to aggro allies on	required
`_call_radius`	`int`	Rooms to check for allies	`1`

Returns:

Type	Description
`list[UUID]`	List of ally IDs that responded

CacheRefreshSystem ¶

CacheRefreshSystem(
    world: World,
    refresh_interval: float = DEFAULT_REFRESH_INTERVAL,
)

Bases: System

Periodically refreshes cached memory and relationship adapters.

Uses delta-time accumulation so the interval is wall-clock based regardless of the engine tick rate.

update `async` ¶

update(delta: float) -> None

Accumulate delta and refresh caches when the interval elapses.

NPCDialogueSystem ¶

NPCDialogueSystem(world: World)

Bases: System, StorageAware

System for handling AI-powered NPC dialogue.

Manages conversations between players and NPCs, including: - AI provider integration for generating responses - Rate limiting to prevent abuse - Conversation history tracking - Context building from world state

Example

system = NPCDialogueSystem(world)

Process a player's message to an NPC¶

success = await system.process_dialogue( player_entity_id=player.id, npc_entity_id=npc.id, message="Hello, how are you?", session=player_session, )

Parameters:

Name	Type	Description	Default
`world`	`World`	The game world instance	required

set_storage ¶

set_storage(store: DocumentStore) -> None

Set the document store for conversation persistence.

Called automatically by the engine when the system is registered if the system implements StorageAware.

Parameters:

Name	Type	Description	Default
`store`	`DocumentStore`	The document store instance	required

startup `async` ¶

startup() -> None

Initialize the dialogue system.

Sets up rate limiting and gets provider registry from engine.

shutdown `async` ¶

shutdown() -> None

Shut down the dialogue system.

Saves conversations to storage (if persistence enabled) and cleans up. Submits extraction jobs for any remaining conversations so that dialogue is converted into memories before data is discarded.

update `async` ¶

update(delta: float) -> None

Periodic update for cleanup of stale conversations.

Parameters:

Name	Type	Description	Default
`delta`	`float`	Time since last update in seconds	required

process_dialogue `async` ¶

process_dialogue(
    player_entity_id: UUID,
    npc_entity_id: UUID,
    message: str,
    session: Session,
) -> bool

Process a dialogue request from player to NPC.

This is the main entry point for NPC dialogue. It handles: - Validating the NPC has a DialogueComponent - Checking rate limits and cooldowns - Building context from world state - Generating AI response - Sending the complete response to player (buffered for content filtering)

Parameters:

Name	Type	Description	Default
`player_entity_id`	`UUID`	UUID of the player entity	required
`npc_entity_id`	`UUID`	UUID of the NPC entity	required
`message`	`str`	The player's message to the NPC	required
`session`	`Session`	Player's session for sending responses	required

Returns:

Type	Description
`bool`	True if dialogue was processed, False on error

find_npc_in_room ¶

find_npc_in_room(
    room_id: UUID, keyword: str
) -> Entity | None

Find an NPC in a room by keyword.

Searches for NPCs in the specified room that match the given keyword in their name or keywords list.

Parameters:

Name	Type	Description	Default
`room_id`	`UUID`	The room to search in	required
`keyword`	`str`	Keyword to match against NPC names/keywords	required

Returns:

Type	Description
`Entity \| None`	The matching Entity or None if not found

get_conversation `async` ¶

get_conversation(
    player_id: UUID, npc_id: UUID
) -> Conversation | None

Get an existing conversation between player and NPC.

Parameters:

Name	Type	Description	Default
`player_id`	`UUID`	UUID of the player	required
`npc_id`	`UUID`	UUID of the NPC	required

Returns:

Type	Description
`Conversation \| None`	The Conversation if it exists, None otherwise

get_player_conversations `async` ¶

get_player_conversations(
    player_id: UUID,
) -> list[Conversation]

Get all active conversations for a player.

Parameters:

Name	Type	Description	Default
`player_id`	`UUID`	UUID of the player	required

Returns:

Type	Description
`list[Conversation]`	List of Conversation objects for this player

get_entity_name ¶

get_entity_name(entity_id: UUID) -> str

Get the display name for an entity.

Parameters:

Name	Type	Description	Default
`entity_id`	`UUID`	UUID of the entity	required

Returns:

Type	Description
`str`	Entity's name or "Someone" if not found

end_conversation `async` ¶

end_conversation(player_id: UUID, npc_id: UUID) -> bool

End a conversation between player and NPC.

Before removing the conversation, pushes exchanges into the working buffer and submits an extraction job so the dialogue can be converted into episodic memories.

Parameters:

Name	Type	Description	Default
`player_id`	`UUID`	UUID of the player	required
`npc_id`	`UUID`	UUID of the NPC	required

Returns:

Type	Description
`bool`	True if a conversation was ended, False if not found

enqueue_background_dialogue ¶

enqueue_background_dialogue(
    handler: Callable[[], Awaitable[str]],
    *,
    npc_id: UUID | None = None,
    room_id: UUID | None = None,
    player_witnessed: bool = False,
) -> UUID | None

Submit an NPC dialogue LLM request to the off-tick queue.

Use this for NPC-initiated dialogue that should not block the tick loop. Results are dispatched as NPCBarkEvent by the autonomy system and delivered to players in the NPC's room.

Parameters:

Name	Type	Description	Default
`handler`	`Callable[[], Awaitable[str]]`	Async callable that performs the LLM request.	required
`npc_id`	`UUID \| None`	Optional NPC entity UUID for result routing.	`None`
`room_id`	`UUID \| None`	Optional room UUID for result delivery.	`None`
`player_witnessed`	`bool`	Bump priority when a player is present.	`False`

Returns:

Type	Description
`UUID \| None`	Request UUID if enqueued, None if queue unavailable or full.

ActionExecutor ¶

Bases: BaseActionExecutor

Executes autonomy actions by mutating intent components.

Extends BaseActionExecutor with classic-rpg-specific action handling: USE_ITEM with ConsumableComponent, SOCIALIZE/GOSSIP with SocialIntent, CRAFT_ITEM, and classic-rpg event emission.

execute_game_action `async` ¶

execute_game_action(
    npc_id: UUID,
    entity: Entity,
    action_type: ActionType,
    action: object,
    world: World,
) -> ActionResult | None

Handle all classic-rpg action types.

NPCMovementHandler ¶

Bases: System

Moves NPCs one step per tick based on NavigationIntent.

update `async` ¶

update(delta: float) -> None

Advance NPC navigation intents by one path step.

FallbackResponseProvider ¶

Provides contextually appropriate canned responses when AI is unavailable.

This provider selects responses based on the NPC's personality type (derived from their role) and the interaction context. Responses are chosen randomly within the matching category to avoid repetition.

When a fallback is used, the provider logs the event with the reason for monitoring and debugging AI provider health.

Example

provider = FallbackResponseProvider()

Get a fallback response for a merchant greeting¶

response = provider.get_response( personality="merchant", context="greeting", )

Log the fallback usage¶

provider.log_fallback( npc_name="Grimjaw", reason=FallbackReason.TIMEOUT, personality="merchant", context="greeting", )

get_response ¶

get_response(
    personality: str = "", context: str = ""
) -> str

Get a contextually appropriate fallback response.

Resolves the personality type from the NPC role string and the interaction context, then selects a random response from the matching category.

Parameters:

Name	Type	Description	Default
`personality`	`str`	NPC role/personality string (e.g., "merchant", "guard"). Maps to a PersonalityType via resolve_personality().	`''`
`context`	`str`	Interaction context string (e.g., "greeting", "trade"). Maps to an InteractionContext via resolve_context().	`''`

Returns:

Type	Description
`str`	A fallback response string appropriate for the NPC and context.

log_fallback ¶

log_fallback(
    npc_name: str,
    reason: FallbackReason,
    personality: str = "",
    context: str = "",
    error_detail: str = "",
) -> None

Log that a fallback response was used.

This method logs at WARNING level to aid in monitoring AI provider health. The log includes the NPC name, the reason for fallback, and additional context about the failure.

Parameters:

Name	Type	Description	Default
`npc_name`	`str`	Display name of the NPC.	required
`reason`	`FallbackReason`	Why the fallback was triggered (from FallbackReason).	required
`personality`	`str`	NPC personality/role string.	`''`
`context`	`str`	Interaction context string.	`''`
`error_detail`	`str`	Additional detail about the error (e.g., exception message).	`''`

get_response_and_log ¶

get_response_and_log(
    npc_name: str,
    reason: FallbackReason,
    personality: str = "",
    context: str = "",
    error_detail: str = "",
) -> str

Get a fallback response and log the event in one call.

Convenience method that combines get_response() and log_fallback().

Parameters:

Name	Type	Description	Default
`npc_name`	`str`	Display name of the NPC.	required
`reason`	`FallbackReason`	Why the fallback was triggered.	required
`personality`	`str`	NPC role/personality string.	`''`
`context`	`str`	Interaction context string.	`''`
`error_detail`	`str`	Additional error detail.	`''`

Returns:

Type	Description
`str`	A fallback response string appropriate for the NPC and context.

MemoryEventBridge ¶

MemoryEventBridge(world: World)

Bases: System

Bridges memory/relationship lifecycle events to NPC subsystems.

Responsibilities: * Log events at debug level for admin observability. * Invalidate cached adapters when new data arrives. * Emit :class:StorySignalEvent for dramatic relationship changes.

startup `async` ¶

startup() -> None

Subscribe to memory and relationship lifecycle events.

shutdown `async` ¶

shutdown() -> None

Unsubscribe all handlers.

update `async` ¶

update(delta: float) -> None

No per-tick work; all logic is event-driven.

ScheduleSystem ¶

ScheduleSystem(world: World)

Bases: System

Applies schedule blocks to NPCs as world time changes.

startup `async` ¶

startup() -> None

Subscribe to time and weather events.

shutdown `async` ¶

shutdown() -> None

Unsubscribe event handlers.

update `async` ¶

update(delta: float) -> None

Schedule reacts via event handlers only.

StorySignalSystem ¶

StorySignalSystem(
    world: World,
    detector: StorySignalDetector | None = None,
)

Bases: System

Periodic detector execution and signal emission.

SocialFabricSystem ¶

SocialFabricSystem(
    world: World,
    relationship_provider: RelationshipProvider
    | None = None,
    resolver: SocialInteractionResolver | None = None,
    relationship_manager: RelationshipManager | None = None,
)

Bases: System

Processes social intents and relationship-derived interaction state.

set_relationship_provider ¶

set_relationship_provider(
    provider: RelationshipProvider,
) -> None

Replace the relationship provider for social lookups.

set_gossip_processor ¶

set_gossip_processor(
    processor: GossipReactionProcessor,
) -> None

Replace the gossip reaction processor in the resolver.

set_relationship_manager ¶

set_relationship_manager(
    manager: RelationshipManager,
) -> None

Inject a RelationshipManager for NPC→NPC relationship tracking.

update `async` ¶

update(delta: float) -> None

Sweep reservations and process social intents.

get_friends ¶

get_friends(npc_id: UUID) -> list[UUID]

Return friend IDs where trust > 0.7.

get_rivals ¶

get_rivals(npc_id: UUID) -> list[UUID]

Return rival IDs where trust and respect are both low.

get_allies ¶

get_allies(npc_id: UUID) -> list[UUID]

Return allies based on loyalty relation if available.

reserve_interaction ¶

reserve_interaction(
    initiator_id: UUID, target_id: UUID, duration: float
) -> bool

Reserve interaction target for a duration.

SpawnerSystem ¶

SpawnerSystem(world: 'World')

Bases: System

System for spawning and respawning NPCs/monsters.

Handles: - Initial spawning at startup - Respawning after death (with timers) - Population limits per room and area - Spawn conditions (time, player presence)

startup `async` ¶

startup() -> None

Initialize spawner and do initial spawns.

shutdown `async` ¶

shutdown() -> None

Shutdown spawner system.

register_spawn_point ¶

register_spawn_point(spawn_point: SpawnPoint) -> None

Register a spawn point.

unregister_spawn_point ¶

unregister_spawn_point(
    spawn_point_id: UUID,
) -> SpawnPoint | None

Unregister a spawn point.

register_template ¶

register_template(template: MonsterTemplate) -> None

Register a monster template.

unregister_template ¶

unregister_template(
    template_id: UUID,
) -> MonsterTemplate | None

Unregister a monster template.

get_spawn_point ¶

get_spawn_point(spawn_point_id: UUID) -> SpawnPoint | None

Get a spawn point by ID.

get_template ¶

get_template(template_id: UUID) -> MonsterTemplate | None

Get a monster template by ID.

get_monster ¶

get_monster(entity_id: UUID) -> Monster | None

Get a spawned monster by ID.

register_spawn_rule ¶

register_spawn_rule(rule: SpawnRule) -> None

Register a spawn rule that governs spawning behavior.

unregister_spawn_rule ¶

unregister_spawn_rule(rule_id: UUID) -> SpawnRule | None

Unregister a spawn rule.

register_spawn_group ¶

register_spawn_group(group: SpawnGroup) -> None

Register a coordinated spawn group.

unregister_spawn_group ¶

unregister_spawn_group(group_id: UUID) -> SpawnGroup | None

Unregister a spawn group.

get_spawn_rule ¶

get_spawn_rule(rule_id: UUID) -> SpawnRule | None

Get a spawn rule by ID.

get_spawn_group ¶

get_spawn_group(group_id: UUID) -> SpawnGroup | None

Get a spawn group by ID.

update `async` ¶

update(delta: float) -> None

Process spawn timers and respawns.

on_entity_death `async` ¶

on_entity_death(entity_id: UUID) -> None

Handle entity death for respawn tracking.

Parameters:

Name	Type	Description	Default
`entity_id`	`UUID`	Entity that died	required

despawn_entity `async` ¶

despawn_entity(
    entity_id: UUID, reason: str = "cleanup"
) -> bool

Manually despawn an entity.

Parameters:

Name	Type	Description	Default
`entity_id`	`UUID`	Entity to despawn	required
`reason`	`str`	Reason for despawn	`'cleanup'`

Returns:

Type	Description
`bool`	True if entity was despawned

get_spawn_point_status ¶

get_spawn_point_status(
    spawn_point_id: UUID,
) -> dict[str, object] | None

Get status of a spawn point.

Returns:

Type	Description
`dict[str, object] \| None`	Dict with spawn point status or None

get_all_spawn_points ¶

get_all_spawn_points() -> list[SpawnPoint]

Get all registered spawn points.

get_all_monsters ¶

get_all_monsters() -> list[Monster]

Get all spawned monsters.

get_monsters_in_room ¶

get_monsters_in_room(room_id: UUID) -> list[Monster]

Get all monsters in a specific room.

NPC Autonomy¶

NPC autonomy systems providing independent NPC behavior including needs, goals, schedules, social interactions, ambient barks, action scoring, signal processing, gossip propagation, and LLM-driven decision queuing.

Submodules: autonomy, needs, goals, schedule, social, barks, executor, scoring, signals, gossip, llm_queue.

autonomy ¶

Classic RPG autonomy system built on the stdlib base loop.

AutonomySystem ¶

AutonomySystem(world: World)

Bases: BaseAutonomySystem

Selects and executes NPC autonomous actions each tick.

Extends BaseAutonomySystem with classic-RPG goal generation, utility scoring, LLM integration, and game-specific events.

startup `async` ¶

startup() -> None

Restore persisted state and wire goal/LLM subsystems.

shutdown `async` ¶

shutdown() -> None

Stop the off-tick LLM queue.

Quest Generation¶

Automated quest generation integration, connecting world state and NPC needs to the quest generation framework in maid-stdlib.

generation ¶

Automated quest generation package.

QuestHistory ¶

QuestHistory(document_store: DocumentStore | None)

In-memory history index with optional DocumentStore persistence.

record ¶

record(quest: GeneratedQuest) -> None

Record newly generated quest metadata.

mark_offered ¶

mark_offered(character_id: UUID, quest_id: str) -> None

Track a pending quest offer for a character.

mark_accepted ¶

mark_accepted(character_id: UUID, quest_id: str) -> None

Mark quest as accepted by character.

set_outcome ¶

set_outcome(
    quest_id: str,
    outcome: QuestOutcome,
    character_id: UUID | None = None,
) -> None

Set quest outcome and completion timestamp.

recent_quest_from_npc ¶

recent_quest_from_npc(
    npc_id: UUID, window_seconds: float
) -> bool

Return True when NPC gave a quest within cooldown window.

recent_types ¶

recent_types(count: int) -> list[QuestSeedType]

Return most recent seed types.

active_generated_quest_ids ¶

active_generated_quest_ids() -> set[str]

Get active generated quest identifiers.

get ¶

get(quest_id: str) -> QuestHistoryEntry | None

Get quest history entry by quest ID.

get_chain ¶

get_chain(chain_id: str) -> list[QuestHistoryEntry]

Return history entries belonging to a quest chain.

count_active_in_region ¶

count_active_in_region(region_id: str) -> int

Count active generated quests in a region.

count_pending_offers ¶

count_pending_offers(character_id: UUID) -> int

Count pending generated quest offers for character.

entries ¶

entries() -> list[QuestHistoryEntry]

Return all history entries (newest first).

entries_for_character ¶

entries_for_character(
    character_id: UUID,
) -> list[QuestHistoryEntry]

Return history entries for a specific character.

recent_entries ¶

recent_entries(count: int) -> list[QuestHistoryEntry]

Return the count most recent history entries.

GeneratedQuest `dataclass` ¶

GeneratedQuest(
    id: str,
    seed_id: UUID,
    seed_type: QuestSeedType,
    archetype_id: str,
    importance: float,
    title: str,
    summary: str,
    detailed_description: str,
    objectives: list[QuestObjective],
    quest_giver_id: UUID,
    quest_giver_motivation: str,
    quest_giver_dialogue: QuestDialogue,
    involved_npcs: dict[UUID, str],
    involved_locations: list[UUID],
    rewards: QuestReward,
    consequences: list[QuestConsequence],
    failure_consequences: list[QuestConsequence],
    difficulty: DifficultyTier,
    time_limit: float | None,
    expiry: datetime,
    chain_potential: float,
    chain_depth: int = 0,
    grounding_score: float = 1.0,
    coherence_score: float = 1.0,
    novelty_score: float = 1.0,
    delivery_method: QuestDeliveryMethod = DIRECT_OFFER,
)

Fully built generated quest payload before conversion to Quest model.

QuestOutcome ¶

Bases: str, Enum

Outcome grade emitted for quest completion/failure.

QuestSeed `dataclass` ¶

QuestSeed(
    source_signals: list[str],
    seed_type: QuestSeedType,
    importance: float,
    urgency: float,
    primary_npc: UUID | None,
    primary_npc_motivation: str,
    backstory: str,
    stakes: str,
    location: UUID | None,
    threat_source: str,
    evidence_location: UUID | None,
    involved_npcs: list[UUID],
    context_keys: dict[str, str],
    created_at: datetime,
    expires_at: datetime | None,
    attempt_count: int = 0,
    chain_depth: int = 0,
    seed_id: UUID = uuid4(),
)

Input signal bundle used to generate a quest.

QuestSeedTypes ¶

Known quest seed type identifiers.

Skills System¶

Character skills, experience, and progression.

from maid_classic_rpg.systems.skills import SkillSystem

# Skills system handles:
# - Skill learning
# - Experience tracking
# - Level progression
# - Skill trees and prerequisites

skills ¶

Skill systems for training and skill checks.

CheckDifficulty ¶

Bases: Enum

Difficulty levels for skill checks.

SkillCheckResult `dataclass` ¶

SkillCheckResult(
    success: bool = False,
    roll: int = 0,
    target: int = 0,
    margin: int = 0,
    critical: bool = False,
    skill_level: int = 0,
)

Result of a skill check.

SkillLearnedEvent `dataclass` ¶

SkillLearnedEvent(character_id: UUID, skill_name: str)

Bases: Event

Emitted when a character learns a skill.

SkillLevelUpEvent `dataclass` ¶

SkillLevelUpEvent(
    character_id: UUID, skill_name: str, new_level: int
)

Bases: Event

Emitted when a character's skill levels up.

SkillSystem ¶

SkillSystem(world: World)

Bases: System, StorageAware

Manages skill training, checks, and progression.

set_storage ¶

set_storage(store: DocumentStore) -> None

Set the document store for persistence.

startup `async` ¶

startup() -> None

Load skill data from storage.

shutdown `async` ¶

shutdown() -> None

Save skill data and clean up.

update `async` ¶

update(_delta: float) -> None

Process any time-based skill updates and clean up cooldowns.

learn_skill ¶

learn_skill(character_id: UUID, skill_name: str) -> bool

Teach a skill to a character.

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	Character to teach	required
`skill_name`	`str`	Skill internal name	required

Returns:

Type	Description
`bool`	True if skill was learned (not already known)

get_skill_level ¶

get_skill_level(character_id: UUID, skill_name: str) -> int

Get a character's level in a skill.

get_character_skill ¶

get_character_skill(
    character_id: UUID, skill_name: str
) -> CharacterSkill | None

Get a character's skill data.

get_all_skills ¶

get_all_skills(character_id: UUID) -> list[CharacterSkill]

Get all skills for a character.

get_skills_by_category ¶

get_skills_by_category(
    character_id: UUID, category: SkillCategory
) -> list[CharacterSkill]

Get skills in a category for a character.

add_experience ¶

add_experience(
    character_id: UUID,
    skill_name: str,
    amount: float,
    respect_cooldown: bool = True,
) -> tuple[float, int]

Add experience to a skill.

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	Character ID	required
`skill_name`	`str`	Skill internal name	required
`amount`	`float`	Experience to add	required
`respect_cooldown`	`bool`	If True, check cooldown (1 second between gains)	`True`

Returns:

Type	Description
`tuple[float, int]`	Tuple of (actual exp added, levels gained)

train_skill ¶

train_skill(
    character_id: UUID,
    skill_name: str,
    training_points: int = 1,
) -> tuple[bool, int]

Train a skill using training points.

Parameters:

Name	Type	Description	Default
`character_id`	`UUID`	Character ID	required
`skill_name`	`str`	Skill to train	required
`training_points`	`int`	Points to spend	`1`

Returns:

Type	Description
`tuple[bool, int]`	Tuple of (success, levels gained)

skill_check ¶

skill_check(
    character: Character,
    skill_name: str,
    difficulty: CheckDifficulty | int,
    stat_bonus: bool = True,
) -> SkillCheckResult

Perform a skill check.

Parameters:

Name	Type	Description	Default
`character`	`Character`	Character making the check	required
`skill_name`	`str`	Skill being checked	required
`difficulty`	`CheckDifficulty \| int`	Target number or CheckDifficulty enum	required
`stat_bonus`	`bool`	Whether to add stat bonuses	`True`

Returns:

Type	Description
`SkillCheckResult`	SkillCheckResult with outcome details

opposed_check ¶

opposed_check(
    attacker: Character,
    defender: Character,
    attack_skill: str,
    defend_skill: str,
) -> tuple[SkillCheckResult, SkillCheckResult]

Perform an opposed skill check.

Parameters:

Name	Type	Description	Default
`attacker`	`Character`	Character initiating the check	required
`defender`	`Character`	Character defending	required
`attack_skill`	`str`	Attacker's skill	required
`defend_skill`	`str`	Defender's skill	required

Returns:

Type	Description
`tuple[SkillCheckResult, SkillCheckResult]`	Tuple of (attacker_result, defender_result)

get_available_skills ¶

get_available_skills(
    character: Character,
) -> list[SkillDefinition]

Get all skills available to a character based on class.

Parameters:

Name	Type	Description	Default
`character`	`Character`	Character to check	required

Returns:

Type	Description
`list[SkillDefinition]`	List of available skill definitions

get_trainable_skills ¶

get_trainable_skills(
    character: Character,
) -> list[SkillDefinition]

Get skills that can be trained at a guild.

Parameters:

Name	Type	Description	Default
`character`	`Character`	Character to check	required

Returns:

Type	Description
`list[SkillDefinition]`	List of trainable skill definitions

Commands¶

Dialogue Commands¶

Commands for interacting with AI-powered NPCs:

talk <npc> <message>      - Talk to an NPC
ask <npc> about <topic>   - Ask NPC about a topic
greet <npc>               - Greet an NPC (aliases: hello, hi)
conversations             - List active conversations (alias: convos)
endconversation <npc>     - End conversation (aliases: endconv, bye)

dialogue ¶

Dialogue command handlers for AI NPC conversations.

cmd_talk `async` ¶

cmd_talk(ctx: CommandContext) -> None

Talk to an NPC using AI dialogue.

Usage: talk Example: talk bartender Hello, do you have any news?

cmd_ask `async` ¶

cmd_ask(ctx: CommandContext) -> None

Ask an NPC about a specific topic.

Usage: ask about Example: ask guard about trouble in town

cmd_conversations `async` ¶

cmd_conversations(ctx: CommandContext) -> None

List active conversations with NPCs.

cmd_endconversation `async` ¶

cmd_endconversation(ctx: CommandContext) -> None

End a conversation with an NPC.

Usage: endconversation Example: endconversation bartender

cmd_greet `async` ¶

cmd_greet(ctx: CommandContext) -> None

Greet an NPC to initiate conversation.

This sends a generic greeting to the NPC, which can be useful to start a conversation or see the NPC's introduction.

Examples:

greet blacksmith hello merchant hi guard

Economy Commands¶

Commands for economic activities:

buy <item> [from <npc>]   - Purchase an item
sell <item> [to <npc>]    - Sell an item
trade <player>            - Initiate a trade
bank deposit <amount>     - Deposit currency
bank withdraw <amount>    - Withdraw currency
auction list              - View auction house
auction bid <id> <amount> - Bid on an item

economy ¶

Economy command handlers for shops, banking, and trading.

cmd_gold `async` ¶

cmd_gold(ctx: CommandContext) -> None

Display current gold on hand.

cmd_list `async` ¶

cmd_list(ctx: CommandContext) -> None

List shop inventory.

cmd_buy `async` ¶

cmd_buy(ctx: CommandContext) -> None

Buy an item from shop.

cmd_sell `async` ¶

cmd_sell(ctx: CommandContext) -> None

Sell an item to shop.

cmd_value `async` ¶

cmd_value(ctx: CommandContext) -> None

Get sale price for an item.

cmd_appraise `async` ¶

cmd_appraise(ctx: CommandContext) -> None

Get detailed info on shop item.

cmd_balance `async` ¶

cmd_balance(ctx: CommandContext) -> None

Check bank balance.

cmd_deposit `async` ¶

cmd_deposit(ctx: CommandContext) -> None

Deposit gold into bank.

cmd_withdraw `async` ¶

cmd_withdraw(ctx: CommandContext) -> None

Withdraw gold from bank.

cmd_trade `async` ¶

cmd_trade(ctx: CommandContext) -> None

Start a trade with another player.

cmd_accept `async` ¶

cmd_accept(ctx: CommandContext) -> None

Accept trade request or confirm trade.

cmd_decline `async` ¶

cmd_decline(ctx: CommandContext) -> None

Decline or cancel trade.

cmd_offer `async` ¶

cmd_offer(ctx: CommandContext) -> None

Add to trade offer.

cmd_remove `async` ¶

cmd_remove(ctx: CommandContext) -> None

Remove from trade offer.

cmd_lock `async` ¶

cmd_lock(ctx: CommandContext) -> None

Lock trade offer.

cmd_unlock `async` ¶

cmd_unlock(ctx: CommandContext) -> None

Unlock trade offer.

cmd_tradestatus `async` ¶

cmd_tradestatus(ctx: CommandContext) -> None

View trade status.

Information Commands¶

Commands for getting information:

score                     - View character sheet
skills                    - List your skills
quests                    - View quest log
who                       - List online players
time                      - View in-game time
weather                   - View current weather

information ¶

Information command handlers.

cmd_score `async` ¶

cmd_score(ctx: CommandContext) -> None

Show character stats and vitals.

cmd_inventory `async` ¶

cmd_inventory(ctx: CommandContext) -> None

Show the player's inventory.

cmd_equipment `async` ¶

cmd_equipment(ctx: CommandContext) -> None

Show equipped items.

cmd_examine `async` ¶

cmd_examine(ctx: CommandContext) -> None

Examine something in detail.

cmd_who `async` ¶

cmd_who(ctx: CommandContext) -> None

Show list of online players.

cmd_help `async` ¶

cmd_help(ctx: CommandContext) -> None

Show help information.

Displays command documentation using the HelpGenerator system. When called without arguments, lists all available commands grouped by category. When called with a command name, shows detailed help for that specific command.

Examples:

help help look help attack help teleport

Content Pack¶

The Classic RPG content pack registers all systems, commands, and game content:

from maid_engine import GameEngine
from maid_stdlib import StdlibContentPack
from maid_classic_rpg import ClassicRPGContentPack

engine = GameEngine(settings)

# Load packs in dependency order
engine.load_content_pack(StdlibContentPack())
engine.load_content_pack(ClassicRPGContentPack())

await engine.start()

ClassicRPGContentPack ¶

ClassicRPGContentPack()

Classic RPG content pack.

Provides a complete traditional MUD experience with: - Combat system with tactical positioning - Magic system with spells and effects - Crafting and gathering - Economy (shops, trading, banking, auctions) - Social features (guilds, factions, achievements) - Quest system with objectives and rewards - PvP (arenas, duels, rankings) - World dynamics (time, weather, ecosystems)

manifest `property` ¶

manifest: ContentPackManifest

Load the authoritative packaged manifest.

get_dependencies ¶

get_dependencies() -> list[str]

Get pack dependencies.

get_systems ¶

get_systems(world: World) -> list[System]

Get systems provided by this pack.

get_events ¶

get_events() -> list[type[Event]]

Get events defined by this pack.

register_commands ¶

register_commands(registry: AnyCommandRegistry) -> None

Register commands provided by this pack.

register_document_schemas ¶

register_document_schemas(store: DocumentStore) -> None

Register document schemas used by this pack.

register_component_types ¶

register_component_types(
    registry: ComponentRegistry,
) -> None

Register component types for persistence serialization.

register_api_routes ¶

register_api_routes(router: APIRouter) -> None

Register API routes. Classic RPG has no custom admin routes.

on_load `async` ¶

on_load(engine: GameEngine) -> None

Called when pack is loaded.

Initializes game data including: - Quest definitions from data files - Crafting recipes - Default world setup if needed - RPG-specific error patterns for anomaly detection

on_unload `async` ¶

on_unload(engine: GameEngine) -> None

Called when pack is unloaded.

Performs cleanup: - Saves any pending data - Releases resources

Configuration¶

The maid-classic-rpg pack does not define its own MAID_-prefixed settings family. Gameplay tuning values (combat formulas, starting gold, day/night and weather timings, etc.) live as constants and defaults inside the individual systems rather than as environment variables. Engine-level settings such as MAID_GAME_* and MAID_AI_* still apply.

Package API¶

Full package API documentation:

maid_classic_rpg ¶

MAID Classic RPG - Traditional MUD gameplay content pack.

This package provides a complete classic MUD experience:

Combat system with tactical grid positioning
Magic system with spells and effects
Crafting and gathering
Economy (shops, trading, banking, auctions)
Social features (guilds, factions, achievements)
Quest system with objectives and rewards
PvP (arenas, duels, rankings)
World dynamics (time, weather, ecosystems)

ClassicRPGContentPack ¶

ClassicRPGContentPack()

Classic RPG content pack.

Provides a complete traditional MUD experience with: - Combat system with tactical positioning - Magic system with spells and effects - Crafting and gathering - Economy (shops, trading, banking, auctions) - Social features (guilds, factions, achievements) - Quest system with objectives and rewards - PvP (arenas, duels, rankings) - World dynamics (time, weather, ecosystems)

Source code in packages/maid-classic-rpg/src/maid_classic_rpg/pack.py

def __init__(self) -> None:
    """Initialize the content pack."""

    self._recipe_manager: Any = None  # RecipeManager stored to prevent GC
    self._gossip_system: Any = None  # Track for cleanup (#12)
    self._event_subscription_ids: list[Any] = []  # Track for cleanup (#15)
    self._pack_systems: list[Any] = []  # Track systems for on_unload (#R2-3)
    self._ai_setup_hook: Any = None  # Track hook for on_unload (#R2-4)
    self._wilderness_spawn_adapter: Any = None  # Track for cleanup
    self._previous_connection_handler: Any = None  # Restore on unload
    self._engine_registry: Any = None  # I-03: engine command registry ref
    self._pending_engine: Any = None  # I-04: deferred wilderness check
    self._deferred_task: Any = None  # I-04: deferred wilderness task handle

manifest `property` ¶

manifest: ContentPackManifest

Load the authoritative packaged manifest.

get_dependencies ¶

get_dependencies() -> list[str]

Get pack dependencies.

Source code in packages/maid-classic-rpg/src/maid_classic_rpg/pack.py

def get_dependencies(self) -> list[str]:
    """Get pack dependencies."""
    return ["stdlib"]

get_events ¶

get_events() -> list[type[Event]]

Get events defined by this pack.

Source code in packages/maid-classic-rpg/src/maid_classic_rpg/pack.py

def get_events(self) -> list[type[Event]]:
    """Get events defined by this pack."""
    from maid_classic_rpg.events import get_classic_rpg_events

    return get_classic_rpg_events()

get_systems ¶

get_systems(world: World) -> list[System]

Get systems provided by this pack.

Source code in packages/maid-classic-rpg/src/maid_classic_rpg/pack.py

def get_systems(self, world: World) -> list[System]:
    """Get systems provided by this pack."""
    from maid_classic_rpg.systems import get_classic_rpg_systems

    systems = get_classic_rpg_systems(world)
    self._pack_systems = list(systems)
    return systems

on_load `async` ¶

on_load(engine: GameEngine) -> None

Called when pack is loaded.

Initializes game data including: - Quest definitions from data files - Crafting recipes - Default world setup if needed - RPG-specific error patterns for anomaly detection

Source code in packages/maid-classic-rpg/src/maid_classic_rpg/pack.py

async def on_load(self, engine: GameEngine) -> None:
    """Called when pack is loaded.

    Initializes game data including:
    - Quest definitions from data files
    - Crafting recipes
    - Default world setup if needed
    - RPG-specific error patterns for anomaly detection
    """
    import logging
    import re
    from importlib.resources import files

    logger = logging.getLogger(__name__)
    logger.info("Loading Classic RPG content pack...")

    # Register RPG-specific error patterns for AI Player anomaly detection
    from maid_engine.ai_players.bugs.detector import AnomalyDetector

    rpg_error_patterns: list[re.Pattern[str]] = [
        re.compile(p, re.IGNORECASE)
        for p in (
            r"you\s+don'?t\s+have\s+enough",
            r"you\s+are\s+too\s+exhausted",
            r"not\s+in\s+combat",
            r"already\s+in\s+combat",
            r"cooldown",
            r"you\s+need\s+to\s+be\s+level",
            r"requires?\s+a\s+weapon",
            r"you\s+don'?t\s+have\s+a\s+weapon\s+equipped",
            r"you\s+need\s+a\s+ranged\s+weapon",
            r"you\s+aren'?t\s+fighting\s+anyone",
            r"can'?t\s+rest\s+while\s+in\s+combat",
            r"can'?t\s+sleep\s+while\s+in\s+combat",
            r"you\s+are\s+dead\s+and\s+cannot",
            r"you\s+are\s+a\s+ghost\s+and\s+cannot",
            r"you\s+are\s+not\s+in\s+a\s+guild",
            r"you\s+don'?t\s+have\s+permission",
            r"you\s+are\s+not\s+in\s+a\s+trade",
            r"peaceful\s+feeling\s+overwhelms\s+you",
            r"you\s+are\s+(?:still\s+)?reloading",
        )
    ]
    AnomalyDetector.register_error_patterns(rpg_error_patterns, category="classic-rpg")
    logger.debug("Registered %d Classic RPG error patterns for anomaly detection", len(rpg_error_patterns))

    # Initialize Managers
    from pathlib import Path

    from maid_classic_rpg.auth.manager import AccountManager
    from maid_classic_rpg.managers.character_manager import CharacterManager
    from maid_classic_rpg.managers.item_manager import ItemManager
    from maid_stdlib.managers.room_manager import RoomManager

    # Attach managers to engine/world for global access
    # Use data directory from settings or default
    data_dir = Path.cwd() / "data"
    if (
        hasattr(engine, "settings")
        and engine.settings
        and hasattr(engine.settings, "storage")
        and hasattr(engine.settings.storage, "data_dir")
    ):
        data_dir = Path(engine.settings.storage.data_dir)
    account_dir = data_dir / "accounts"
    char_dir = data_dir / "characters"
    items_dir = data_dir / "items"
    rooms_dir = data_dir / "rooms"

    # Reuse the authoritative engine AccountManager if the server
    # bootstrap already created one (cli/app.py sets
    # ``engine._account_manager`` before loading content packs), so
    # classic-rpg shares a single account cache/store/password
    # verifier instead of creating a second identity store. Falls back
    # to a fresh engine-backed manager for standalone/test usage, in
    # which case that fresh manager is published back onto
    # ``engine._account_manager`` so other subsystems (e.g. the admin
    # panel) that resolve accounts via the engine manager also see it.
    engine_account_manager = getattr(engine, "_account_manager", None)
    self.account_manager = AccountManager(
        data_dir=account_dir, engine_manager=engine_account_manager
    )
    if engine_account_manager is None:
        engine._account_manager = self.account_manager.engine_manager  # type: ignore[attr-defined]
    self.character_manager = CharacterManager(data_dir=char_dir)
    self.item_manager = ItemManager(data_dir=items_dir)
    self.room_manager = RoomManager(data_dir=rooms_dir)

    # Attach managers to engine/world for access by systems/commands
    # Note: engine.characters, engine.items, engine.rooms are properties
    # that delegate to world, so only set on world
    engine.accounts = self.account_manager  # type: ignore[attr-defined]  # content-pack attaches manager to engine
    engine._character_manager = self.character_manager  # type: ignore[attr-defined]  # content-pack attaches manager to engine
    if engine.world:
        engine.world.characters = self.character_manager  # type: ignore[attr-defined]  # content-pack attaches manager to world
        engine.world.items = self.item_manager  # type: ignore[attr-defined]  # content-pack attaches manager to world
        engine.world.rooms = self.room_manager  # type: ignore[attr-defined]  # content-pack attaches manager to world
        # Give the CharacterManager a back-reference to the world
        # so its ``update()`` override can mirror legacy
        # ``character.inventory``/``equipment`` mutations into
        # the character's ECS ``InventoryComponent`` /
        # ``EquipmentComponent`` — keeping stdlib commands
        # (wear/wield/inventory) and classic-rpg commands
        # (get/drop/give/put) consistent. See
        # ``CharacterManager._sync_to_ecs`` for the contract.
        self.character_manager.world = engine.world

    # Load manager data
    await self.item_manager.load()
    await self.room_manager.load()

    # Load starter area from YAML files (expanded Millbrook)
    from maid_classic_rpg.world.load_yaml_starter_area import (
        load_expanded_starter_area,
    )

    if engine.world:
        await load_expanded_starter_area(
            self.room_manager, self.item_manager, engine.world,
        )

    # Sync all loaded items to ECS entities
    if engine.world:
        from maid_classic_rpg.managers.ecs_item_bridge import (
            sync_all_items_to_ecs,
        )

        synced = await sync_all_items_to_ecs(engine.world, self.item_manager)
        logger.info("Synced %d items to ECS entities", synced)

    # Fall back to legacy starter area if YAML loading produced no rooms
    from maid_classic_rpg.world.setup_starter_area import (
        STARTING_ROOM_ID,
        setup_starter_area,
        setup_starter_npcs,
    )

    if not await self.room_manager.get_by_id(STARTING_ROOM_ID):
        await setup_starter_area(self.room_manager)
        logger.info("Created legacy starter area (Town Square and surroundings)")

    # Create starter NPCs (Guard Captain Aldric, etc.) if not already present
    if engine.world:
        setup_starter_npcs(engine.world)

    # Sync rooms from room_manager into the World room index
    if engine.world:
        all_rooms = await self.room_manager.list_rooms()
        for room in all_rooms:
            engine.world.register_room(room.id, room)
        logger.debug("Registered %d rooms with World", len(all_rooms))

    # Load NPC autonomy archetypes and bark templates
    from maid_classic_rpg.models.npc.archetypes import ArchetypeRegistry
    from maid_classic_rpg.systems.npc.barks import load_barks_from_yaml

    npc_data_dir = files("maid_classic_rpg") / "data" / "npcs"
    archetypes_path = Path(str(npc_data_dir / "archetypes.yaml"))
    barks_path = Path(str(npc_data_dir / "barks.yaml"))

    archetype_registry = ArchetypeRegistry()
    if archetypes_path.exists():
        loaded = archetype_registry.load_from_yaml(archetypes_path)
        if engine.world:
            engine.world.set_data("npc_archetype_registry", archetype_registry)
            engine.world.set_data(
                "npc_archetype_max_need_delta",
                {
                    archetype.archetype_id: archetype.max_need_delta_per_tick
                    for archetype in loaded
                },
            )
            logger.debug("Loaded %d NPC archetypes", len(loaded))

    if barks_path.exists() and engine.world:
        bark_templates = load_barks_from_yaml(barks_path)
        engine.world.set_data("npc_bark_templates", bark_templates)
        logger.debug("Loaded bark templates for %d archetypes", len(bark_templates))

    # Load loot tables from YAML into world data so LootResolver can
    # look up tables by ID when processing monster death drops.
    if engine.world:
        import yaml as _yaml

        from maid_classic_rpg.models.entities.monster import LootTable

        loot_data_path = Path(
            str(files("maid_classic_rpg") / "data" / "monsters" / "loot_tables.yaml")
        )
        if loot_data_path.exists():
            try:
                raw_loot = _yaml.safe_load(loot_data_path.read_text(encoding="utf-8"))
                loot_count = 0
                for entry in (raw_loot or {}).get("loot_tables", []):
                    try:
                        table = LootTable.model_validate(entry)
                        engine.world.set_data(
                            "loot_tables", str(table.id), table,
                        )
                        loot_count += 1
                    except Exception:  # noqa: BLE001
                        logger.warning(
                            "Failed to parse loot table entry: %s",
                            entry.get("name", "<unknown>"),
                        )
                if loot_count:
                    logger.debug("Loaded %d loot tables from YAML", loot_count)
            except Exception:  # noqa: BLE001
                logger.exception("Failed to load loot_tables.yaml")

    # Wire cached memory/relationship adapters into NPC systems.
    # Systems are created with null providers; replace them now if
    # the real services are available on the world.
    if engine.world:
        from maid_classic_rpg.systems.npc.social import SocialFabricSystem
        from maid_stdlib.systems.npc.needs import NeedDecaySystem
        from maid_stdlib.systems.npc.stubs import (
            CachedMemoryAdapter,
            CachedRelationshipAdapter,
        )

        memory_adapter = CachedMemoryAdapter(engine.world)
        relationship_adapter = CachedRelationshipAdapter(engine.world)

        systems_mgr_npc = getattr(engine.world, "systems", None)
        if systems_mgr_npc:
            for system in getattr(systems_mgr_npc, "systems", []):
                if isinstance(system, NeedDecaySystem):
                    system.set_memory_provider(memory_adapter)
                elif isinstance(system, SocialFabricSystem):
                    system.set_relationship_provider(relationship_adapter)

        engine.world.set_data("_memory_adapter", memory_adapter)
        engine.world.set_data("_relationship_adapter", relationship_adapter)
        logger.debug("Wired cached memory/relationship adapters into NPC systems")

    # Wire GossipSystem with knowledge and relationship managers.
    # These managers may already exist on world data (created by the
    # memory system startup in stdlib).  Create fresh instances when
    # they are unavailable so gossip works even without the full
    # memory subsystem.
    if engine.world:
        from maid_stdlib.knowledge.gossip import GossipSystem
        from maid_stdlib.knowledge.manager import KnowledgeManager
        from maid_stdlib.relationships.manager import RelationshipManager

        knowledge_manager: KnowledgeManager | None = engine.world.get_data(
            "knowledge_manager",
        )
        if knowledge_manager is None:
            knowledge_manager = KnowledgeManager(
                document_store=engine.document_store,
                event_bus=engine.world.events,
            )
            engine.world.set_data("knowledge_manager", knowledge_manager)
        else:
            knowledge_manager.set_event_bus(engine.world.events)

        relationship_manager: RelationshipManager | None = engine.world.get_data(
            "relationship_manager",
        )
        if relationship_manager is None:
            relationship_manager = RelationshipManager(
                document_store=engine.document_store,
            )
            engine.world.set_data("relationship_manager", relationship_manager)

        gossip_system = GossipSystem(
            world=engine.world,
            knowledge_manager=knowledge_manager,
            relationship_manager=relationship_manager,
        )
        # Use register_and_start so startup() runs if world is live (#13)
        await engine.world.systems.register_and_start(gossip_system, source_pack="classic-rpg")
        self._gossip_system = gossip_system  # Track for cleanup (#12)
        logger.debug("Registered GossipSystem with knowledge and relationship managers")

        # Inject KnowledgeManager into BarkSystem so it can generate
        # knowledge-aware barks before falling back to templates.
        from maid_classic_rpg.systems.npc.barks import BarkSystem as _BarkSystem

        bark_system = engine.world.systems.get_system(_BarkSystem)
        if bark_system is not None:
            bark_system.set_knowledge_manager(knowledge_manager)
            logger.debug("Wired KnowledgeManager into BarkSystem")

        # Wire KnowledgeGossipProcessor into SocialFabricSystem so
        # NPC-to-NPC gossip interactions transfer knowledge entries.
        from maid_classic_rpg.systems.npc.gossip import KnowledgeGossipProcessor

        gossip_processor = KnowledgeGossipProcessor(
            knowledge_manager=knowledge_manager,
            relationship_manager=relationship_manager,
            event_bus=engine.world.events,
        )
        if systems_mgr_npc:
            for system in getattr(systems_mgr_npc, "systems", []):
                if isinstance(system, SocialFabricSystem):
                    system.set_gossip_processor(gossip_processor)
                    system.set_relationship_manager(relationship_manager)
                    break
        logger.debug("Wired KnowledgeGossipProcessor into SocialFabricSystem")

        # Ensure a ContextOrchestrator exists so that knowledge
        # reaches NPC prompts even when the full memory subsystem
        # (memory.enabled) is disabled.
        if getattr(engine, "context_orchestrator", None) is None:
            from maid_engine.ai import ContextOrchestrator
            from maid_stdlib.knowledge.context_provider import KnowledgeContextProvider

            orchestrator = ContextOrchestrator(provider_timeout_ms=40)
            knowledge_provider = KnowledgeContextProvider(
                knowledge_manager=knowledge_manager,
            )
            orchestrator.register_provider(
                name=knowledge_provider.get_name(),
                provider=knowledge_provider,
                priority=knowledge_provider.get_priority(),
                default_budget_pct=1.0,
            )
            engine.context_orchestrator = orchestrator  # type: ignore[attr-defined]  # content-pack attaches orchestrator to engine
            logger.debug(
                "Created minimal ContextOrchestrator with KnowledgeContextProvider "
                "(memory system not enabled)"
            )

    # Wire ItemManager into RewardDistributor and CraftingSystem.
    # These are created during get_classic_rpg_systems() before the
    # ItemManager exists, so we connect them now.
    if engine.world:
        # Wire into RewardDistributor
        reward_distributor = getattr(engine.world, "reward_distributor", None)
        if reward_distributor and hasattr(reward_distributor, "set_item_manager"):
            reward_distributor.set_item_manager(self.item_manager)
            logger.debug("Wired ItemManager into RewardDistributor")

        # Wire into CraftingSystem
        from maid_classic_rpg.systems.crafting.crafting import CraftingSystem

        systems_manager = getattr(engine.world, "systems", None)
        if systems_manager:
            systems_list = getattr(systems_manager, "systems", [])
            from maid_classic_rpg.systems.quests.generation.models import (
                QuestSeedTypes,
            )
            from maid_classic_rpg.systems.quests.generation.system import (
                QuestGenerationSystem,
            )

            for system in systems_list:
                if isinstance(system, CraftingSystem):
                    system.set_item_manager(self.item_manager)
                    logger.debug("Wired ItemManager into CraftingSystem")
                if isinstance(system, QuestGenerationSystem):
                    for seed_type in QuestSeedTypes.EXTENSION_TYPES:
                        system.register_seed_type(
                            seed_type,
                            f"Registered extension seed type: {seed_type}",
                        )
                    from maid_classic_rpg.systems.quests.generation.archetypes import (
                        DEFAULT_ARCHETYPES,
                    )

                    system.register_archetypes(DEFAULT_ARCHETYPES)
                    logger.debug("Registered %d quest archetypes", len(DEFAULT_ARCHETYPES))

    # Register Connection Handler
    if engine.server:
        from maid_classic_rpg.auth.character_handler import CharacterHandler
        from maid_classic_rpg.auth.login_handler import LoginHandler
        from maid_classic_rpg.entities.character_entity import create_character_entity
        from maid_engine.net.server import MAIDServer
        from maid_engine.net.session import Session, SessionState

        # Save previous handler so we can restore it on unload
        self._previous_connection_handler = getattr(
            engine.server, "_connection_handler", None,
        )

        async def classic_rpg_connection_handler(session: Session, server: MAIDServer) -> None:
            """Handle new connections for Classic RPG."""
            character = None
            try:
                # 1. Login
                login_handler = LoginHandler(session, self.account_manager)
                account = await login_handler.run()

                if not account:
                    await session.close()
                    return

                session.metadata["account_id"] = account.id
                # R9 critic: store the full account so classic admin/builder/
                # moderation handlers can read ``ctx.session.metadata["account"]``
                # (their _has_permission helper requires the live account
                # object, not just the id).  Also derive the engine-level
                # AccessLevel and stash it on the session so the dispatch
                # call below can pass it to ``command_registry.execute()``
                # — without this, every BUILDER/ADMIN/IMPLEMENTOR-gated
                # classic command is rejected at registry.py:795 by the
                # default PLAYER access check.
                from maid_classic_rpg.auth.access_level import (
                    access_level_for_classic_account,
                )
                session.metadata["account"] = account
                session.access_level = access_level_for_classic_account(account)  # type: ignore[attr-defined]  # access_level is attached to session post-login

                # Maintenance gate (PR-B): close non-admin sessions during a window.
                # R1 BLOCKER 3: bypass requires admin-equivalent role (SYSOP) — NOT
                # CREATOR (the builder-equivalent). Builders may have triggered the
                # window themselves; they should reconnect when it ends just like a
                # player. The shared ``account_bypasses_maintenance`` helper handles
                # both the engine and classic-rpg Account shapes.
                from maid_engine.maintenance import (
                    account_bypasses_maintenance,
                    telnet_maintenance_gate,
                )
                # CAVEAT 1: only emit ANSI to clients whose TTYPE
                # negotiation reported a colour-capable terminal.
                # Default to True when negotiation never completed
                # so legacy clients (which usually grok SGR) still
                # get the rich rendering.
                term = getattr(session, "terminal_type", None) or ""
                term_lc = term.lower()
                supports_ansi = (
                    not term
                    or "ansi" in term_lc
                    or "xterm" in term_lc
                    or "vt" in term_lc
                    or "color" in term_lc
                    or "screen" in term_lc
                    or "tmux" in term_lc
                )
                if await telnet_maintenance_gate(
                    is_admin=account_bypasses_maintenance(account),
                    send=session.send,
                    close=lambda: session.close(reason="maintenance"),
                    actor=getattr(account, "username", None),
                    client_supports_ansi=supports_ansi,
                ):
                    return

                # 2. Character Selection
                from maid_classic_rpg.world.setup_starter_area import (
                    STARTING_ROOM_ID,
                )

                char_handler = CharacterHandler(
                    session, account, self.character_manager,
                    starting_room_id=STARTING_ROOM_ID,
                )
                character = await char_handler.run()

                if not character:
                    await session.close()
                    return

                # 3. Enter World
                # Create ECS entity for the character
                # Check if entity already exists (reconnect), otherwise create
                entity = engine.world.get_entity(character.id)
                if not entity:
                    # Create new entity from character model
                    entity = create_character_entity(engine.world.entities, character)

                    # Recalculate inventory weight on load
                    if character.inventory and self.item_manager:
                        weight = await self.item_manager.recalculate_inventory_weight(character.inventory)
                        character.current_weight = weight
                        # Also update the inventory component
                        from maid_stdlib.components import InventoryComponent
                        inv_comp = entity.try_get(InventoryComponent)
                        if inv_comp:
                            inv_comp.current_weight = weight

                # Link session to player entity
                session.metadata["character_name"] = character.name
                session.metadata["character"] = character
                server.sessions.link_player(session.id, character.id)
                session.state = SessionState.PLAYING

                # Place entity in room (via World room index)
                if character.room_id:
                    engine.world.place_entity_in_room(character.id, character.room_id)
                    # Emit RoomEnterEvent so wilderness occupancy tracking
                    # (and other listeners) recognise the player is present.
                    from maid_engine.core.events import RoomEnterEvent

                    await engine.world.events.emit(
                        RoomEnterEvent(
                            entity_id=character.id,
                            room_id=character.room_id,
                            from_room_id=None,
                        )
                    )

                # Notify login
                await session.send_line("\r\n[Entering World...]")

                # Initial look
                from maid_engine.commands.registry import CommandContext

                ctx = CommandContext(
                    session=session,
                    player_id=character.id,
                    command="look",
                    args=[],
                    raw_input="look",
                    world=engine.world,
                    document_store=engine.document_store,
                    metadata={"engine": engine, "server": server},
                )
                # R9 critic: pass the resolved access level so admin/builder
                # classic commands reach their handlers; default to PLAYER
                # if access_level was never set (defensive only — login
                # always sets it above).
                from maid_engine.commands.registry import (
                    AccessLevel as _RPGAccessLevel,
                )
                _session_access = getattr(
                    session, "access_level", _RPGAccessLevel.PLAYER,
                )
                await engine.command_registry.execute(ctx, _session_access)
                await session.send_prompt()

                # 4. Main Loop
                while session.state not in (SessionState.DISCONNECTING, SessionState.DISCONNECTED):
                    try:
                        # Wait for input
                        line = await session.receive()
                        line = line.strip()

                        if not line:
                            await session.send_prompt()
                            continue

                        if line.lower() in ("quit", "exit"):
                            await session.send_line("Goodbye!")
                            break

                        # Execute command
                        parts = line.split()
                        cmd = parts[0].lower()
                        args = parts[1:]

                        ctx = CommandContext(
                            session=session,
                            player_id=character.id,
                            command=cmd,
                            args=args,
                            raw_input=line,
                            world=engine.world,
                            document_store=engine.document_store,
                            target=None,
                            metadata={"engine": engine, "server": server},
                        )

                        try:
                            # R9 critic: pass the session's resolved access
                            # level (set at login) so BUILDER/ADMIN/IMPLEMENTOR
                            # classic commands actually reach their handlers.
                            handled = await engine.command_registry.execute(
                                ctx,
                                getattr(
                                    session,
                                    "access_level",
                                    _RPGAccessLevel.PLAYER,
                                ),
                            )
                            if not handled:
                                await session.send_line("I don't understand that.")
                        except PermissionError as pe:
                            await session.send_line(str(pe))
                        except Exception as cmd_err:  # noqa: BLE001
                            logger.exception("Error executing command %r", cmd)
                            await session.send_line(f"An error occurred: {cmd_err}")

                        await session.send_prompt()

                    except (ConnectionError, asyncio.CancelledError):
                        break

            except Exception as e:  # noqa: BLE001
                logger.error(f"Error in connection handler: {e}")
                try:
                    await session.send_line(f"An error occurred: {e}")
                    await session.close()
                except (OSError, ConnectionError):
                    pass
            finally:
                # Always clean up: remove character entity from room on disconnect
                if character and character.id:
                    # Remove ECS entity entirely to prevent PositionSyncSystem
                    # from re-adding it to the room index on the next tick
                    engine.world.destroy_entity(character.id)
                    # Clear character's room_id so they don't appear
                    # in CharacterManager.get_in_room() queries
                    character.room_id = None
                    cached = self.character_manager._cache.get(character.id)
                    if cached and cached is not character:
                        cached.room_id = None

        engine.server.set_connection_handler(classic_rpg_connection_handler)

    # Load quest data
    # Quest data is loaded by QuestManager when systems initialize
    logger.debug("Quest system initialized")

    # Load crafting recipes (non-critical - game can run without recipes)
    try:
        from maid_classic_rpg.systems.crafting.recipes import RecipeManager

        # Get the data directory path using importlib.resources
        recipe_data_dir = files("maid_classic_rpg") / "data" / "crafting" / "recipes"
        if recipe_data_dir.is_dir():
            # Convert Traversable to Path for RecipeManager
            recipe_path = Path(str(recipe_data_dir))
            self._recipe_manager = RecipeManager(data_dir=recipe_path)
            count = await self._recipe_manager.load()
            logger.debug(f"Loaded {count} crafting recipes")
        else:
            logger.debug("No crafting recipes directory found, skipping")
    except ImportError:
        logger.debug("RecipeManager not available, crafting disabled")
    except Exception:  # noqa: BLE001
        # Log full traceback for unexpected errors, but don't crash
        logger.exception("Failed to load crafting recipes")

    # Note: Storage injection is handled automatically by the engine
    # for systems implementing StorageAware protocol

    # QuestManager is not a System, so we need to manually inject storage
    if engine.world and hasattr(engine.world, "quest_manager"):
        quest_manager = engine.world.quest_manager
        if hasattr(quest_manager, "set_storage"):
            quest_manager.set_storage(engine.document_store)
            logger.debug("Injected storage into QuestManager")

    # Ecosystem populations are now auto-seeded by
    # EcosystemSystem.startup() after loading definitions.

    # Wire WildernessSpawnAdapter to convert biome encounter/resource
    # strings into real ECS entities when wilderness rooms are generated.
    if engine.world:
        wilderness = getattr(engine.world, "wilderness", None)
        if wilderness is None:
            wilderness = engine.world.get_data("wilderness")
        if wilderness is not None:
            self._attach_wilderness_adapter(engine.world, wilderness, logger)
        else:
            # I-04: Wilderness may load after classic-rpg; schedule a
            # deferred check that runs once the remaining packs finish.
            self._pending_engine = engine
            self._deferred_task = asyncio.ensure_future(
                self._deferred_wilderness_setup()
            )

    # Load loot tables and loot item definitions into world data
    # so DeathSystem can resolve loot drops for spawned monsters.
    if engine.world:
        import yaml

        monster_data_dir = files("maid_classic_rpg") / "data" / "monsters"

        # Load loot tables
        loot_tables_path = Path(str(monster_data_dir / "loot_tables.yaml"))
        if loot_tables_path.exists():
            from maid_classic_rpg.models.entities.monster import LootTable

            with open(loot_tables_path) as f:
                loot_data = yaml.safe_load(f)
            for table_raw in loot_data.get("loot_tables", []):
                table = LootTable(**table_raw)
                engine.world.set_data(
                    "loot_tables", str(table.id), table
                )
            logger.debug(
                "Loaded %d loot tables",
                len(loot_data.get("loot_tables", [])),
            )

        # Load loot item definitions (names, descriptions, weights)
        loot_items_path = Path(str(monster_data_dir / "loot_items.yaml"))
        if loot_items_path.exists():
            with open(loot_items_path) as f:
                items_data = yaml.safe_load(f)
            for item_def in items_data.get("loot_items", []):
                engine.world.set_data(
                    "loot_item_defs", item_def["id"], item_def
                )
            logger.debug(
                "Loaded %d loot item definitions",
                len(items_data.get("loot_items", [])),
            )

    logger.info("Classic RPG content pack loaded")

    # I-02: Load monster templates into SpawnerSystem so startup()
    # has data for its initial spawn pass.
    if engine.world:
        from maid_classic_rpg.systems.npc.spawner import SpawnerSystem

        spawner = engine.world.systems.get(SpawnerSystem)
        if isinstance(spawner, SpawnerSystem):
            from maid_classic_rpg.systems.world.wilderness_spawn import (
                _load_monster_templates_from_yaml,
            )

            templates = _load_monster_templates_from_yaml()
            for template in templates.values():
                spawner.register_template(template)
            logger.debug(
                "Registered %d monster templates with SpawnerSystem",
                len(templates),
            )

            # Create spawn points for rooms tagged with spawn data
            for entity in engine.world.get_all_entities():
                spawn_tag = engine.world.get_data(
                    "room_spawn_config", str(entity.id)
                )
                if spawn_tag is not None:
                    from maid_classic_rpg.models.world.spawn import SpawnPoint

                    try:
                        sp = SpawnPoint.model_validate(spawn_tag)
                        spawner.register_spawn_point(sp)
                    except Exception:  # noqa: BLE001
                        logger.debug(
                            "Skipping invalid spawn config on room %s",
                            entity.id,
                        )
        else:
            logger.debug("SpawnerSystem not found; skipping template loading")

    # Register social commands (requires system instances from the world)
    if engine.world:
        systems_mgr = getattr(engine.world, "systems", None)
        if systems_mgr:
            from maid_classic_rpg.systems.social.groups import GroupSystem
            from maid_stdlib.systems.social.channels import ChannelSystem
            from maid_stdlib.systems.social.friends import FriendsSystem
            from maid_stdlib.systems.social.mail import MailSystem

            all_systems = getattr(systems_mgr, "systems", [])
            channels = None
            groups = None
            friends = None
            mail = None
            for system in all_systems:
                if isinstance(system, ChannelSystem):
                    channels = system
                elif isinstance(system, GroupSystem):
                    groups = system
                elif isinstance(system, FriendsSystem):
                    friends = system
                elif isinstance(system, MailSystem):
                    mail = system

            from maid_classic_rpg.commands.registry import get_all_commands

            if channels and groups and friends and mail:
                from maid_classic_rpg.systems.social.commands import (
                    register_social_commands,
                )

                register_social_commands(channels, groups, friends, mail)

                # Bridge social commands into the engine's
                # LayeredCommandRegistry.  On reload, commands may already
                # exist in the legacy registry so a pre/post diff finds
                # nothing "new".  To stay idempotent we first unregister
                # any prior classic-rpg social entries, then
                # unconditionally re-register every social command.
                if self._engine_registry is not None:
                    from maid_engine.commands.registry import (
                        LayeredCommandRegistry,
                    )

                    if isinstance(
                        self._engine_registry, LayeredCommandRegistry
                    ):
                        _social_command_names = {
                            "ooc", "newbie", "auctionchat", "shout",
                            "channel", "tell", "reply", "gtell",
                            "group", "follow", "ungroup", "assist",
                            "friend", "ignore", "privacy", "whois",
                            "finger", "mail",
                        }

                        # Unregister all known names first (cleans up stale commands)
                        for name in _social_command_names:
                            self._engine_registry.unregister(
                                name, "classic-rpg"
                            )

                        social_cmd_defs = [
                            cmd for cmd in get_all_commands()
                            if cmd.name in _social_command_names
                        ]

                        for cmd_def in social_cmd_defs:
                            try:
                                from maid_classic_rpg.commands import (
                                    _adapt_classic_handler,
                                )

                                self._engine_registry.register(
                                    name=cmd_def.name,
                                    handler=_adapt_classic_handler(cmd_def.handler),
                                    pack_name="classic-rpg",
                                    aliases=cmd_def.aliases,
                                    category=cmd_def.category.value,
                                    description=cmd_def.help_text,
                                    usage=cmd_def.usage,
                                    access_level=cmd_def.access_level,
                                    hidden=cmd_def.hidden,
                                    locks=cmd_def.locks,
                                    metadata=cmd_def.metadata,
                                    priority=cmd_def.priority if cmd_def.priority else None,
                                )
                            except Exception:  # noqa: BLE001
                                logger.debug(
                                    "Could not bridge social command '%s'",
                                    cmd_def.name,
                                )

                logger.debug("Registered social commands")
            else:
                logger.warning("Could not find all social systems for command registration")

            # Guild commands are independent of social systems
            from maid_classic_rpg.systems.social.guild import GuildSystem as GuildSystemType

            guild_system = None
            for system in all_systems:
                if isinstance(system, GuildSystemType):
                    guild_system = system
                    break

            if guild_system is not None:
                from maid_classic_rpg.systems.social.guild_commands import (
                    register_guild_commands,
                )

                register_guild_commands(guild_system)

                # Wire session provider for guild chat delivery
                def _get_sessions() -> list[Any]:
                    server = getattr(engine, "server", None)
                    if server is None:
                        return []
                    sessions_mgr = getattr(server, "sessions", None)
                    if sessions_mgr is None:
                        return []
                    all_fn = getattr(sessions_mgr, "all", None)
                    if callable(all_fn):
                        result = all_fn()
                        return list(result) if result else []
                    return []

                guild_system.set_session_provider(_get_sessions)

                # Bridge guild commands into the engine's LayeredCommandRegistry.
                # On reload, commands may already exist in the legacy registry so
                # the pre/post diff finds nothing "new".  To stay idempotent we
                # first unregister any prior classic-rpg guild entries, then
                # unconditionally re-register every guild command.
                if self._engine_registry is not None:
                    from maid_engine.commands.registry import (
                        LayeredCommandRegistry,
                    )

                    if isinstance(
                        self._engine_registry, LayeredCommandRegistry
                    ):
                        _guild_command_names = {"guild"}

                        # Unregister all known names first (cleans up stale commands)
                        for name in _guild_command_names:
                            self._engine_registry.unregister(
                                name, "classic-rpg"
                            )

                        guild_cmd_defs = [
                            cmd for cmd in get_all_commands()
                            if cmd.name in _guild_command_names
                        ]

                        for cmd_def in guild_cmd_defs:
                            try:
                                from maid_classic_rpg.commands import (
                                    _adapt_classic_handler,
                                )

                                self._engine_registry.register(
                                    name=cmd_def.name,
                                    handler=_adapt_classic_handler(cmd_def.handler),
                                    pack_name="classic-rpg",
                                    aliases=cmd_def.aliases,
                                    category=cmd_def.category.value,
                                    description=cmd_def.help_text,
                                    usage=cmd_def.usage,
                                    access_level=cmd_def.access_level,
                                    hidden=cmd_def.hidden,
                                    locks=cmd_def.locks,
                                    metadata=cmd_def.metadata,
                                    priority=cmd_def.priority if cmd_def.priority else None,
                                )
                            except Exception:  # noqa: BLE001
                                logger.debug(
                                    "Could not bridge guild command '%s'",
                                    cmd_def.name,
                                )

                logger.debug("Registered guild commands")

    # Subscribe audit logger to player connect/disconnect events
    if engine.world and engine.world.events:
        from maid_engine.core.events import (
            PlayerConnectedEvent,
            PlayerDisconnectedEvent,
        )
        from maid_engine.logging.audit import (
            AuditEntry,
            AuditEventType,
            get_audit_logger,
        )

        audit = get_audit_logger()

        async def _on_player_connected(event: PlayerConnectedEvent) -> None:
            await audit.log(AuditEntry(
                event_type=AuditEventType.ACCOUNT_PROMOTE,  # closest available type
                action="player_login",
                actor_id=event.player_id,
                details=f"session={event.session_id}",
            ))

        async def _on_player_disconnected(event: PlayerDisconnectedEvent) -> None:
            await audit.log(AuditEntry(
                event_type=AuditEventType.ACCOUNT_DEMOTE,  # closest available type
                action="player_logout",
                actor_id=event.player_id,
                details=f"session={event.session_id}",
            ))

        sub_id1 = engine.world.events.subscribe(
            PlayerConnectedEvent, _on_player_connected,
        )
        sub_id2 = engine.world.events.subscribe(
            PlayerDisconnectedEvent, _on_player_disconnected,
        )
        self._event_subscription_ids.extend([sub_id1, sub_id2])  # Track for cleanup (#15)
        logger.debug("Subscribed audit logger to player connect/disconnect events")

    # Register AI player entity setup hook so AI bots get RPG components
    if hasattr(engine, "ai_player_manager") and engine.ai_player_manager is not None:
        from maid_classic_rpg.entities.ai_player_setup import setup_rpg_components

        engine.ai_player_manager.register_entity_setup_hook(setup_rpg_components)
        self._ai_setup_hook = setup_rpg_components
        logger.debug("Registered AI player entity setup hook for Classic RPG components")

on_unload `async` ¶

on_unload(engine: GameEngine) -> None

Called when pack is unloaded.

Performs cleanup: - Saves any pending data - Releases resources

Source code in packages/maid-classic-rpg/src/maid_classic_rpg/pack.py

async def on_unload(self, engine: GameEngine) -> None:
    """Called when pack is unloaded.

    Performs cleanup:
    - Saves any pending data
    - Releases resources
    """
    import logging

    logger = logging.getLogger(__name__)
    logger.info("Unloading Classic RPG content pack...")

    # Build a set of system instances we own (from get_systems + GossipSystem).
    # Use identity checks to avoid double-shutdown.
    owned: set[int] = set()

    for system in reversed(self._pack_systems):
        obj_id = id(system)
        if obj_id in owned:
            continue
        owned.add(obj_id)
        if hasattr(system, "shutdown"):
            try:
                await system.shutdown()
            except Exception as e:  # noqa: BLE001
                logger.warning(f"Error during system shutdown: {e}")

    # I-01: Unregister each owned system from the SystemManager so the
    # engine tick loop no longer references them after hot-unload.
    systems_manager = engine.world.systems
    for system in self._pack_systems:
        with suppress(Exception):
            systems_manager.unregister(type(system))
    self._pack_systems.clear()

    # Also shut down and unregister source_pack-tagged systems
    # (e.g. GossipSystem registered via register_and_start).
    pack_systems = systems_manager.get_systems_for_pack("classic-rpg")
    for system in pack_systems:
        obj_id = id(system)
        if obj_id in owned:
            continue
        owned.add(obj_id)
        if hasattr(system, "shutdown"):
            try:
                await system.shutdown()
            except Exception as e:  # noqa: BLE001
                logger.warning(f"Error during system shutdown: {e}")
        with suppress(Exception):
            systems_manager.unregister(type(system))

    # Clean up GossipSystem tracked during on_load (#12).
    if self._gossip_system is not None and id(self._gossip_system) not in owned:
        try:
            await self._gossip_system.shutdown()
        except Exception as e:  # noqa: BLE001
            logger.warning(f"Error during GossipSystem shutdown: {e}")
        self._gossip_system = None

    # Unsubscribe event handlers to prevent leaks on reload (#15).
    if engine.world and engine.world.events:
        for sub_id in self._event_subscription_ids:
            engine.world.events.unsubscribe(sub_id)
        self._event_subscription_ids.clear()

    # Remove AI player entity setup hook (#R2-4).
    if (
        self._ai_setup_hook is not None
        and hasattr(engine, "ai_player_manager")
        and engine.ai_player_manager is not None
    ):
        hooks = getattr(engine.ai_player_manager, "_entity_setup_hooks", None)
        if hooks is not None:
            import contextlib

            with contextlib.suppress(ValueError):
                hooks.remove(self._ai_setup_hook)
        self._ai_setup_hook = None

    # Cancel deferred wilderness setup if still running.
    if self._deferred_task is not None and not self._deferred_task.done():
        self._deferred_task.cancel()
    self._deferred_task = None

    # Unregister WildernessSpawnAdapter callback.
    if self._wilderness_spawn_adapter is not None:
        # I-05: Clean up all spawned entities before dropping the adapter.
        try:
            self._wilderness_spawn_adapter.teardown()
        except Exception as e:  # noqa: BLE001
            logger.warning(f"Error during WildernessSpawnAdapter teardown: {e}")

        wilderness = getattr(engine.world, "wilderness", None)
        if wilderness is None and engine.world:
            wilderness = engine.world.get_data("wilderness")
        if wilderness is not None:
            if hasattr(wilderness, "unregister_room_callback"):
                wilderness.unregister_room_callback(
                    self._wilderness_spawn_adapter.on_room_generated,
                )
            if hasattr(wilderness, "unregister_cleanup_callback"):
                wilderness.unregister_cleanup_callback(
                    self._wilderness_spawn_adapter.on_room_destroyed,
                )
        self._wilderness_spawn_adapter = None

    # Shutdown QuestManager (not a System, but has pending saves to await)
    if engine.world and hasattr(engine.world, "quest_manager"):
        quest_manager = engine.world.quest_manager
        if hasattr(quest_manager, "shutdown"):
            try:
                await quest_manager.shutdown()
            except Exception as e:  # noqa: BLE001
                logger.warning(f"Error during QuestManager shutdown: {e}")

    # Clean up registered error patterns
    from maid_engine.ai_players.bugs.detector import AnomalyDetector

    AnomalyDetector.clear_custom_patterns("classic-rpg")

    # Restore previous connection handler so the server doesn't keep
    # a stale reference to this pack's handler after unload.
    if engine.server:
        engine.server.set_connection_handler(self._previous_connection_handler)
        self._previous_connection_handler = None

    logger.info("Classic RPG content pack unloaded")

register_api_routes ¶

register_api_routes(router: APIRouter) -> None

Register API routes. Classic RPG has no custom admin routes.

Source code in packages/maid-classic-rpg/src/maid_classic_rpg/pack.py

def register_api_routes(self, router: APIRouter) -> None:
    """Register API routes. Classic RPG has no custom admin routes."""
    pass

register_commands ¶

register_commands(registry: AnyCommandRegistry) -> None

Register commands provided by this pack.

Source code in packages/maid-classic-rpg/src/maid_classic_rpg/pack.py

def register_commands(self, registry: AnyCommandRegistry) -> None:
    """Register commands provided by this pack."""
    # Ensure stdlib commands (look, examine, inventory, etc.) are always registered
    from maid_stdlib.commands import register_stdlib_commands
    register_stdlib_commands(registry)

    from maid_classic_rpg.commands import register_classic_rpg_commands
    register_classic_rpg_commands(registry, pack_name="classic-rpg")

    # Store engine registry reference for social command bridging (I-03)
    self._engine_registry = registry

    # Register hooks for command execution
    # These hooks are defined in maid-engine but should be configured and
    # registered by content packs that implement gameplay
    from maid_engine.commands import LayeredCommandRegistry

    if isinstance(registry, LayeredCommandRegistry):
        # Block most commands while the player is a ghost
        from maid_classic_rpg.commands.handlers.ghost import (
            ghost_restriction_hook,
        )
        from maid_classic_rpg.commands.restrictions import (
            frozen_restriction_hook,
            mute_restriction_hook,
        )
        from maid_engine.commands import (
            CombatStateHook,
            CooldownHook,
            CooldownRecordHook,
            HookPriority,
        )

        registry.register_pre_hook(
            name="ghost_restriction",
            handler=ghost_restriction_hook,
            priority=HookPriority.FIRST,
        )

        registry.register_pre_hook(
            name="frozen_restriction",
            handler=frozen_restriction_hook,
            priority=HookPriority.FIRST,
        )

        registry.register_pre_hook(
            name="mute_restriction",
            handler=mute_restriction_hook,
            priority=HookPriority.HIGH,
        )

        # Block commands that shouldn't be used during combat:
        # - quit/logout: Can't leave the game while fighting
        # - crafting category: Can't craft items while fighting
        # - economy category: Can't shop/trade while fighting
        # - builder category: Can't build while fighting
        combat_hook = CombatStateHook(
            blocked_commands=["quit", "logout"],
            blocked_categories=["crafting", "economy", "builder"],
        )
        registry.register_pre_hook(
            name="combat_state_block",
            handler=combat_hook,
            priority=HookPriority.HIGH,
        )

        # Register CooldownHook to enforce command cooldowns
        # Commands can specify cooldown times in their metadata (in seconds)
        # e.g., @command(name="teleport", metadata={"cooldown": 60})
        cooldown_hook = CooldownHook()
        registry.register_pre_hook(
            name="cooldown_check",
            handler=cooldown_hook,
            priority=HookPriority.HIGH,
        )

        # Register CooldownRecordHook to record successful command usage
        # This must share the same CooldownHook instance to track cooldowns
        cooldown_record_hook = CooldownRecordHook(cooldown_hook)
        registry.register_post_hook(
            name="cooldown_record",
            handler=cooldown_record_hook,
            priority=HookPriority.NORMAL,
        )

register_component_types ¶

register_component_types(
    registry: ComponentRegistry,
) -> None

Register component types for persistence serialization.

Source code in packages/maid-classic-rpg/src/maid_classic_rpg/pack.py

def register_component_types(self, registry: ComponentRegistry) -> None:
    """Register component types for persistence serialization."""
    from maid_classic_rpg import components as classic_components
    from maid_engine.core.ecs.component import Component

    stdlib_reexports = {"CombatComponent", "GoldComponent", "StatsComponent"}
    for name in getattr(classic_components, "__all__", []):
        if name in stdlib_reexports:
            continue
        candidate = getattr(classic_components, name, None)
        if (
            isinstance(candidate, type)
            and issubclass(candidate, Component)
            and candidate is not Component
        ):
            registry.register(
                candidate,
                pack_name="classic-rpg",
                allow_overwrite=True,
            )

register_document_schemas ¶

register_document_schemas(store: DocumentStore) -> None

Register document schemas used by this pack.

Source code in packages/maid-classic-rpg/src/maid_classic_rpg/pack.py

def register_document_schemas(self, store: DocumentStore) -> None:
    """Register document schemas used by this pack."""
    from maid_classic_rpg.models import register_classic_rpg_schemas
    from maid_classic_rpg.systems.quests.generation.history import (
        register_generation_schemas,
    )

    register_classic_rpg_schemas(store)
    has_collection = getattr(store, "has_collection", None)
    if not callable(has_collection) or not any(
        has_collection(name)
        for name in (
            "generated_quests",
            "quest_history",
            "quest_consequences",
        )
    ):
        register_generation_schemas(store)

maid-classic-rpg¶

Installation¶

Module Overview¶

Systems (maid_classic_rpg.systems)¶

Commands (maid_classic_rpg.commands)¶

Models (maid_classic_rpg.models)¶

Events (maid_classic_rpg.events)¶

Components (maid_classic_rpg.components)¶

Systems¶

Combat System¶

combat ¶

BaseCombatSystem ¶

start_combat abstractmethod async ¶

end_combat abstractmethod async ¶

process_attack abstractmethod async ¶

calculate_damage abstractmethod async ¶

is_in_combat ¶

BodyDamageResult ¶

is_fatal property ¶

BodySystem ¶

update async ¶

apply_damage_to_location ¶

heal_location ¶

restore_part ¶

get_total_health ¶

get_health_percentage ¶

is_vital_destroyed ¶

get_status_summary ¶

add_part_effect ¶

remove_part_effect ¶

clear_part_effects ¶

DamageSystem ¶

startup async ¶

update async ¶

shutdown async ¶

remove_body ¶

get_or_create_body ¶

apply_attack_result async ¶

apply_direct_damage ¶

heal_character ¶

is_alive ¶

get_health_status ¶

get_body_status ¶

StatusEffectManager ¶

shutdown async ¶

update async ¶

add_effect ¶

remove_effect ¶

remove_all_effects ¶

get_effects ¶

has_effect ¶

get_effect_stacks ¶

get_effect_modifier ¶

MeleeCombatSystem ¶

update async ¶

resolve_attack async ¶

RangedCombatSystem ¶

update async ¶

shutdown async ¶

resolve_attack async ¶

get_reload_remaining ¶

is_reloading ¶

force_reload ¶

cancel_reload ¶

Magic System¶

magic ¶

ActiveBuff dataclass ¶

BuffAppliedEvent dataclass ¶

BuffDispelledEvent dataclass ¶

BuffExpiredEvent dataclass ¶

SpellEffectSystem ¶

update async ¶

register_handler ¶

apply_effect async ¶

add_buff ¶

remove_buff ¶

get_buffs ¶

get_buff ¶

has_buff ¶

get_stat_modifier ¶

Systems (`maid_classic_rpg.systems`)¶

Commands (`maid_classic_rpg.commands`)¶

Models (`maid_classic_rpg.models`)¶

Events (`maid_classic_rpg.events`)¶

Components (`maid_classic_rpg.components`)¶

start_combat `abstractmethod` `async` ¶

end_combat `abstractmethod` `async` ¶

process_attack `abstractmethod` `async` ¶

calculate_damage `abstractmethod` `async` ¶

is_fatal `property` ¶

update `async` ¶

startup `async` ¶

update `async` ¶

shutdown `async` ¶

apply_attack_result `async` ¶

shutdown `async` ¶

update `async` ¶

update `async` ¶

resolve_attack `async` ¶

update `async` ¶

shutdown `async` ¶

resolve_attack `async` ¶

ActiveBuff `dataclass` ¶

BuffAppliedEvent `dataclass` ¶

BuffDispelledEvent `dataclass` ¶

BuffExpiredEvent `dataclass` ¶

update `async` ¶

apply_effect `async` ¶

dispel `async` ¶

SpellCastCompleteEvent `dataclass` ¶

SpellCastInterruptedEvent `dataclass` ¶

SpellCastStartEvent `dataclass` ¶

SpellLearnedEvent `dataclass` ¶

CastingState `dataclass` ¶

SpellCastResult `dataclass` ¶

startup `async` ¶

shutdown `async` ¶

update `async` ¶

start_cast `async` ¶

interrupt_cast `async` ¶

CraftingResult `dataclass` ¶

recipe_manager `property` ¶

station_registry `property` ¶

startup `async` ¶

shutdown `async` ¶

update `async` ¶

attempt_craft `async` ¶

enchant_item `async` ¶

EnchantResult `dataclass` ¶

RepairResult `dataclass` ¶

repair_item `async` ¶

socket_gem `async` ¶

remove_gem `async` ¶

SocketResult `dataclass` ¶

GatheringResult `dataclass` ¶

startup `async` ¶

shutdown `async` ¶

update `async` ¶

attempt_gather `async` ¶

attempt_skinning `async` ¶