MAID Command System Enhancements Design Specification¶
Document Version: 1.0
Date: January 30, 2026
Status: Draft
Authors: MAID Development Team
Executive Summary¶
This document specifies enhancements to MAID's command system to achieve feature parity with Evennia's mature command infrastructure. The four major enhancement areas are:
- Advanced Argument Parsing - Pattern-based argument extraction with type coercion
- Auto-Generated Help from Docstrings - Automatic help text from command handler documentation
- Pre/Post Command Hooks - Middleware-style execution pipeline
- Enhanced Lock System - Fine-grained permission expressions beyond access levels
These enhancements address the gaps identified in the MAID vs Evennia comparison where Evennia leads in argument parsing, auto-help generation, pre/post hooks, and lock string expressions.
Table of Contents¶
- Feature 1: Advanced Argument Parsing
- Feature 2: Auto-Generated Help from Docstrings
- Feature 3: Pre/Post Command Hooks
- Feature 4: Enhanced Lock System
- Appendix A: Migration Guide
- Appendix B: Testing Requirements
Feature 1: Advanced Argument Parsing¶
1.1 Feature Overview¶
What it does:
Provides a declarative argument definition system that automatically parses, validates, and coerces command arguments based on a schema definition. Supports required/optional arguments, type conversion, choices, defaults, and complex patterns like <target> = <message> or [obj] [in|on] <container>.
Why it's needed:
- Current state: Manual args = context.args[0] access with no validation
- Commands must manually handle missing args, type conversion, and errors
- No consistency across commands for error messages
- Evennia provides MuxCommand.parse() with sophisticated pattern matching
Current Limitation:
MAID's CommandParser in maid-stdlib provides basic tokenization and target indexing (2.sword, all.sword), but no type coercion, schema validation, or pattern matching.
1.2 User Stories¶
US-1.1: Typed Argument Declaration
As a command developer, I want to declare arguments with types so that the parser automatically converts "5" to an integer and reports errors for invalid input.
US-1.2: Required vs Optional Arguments
As a command developer, I want to mark arguments as required or optional with defaults so that users get helpful error messages for missing required arguments.
US-1.3: Choice Arguments
As a command developer, I want to restrict an argument to a set of valid choices so that invalid options are rejected with a helpful list of alternatives.
US-1.4: Multi-Word Arguments
As a command developer, I want to capture multi-word arguments (like "a long message") so that users can input natural text without special quoting.
US-1.5: Complex Patterns
As a command developer, I want to define patterns like
<player> = <message>for thepagecommand so that the parser extracts both parts automatically.
US-1.6: Entity Resolution
As a command developer, I want arguments that automatically resolve to entities in the room or inventory so that I don't have to write matching logic in every command.
1.3 Technical Requirements¶
1.3.1 Core Requirements¶
| ID | Requirement | Priority |
|---|---|---|
| ARG-001 | System SHALL support typed argument declarations (str, int, float, bool) | P0 |
| ARG-002 | System SHALL support required and optional arguments with defaults | P0 |
| ARG-003 | System SHALL support choice constraints with enumerated valid values | P0 |
| ARG-004 | System SHALL provide automatic type coercion with error messages | P0 |
| ARG-005 | System SHALL support multi-word "rest of line" argument capture | P0 |
| ARG-006 | System SHALL support pattern-based parsing with delimiters | P1 |
| ARG-007 | System SHALL support entity reference resolution (room/inventory) | P1 |
| ARG-008 | System SHALL support regex-based argument validation | P1 |
| ARG-009 | System SHALL support argument grouping (mutually exclusive) | P2 |
| ARG-010 | System SHALL support repeatable arguments (collect all) | P1 |
| ARG-011 | System SHALL preserve backward compatibility with raw args access | P0 |
| ARG-012 | System SHALL provide helpful, consistent error messages | P0 |
1.3.2 Entity Resolution Requirements¶
| ID | Requirement | Priority |
|---|---|---|
| ENT-001 | System SHALL resolve "sword" to matching entity in room or inventory | P0 |
| ENT-002 | System SHALL resolve "2.sword" to second matching entity | P0 |
| ENT-003 | System SHALL resolve "all.sword" to all matching entities | P0 |
| ENT-004 | System SHALL resolve "me" to the player entity | P0 |
| ENT-005 | System SHALL resolve "here" to the current room | P0 |
| ENT-006 | System SHALL support search scope: room, inventory, equipment, all | P1 |
| ENT-007 | System SHALL emit appropriate "not found" messages | P0 |
1.4 API/Interface Design¶
1.4.1 Argument Definition Classes¶
# packages/maid-engine/src/maid_engine/commands/arguments.py
from __future__ import annotations
import re
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum, auto
from typing import TYPE_CHECKING, Any, Callable, Generic, TypeVar
from uuid import UUID
if TYPE_CHECKING:
from maid_engine.commands.registry import CommandContext
from maid_engine.core.ecs.entity import Entity
T = TypeVar("T")
class ArgumentType(Enum):
"""Built-in argument types."""
STRING = auto()
INTEGER = auto()
FLOAT = auto()
BOOLEAN = auto()
ENTITY = auto()
ENTITY_LIST = auto()
DIRECTION = auto()
PLAYER = auto()
REST = auto() # Capture rest of line
class SearchScope(Enum):
"""Where to search for entity arguments."""
ROOM = auto()
INVENTORY = auto()
EQUIPMENT = auto()
ROOM_AND_INVENTORY = auto()
ALL = auto()
@dataclass
class ArgumentError:
"""Error from argument parsing."""
argument_name: str
message: str
value: str | None = None
expected_type: str | None = None
valid_choices: list[str] | None = None
@dataclass
class ArgumentSpec(Generic[T]):
"""Specification for a command argument."""
name: str
arg_type: ArgumentType
required: bool = True
default: T | None = None
description: str = ""
choices: list[T] | None = None
pattern: str | None = None # Regex pattern for validation
search_scope: SearchScope = SearchScope.ROOM_AND_INVENTORY
min_value: float | None = None
max_value: float | None = None
def validate(self, value: Any) -> tuple[bool, str | None]:
"""Validate a parsed value. Returns (valid, error_message)."""
if self.choices and value not in self.choices:
choices_str = ", ".join(str(c) for c in self.choices)
return False, f"Must be one of: {choices_str}"
if self.pattern and isinstance(value, str):
if not re.match(self.pattern, value):
return False, f"Invalid format, expected pattern: {self.pattern}"
if self.min_value is not None and isinstance(value, (int, float)):
if value < self.min_value:
return False, f"Must be at least {self.min_value}"
if self.max_value is not None and isinstance(value, (int, float)):
if value > self.max_value:
return False, f"Must be at most {self.max_value}"
return True, None
@dataclass
class ArgumentPattern:
"""Pattern-based argument extraction.
Examples:
"<target> = <message>" for 'page bob = hello there'
"[obj] [in|on] <container>" for 'put sword in bag' or 'put sword bag'
"""
pattern: str
arguments: list[ArgumentSpec]
delimiter_optional: bool = False
def compile(self) -> re.Pattern:
"""Compile pattern to regex for extraction."""
# Implementation converts pattern DSL to regex
...
@dataclass
class ParsedArguments:
"""Result of argument parsing."""
success: bool
arguments: dict[str, Any] = field(default_factory=dict)
errors: list[ArgumentError] = field(default_factory=list)
raw_args: list[str] = field(default_factory=list)
raw_input: str = ""
def get(self, name: str, default: Any = None) -> Any:
"""Get argument value with optional default."""
return self.arguments.get(name, default)
def __getitem__(self, name: str) -> Any:
"""Get argument value."""
return self.arguments[name]
def __contains__(self, name: str) -> bool:
"""Check if argument was provided."""
return name in self.arguments
class EntityReference:
"""Resolved entity reference from argument."""
def __init__(
self,
entities: list[Entity],
original_text: str,
index: int | None = None,
all_matching: bool = False,
):
self.entities = entities
self.original_text = original_text
self.index = index
self.all_matching = all_matching
@property
def entity(self) -> Entity | None:
"""Get single entity (first if multiple)."""
return self.entities[0] if self.entities else None
@property
def found(self) -> bool:
"""Whether any entity was found."""
return len(self.entities) > 0
def __iter__(self):
return iter(self.entities)
def __len__(self):
return len(self.entities)
1.4.2 Argument Parser Implementation¶
# packages/maid-engine/src/maid_engine/commands/arguments.py (continued)
class ArgumentParser:
"""Parses command arguments against a specification.
Usage:
parser = ArgumentParser([
ArgumentSpec("target", ArgumentType.ENTITY, required=True),
ArgumentSpec("amount", ArgumentType.INTEGER, default=1),
])
result = await parser.parse(context)
if not result.success:
await context.session.send(result.errors[0].message)
return
target = result["target"]
amount = result["amount"]
"""
def __init__(
self,
arguments: list[ArgumentSpec] | None = None,
pattern: ArgumentPattern | None = None,
):
self.arguments = arguments or []
self.pattern = pattern
self._compiled_pattern: re.Pattern | None = None
if pattern:
self._compiled_pattern = pattern.compile()
async def parse(self, context: CommandContext) -> ParsedArguments:
"""Parse arguments from command context."""
result = ParsedArguments(
success=True,
raw_args=context.args,
raw_input=context.raw_input,
)
if self.pattern:
return await self._parse_pattern(context, result)
else:
return await self._parse_positional(context, result)
async def _parse_positional(
self,
context: CommandContext,
result: ParsedArguments
) -> ParsedArguments:
"""Parse positional arguments."""
args = context.args
arg_index = 0
for spec in self.arguments:
# Handle REST type - captures remaining arguments
if spec.arg_type == ArgumentType.REST:
if arg_index < len(args):
result.arguments[spec.name] = " ".join(args[arg_index:])
elif spec.required:
result.success = False
result.errors.append(ArgumentError(
argument_name=spec.name,
message=f"Missing required argument: {spec.name}",
))
else:
result.arguments[spec.name] = spec.default or ""
break
# Get raw value
if arg_index < len(args):
raw_value = args[arg_index]
arg_index += 1
elif spec.required:
result.success = False
result.errors.append(ArgumentError(
argument_name=spec.name,
message=f"Missing required argument: {spec.name}",
))
continue
else:
result.arguments[spec.name] = spec.default
continue
# Parse and validate
parsed = await self._parse_value(spec, raw_value, context)
if isinstance(parsed, ArgumentError):
result.success = False
result.errors.append(parsed)
else:
result.arguments[spec.name] = parsed
return result
async def _parse_pattern(
self,
context: CommandContext,
result: ParsedArguments,
) -> ParsedArguments:
"""Parse arguments using pattern matching."""
# Extract the args portion (everything after command)
full_args = context.raw_input
if context.command in full_args.lower():
# Remove command prefix
idx = full_args.lower().index(context.command) + len(context.command)
full_args = full_args[idx:].strip()
match = self._compiled_pattern.match(full_args)
if not match:
result.success = False
result.errors.append(ArgumentError(
argument_name="",
message=f"Invalid syntax. Usage: {context.command} {self.pattern.pattern}",
))
return result
# Extract matched groups
groups = match.groupdict()
for spec in self.pattern.arguments:
if spec.name in groups:
raw_value = groups[spec.name]
if raw_value is None:
if spec.required:
result.success = False
result.errors.append(ArgumentError(
argument_name=spec.name,
message=f"Missing required argument: {spec.name}",
))
else:
result.arguments[spec.name] = spec.default
else:
parsed = await self._parse_value(spec, raw_value, context)
if isinstance(parsed, ArgumentError):
result.success = False
result.errors.append(parsed)
else:
result.arguments[spec.name] = parsed
return result
async def _parse_value(
self,
spec: ArgumentSpec,
raw_value: str,
context: CommandContext,
) -> Any | ArgumentError:
"""Parse a single value according to its type."""
try:
match spec.arg_type:
case ArgumentType.STRING:
value = raw_value
case ArgumentType.INTEGER:
try:
value = int(raw_value)
except ValueError:
return ArgumentError(
argument_name=spec.name,
message=f"'{raw_value}' is not a valid number",
value=raw_value,
expected_type="integer",
)
case ArgumentType.FLOAT:
try:
value = float(raw_value)
except ValueError:
return ArgumentError(
argument_name=spec.name,
message=f"'{raw_value}' is not a valid decimal number",
value=raw_value,
expected_type="decimal",
)
case ArgumentType.BOOLEAN:
if raw_value.lower() in ("true", "yes", "on", "1"):
value = True
elif raw_value.lower() in ("false", "no", "off", "0"):
value = False
else:
return ArgumentError(
argument_name=spec.name,
message=f"'{raw_value}' is not a valid boolean (use yes/no, true/false)",
value=raw_value,
expected_type="boolean",
)
case ArgumentType.ENTITY:
value = await self._resolve_entity(raw_value, spec, context)
if not value.found:
return ArgumentError(
argument_name=spec.name,
message=f"You don't see '{raw_value}' here.",
value=raw_value,
)
case ArgumentType.ENTITY_LIST:
value = await self._resolve_entity(raw_value, spec, context, allow_all=True)
if not value.found:
return ArgumentError(
argument_name=spec.name,
message=f"You don't see '{raw_value}' here.",
value=raw_value,
)
case ArgumentType.DIRECTION:
value = self._normalize_direction(raw_value)
if value is None:
return ArgumentError(
argument_name=spec.name,
message=f"'{raw_value}' is not a valid direction",
value=raw_value,
valid_choices=["north", "south", "east", "west", "up", "down"],
)
case ArgumentType.PLAYER:
value = await self._resolve_player(raw_value, context)
if value is None:
return ArgumentError(
argument_name=spec.name,
message=f"Player '{raw_value}' is not online.",
value=raw_value,
)
case ArgumentType.REST:
value = raw_value
case _:
value = raw_value
# Validate against spec constraints
valid, error_msg = spec.validate(value)
if not valid:
return ArgumentError(
argument_name=spec.name,
message=error_msg,
value=raw_value,
valid_choices=spec.choices,
)
return value
except Exception as e:
return ArgumentError(
argument_name=spec.name,
message=f"Error parsing argument: {e}",
value=raw_value,
)
async def _resolve_entity(
self,
text: str,
spec: ArgumentSpec,
context: CommandContext,
allow_all: bool = False,
) -> EntityReference:
"""Resolve text to entity reference."""
# Parse index/all prefix
index = None
all_matching = False
search_text = text
if "." in text:
prefix, name = text.split(".", 1)
if prefix.isdigit():
index = int(prefix)
search_text = name
elif prefix.lower() == "all" and allow_all:
all_matching = True
search_text = name
# Special keywords
if search_text.lower() == "me":
player_entity = context.world.get_entity(context.player_id)
return EntityReference([player_entity], text) if player_entity else EntityReference([], text)
if search_text.lower() == "here":
# Get current room entity
room_id = context.world.room_index.get_room(context.player_id)
if room_id:
room_entity = context.world.get_entity(room_id)
return EntityReference([room_entity], text) if room_entity else EntityReference([], text)
return EntityReference([], text)
# Search for matching entities
matches = []
# Get player's room
room_id = context.world.room_index.get_room(context.player_id)
if spec.search_scope in (SearchScope.ROOM, SearchScope.ROOM_AND_INVENTORY, SearchScope.ALL):
if room_id:
for entity_id in context.world.room_index.get_entities(room_id):
entity = context.world.get_entity(entity_id)
if entity and self._entity_matches_keyword(entity, search_text):
matches.append(entity)
if spec.search_scope in (SearchScope.INVENTORY, SearchScope.ROOM_AND_INVENTORY, SearchScope.ALL):
player = context.world.get_entity(context.player_id)
if player:
inventory = player.get_component("InventoryComponent")
if inventory:
for item_id in inventory.items:
item = context.world.get_entity(item_id)
if item and self._entity_matches_keyword(item, search_text):
matches.append(item)
if spec.search_scope in (SearchScope.EQUIPMENT, SearchScope.ALL):
player = context.world.get_entity(context.player_id)
if player:
equipment = player.get_component("EquipmentComponent")
if equipment:
for slot, item_id in equipment.slots.items():
if item_id:
item = context.world.get_entity(item_id)
if item and self._entity_matches_keyword(item, search_text):
matches.append(item)
# Apply index filter
if index is not None and index <= len(matches):
matches = [matches[index - 1]] # 1-indexed
elif not all_matching and matches:
matches = [matches[0]] # Default to first match
return EntityReference(matches, text, index, all_matching)
def _entity_matches_keyword(self, entity: Entity, keyword: str) -> bool:
"""Check if entity matches a search keyword."""
keyword = keyword.lower()
# Check DescriptionComponent keywords
desc = entity.get_component("DescriptionComponent")
if desc:
if hasattr(desc, "keywords") and keyword in [k.lower() for k in desc.keywords]:
return True
if hasattr(desc, "name") and keyword in desc.name.lower():
return True
# Check name directly on entity
if hasattr(entity, "name") and keyword in entity.name.lower():
return True
return False
async def _resolve_player(self, name: str, context: CommandContext) -> Entity | None:
"""Resolve player name to online player entity."""
name_lower = name.lower()
# Search all entities with PlayerComponent
for entity_id, entity in context.world._entities.items():
if entity.has_component("PlayerComponent"):
player_comp = entity.get_component("PlayerComponent")
if hasattr(player_comp, "name") and player_comp.name.lower() == name_lower:
return entity
return None
def _normalize_direction(self, text: str) -> str | None:
"""Normalize direction alias to full direction name."""
DIRECTIONS = {
"n": "north", "north": "north",
"s": "south", "south": "south",
"e": "east", "east": "east",
"w": "west", "west": "west",
"u": "up", "up": "up",
"d": "down", "down": "down",
"ne": "northeast", "northeast": "northeast",
"nw": "northwest", "northwest": "northwest",
"se": "southeast", "southeast": "southeast",
"sw": "southwest", "southwest": "southwest",
"in": "in", "out": "out",
}
return DIRECTIONS.get(text.lower())
1.4.3 Decorator-Based Argument Definition¶
# packages/maid-engine/src/maid_engine/commands/decorators.py
from functools import wraps
from typing import Callable, ParamSpec, TypeVar
from .arguments import ArgumentParser, ArgumentSpec, ArgumentType, ParsedArguments
from .registry import CommandContext
P = ParamSpec("P")
R = TypeVar("R")
def arguments(*specs: ArgumentSpec):
"""Decorator to define command arguments.
Usage:
@arguments(
ArgumentSpec("target", ArgumentType.ENTITY, description="What to examine"),
ArgumentSpec("detail", ArgumentType.STRING, required=False, default="brief"),
)
async def cmd_examine(ctx: CommandContext, args: ParsedArguments) -> bool:
target = args["target"]
detail = args["detail"]
...
"""
parser = ArgumentParser(arguments=list(specs))
def decorator(func: Callable[..., R]) -> Callable[..., R]:
@wraps(func)
async def wrapper(ctx: CommandContext, *args, **kwargs):
parsed = await parser.parse(ctx)
if not parsed.success:
for error in parsed.errors:
await ctx.session.send_line(f"Error: {error.message}")
return False
return await func(ctx, parsed, *args, **kwargs)
# Attach parser for introspection (help generation)
wrapper.__argument_parser__ = parser
wrapper.__argument_specs__ = list(specs)
return wrapper
return decorator
def pattern(pattern_string: str, *specs: ArgumentSpec, delimiter_optional: bool = False):
"""Decorator for pattern-based argument parsing.
Usage:
@pattern(
"<target> = <message>",
ArgumentSpec("target", ArgumentType.PLAYER),
ArgumentSpec("message", ArgumentType.REST),
)
async def cmd_page(ctx: CommandContext, args: ParsedArguments) -> bool:
target = args["target"]
message = args["message"]
...
"""
from .arguments import ArgumentPattern
arg_pattern = ArgumentPattern(
pattern=pattern_string,
arguments=list(specs),
delimiter_optional=delimiter_optional,
)
parser = ArgumentParser(pattern=arg_pattern)
def decorator(func: Callable[..., R]) -> Callable[..., R]:
@wraps(func)
async def wrapper(ctx: CommandContext, *args, **kwargs):
parsed = await parser.parse(ctx)
if not parsed.success:
for error in parsed.errors:
await ctx.session.send_line(f"Error: {error.message}")
return False
return await func(ctx, parsed, *args, **kwargs)
wrapper.__argument_parser__ = parser
wrapper.__argument_pattern__ = pattern_string
wrapper.__argument_specs__ = list(specs)
return wrapper
return decorator
1.5 Usage Examples¶
Basic Command with Arguments¶
from maid_engine.commands.arguments import ArgumentSpec, ArgumentType, ParsedArguments
from maid_engine.commands.decorators import arguments
from maid_engine.commands.registry import CommandContext
@arguments(
ArgumentSpec("item", ArgumentType.ENTITY, description="Item to give"),
ArgumentSpec("target", ArgumentType.PLAYER, description="Player to give item to"),
ArgumentSpec("amount", ArgumentType.INTEGER, required=False, default=1,
min_value=1, max_value=100, description="How many to give"),
)
async def cmd_give(ctx: CommandContext, args: ParsedArguments) -> bool:
"""Give an item from your inventory to another player."""
item = args["item"].entity
target = args["target"]
amount = args["amount"]
# Transfer logic...
await ctx.session.send_line(f"You give {item.name} to {target.name}.")
return True
Pattern-Based Command¶
from maid_engine.commands.decorators import pattern
@pattern(
"<target> = <message>",
ArgumentSpec("target", ArgumentType.PLAYER, description="Player to message"),
ArgumentSpec("message", ArgumentType.REST, description="Message to send"),
)
async def cmd_page(ctx: CommandContext, args: ParsedArguments) -> bool:
"""Send a private message to another player.
Usage: page <player> = <message>
Example: page Bob = Hello, how are you?
"""
target = args["target"]
message = args["message"]
await target.session.send_line(f"[Page from {ctx.player.name}]: {message}")
await ctx.session.send_line(f"[Page to {target.name}]: {message}")
return True
Entity List Command¶
@arguments(
ArgumentSpec("items", ArgumentType.ENTITY_LIST, description="Items to drop"),
)
async def cmd_drop(ctx: CommandContext, args: ParsedArguments) -> bool:
"""Drop items from your inventory.
Usage: drop <item>
drop all.sword (drops all swords)
drop 2.potion (drops the second potion)
"""
items = args["items"] # EntityReference with list of entities
for item in items:
# Drop logic...
await ctx.session.send_line(f"You drop {item.name}.")
return True
1.6 Data Model¶
┌──────────────────────────────────────────────────────────────────┐
│ ArgumentParser │
│ ┌──────────────────┐ ┌─────────────────────────────────────┐ │
│ │ ArgumentSpec[] │ │ ArgumentPattern (optional) │ │
│ │ - name │ │ - pattern: "<target> = <message>" │ │
│ │ - arg_type │ │ - arguments: ArgumentSpec[] │ │
│ │ - required │ │ - compiled_regex │ │
│ │ - default │ └─────────────────────────────────────┘ │
│ │ - choices │ │
│ │ - search_scope │ │
│ └──────────────────┘ │
└────────────────────────────┬─────────────────────────────────────┘
│ parse(context)
▼
┌──────────────────────────────────────────────────────────────────┐
│ ParsedArguments │
│ - success: bool │
│ - arguments: dict[str, Any] │
│ - errors: list[ArgumentError] │
│ - raw_args: list[str] │
└──────────────────────────────────────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────────┐
│ EntityReference │
│ - entities: list[Entity] │
│ - original_text: str │
│ - index: int | None (for "2.sword") │
│ - all_matching: bool (for "all.sword") │
└──────────────────────────────────────────────────────────────────┘
1.7 Acceptance Criteria¶
| ID | Criterion | Verification |
|---|---|---|
| AC-1.1 | Commands can declare typed arguments and receive parsed values | Unit test with all types |
| AC-1.2 | Missing required arguments produce descriptive error messages | Unit test for error formatting |
| AC-1.3 | Optional arguments use default values when not provided | Unit test defaults |
| AC-1.4 | Choice constraints reject invalid values with helpful message | Unit test choices |
| AC-1.5 | Entity resolution finds items by keyword in room and inventory | Integration test |
| AC-1.6 | "2.sword" syntax resolves to correct indexed entity | Unit test |
| AC-1.7 | "all.sword" syntax returns all matching entities | Unit test |
| AC-1.8 | Pattern-based parsing extracts delimited arguments | Unit test patterns |
| AC-1.9 | Existing commands continue to work with raw args access | Regression test |
| AC-1.10 | Type coercion errors are human-readable | Manual review |
Feature 2: Auto-Generated Help from Docstrings¶
2.1 Feature Overview¶
What it does:
Automatically generates command help text by parsing Python docstrings from command handlers. Combines docstring content with argument specifications to produce comprehensive help output.
Why it's needed:
- Current state: Help text must be manually duplicated in both handler docstring and registration
- Evennia extracts help from __doc__ attributes automatically
- Reduces documentation maintenance burden and prevents drift
2.2 User Stories¶
US-2.1: Docstring as Help Source
As a command developer, I want my handler's docstring to become the help text so that I only write documentation once.
US-2.2: Argument Documentation in Help
As a player, I want help to show argument descriptions so that I understand what each argument does.
US-2.3: Usage Pattern Generation
As a command developer, I want usage patterns auto-generated from argument specs so that I don't have to manually maintain usage strings.
US-2.4: Category Inheritance
As a command developer, I want help to include the command category so that players can browse related commands.
2.3 Technical Requirements¶
| ID | Requirement | Priority |
|---|---|---|
| HELP-001 | System SHALL extract short description from first line of docstring | P0 |
| HELP-002 | System SHALL extract long description from remaining docstring | P0 |
| HELP-003 | System SHALL auto-generate usage from argument specs | P0 |
| HELP-004 | System SHALL include argument descriptions in help output | P0 |
| HELP-005 | System SHALL include aliases in help output | P0 |
| HELP-006 | System SHALL include access level requirement in help output | P1 |
| HELP-007 | System SHALL support explicit help_text override | P0 |
| HELP-008 | System SHALL support "See also:" cross-references | P2 |
| HELP-009 | System SHALL format output for terminal display (wrapping) | P1 |
2.4 API/Interface Design¶
# packages/maid-engine/src/maid_engine/commands/help.py
from __future__ import annotations
import inspect
import textwrap
from dataclasses import dataclass
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .registry import CommandDefinition
@dataclass
class HelpEntry:
"""Parsed help information for a command."""
command_name: str
short_description: str
long_description: str
usage: str
aliases: list[str]
arguments: list[ArgumentHelp]
category: str
access_level: str
see_also: list[str]
examples: list[str]
@dataclass
class ArgumentHelp:
"""Help information for a single argument."""
name: str
arg_type: str
required: bool
default: str | None
description: str
choices: list[str] | None
class HelpGenerator:
"""Generates help text from command definitions and handlers."""
def __init__(self, line_width: int = 78):
self.line_width = line_width
def generate(self, definition: CommandDefinition) -> HelpEntry:
"""Generate help entry from command definition."""
handler = definition.handler
# Extract from docstring
docstring = inspect.getdoc(handler) or ""
short_desc, long_desc, see_also, examples = self._parse_docstring(docstring)
# Use explicit description if provided
if definition.description:
short_desc = definition.description
# Generate usage string
usage = self._generate_usage(definition)
# Extract argument help from decorators
arguments = self._extract_argument_help(handler)
return HelpEntry(
command_name=definition.name,
short_description=short_desc,
long_description=long_desc,
usage=usage,
aliases=definition.aliases,
arguments=arguments,
category=definition.category,
access_level=definition.access_level.name,
see_also=see_also,
examples=examples,
)
def _parse_docstring(self, docstring: str) -> tuple[str, str, list[str], list[str]]:
"""Parse docstring into components.
Returns:
(short_description, long_description, see_also, examples)
"""
if not docstring:
return ("No description available.", "", [], [])
lines = docstring.strip().split("\n")
# First non-empty line is short description
short_desc = lines[0].strip() if lines else ""
# Remaining is long description
long_desc_lines = []
see_also = []
examples = []
current_section = "description"
for line in lines[1:]:
line_stripped = line.strip()
if line_stripped.lower().startswith("see also:"):
current_section = "see_also"
content = line_stripped[9:].strip()
if content:
see_also.extend(c.strip() for c in content.split(","))
elif line_stripped.lower().startswith("example"):
current_section = "examples"
elif current_section == "see_also":
if line_stripped:
see_also.extend(c.strip() for c in line_stripped.split(","))
elif current_section == "examples":
if line_stripped:
examples.append(line_stripped)
else:
long_desc_lines.append(line)
long_desc = "\n".join(long_desc_lines).strip()
return (short_desc, long_desc, see_also, examples)
def _generate_usage(self, definition: CommandDefinition) -> str:
"""Generate usage string from argument specs."""
handler = definition.handler
# Check for pattern
if hasattr(handler, "__argument_pattern__"):
return f"{definition.name} {handler.__argument_pattern__}"
# Check for argument specs
if hasattr(handler, "__argument_specs__"):
specs = handler.__argument_specs__
parts = [definition.name]
for spec in specs:
if spec.required:
parts.append(f"<{spec.name}>")
else:
if spec.default is not None:
parts.append(f"[{spec.name}={spec.default}]")
else:
parts.append(f"[{spec.name}]")
return " ".join(parts)
# Use explicit usage if provided
if definition.usage:
return definition.usage
return definition.name
def _extract_argument_help(self, handler) -> list[ArgumentHelp]:
"""Extract argument help from handler decorators."""
if not hasattr(handler, "__argument_specs__"):
return []
specs = handler.__argument_specs__
return [
ArgumentHelp(
name=spec.name,
arg_type=spec.arg_type.name.lower(),
required=spec.required,
default=str(spec.default) if spec.default is not None else None,
description=spec.description,
choices=[str(c) for c in spec.choices] if spec.choices else None,
)
for spec in specs
]
def format_terminal(self, entry: HelpEntry) -> str:
"""Format help entry for terminal display."""
lines = []
# Header
lines.append(f"{'=' * self.line_width}")
lines.append(f"Help: {entry.command_name.upper()}")
lines.append(f"{'=' * self.line_width}")
lines.append("")
# Short description
lines.append(entry.short_description)
lines.append("")
# Usage
lines.append(f"Usage: {entry.usage}")
lines.append("")
# Aliases
if entry.aliases:
lines.append(f"Aliases: {', '.join(entry.aliases)}")
lines.append("")
# Arguments
if entry.arguments:
lines.append("Arguments:")
for arg in entry.arguments:
req_marker = "*" if arg.required else " "
type_info = f"({arg.arg_type})"
if arg.default:
type_info += f" [default: {arg.default}]"
lines.append(f" {req_marker} {arg.name} {type_info}")
if arg.description:
wrapped = textwrap.wrap(arg.description, width=self.line_width - 6)
for wrap_line in wrapped:
lines.append(f" {wrap_line}")
if arg.choices:
lines.append(f" Choices: {', '.join(arg.choices)}")
lines.append("")
lines.append(" (* = required)")
lines.append("")
# Long description
if entry.long_description:
wrapped = textwrap.wrap(entry.long_description, width=self.line_width)
lines.extend(wrapped)
lines.append("")
# Examples
if entry.examples:
lines.append("Examples:")
for example in entry.examples:
lines.append(f" {example}")
lines.append("")
# See also
if entry.see_also:
lines.append(f"See also: {', '.join(entry.see_also)}")
lines.append("")
# Access level
if entry.access_level != "PLAYER":
lines.append(f"Requires: {entry.access_level} access")
lines.append("")
# Category
lines.append(f"Category: {entry.category}")
return "\n".join(lines)
2.5 Usage Example¶
@command(
"give",
category="items",
aliases=["hand", "transfer"],
)
@arguments(
ArgumentSpec("item", ArgumentType.ENTITY, description="Item from your inventory to give"),
ArgumentSpec("target", ArgumentType.PLAYER, description="Player to receive the item"),
ArgumentSpec("amount", ArgumentType.INTEGER, required=False, default=1,
min_value=1, max_value=100, description="Quantity to give"),
)
async def cmd_give(ctx: CommandContext, args: ParsedArguments) -> bool:
"""Give an item from your inventory to another player.
Transfer ownership of items between players. Both players must be
in the same room. The receiving player must have inventory space.
Example:
give sword Bob
give 5.potion Alice
give gold Bob 100
See also: drop, get, trade
"""
...
# Help output would be:
# ==============================================================================
# Help: GIVE
# ==============================================================================
#
# Give an item from your inventory to another player.
#
# Usage: give <item> <target> [amount=1]
#
# Aliases: hand, transfer
#
# Arguments:
# * item (entity)
# Item from your inventory to give
# * target (player)
# Player to receive the item
# amount (integer) [default: 1]
# Quantity to give
#
# (* = required)
#
# Transfer ownership of items between players. Both players must be
# in the same room. The receiving player must have inventory space.
#
# Examples:
# give sword Bob
# give 5.potion Alice
# give gold Bob 100
#
# See also: drop, get, trade
#
# Category: items
2.6 Acceptance Criteria¶
| ID | Criterion | Verification |
|---|---|---|
| AC-2.1 | First line of docstring becomes short description | Unit test |
| AC-2.2 | Remaining docstring becomes long description | Unit test |
| AC-2.3 | Usage is auto-generated from argument specs | Unit test |
| AC-2.4 | Argument descriptions appear in help output | Unit test |
| AC-2.5 | "See also:" section is extracted and formatted | Unit test |
| AC-2.6 | Examples section is extracted and formatted | Unit test |
| AC-2.7 | Output is properly wrapped for terminal width | Visual verification |
| AC-2.8 | Explicit description overrides docstring | Unit test |
Feature 3: Pre/Post Command Hooks¶
3.1 Feature Overview¶
What it does:
Provides a middleware pipeline for command execution with hooks that run before and after the command handler. Enables cross-cutting concerns like logging, cooldowns, resource validation, and state cleanup.
Why it's needed:
- Current state: All validation must be duplicated in each command handler
- Evennia provides at_pre_cmd() and at_post_cmd() methods
- Enables: logging, cooldowns, combat state checks, transaction rollback, audit trails
3.2 User Stories¶
US-3.1: Global Command Logging
As a server administrator, I want all commands logged for audit purposes so that I can review player actions.
US-3.2: Command Cooldowns
As a game designer, I want to apply cooldowns to commands so that players can't spam abilities.
US-3.3: Combat State Validation
As a game designer, I want commands to automatically check if the player is in combat so that certain commands are blocked during fights.
US-3.4: Resource Cleanup
As a command developer, I want automatic cleanup after command execution so that temporary state is always released.
US-3.5: Command Cancellation
As a game designer, I want pre-hooks to be able to cancel command execution so that conditions can prevent actions.
3.3 Technical Requirements¶
| ID | Requirement | Priority |
|---|---|---|
| HOOK-001 | System SHALL support pre-command hooks that run before handler | P0 |
| HOOK-002 | System SHALL support post-command hooks that run after handler | P0 |
| HOOK-003 | Pre-hooks SHALL be able to cancel command execution | P0 |
| HOOK-004 | Post-hooks SHALL receive command result (success/failure) | P0 |
| HOOK-005 | Hooks SHALL be registrable globally (all commands) | P0 |
| HOOK-006 | Hooks SHALL be registrable per-command | P1 |
| HOOK-007 | Hooks SHALL be registrable per-category | P1 |
| HOOK-008 | Hooks SHALL execute in priority order | P0 |
| HOOK-009 | Post-hooks SHALL run even if command raises exception | P0 |
| HOOK-010 | Hooks SHALL be async-compatible | P0 |
| HOOK-011 | System SHALL provide built-in hooks for common patterns | P1 |
3.4 API/Interface Design¶
# packages/maid-engine/src/maid_engine/commands/hooks.py
from __future__ import annotations
import asyncio
import logging
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum, auto
from typing import TYPE_CHECKING, Any, Callable
if TYPE_CHECKING:
from .registry import CommandContext, CommandDefinition
logger = logging.getLogger(__name__)
class HookPriority(Enum):
"""Priority levels for hook execution order."""
FIRST = 0 # Authentication, rate limiting
HIGH = 25 # Validation
NORMAL = 50 # Default
LOW = 75 # Logging, metrics
LAST = 100 # Cleanup
class HookResult(Enum):
"""Result of a pre-command hook."""
CONTINUE = auto() # Continue to next hook/command
CANCEL = auto() # Cancel command execution (with message)
SKIP = auto() # Skip remaining hooks, execute command
@dataclass
class PreHookContext:
"""Context passed to pre-command hooks."""
command_context: CommandContext
definition: CommandDefinition
metadata: dict[str, Any] = field(default_factory=dict)
def cancel(self, message: str) -> HookResult:
"""Cancel command with message to user."""
self.metadata["cancel_message"] = message
return HookResult.CANCEL
@dataclass
class PostHookContext:
"""Context passed to post-command hooks."""
command_context: CommandContext
definition: CommandDefinition
result: bool # Command handler return value
exception: Exception | None # Exception if handler raised
execution_time_ms: float # How long command took
metadata: dict[str, Any] = field(default_factory=dict)
PreHookFn = Callable[[PreHookContext], HookResult | None]
PostHookFn = Callable[[PostHookContext], None]
@dataclass
class RegisteredHook:
"""A registered hook with metadata."""
name: str
priority: HookPriority
handler: PreHookFn | PostHookFn
commands: set[str] | None = None # None = all commands
categories: set[str] | None = None # None = all categories
def matches(self, definition: CommandDefinition) -> bool:
"""Check if hook applies to this command."""
if self.commands is not None:
if definition.name not in self.commands:
return False
if self.categories is not None:
if definition.category not in self.categories:
return False
return True
class HookRegistry:
"""Registry for command hooks."""
def __init__(self):
self._pre_hooks: list[RegisteredHook] = []
self._post_hooks: list[RegisteredHook] = []
def register_pre_hook(
self,
name: str,
handler: PreHookFn,
priority: HookPriority = HookPriority.NORMAL,
commands: set[str] | None = None,
categories: set[str] | None = None,
) -> None:
"""Register a pre-command hook."""
hook = RegisteredHook(
name=name,
priority=priority,
handler=handler,
commands=commands,
categories=categories,
)
self._pre_hooks.append(hook)
self._pre_hooks.sort(key=lambda h: h.priority.value)
def register_post_hook(
self,
name: str,
handler: PostHookFn,
priority: HookPriority = HookPriority.NORMAL,
commands: set[str] | None = None,
categories: set[str] | None = None,
) -> None:
"""Register a post-command hook."""
hook = RegisteredHook(
name=name,
priority=priority,
handler=handler,
commands=commands,
categories=categories,
)
self._post_hooks.append(hook)
self._post_hooks.sort(key=lambda h: h.priority.value)
def unregister(self, name: str) -> None:
"""Unregister hook by name."""
self._pre_hooks = [h for h in self._pre_hooks if h.name != name]
self._post_hooks = [h for h in self._post_hooks if h.name != name]
def get_pre_hooks(self, definition: CommandDefinition) -> list[RegisteredHook]:
"""Get pre-hooks that apply to a command."""
return [h for h in self._pre_hooks if h.matches(definition)]
def get_post_hooks(self, definition: CommandDefinition) -> list[RegisteredHook]:
"""Get post-hooks that apply to a command."""
return [h for h in self._post_hooks if h.matches(definition)]
class CommandExecutor:
"""Executes commands with hook pipeline."""
def __init__(self, hook_registry: HookRegistry):
self.hooks = hook_registry
async def execute(
self,
context: CommandContext,
definition: CommandDefinition,
) -> bool:
"""Execute command with pre/post hooks."""
start_time = time.perf_counter()
exception: Exception | None = None
result = False
# Create hook context with shared metadata
metadata: dict[str, Any] = {}
# Run pre-hooks
pre_ctx = PreHookContext(
command_context=context,
definition=definition,
metadata=metadata,
)
for hook in self.hooks.get_pre_hooks(definition):
try:
hook_result = await self._call_hook(hook.handler, pre_ctx)
if hook_result == HookResult.CANCEL:
message = metadata.get("cancel_message", "Command cancelled.")
await context.session.send_line(message)
return False
if hook_result == HookResult.SKIP:
break
except Exception as e:
logger.exception(f"Pre-hook '{hook.name}' raised exception")
# Continue with other hooks
# Run command handler
try:
result = await definition.handler(context)
except Exception as e:
exception = e
logger.exception(f"Command '{definition.name}' raised exception")
# Calculate execution time
execution_time_ms = (time.perf_counter() - start_time) * 1000
# Run post-hooks (always, even on exception)
post_ctx = PostHookContext(
command_context=context,
definition=definition,
result=result,
exception=exception,
execution_time_ms=execution_time_ms,
metadata=metadata,
)
for hook in self.hooks.get_post_hooks(definition):
try:
await self._call_hook(hook.handler, post_ctx)
except Exception as e:
logger.exception(f"Post-hook '{hook.name}' raised exception")
# Re-raise command exception if not handled
if exception:
raise exception
return result
async def _call_hook(self, handler: Callable, context: Any) -> Any:
"""Call hook handler, supporting both sync and async."""
if asyncio.iscoroutinefunction(handler):
return await handler(context)
else:
return handler(context)
# ============================================================================
# Built-in Hooks
# ============================================================================
async def logging_pre_hook(ctx: PreHookContext) -> HookResult:
"""Log command execution attempts."""
logger.info(
f"Command: player={ctx.command_context.player_id} "
f"cmd={ctx.definition.name} args={ctx.command_context.args}"
)
return HookResult.CONTINUE
async def logging_post_hook(ctx: PostHookContext) -> None:
"""Log command completion."""
status = "success" if ctx.result else "failed"
if ctx.exception:
status = f"exception: {ctx.exception}"
logger.info(
f"Command completed: cmd={ctx.definition.name} "
f"status={status} time_ms={ctx.execution_time_ms:.2f}"
)
class CooldownHook:
"""Pre-hook that enforces command cooldowns."""
def __init__(self):
# player_id -> command_name -> last_use_timestamp
self._cooldowns: dict[str, dict[str, float]] = {}
async def __call__(self, ctx: PreHookContext) -> HookResult:
player_id = str(ctx.command_context.player_id)
cmd_name = ctx.definition.name
# Check for cooldown metadata on command
cooldown_seconds = ctx.definition.metadata.get("cooldown", 0)
if cooldown_seconds <= 0:
return HookResult.CONTINUE
now = time.time()
player_cooldowns = self._cooldowns.setdefault(player_id, {})
last_use = player_cooldowns.get(cmd_name, 0)
remaining = cooldown_seconds - (now - last_use)
if remaining > 0:
return ctx.cancel(f"You must wait {remaining:.1f} seconds before using {cmd_name} again.")
# Update last use time
player_cooldowns[cmd_name] = now
return HookResult.CONTINUE
class CombatStateHook:
"""Pre-hook that blocks commands during combat."""
def __init__(self, blocked_commands: set[str] | None = None):
self.blocked_commands = blocked_commands or {
"quit", "logout", "recall", "teleport", "home"
}
async def __call__(self, ctx: PreHookContext) -> HookResult:
if ctx.definition.name not in self.blocked_commands:
return HookResult.CONTINUE
# Check if player is in combat
player = ctx.command_context.world.get_entity(ctx.command_context.player_id)
if player:
combat = player.get_component("CombatStateComponent")
if combat and combat.in_combat:
return ctx.cancel("You cannot do that while in combat!")
return HookResult.CONTINUE
class MetricsHook:
"""Post-hook that collects command metrics."""
def __init__(self):
self.command_counts: dict[str, int] = {}
self.command_times: dict[str, list[float]] = {}
self.error_counts: dict[str, int] = {}
async def __call__(self, ctx: PostHookContext) -> None:
cmd = ctx.definition.name
# Count executions
self.command_counts[cmd] = self.command_counts.get(cmd, 0) + 1
# Track execution times
if cmd not in self.command_times:
self.command_times[cmd] = []
self.command_times[cmd].append(ctx.execution_time_ms)
# Keep only last 100 times
if len(self.command_times[cmd]) > 100:
self.command_times[cmd] = self.command_times[cmd][-100:]
# Count errors
if ctx.exception:
self.error_counts[cmd] = self.error_counts.get(cmd, 0) + 1
def get_stats(self, command: str) -> dict:
"""Get statistics for a command."""
times = self.command_times.get(command, [])
return {
"executions": self.command_counts.get(command, 0),
"errors": self.error_counts.get(command, 0),
"avg_time_ms": sum(times) / len(times) if times else 0,
"max_time_ms": max(times) if times else 0,
}
3.5 Integration with Registry¶
# Additions to packages/maid-engine/src/maid_engine/commands/registry.py
class LayeredCommandRegistry:
"""Extended to support hooks."""
def __init__(self):
# ... existing initialization ...
self.hook_registry = HookRegistry()
self.executor = CommandExecutor(self.hook_registry)
async def execute(
self,
context: CommandContext,
player_access_level: AccessLevel = AccessLevel.PLAYER,
) -> bool:
"""Execute command with hooks pipeline."""
definition = self.get(context.command)
if not definition:
return False
if player_access_level < definition.access_level:
raise PermissionError(
f"Command '{definition.name}' requires {definition.access_level.name} access"
)
return await self.executor.execute(context, definition)
def register_pre_hook(self, *args, **kwargs) -> None:
"""Register a pre-command hook."""
self.hook_registry.register_pre_hook(*args, **kwargs)
def register_post_hook(self, *args, **kwargs) -> None:
"""Register a post-command hook."""
self.hook_registry.register_post_hook(*args, **kwargs)
3.6 Usage Example¶
# In content pack on_load():
async def on_load(self, engine: GameEngine) -> None:
registry = engine.command_registry
# Register global logging hooks
registry.register_pre_hook(
"command_logger",
logging_pre_hook,
priority=HookPriority.LAST,
)
registry.register_post_hook(
"command_logger",
logging_post_hook,
priority=HookPriority.LAST,
)
# Register cooldown hook for combat commands
cooldown_hook = CooldownHook()
registry.register_pre_hook(
"combat_cooldowns",
cooldown_hook,
priority=HookPriority.HIGH,
categories={"combat"},
)
# Block certain commands during combat
combat_block = CombatStateHook()
registry.register_pre_hook(
"combat_block",
combat_block,
priority=HookPriority.FIRST,
)
# Collect metrics
self.metrics = MetricsHook()
registry.register_post_hook(
"metrics",
self.metrics,
priority=HookPriority.LAST,
)
3.7 Acceptance Criteria¶
| ID | Criterion | Verification |
|---|---|---|
| AC-3.1 | Pre-hooks execute before command handler | Unit test order |
| AC-3.2 | Post-hooks execute after command handler | Unit test order |
| AC-3.3 | Pre-hook can cancel execution with message | Unit test cancellation |
| AC-3.4 | Post-hooks run even when handler raises exception | Unit test exception flow |
| AC-3.5 | Hooks execute in priority order | Unit test ordering |
| AC-3.6 | Hooks can filter by command name | Unit test filtering |
| AC-3.7 | Hooks can filter by category | Unit test filtering |
| AC-3.8 | Built-in cooldown hook enforces delays | Integration test |
| AC-3.9 | Built-in combat hook blocks commands | Integration test |
| AC-3.10 | Metrics hook collects accurate statistics | Unit test metrics |
Feature 4: Enhanced Lock System¶
4.1 Feature Overview¶
What it does:
Provides a flexible, expression-based permission system that goes beyond simple access levels. Supports complex lock expressions like perm(admin) OR (owns(obj) AND in_room(home)).
Why it's needed:
- Current state: Simple AccessLevel enum (PLAYER, HELPER, BUILDER, ADMIN, IMPLEMENTOR)
- Cannot express "owner of this object" or "member of this guild" or "in combat"
- Evennia provides full lock string expressions with custom lock functions
4.2 User Stories¶
US-4.1: Object Ownership Locks
As a game designer, I want to lock commands to object owners so that only the player who owns a house can use "lock door".
US-4.2: Guild Membership Locks
As a game designer, I want to lock commands to guild members so that guild commands only work for members.
US-4.3: Location-Based Locks
As a game designer, I want to lock commands to specific locations so that "bank" only works in bank rooms.
US-4.4: State-Based Locks
As a game designer, I want to lock commands based on player state so that "rest" only works when not in combat.
US-4.5: Combinable Expressions
As a game designer, I want to combine lock conditions with AND/OR/NOT so that I can express complex requirements.
4.3 Technical Requirements¶
| ID | Requirement | Priority |
|---|---|---|
| LOCK-001 | System SHALL support lock expression strings | P0 |
| LOCK-002 | System SHALL support AND/OR/NOT operators | P0 |
| LOCK-003 | System SHALL support parentheses for grouping | P0 |
| LOCK-004 | System SHALL support pluggable lock functions | P0 |
| LOCK-005 | System SHALL provide built-in lock functions for common cases | P0 |
| LOCK-006 | System SHALL cache parsed lock expressions | P1 |
| LOCK-007 | System SHALL provide helpful error messages for invalid expressions | P0 |
| LOCK-008 | System SHALL integrate with command registration | P0 |
| LOCK-009 | System SHALL support lock inheritance from parent | P2 |
| LOCK-010 | Lock functions SHALL receive full context (player, target, world) | P0 |
4.4 API/Interface Design¶
# packages/maid-engine/src/maid_engine/commands/locks.py
from __future__ import annotations
import re
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Callable
from uuid import UUID
if TYPE_CHECKING:
from maid_engine.core.world import World
from maid_engine.core.ecs.entity import Entity
@dataclass
class LockContext:
"""Context passed to lock functions."""
player: Entity
player_id: UUID
world: World
target: Entity | None = None
command_name: str | None = None
args: dict[str, Any] | None = None
LockFunction = Callable[[LockContext, list[str]], bool]
class LockFunctionRegistry:
"""Registry for lock functions."""
_functions: dict[str, LockFunction] = {}
@classmethod
def register(cls, name: str, func: LockFunction) -> None:
"""Register a lock function."""
cls._functions[name.lower()] = func
@classmethod
def get(cls, name: str) -> LockFunction | None:
"""Get a lock function by name."""
return cls._functions.get(name.lower())
@classmethod
def evaluate(cls, name: str, context: LockContext, args: list[str]) -> bool:
"""Evaluate a lock function."""
func = cls.get(name)
if func is None:
raise ValueError(f"Unknown lock function: {name}")
return func(context, args)
# ============================================================================
# Lock Expression Parser
# ============================================================================
class LockNode(ABC):
"""Base class for lock expression AST nodes."""
@abstractmethod
def evaluate(self, context: LockContext) -> bool:
"""Evaluate this node against context."""
pass
@dataclass
class LockFunctionNode(LockNode):
"""A lock function call like 'perm(admin)'."""
name: str
args: list[str]
def evaluate(self, context: LockContext) -> bool:
return LockFunctionRegistry.evaluate(self.name, context, self.args)
@dataclass
class LockAndNode(LockNode):
"""AND combination of lock nodes."""
left: LockNode
right: LockNode
def evaluate(self, context: LockContext) -> bool:
return self.left.evaluate(context) and self.right.evaluate(context)
@dataclass
class LockOrNode(LockNode):
"""OR combination of lock nodes."""
left: LockNode
right: LockNode
def evaluate(self, context: LockContext) -> bool:
return self.left.evaluate(context) or self.right.evaluate(context)
@dataclass
class LockNotNode(LockNode):
"""NOT negation of lock node."""
child: LockNode
def evaluate(self, context: LockContext) -> bool:
return not self.child.evaluate(context)
@dataclass
class LockTrueNode(LockNode):
"""Always true (no lock)."""
def evaluate(self, context: LockContext) -> bool:
return True
@dataclass
class LockFalseNode(LockNode):
"""Always false (completely locked)."""
def evaluate(self, context: LockContext) -> bool:
return False
class LockExpressionParser:
"""Parser for lock expression strings.
Grammar:
expression := or_expr
or_expr := and_expr ('OR' and_expr)*
and_expr := not_expr ('AND' not_expr)*
not_expr := 'NOT' not_expr | primary
primary := function | '(' expression ')'
function := name '(' args ')'
args := arg (',' arg)*
arg := [^,)]+
"""
# Token patterns
TOKEN_PATTERNS = [
(r'\s+', None), # Whitespace (skip)
(r'\(', 'LPAREN'),
(r'\)', 'RPAREN'),
(r',', 'COMMA'),
(r'AND\b', 'AND'),
(r'OR\b', 'OR'),
(r'NOT\b', 'NOT'),
(r'all\b', 'ALL'), # Special: always true
(r'none\b', 'NONE'), # Special: always false
(r'[a-zA-Z_][a-zA-Z0-9_]*', 'IDENT'),
(r'[^,()]+', 'ARG'), # Argument value
]
def __init__(self):
self._cache: dict[str, LockNode] = {}
def parse(self, expression: str) -> LockNode:
"""Parse lock expression string into AST."""
if not expression or expression.strip() == "":
return LockTrueNode()
# Check cache
if expression in self._cache:
return self._cache[expression]
# Tokenize
tokens = self._tokenize(expression)
# Parse
result, remaining = self._parse_or_expr(tokens)
if remaining:
raise ValueError(f"Unexpected tokens at end: {remaining}")
# Cache result
self._cache[expression] = result
return result
def _tokenize(self, expression: str) -> list[tuple[str, str]]:
"""Tokenize expression string."""
tokens = []
pos = 0
while pos < len(expression):
matched = False
for pattern, token_type in self.TOKEN_PATTERNS:
regex = re.compile(pattern, re.IGNORECASE)
match = regex.match(expression, pos)
if match:
if token_type: # Skip whitespace
tokens.append((token_type, match.group()))
pos = match.end()
matched = True
break
if not matched:
raise ValueError(f"Invalid character at position {pos}: {expression[pos]}")
return tokens
def _parse_or_expr(self, tokens: list) -> tuple[LockNode, list]:
"""Parse OR expression."""
left, tokens = self._parse_and_expr(tokens)
while tokens and tokens[0][0] == 'OR':
tokens = tokens[1:] # consume OR
right, tokens = self._parse_and_expr(tokens)
left = LockOrNode(left, right)
return left, tokens
def _parse_and_expr(self, tokens: list) -> tuple[LockNode, list]:
"""Parse AND expression."""
left, tokens = self._parse_not_expr(tokens)
while tokens and tokens[0][0] == 'AND':
tokens = tokens[1:] # consume AND
right, tokens = self._parse_not_expr(tokens)
left = LockAndNode(left, right)
return left, tokens
def _parse_not_expr(self, tokens: list) -> tuple[LockNode, list]:
"""Parse NOT expression."""
if tokens and tokens[0][0] == 'NOT':
tokens = tokens[1:] # consume NOT
child, tokens = self._parse_not_expr(tokens)
return LockNotNode(child), tokens
return self._parse_primary(tokens)
def _parse_primary(self, tokens: list) -> tuple[LockNode, list]:
"""Parse primary expression (function or parenthesized)."""
if not tokens:
raise ValueError("Unexpected end of expression")
token_type, token_value = tokens[0]
if token_type == 'ALL':
return LockTrueNode(), tokens[1:]
if token_type == 'NONE':
return LockFalseNode(), tokens[1:]
if token_type == 'LPAREN':
tokens = tokens[1:] # consume (
expr, tokens = self._parse_or_expr(tokens)
if not tokens or tokens[0][0] != 'RPAREN':
raise ValueError("Expected closing parenthesis")
return expr, tokens[1:]
if token_type == 'IDENT':
# Function call
func_name = token_value
tokens = tokens[1:]
if not tokens or tokens[0][0] != 'LPAREN':
raise ValueError(f"Expected '(' after function name '{func_name}'")
tokens = tokens[1:] # consume (
# Parse arguments
args = []
while tokens and tokens[0][0] != 'RPAREN':
if tokens[0][0] == 'COMMA':
tokens = tokens[1:] # consume ,
continue
arg_value = tokens[0][1].strip()
args.append(arg_value)
tokens = tokens[1:]
if not tokens or tokens[0][0] != 'RPAREN':
raise ValueError(f"Expected ')' after function arguments")
tokens = tokens[1:] # consume )
return LockFunctionNode(func_name, args), tokens
raise ValueError(f"Unexpected token: {token_type} '{token_value}'")
# ============================================================================
# Lock Evaluator
# ============================================================================
class LockEvaluator:
"""Evaluates lock expressions against context."""
def __init__(self):
self.parser = LockExpressionParser()
def check(
self,
expression: str,
player: Entity,
world: World,
target: Entity | None = None,
command_name: str | None = None,
args: dict | None = None,
) -> bool:
"""Check if player passes lock expression."""
context = LockContext(
player=player,
player_id=player.id,
world=world,
target=target,
command_name=command_name,
args=args,
)
try:
node = self.parser.parse(expression)
return node.evaluate(context)
except Exception as e:
# Log error and deny by default
import logging
logging.getLogger(__name__).error(f"Lock evaluation failed: {e}")
return False
# ============================================================================
# Built-in Lock Functions
# ============================================================================
def lock_all(context: LockContext, args: list[str]) -> bool:
"""Always passes. Usage: all()"""
return True
def lock_none(context: LockContext, args: list[str]) -> bool:
"""Never passes. Usage: none()"""
return False
def lock_perm(context: LockContext, args: list[str]) -> bool:
"""Check access level. Usage: perm(admin), perm(builder)"""
if not args:
return False
required_level = args[0].upper()
from .registry import AccessLevel
try:
required = AccessLevel[required_level]
except KeyError:
return False
# Get player's access level
player_comp = context.player.get_component("PlayerComponent")
if not player_comp:
return False
player_level = getattr(player_comp, "access_level", AccessLevel.PLAYER)
return player_level >= required
def lock_owns(context: LockContext, args: list[str]) -> bool:
"""Check if player owns target object. Usage: owns()"""
if not context.target:
return False
owner_comp = context.target.get_component("OwnershipComponent")
if not owner_comp:
return False
return owner_comp.owner_id == context.player_id
def lock_holds(context: LockContext, args: list[str]) -> bool:
"""Check if player is holding target object. Usage: holds()"""
if not context.target:
return False
inventory = context.player.get_component("InventoryComponent")
if not inventory:
return False
return context.target.id in inventory.items
def lock_in_room(context: LockContext, args: list[str]) -> bool:
"""Check if player is in specific room. Usage: in_room(bank), in_room(guild_hall)"""
if not args:
return False
room_tag = args[0].lower()
# Get player's current room
room_id = context.world.room_index.get_room(context.player_id)
if not room_id:
return False
room = context.world.get_entity(room_id)
if not room:
return False
# Check room tags/flags
if room_tag in room.tags:
return True
room_comp = room.get_component("RoomComponent")
if room_comp and hasattr(room_comp, "flags"):
if room_tag.upper() in room_comp.flags:
return True
return False
def lock_has_flag(context: LockContext, args: list[str]) -> bool:
"""Check if player has a flag/tag. Usage: has_flag(vip), has_flag(beta_tester)"""
if not args:
return False
flag = args[0].lower()
return flag in context.player.tags
def lock_in_guild(context: LockContext, args: list[str]) -> bool:
"""Check guild membership. Usage: in_guild(warriors), in_guild()"""
guild_comp = context.player.get_component("GuildMemberComponent")
if not guild_comp:
return False
if not args:
# Any guild
return True
required_guild = args[0].lower()
return guild_comp.guild_name.lower() == required_guild
def lock_guild_rank(context: LockContext, args: list[str]) -> bool:
"""Check guild rank. Usage: guild_rank(officer), guild_rank(3)"""
if not args:
return False
guild_comp = context.player.get_component("GuildMemberComponent")
if not guild_comp:
return False
required = args[0]
# Check by name or numeric rank
if required.isdigit():
return guild_comp.rank >= int(required)
else:
return guild_comp.rank_name.lower() == required.lower()
def lock_char_level(context: LockContext, args: list[str]) -> bool:
"""Check character (gameplay) level. Usage: char_level(10), char_level(50).
Note: this is the character-tier check. For account-tier admin gates,
use ``level(name)`` / ``account_level(name)`` instead, which reads
the account's AccessLevel rather than a stats component.
"""
if not args:
return False
required_level = int(args[0])
char_comp = context.player.get_component("CharacterComponent")
if not char_comp:
return False
return char_comp.level >= required_level
def lock_in_combat(context: LockContext, args: list[str]) -> bool:
"""Check if player is in combat. Usage: in_combat()"""
combat = context.player.get_component("CombatStateComponent")
return combat is not None and combat.in_combat
def lock_has_item(context: LockContext, args: list[str]) -> bool:
"""Check if player has specific item. Usage: has_item(key_123), has_item(gold, 100)"""
if not args:
return False
item_keyword = args[0]
required_amount = int(args[1]) if len(args) > 1 else 1
inventory = context.player.get_component("InventoryComponent")
if not inventory:
return False
count = 0
for item_id in inventory.items:
item = context.world.get_entity(item_id)
if item:
desc = item.get_component("DescriptionComponent")
if desc and item_keyword.lower() in [k.lower() for k in desc.keywords]:
count += 1
return count >= required_amount
def lock_has_skill(context: LockContext, args: list[str]) -> bool:
"""Check if player has skill at level. Usage: has_skill(lockpicking, 50)"""
if not args:
return False
skill_name = args[0].lower()
required_level = int(args[1]) if len(args) > 1 else 1
skills = context.player.get_component("SkillsComponent")
if not skills:
return False
return skills.get_level(skill_name) >= required_level
# Register built-in lock functions
LockFunctionRegistry.register("all", lock_all)
LockFunctionRegistry.register("none", lock_none)
LockFunctionRegistry.register("perm", lock_perm)
LockFunctionRegistry.register("owns", lock_owns)
LockFunctionRegistry.register("holds", lock_holds)
LockFunctionRegistry.register("in_room", lock_in_room)
LockFunctionRegistry.register("has_flag", lock_has_flag)
LockFunctionRegistry.register("in_guild", lock_in_guild)
LockFunctionRegistry.register("guild_rank", lock_guild_rank)
LockFunctionRegistry.register("level", lock_level)
LockFunctionRegistry.register("in_combat", lock_in_combat)
LockFunctionRegistry.register("has_item", lock_has_item)
LockFunctionRegistry.register("has_skill", lock_has_skill)
4.5 Integration with Command Registration¶
# Additions to packages/maid-engine/src/maid_engine/commands/registry.py
@dataclass
class CommandDefinition:
"""Extended with lock support."""
name: str
handler: CommandHandler
pack_name: str
priority: int = 0
aliases: list[str] = field(default_factory=list)
category: str = "general"
description: str = ""
usage: str = ""
access_level: AccessLevel = AccessLevel.PLAYER
hidden: bool = False
locks: str = "" # NEW: Lock expression string
metadata: dict[str, Any] = field(default_factory=dict)
class LayeredCommandRegistry:
"""Extended with lock evaluation."""
def __init__(self):
# ... existing ...
self.lock_evaluator = LockEvaluator()
async def execute(
self,
context: CommandContext,
player_access_level: AccessLevel = AccessLevel.PLAYER,
) -> bool:
"""Execute with access level and lock evaluation."""
definition = self.get(context.command)
if not definition:
return False
player = context.world.get_entity(context.player_id)
# Check access level (backward compatibility)
if player_access_level < definition.access_level:
raise PermissionError(
f"Command '{definition.name}' requires {definition.access_level.name} access"
)
# Check lock expression
if definition.locks:
if not self.lock_evaluator.check(
definition.locks,
player,
context.world,
command_name=definition.name,
):
raise PermissionError(
f"You don't have permission to use '{definition.name}'"
)
return await self.executor.execute(context, definition)
4.6 Usage Examples¶
# Command requiring admin OR owner
@command(
"delete",
category="admin",
locks="perm(admin) OR owns()",
)
async def cmd_delete(ctx: CommandContext) -> bool:
"""Delete an object you own or have admin access to."""
...
# Guild-only command with rank requirement
@command(
"guild_promote",
category="guild",
locks="in_guild() AND guild_rank(officer)",
)
async def cmd_guild_promote(ctx: CommandContext) -> bool:
"""Promote a guild member (requires officer rank)."""
...
# Location-based command
@command(
"deposit",
category="economy",
locks="in_room(bank)",
)
async def cmd_deposit(ctx: CommandContext) -> bool:
"""Deposit gold in the bank (must be at bank)."""
...
# Complex expression
@command(
"enter_dungeon",
category="adventure",
locks="char_level(20) AND NOT in_combat() AND (has_item(dungeon_key) OR in_guild(adventurers))",
)
async def cmd_enter_dungeon(ctx: CommandContext) -> bool:
"""Enter the dungeon (requires key or guild membership, character level 20+, not in combat)."""
...
# Register custom lock function
def lock_is_vampire(context: LockContext, args: list[str]) -> bool:
"""Check if player is a vampire."""
char = context.player.get_component("CharacterComponent")
return char and char.race.lower() == "vampire"
LockFunctionRegistry.register("is_vampire", lock_is_vampire)
# Use custom lock
@command(
"blood_drain",
category="vampire",
locks="is_vampire()",
)
async def cmd_blood_drain(ctx: CommandContext) -> bool:
"""Drain blood from a victim (vampires only)."""
...
4.7 Acceptance Criteria¶
| ID | Criterion | Verification |
|---|---|---|
| AC-4.1 | Lock expressions parse correctly | Unit test parser |
| AC-4.2 | AND/OR/NOT operators work correctly | Unit test logic |
| AC-4.3 | Parentheses affect operator precedence | Unit test grouping |
| AC-4.4 | Built-in lock functions evaluate correctly | Unit test each function |
| AC-4.5 | Custom lock functions can be registered | Unit test registration |
| AC-4.6 | Invalid expressions produce clear errors | Unit test error messages |
| AC-4.7 | Lock expressions integrate with command execution | Integration test |
| AC-4.8 | Backward compatibility with access_level preserved | Regression test |
| AC-4.9 | Lock evaluation failure denies by default | Unit test |
| AC-4.10 | Complex real-world expressions work correctly | Integration test |
Appendix A: Migration Guide¶
Migrating Existing Commands¶
Before (current approach):
async def cmd_give(ctx: CommandContext) -> bool:
if len(ctx.args) < 2:
await ctx.session.send("Usage: give <item> <player>")
return False
item_name = ctx.args[0]
player_name = ctx.args[1]
# Manual entity resolution...
# Manual player lookup...
After (with new features):
@arguments(
ArgumentSpec("item", ArgumentType.ENTITY),
ArgumentSpec("target", ArgumentType.PLAYER),
)
async def cmd_give(ctx: CommandContext, args: ParsedArguments) -> bool:
"""Give an item to another player."""
item = args["item"].entity
target = args["target"]
# Item and player are already resolved
Backward Compatibility¶
All new features are additive:
- Commands without @arguments decorator continue to receive raw ctx.args
- Commands without locks attribute use only access_level
- Commands without docstrings use explicit description
- Hooks are optional and don't affect commands that don't register them
Appendix B: Testing Requirements¶
Unit Test Coverage¶
| Component | Minimum Coverage | Key Test Cases |
|---|---|---|
| ArgumentParser | 90% | All types, edge cases, errors |
| EntityReference | 90% | Index syntax, all. syntax, special keywords |
| HelpGenerator | 85% | Docstring parsing, formatting |
| HookRegistry | 90% | Registration, ordering, filtering |
| CommandExecutor | 90% | Hook pipeline, exception handling |
| LockExpressionParser | 95% | All operators, nesting, errors |
| Lock Functions | 90% | Each built-in function |
Integration Test Scenarios¶
- Full Command Flow: Register command with arguments, hooks, and locks; execute with various inputs
- Entity Resolution: Create entities in room and inventory; verify correct resolution
- Hook Cancellation: Pre-hook cancels command; verify handler not called
- Lock Combinations: Complex lock expressions with multiple conditions
- Help Generation: Commands with full docstrings and argument specs
Performance Benchmarks¶
| Operation | Target | Measurement |
|---|---|---|
| Argument parsing (5 args) | < 1ms | pytest-benchmark |
| Lock expression parsing | < 0.5ms | pytest-benchmark |
| Lock evaluation (complex) | < 0.1ms | pytest-benchmark |
| Hook pipeline (5 hooks) | < 2ms | pytest-benchmark |
| Help generation | < 5ms | pytest-benchmark |
Document History¶
| Version | Date | Author | Changes |
|---|---|---|---|
| 1.0 | 2026-01-30 | MAID Team | Initial specification |