MAID Admin & Developer Tools Design Specification¶
Version: 1.0.0
Date: January 2025
Status: Draft
Author: MAID Development Team
Table of Contents¶
- Executive Summary
- Web Admin Interface
- In-Game Building Commands
- Hot Reload System
- Profiling Tools
- MaidEditor (In-Game Text Editor)
- MaidMenu (Dynamic Menu System)
- MaidTable (Formatted Table Display)
- Batch Command/Code Processors
- Internationalization (i18n) Support
- Implementation Roadmap
- Appendices
1. Executive Summary¶
This document specifies the design and implementation of admin and developer tools required to bring MAID to feature parity with Evennia. These tools are essential for:
- Game Administrators: Managing servers, players, and content through intuitive interfaces
- World Builders: Creating and modifying game content in real-time
- Developers: Debugging, profiling, and iterating on code efficiently
- Content Creators: Writing and editing in-game text content
- International Teams: Supporting multiple languages for global player bases
Scope¶
| Feature | Priority | Estimated Effort | Package Location |
|---|---|---|---|
| Web Admin Interface | P0 | 4-6 weeks | maid-engine |
| In-Game Building Commands | P0 | 3-4 weeks | maid-stdlib |
| Hot Reload System | P1 | 2-3 weeks | maid-engine |
| Profiling Tools | P1 | 2 weeks | maid-engine |
| MaidEditor | P2 | 2 weeks | maid-stdlib |
| MaidMenu | P2 | 2 weeks | maid-stdlib |
| MaidTable | P2 | 1 week | maid-stdlib |
| Batch Processors | P2 | 2 weeks | maid-engine |
| i18n Support | P1 | 3-4 weeks | maid-engine |
2. Web Admin Interface¶
2.1 Feature Overview¶
The Web Admin Interface provides a comprehensive browser-based administration panel for managing MAID servers. Unlike Evennia's Django Admin which is database-centric, MAID's admin will be API-first with a modern React-based frontend, leveraging the existing FastAPI infrastructure.
Why it's needed: - Enables remote server administration without in-game access - Provides visual tools for complex operations (world editing, player management) - Reduces barrier to entry for non-technical administrators - Enables real-time monitoring and alerting
2.2 User Stories¶
US-2.1: Server Dashboard
As a server administrator, I want to view real-time server statistics on a dashboard so that I can monitor server health and player activity at a glance.
US-2.2: Player Management
As a server administrator, I want to search, view, edit, ban, and unban player accounts so that I can manage the player community effectively.
US-2.3: Entity Browser
As a world builder, I want to browse all entities in the world with their components so that I can understand and debug the game state.
US-2.4: World Editor
As a world builder, I want to create and edit rooms, exits, and items through a visual interface so that I can build game content without using in-game commands.
US-2.5: Configuration Editor
As a server administrator, I want to modify server configuration through the web interface so that I can tune server settings without restarting.
US-2.6: Log Viewer
As a server administrator, I want to view and search server logs in real-time so that I can diagnose issues and monitor activity.
US-2.7: Content Pack Management
As a server administrator, I want to view, enable, disable, and configure content packs so that I can customize the game without code changes.
2.3 Technical Requirements¶
2.3.1 Backend Architecture¶
packages/maid-engine/src/maid_engine/
├── api/
│ ├── admin/
│ │ ├── __init__.py
│ │ ├── router.py # Main admin API router
│ │ ├── auth.py # Admin authentication/authorization
│ │ ├── dashboard.py # Dashboard endpoints
│ │ ├── entities.py # Entity/Component CRUD
│ │ ├── players.py # Player management
│ │ ├── world.py # World editing endpoints
│ │ ├── config.py # Configuration management
│ │ ├── logs.py # Log streaming endpoints
│ │ └── packs.py # Content pack management
│ └── websocket/
│ └── admin_ws.py # WebSocket for real-time updates
2.3.2 Frontend Architecture¶
packages/maid-engine/src/maid_engine/
├── static/
│ └── admin/
│ ├── index.html
│ └── dist/ # Built React app
├── admin_frontend/ # React source (separate build)
├── src/
│ ├── components/
│ │ ├── Dashboard/
│ │ ├── EntityBrowser/
│ │ ├── PlayerManager/
│ │ ├── WorldEditor/
│ │ ├── ConfigEditor/
│ │ └── LogViewer/
│ ├── hooks/
│ ├── services/
│ └── App.tsx
└── package.json
2.3.3 Authentication & Authorization¶
# packages/maid-engine/src/maid_engine/api/admin/auth.py
from enum import IntEnum
from typing import Annotated
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
from pydantic import BaseModel
class AdminRole(IntEnum):
"""Admin permission levels.
Values are spaced by 10 to allow future roles to be inserted
between existing ones without breaking ordering or existing tokens.
"""
VIEWER = 10 # Read-only access
MODERATOR = 20 # Player management
BUILDER = 30 # World editing
ADMIN = 40 # Full configuration
SUPERADMIN = 50 # All permissions
class AdminUser(BaseModel):
"""Authenticated admin user."""
user_id: str
username: str
role: AdminRole
permissions: set[str]
class AdminAuthConfig(BaseModel):
"""Admin authentication configuration."""
secret_key: str
algorithm: str = "HS256"
token_expiry_hours: int = 24
require_2fa: bool = False # Placeholder - 2FA not yet implemented (see Phase 4)
security = HTTPBearer()
async def get_current_admin(
credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)],
config: AdminAuthConfig = Depends(get_admin_config),
) -> AdminUser:
"""Validate admin JWT token and return user."""
try:
payload = jwt.decode(
credentials.credentials,
config.secret_key,
algorithms=[config.algorithm]
)
return AdminUser(**payload)
except jwt.InvalidTokenError as e:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=f"Invalid authentication token: {e}"
)
def require_role(min_role: AdminRole):
"""Dependency to require minimum admin role."""
async def role_checker(admin: AdminUser = Depends(get_current_admin)):
if admin.role < min_role:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"Requires {min_role.name} role or higher"
)
return admin
return role_checker
2.4 API/Interface Design¶
2.4.1 Dashboard API¶
# packages/maid-engine/src/maid_engine/api/admin/dashboard.py
from datetime import datetime, timedelta
from fastapi import APIRouter, Depends, WebSocket
from pydantic import BaseModel
router = APIRouter(prefix="/admin/dashboard", tags=["admin-dashboard"])
class ServerMetrics(BaseModel):
"""Real-time server metrics."""
uptime_seconds: float
tick_rate: float
current_tick: int
memory_usage_mb: float
cpu_percent: float
class PlayerMetrics(BaseModel):
"""Player activity metrics."""
online_count: int
peak_today: int
total_accounts: int
new_today: int
active_sessions: int
class WorldMetrics(BaseModel):
"""World state metrics."""
entity_count: int
component_count: int
system_count: int
room_count: int
npc_count: int
item_count: int
class DashboardData(BaseModel):
"""Complete dashboard data."""
server: ServerMetrics
players: PlayerMetrics
world: WorldMetrics
content_packs: list[str]
recent_events: list[dict]
timestamp: datetime
@router.get("/", response_model=DashboardData)
async def get_dashboard(
admin: AdminUser = Depends(require_role(AdminRole.VIEWER))
) -> DashboardData:
"""Get complete dashboard data."""
...
@router.websocket("/ws")
async def dashboard_websocket(
websocket: WebSocket,
admin: AdminUser = Depends(require_role(AdminRole.VIEWER))
):
"""WebSocket for real-time dashboard updates."""
await websocket.accept()
try:
while True:
data = await get_realtime_metrics()
await websocket.send_json(data.model_dump())
await asyncio.sleep(1) # 1 second update interval
except WebSocketDisconnect:
pass
class TimeSeriesQuery(BaseModel):
"""Query parameters for time series data."""
metric: str
start: datetime
end: datetime
interval: timedelta = timedelta(minutes=5)
@router.post("/metrics/history")
async def get_metrics_history(
query: TimeSeriesQuery,
admin: AdminUser = Depends(require_role(AdminRole.VIEWER))
) -> list[dict]:
"""Get historical metrics for charting."""
...
2.4.2 Entity Browser API¶
# packages/maid-engine/src/maid_engine/api/admin/entities.py
from uuid import UUID
from fastapi import APIRouter, Depends, Query
from pydantic import BaseModel
router = APIRouter(prefix="/admin/entities", tags=["admin-entities"])
class ComponentData(BaseModel):
"""Serialized component data."""
type: str
data: dict
class EntityData(BaseModel):
"""Complete entity with all components."""
id: UUID
components: list[ComponentData]
tags: list[str]
created_at: datetime
updated_at: datetime
class EntityListResponse(BaseModel):
"""Paginated entity list."""
items: list[EntityData]
total: int
page: int
page_size: int
has_more: bool
class EntityFilter(BaseModel):
"""Entity filter criteria."""
component_types: list[str] | None = None
tags: list[str] | None = None
search: str | None = None
room_id: UUID | None = None
@router.get("/", response_model=EntityListResponse)
async def list_entities(
page: int = Query(1, ge=1),
page_size: int = Query(50, ge=1, le=100),
component_types: list[str] | None = Query(None),
tags: list[str] | None = Query(None),
search: str | None = Query(None),
admin: AdminUser = Depends(require_role(AdminRole.VIEWER))
) -> EntityListResponse:
"""List entities with filtering and pagination."""
...
@router.get("/{entity_id}", response_model=EntityData)
async def get_entity(
entity_id: UUID,
admin: AdminUser = Depends(require_role(AdminRole.VIEWER))
) -> EntityData:
"""Get single entity with all components."""
...
@router.post("/", response_model=EntityData)
async def create_entity(
components: list[ComponentData],
admin: AdminUser = Depends(require_role(AdminRole.BUILDER))
) -> EntityData:
"""Create new entity with components."""
...
@router.put("/{entity_id}/components/{component_type}")
async def update_component(
entity_id: UUID,
component_type: str,
data: dict,
admin: AdminUser = Depends(require_role(AdminRole.BUILDER))
) -> ComponentData:
"""Update a specific component on an entity."""
...
@router.delete("/{entity_id}")
async def delete_entity(
entity_id: UUID,
admin: AdminUser = Depends(require_role(AdminRole.ADMIN))
) -> dict:
"""Delete an entity."""
...
@router.post("/{entity_id}/components")
async def add_component(
entity_id: UUID,
component: ComponentData,
admin: AdminUser = Depends(require_role(AdminRole.BUILDER))
) -> EntityData:
"""Add a component to an existing entity."""
...
@router.delete("/{entity_id}/components/{component_type}")
async def remove_component(
entity_id: UUID,
component_type: str,
admin: AdminUser = Depends(require_role(AdminRole.BUILDER))
) -> EntityData:
"""Remove a component from an entity."""
...
2.4.3 Player Management API¶
# packages/maid-engine/src/maid_engine/api/admin/players.py
from uuid import UUID
from enum import Enum
from fastapi import APIRouter, Depends, Query
from pydantic import BaseModel, EmailStr
router = APIRouter(prefix="/admin/players", tags=["admin-players"])
class AccountStatus(str, Enum):
ACTIVE = "active"
BANNED = "banned"
SUSPENDED = "suspended"
PENDING = "pending"
class PlayerAccount(BaseModel):
"""Player account information."""
id: UUID
username: str
email: EmailStr | None
status: AccountStatus
created_at: datetime
last_login: datetime | None
total_playtime_hours: float
characters: list[UUID]
access_level: str
notes: str | None
class CharacterSummary(BaseModel):
"""Character summary for player view."""
id: UUID
name: str
level: int
race: str
character_class: str
last_played: datetime | None
current_room: str | None
class BanRequest(BaseModel):
"""Request to ban a player."""
player_id: UUID
reason: str
duration_hours: int | None = None # None = permanent
ban_ip: bool = False
class PlayerSearchResult(BaseModel):
"""Search result for players."""
items: list[PlayerAccount]
total: int
page: int
page_size: int
@router.get("/", response_model=PlayerSearchResult)
async def search_players(
search: str | None = Query(None),
status: AccountStatus | None = Query(None),
page: int = Query(1, ge=1),
page_size: int = Query(50, ge=1, le=100),
admin: AdminUser = Depends(require_role(AdminRole.MODERATOR))
) -> PlayerSearchResult:
"""Search and list player accounts."""
...
@router.get("/{player_id}", response_model=PlayerAccount)
async def get_player(
player_id: UUID,
admin: AdminUser = Depends(require_role(AdminRole.MODERATOR))
) -> PlayerAccount:
"""Get player account details."""
...
@router.get("/{player_id}/characters", response_model=list[CharacterSummary])
async def get_player_characters(
player_id: UUID,
admin: AdminUser = Depends(require_role(AdminRole.MODERATOR))
) -> list[CharacterSummary]:
"""Get all characters for a player."""
...
@router.post("/{player_id}/ban")
async def ban_player(
player_id: UUID,
request: BanRequest,
admin: AdminUser = Depends(require_role(AdminRole.MODERATOR))
) -> dict:
"""Ban a player account."""
...
@router.post("/{player_id}/unban")
async def unban_player(
player_id: UUID,
admin: AdminUser = Depends(require_role(AdminRole.MODERATOR))
) -> dict:
"""Unban a player account."""
...
@router.post("/{player_id}/kick")
async def kick_player(
player_id: UUID,
reason: str = "Kicked by administrator",
admin: AdminUser = Depends(require_role(AdminRole.MODERATOR))
) -> dict:
"""Disconnect a player's active session."""
...
@router.put("/{player_id}/access-level")
async def set_access_level(
player_id: UUID,
access_level: str,
admin: AdminUser = Depends(require_role(AdminRole.ADMIN))
) -> PlayerAccount:
"""Set player's access level."""
...
@router.post("/{player_id}/message")
async def message_player(
player_id: UUID,
message: str,
admin: AdminUser = Depends(require_role(AdminRole.MODERATOR))
) -> dict:
"""Send a message to an online player."""
...
2.4.4 World Editor API¶
# packages/maid-engine/src/maid_engine/api/admin/world.py
from uuid import UUID
from fastapi import APIRouter, Depends
from pydantic import BaseModel
router = APIRouter(prefix="/admin/world", tags=["admin-world"])
class RoomData(BaseModel):
"""Room definition."""
id: UUID | None = None
name: str
description: str
area: str | None = None
coordinates: tuple[int, int, int] | None = None
grid_size: tuple[int, int] = (1, 1)
terrain: str = "indoor"
flags: list[str] = []
class ExitData(BaseModel):
"""Exit definition."""
id: UUID | None = None
direction: str
source_room_id: UUID
destination_room_id: UUID
description: str | None = None
keywords: list[str] = []
is_door: bool = False
is_locked: bool = False
key_item_id: UUID | None = None
class AreaData(BaseModel):
"""Area/zone definition."""
name: str
description: str
level_range: tuple[int, int] | None = None
rooms: list[UUID] = []
class WorldGraphNode(BaseModel):
"""Node in world visualization graph."""
id: UUID
name: str
area: str | None
x: float # Layout position
y: float
class WorldGraphEdge(BaseModel):
"""Edge in world visualization graph."""
source: UUID
target: UUID
direction: str
bidirectional: bool
class WorldGraph(BaseModel):
"""World graph for visualization."""
nodes: list[WorldGraphNode]
edges: list[WorldGraphEdge]
@router.get("/rooms", response_model=list[RoomData])
async def list_rooms(
area: str | None = None,
admin: AdminUser = Depends(require_role(AdminRole.BUILDER))
) -> list[RoomData]:
"""List all rooms, optionally filtered by area."""
...
@router.post("/rooms", response_model=RoomData)
async def create_room(
room: RoomData,
admin: AdminUser = Depends(require_role(AdminRole.BUILDER))
) -> RoomData:
"""Create a new room."""
...
@router.put("/rooms/{room_id}", response_model=RoomData)
async def update_room(
room_id: UUID,
room: RoomData,
admin: AdminUser = Depends(require_role(AdminRole.BUILDER))
) -> RoomData:
"""Update an existing room."""
...
@router.delete("/rooms/{room_id}")
async def delete_room(
room_id: UUID,
admin: AdminUser = Depends(require_role(AdminRole.ADMIN))
) -> dict:
"""Delete a room and its exits."""
...
@router.post("/exits", response_model=ExitData)
async def create_exit(
exit_data: ExitData,
bidirectional: bool = True,
admin: AdminUser = Depends(require_role(AdminRole.BUILDER))
) -> ExitData:
"""Create an exit (optionally bidirectional)."""
...
@router.delete("/exits/{exit_id}")
async def delete_exit(
exit_id: UUID,
admin: AdminUser = Depends(require_role(AdminRole.BUILDER))
) -> dict:
"""Delete an exit."""
...
@router.get("/graph", response_model=WorldGraph)
async def get_world_graph(
area: str | None = None,
admin: AdminUser = Depends(require_role(AdminRole.BUILDER))
) -> WorldGraph:
"""Get world as graph for visualization."""
...
@router.get("/areas", response_model=list[AreaData])
async def list_areas(
admin: AdminUser = Depends(require_role(AdminRole.BUILDER))
) -> list[AreaData]:
"""List all areas/zones."""
...
2.4.5 Log Viewer API¶
# packages/maid-engine/src/maid_engine/api/admin/logs.py
from datetime import datetime
from enum import Enum
from fastapi import APIRouter, Depends, Query, WebSocket
from pydantic import BaseModel
router = APIRouter(prefix="/admin/logs", tags=["admin-logs"])
class LogLevel(str, Enum):
DEBUG = "DEBUG"
INFO = "INFO"
WARNING = "WARNING"
ERROR = "ERROR"
CRITICAL = "CRITICAL"
class LogEntry(BaseModel):
"""Single log entry."""
timestamp: datetime
level: LogLevel
logger: str
message: str
extra: dict | None = None
class LogSearchResult(BaseModel):
"""Log search results."""
entries: list[LogEntry]
total: int
has_more: bool
class LogFilter(BaseModel):
"""Log filter criteria."""
levels: list[LogLevel] | None = None
loggers: list[str] | None = None
search: str | None = None
start_time: datetime | None = None
end_time: datetime | None = None
@router.get("/", response_model=LogSearchResult)
async def search_logs(
levels: list[LogLevel] | None = Query(None),
loggers: list[str] | None = Query(None),
search: str | None = Query(None),
start_time: datetime | None = Query(None),
end_time: datetime | None = Query(None),
limit: int = Query(100, ge=1, le=1000),
offset: int = Query(0, ge=0),
admin: AdminUser = Depends(require_role(AdminRole.VIEWER))
) -> LogSearchResult:
"""Search historical logs."""
...
@router.get("/loggers", response_model=list[str])
async def list_loggers(
admin: AdminUser = Depends(require_role(AdminRole.VIEWER))
) -> list[str]:
"""List all available logger names."""
...
@router.websocket("/stream")
async def stream_logs(
websocket: WebSocket,
levels: list[LogLevel] | None = Query(None),
loggers: list[str] | None = Query(None),
admin: AdminUser = Depends(require_role(AdminRole.VIEWER))
):
"""WebSocket for real-time log streaming."""
await websocket.accept()
filter_config = LogFilter(levels=levels, loggers=loggers)
async for entry in log_stream(filter_config):
await websocket.send_json(entry.model_dump())
@router.post("/export")
async def export_logs(
filter: LogFilter,
format: str = Query("json", regex="^(json|csv)$"),
admin: AdminUser = Depends(require_role(AdminRole.ADMIN))
) -> StreamingResponse:
"""Export logs to file."""
...
2.4.6 Configuration Editor API¶
# packages/maid-engine/src/maid_engine/api/admin/config.py
from fastapi import APIRouter, Depends
from pydantic import BaseModel
router = APIRouter(prefix="/admin/config", tags=["admin-config"])
class ConfigSection(BaseModel):
"""Configuration section."""
name: str
description: str
settings: dict
schema_json: dict # JSON Schema for validation
class ConfigUpdate(BaseModel):
"""Configuration update request."""
section: str
key: str
value: Any
restart_required: bool = False
class ConfigValidationResult(BaseModel):
"""Result of config validation."""
valid: bool
errors: list[str]
warnings: list[str]
@router.get("/", response_model=list[ConfigSection])
async def get_all_config(
admin: AdminUser = Depends(require_role(AdminRole.ADMIN))
) -> list[ConfigSection]:
"""Get all configuration sections."""
...
@router.get("/{section}", response_model=ConfigSection)
async def get_config_section(
section: str,
admin: AdminUser = Depends(require_role(AdminRole.ADMIN))
) -> ConfigSection:
"""Get specific configuration section."""
...
@router.post("/validate", response_model=ConfigValidationResult)
async def validate_config(
updates: list[ConfigUpdate],
admin: AdminUser = Depends(require_role(AdminRole.ADMIN))
) -> ConfigValidationResult:
"""Validate configuration changes without applying."""
...
@router.put("/{section}/{key}")
async def update_config(
section: str,
key: str,
value: Any,
admin: AdminUser = Depends(require_role(AdminRole.SUPERADMIN))
) -> ConfigSection:
"""Update a configuration value."""
...
@router.post("/reload")
async def reload_config(
admin: AdminUser = Depends(require_role(AdminRole.SUPERADMIN))
) -> dict:
"""Reload configuration from files."""
...
2.5 UI Mockups¶
2.5.1 Dashboard¶
┌─────────────────────────────────────────────────────────────────────────────┐
│ MAID Admin Dashboard [Admin: superuser] │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ PLAYERS │ │ UPTIME │ │ TICK RATE │ │ MEMORY │ │
│ │ 42 │ │ 3d 14h 22m │ │ 4.0 TPS │ │ 1.2 GB │ │
│ │ ▲ +5 today │ │ │ │ ✓ Stable │ │ ▼ 85% peak │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │
│ ┌────────────────────────────────────┐ ┌────────────────────────────────┐│
│ │ Player Activity (24h) │ │ Recent Events ││
│ │ │ │ ││
│ │ 60│ ╭──╮ │ │ • Player "Gandalf" logged in ││
│ │ │ ╭╯ ╰╮ ╭─ │ │ • Zone "Mines" reset ││
│ │ 40│──╯ ╰─╯ │ │ • Player "Frodo" leveled up ││
│ │ │ │ │ • NPC spawner activated ││
│ │ 20│ │ │ • Server auto-saved ││
│ │ └──────────────────────── │ │ • Content pack reloaded ││
│ │ 00:00 06:00 12:00 18:00 │ │ ││
│ └────────────────────────────────────┘ └────────────────────────────────┘│
│ │
│ ┌────────────────────────────────────────────────────────────────────────┐│
│ │ Content Packs ││
│ │ ┌──────────────────┬─────────┬────────────┬─────────────────────────┐ ││
│ │ │ Name │ Version │ Status │ Dependencies │ ││
│ │ ├──────────────────┼─────────┼────────────┼─────────────────────────┤ ││
│ │ │ maid-stdlib │ 1.0.0 │ ✓ Active │ maid-engine │ ││
│ │ │ maid-classic-rpg │ 1.0.0 │ ✓ Active │ maid-stdlib │ ││
│ │ │ custom-world │ 0.5.0 │ ✓ Active │ maid-classic-rpg │ ││
│ │ └──────────────────┴─────────┴────────────┴─────────────────────────┘ ││
│ └────────────────────────────────────────────────────────────────────────┘│
└─────────────────────────────────────────────────────────────────────────────┘
2.5.2 Entity Browser¶
┌─────────────────────────────────────────────────────────────────────────────┐
│ Entity Browser [Filter] [+ New] │
├─────────────────────────────────────────────────────────────────────────────┤
│ Components: [PositionComponent ▼] [PlayerComponent ▼] Search: [________] │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─ Entity List ────────────────────────┐ ┌─ Entity Details ─────────────┐│
│ │ │ │ ││
│ │ ▸ 8f3a2b1c - "Gandalf the Grey" │ │ ID: 8f3a2b1c-4d5e-6f7a... ││
│ │ PlayerComponent, PositionComponent │ │ Created: 2025-01-15 10:30 ││
│ │ │ │ ││
│ │ ▸ 9d4c3b2a - "Town Square" │ │ Components: ││
│ │ RoomComponent, DescriptionComponent│ │ ┌────────────────────────┐ ││
│ │ │ │ │ ▾ PlayerComponent │ ││
│ │ ▸ 1e5f4a3b - "Iron Sword" │ │ │ account_id: "abc123" │ ││
│ │ ItemComponent, EquipmentComponent │ │ │ access_level: PLAYER │ ││
│ │ │ │ │ session_id: "xyz789" │ ││
│ │ ▸ 2a6b5c4d - "Goblin Guard" │ │ │ [Edit] [Remove] │ ││
│ │ NPCComponent, CombatComponent │ │ └────────────────────────┘ ││
│ │ │ │ ┌────────────────────────┐ ││
│ │ ▸ 3b7c6d5e - "Health Potion" │ │ │ ▾ PositionComponent │ ││
│ │ ItemComponent, ConsumableComponent │ │ │ room_id: "9d4c3b2a" │ ││
│ │ │ │ │ x: 5, y: 3 │ ││
│ │ │ │ │ [Edit] [Remove] │ ││
│ │ Page 1 of 50 [<] [1] [2] [3] [>] │ │ └────────────────────────┘ ││
│ └──────────────────────────────────────┘ │ ││
│ │ [+ Add Component] [Delete] ││
│ └──────────────────────────────┘│
└─────────────────────────────────────────────────────────────────────────────┘
2.5.3 World Editor¶
┌─────────────────────────────────────────────────────────────────────────────┐
│ World Editor [Save] [Undo] [Redo] │
├─────────────────────────────────────────────────────────────────────────────┤
│ Area: [Starting Village ▼] Zoom: [100%] │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─ Room Graph ─────────────────────────────────┐ ┌─ Room Properties ────┐│
│ │ │ │ ││
│ │ ┌─────────┐ │ │ Name: ││
│ │ │ Forest │ │ │ [Town Square ] ││
│ │ │ Path │ │ │ ││
│ │ └────┬────┘ │ │ Description: ││
│ │ │ n │ │ ┌──────────────────┐ ││
│ │ ┌───────┐ │ ┌───────┐ │ │ │ A bustling town │ ││
│ │ │ Shop │───┼───│ Inn │ │ │ │ square with a │ ││
│ │ └───────┘ w │ e └───────┘ │ │ │ fountain in the │ ││
│ │ │ │ │ │ center... │ ││
│ │ ┌────┴────┐ │ │ └──────────────────┘ ││
│ │ │ Town │ ← Selected │ │ ││
│ │ │ Square │ │ │ Exits: ││
│ │ └────┬────┘ │ │ • north → Forest Path││
│ │ │ s │ │ • east → Inn ││
│ │ ┌────┴────┐ │ │ • west → Shop ││
│ │ │ Temple │ │ │ • south → Temple ││
│ │ └─────────┘ │ │ [+ Add Exit] ││
│ │ │ │ ││
│ │ Tools: [Select] [Connect] [New Room] │ │ Flags: [indoor ▼] ││
│ └──────────────────────────────────────────────┘ │ [Apply Changes] ││
│ └──────────────────────┘│
└─────────────────────────────────────────────────────────────────────────────┘
2.6 Configuration¶
# packages/maid-engine/src/maid_engine/config/settings.py (additions)
class AdminSettings(BaseSettings):
"""Web admin configuration."""
model_config = SettingsConfigDict(env_prefix="MAID_ADMIN_")
enabled: bool = True
host: str = "0.0.0.0"
port: int = 8081
# Authentication
secret_key: SecretStr = SecretStr("") # Must be set in production
token_expiry_hours: int = 24
require_2fa: bool = False # Placeholder - 2FA not yet implemented (see Phase 4)
# Security
allowed_origins: list[str] = ["http://localhost:8081"]
rate_limit_requests: int = 100
rate_limit_window_seconds: int = 60
# Features
enable_world_editor: bool = True
enable_config_editor: bool = True
enable_log_viewer: bool = True
max_log_retention_days: int = 30
# Dashboard
metrics_retention_hours: int = 168 # 1 week
dashboard_refresh_interval_seconds: int = 5
Environment Variables:
MAID_ADMIN_ENABLED=true
MAID_ADMIN_PORT=8081
MAID_ADMIN_SECRET_KEY=your-secret-key-here
MAID_ADMIN_TOKEN_EXPIRY_HOURS=24
MAID_ADMIN_REQUIRE_2FA=false # Placeholder - 2FA not yet implemented
MAID_ADMIN_ALLOWED_ORIGINS=["https://admin.example.com"]
MAID_ADMIN_RATE_LIMIT_REQUESTS=100
MAID_ADMIN_ENABLE_WORLD_EDITOR=true
MAID_ADMIN_ENABLE_CONFIG_EDITOR=true
2.7 Dependencies¶
Backend:
# packages/maid-engine/pyproject.toml additions
[project.optional-dependencies]
admin = [
"fastapi>=0.109.0",
"uvicorn[standard]>=0.27.0",
"python-jose[cryptography]>=3.3.0", # JWT handling
"passlib[bcrypt]>=1.7.4", # Password hashing
"python-multipart>=0.0.6", # Form handling
"aiofiles>=23.0.0", # Async file operations
"structlog>=24.0.0", # Structured logging for log viewer
]
Frontend:
// packages/maid-engine/admin_frontend/package.json
{
"dependencies": {
"react": "^18.2.0",
"react-dom": "^18.2.0",
"react-router-dom": "^6.21.0",
"@tanstack/react-query": "^5.17.0",
"recharts": "^2.10.0",
"reactflow": "^11.11.0",
"@monaco-editor/react": "^4.6.0",
"axios": "^1.6.0",
"zustand": "^4.4.0"
},
"devDependencies": {
"typescript": "^5.3.0",
"vite": "^5.0.0",
"@vitejs/plugin-react": "^4.2.0",
"tailwindcss": "^3.4.0"
}
}
2.8 Implementation Tasks¶
- [ ] Phase 1: Backend Foundation (Week 1-2)
- [ ] Create admin API router structure
- [ ] Implement JWT authentication system
- [ ] Add role-based authorization middleware
- [ ] Create dashboard metrics collection
- [ ] Implement entity CRUD endpoints
-
[ ] Add WebSocket infrastructure for real-time updates
-
[ ] Phase 2: Core Features (Week 2-3)
- [ ] Implement player management endpoints
- [ ] Create world editor API (rooms, exits)
- [ ] Add log streaming and search
- [ ] Implement configuration management API
-
[ ] Add content pack management endpoints
-
[ ] Phase 3: Frontend Development (Week 3-5)
- [ ] Set up React project with Vite
- [ ] Create authentication flow
- [ ] Build dashboard with real-time metrics
- [ ] Implement entity browser with filtering
- [ ] Create player management interface
- [ ] Build world editor with graph visualization
- [ ] Add log viewer with streaming
-
[ ] Create configuration editor
-
[ ] Phase 4: Polish & Security (Week 5-6)
- [ ] Add rate limiting
- [ ] Implement audit logging
- [ ] Add 2FA support (TOTP-based, requires additional dependencies: pyotp, qrcode)
- Status: Planned - Not Yet Implemented
- Rationale: Deferred to reduce initial complexity. The
require_2fasetting is a placeholder that currently has no effect. Implementation will require: - TOTP secret generation and storage per admin user
- QR code generation for authenticator app setup
- Recovery codes for account recovery
- UI flow for 2FA enrollment and verification
- [ ] Create admin user management
- [ ] Add export/import functionality
- [ ] Performance optimization
- [ ] Documentation
2.9 Testing Requirements¶
# packages/maid-engine/tests/api/admin/test_dashboard.py
import pytest
from httpx import AsyncClient
from maid_engine.api.admin.auth import AdminRole
@pytest.fixture
async def admin_client(test_app, admin_token):
"""Create authenticated admin client."""
async with AsyncClient(app=test_app, base_url="http://test") as client:
client.headers["Authorization"] = f"Bearer {admin_token}"
yield client
@pytest.mark.asyncio
async def test_dashboard_requires_auth(test_app):
"""Test dashboard endpoint requires authentication."""
async with AsyncClient(app=test_app, base_url="http://test") as client:
response = await client.get("/admin/dashboard/")
assert response.status_code == 401
@pytest.mark.asyncio
async def test_dashboard_returns_metrics(admin_client):
"""Test dashboard returns all required metrics."""
response = await admin_client.get("/admin/dashboard/")
assert response.status_code == 200
data = response.json()
assert "server" in data
assert "players" in data
assert "world" in data
assert data["server"]["uptime_seconds"] >= 0
@pytest.mark.asyncio
async def test_entity_crud(admin_client):
"""Test entity create, read, update, delete."""
# Create
create_response = await admin_client.post("/admin/entities/", json={
"components": [{"type": "DescriptionComponent", "data": {"name": "Test"}}]
})
assert create_response.status_code == 200
entity_id = create_response.json()["id"]
# Read
get_response = await admin_client.get(f"/admin/entities/{entity_id}")
assert get_response.status_code == 200
# Update
update_response = await admin_client.put(
f"/admin/entities/{entity_id}/components/DescriptionComponent",
json={"name": "Updated"}
)
assert update_response.status_code == 200
# Delete
delete_response = await admin_client.delete(f"/admin/entities/{entity_id}")
assert delete_response.status_code == 200
@pytest.mark.asyncio
async def test_role_authorization(test_app, viewer_token):
"""Test that viewers cannot access admin endpoints."""
async with AsyncClient(app=test_app, base_url="http://test") as client:
client.headers["Authorization"] = f"Bearer {viewer_token}"
response = await client.post("/admin/players/123/ban", json={
"player_id": "123",
"reason": "Test"
})
assert response.status_code == 403
2.10 Acceptance Criteria¶
- Dashboard
- [ ] Shows real-time server metrics (uptime, tick rate, memory, CPU)
- [ ] Displays current player count with trend indicators
- [ ] Lists loaded content packs with status
- [ ] Shows recent server events
-
[ ] Updates automatically via WebSocket
-
Entity Browser
- [ ] Lists all entities with pagination (50 per page)
- [ ] Filters by component type and tags
- [ ] Full-text search on entity data
- [ ] View/edit individual components
-
[ ] Create and delete entities
-
Player Management
- [ ] Search players by username/email
- [ ] View player details and characters
- [ ] Ban/unban with reason tracking
- [ ] Kick online players
-
[ ] Set access levels
-
World Editor
- [ ] Visual graph of room connections
- [ ] Create/edit/delete rooms
- [ ] Create/edit exits with bidirectional option
- [ ] Edit room descriptions inline
-
[ ] Undo/redo support
-
Log Viewer
- [ ] Real-time log streaming
- [ ] Filter by level and logger
- [ ] Full-text search
-
[ ] Export to JSON/CSV
-
Configuration Editor
- [ ] View all configuration sections
- [ ] Edit configuration with validation
- [ ] Shows which changes require restart
- [ ] Audit log of changes
3. In-Game Building Commands¶
3.1 Feature Overview¶
In-game building commands allow world builders to create and modify game content directly from within the game client. This is essential for rapid iteration and allows builders to test their creations immediately.
Why it's needed: - Enables real-time world building without external tools - Allows builders to see changes immediately in context - Supports collaborative building with other online builders - Provides familiar interface for experienced MUD builders
3.2 User Stories¶
US-3.1: Room Creation
As a builder, I want to create new rooms from within the game so that I can expand the world without using external tools.
US-3.2: Object Creation
As a builder, I want to create items, NPCs, and other objects in my current location so that I can populate the world quickly.
US-3.3: Exit Management
As a builder, I want to dig passages between rooms so that I can connect areas of the world.
US-3.4: Attribute Editing
As a builder, I want to set and modify attributes on any object so that I can customize game entities.
US-3.5: Object Inspection
As a builder, I want to examine the full details of any object, including all components and hidden attributes, so that I can debug and understand the game state.
US-3.6: Batch Operations
As a builder, I want to modify multiple objects at once so that I can make sweeping changes efficiently.
3.3 Technical Requirements¶
3.3.1 Command Structure¶
packages/maid-stdlib/src/maid_stdlib/
├── commands/
│ ├── building/
│ │ ├── __init__.py
│ │ ├── create.py # @create, @spawn
│ │ ├── destroy.py # @destroy, @purge
│ │ ├── dig.py # @dig, @tunnel
│ │ ├── describe.py # @describe, @name
│ │ ├── examine.py # @examine, @stat
│ │ ├── link.py # @link, @unlink
│ │ ├── set.py # @set, @attribute
│ │ ├── teleport.py # @teleport, @goto
│ │ ├── find.py # @find, @search
│ │ ├── copy.py # @copy, @clone
│ │ ├── lock.py # @lock, @unlock
│ │ └── zone.py # @zone, @area
3.3.2 Command Definitions¶
# packages/maid-stdlib/src/maid_stdlib/commands/building/create.py
from dataclasses import dataclass
from uuid import UUID
from maid_engine.commands.registry import (
CommandContext,
CommandHandler,
AccessLevel
)
@dataclass
class CreateOptions:
"""Options for entity creation."""
name: str
template: str | None = None
location: UUID | None = None # None = current room
components: dict | None = None
async def cmd_create(ctx: CommandContext) -> bool:
"""Create a new entity.
Usage:
@create <type> <name> [= template]
@create item "Iron Sword" = weapons/sword
@create npc "Town Guard" = guards/basic
@create room "Dark Cave"
Types: item, npc, room, exit, container, vehicle
"""
if len(ctx.args) < 2:
await ctx.session.send("Usage: @create <type> <name> [= template]")
return False
entity_type = ctx.args[0].lower()
# Parse name and optional template
rest = " ".join(ctx.args[1:])
if "=" in rest:
name, template = rest.split("=", 1)
name = name.strip().strip('"')
template = template.strip()
else:
name = rest.strip().strip('"')
template = None
# Get current room for location
player = ctx.world.entities.get(ctx.player_id)
position = player.get_component(PositionComponent)
current_room = position.room_id if position else None
# Create based on type
factory = get_entity_factory(entity_type)
entity = await factory.create(
ctx.world,
name=name,
template=template,
location=current_room if entity_type != "room" else None
)
await ctx.session.send(
f"Created {entity_type} '{name}' with ID {entity.id}"
)
return True
async def cmd_spawn(ctx: CommandContext) -> bool:
"""Spawn an entity from a template.
Usage:
@spawn <template> [count] [at <location>]
@spawn monsters/goblin 5
@spawn items/gold_coin 100 at treasury
"""
if not ctx.args:
await ctx.session.send("Usage: @spawn <template> [count] [at <location>]")
return False
template = ctx.args[0]
count = 1
location = None
# Parse optional count and location
args = ctx.args[1:]
if args and args[0].isdigit():
count = min(int(args[0]), 100) # Cap at 100
args = args[1:]
if len(args) >= 2 and args[0].lower() == "at":
location = await resolve_location(ctx, " ".join(args[1:]))
# Spawn entities
spawned = []
for _ in range(count):
entity = await spawn_from_template(ctx.world, template, location)
spawned.append(entity)
await ctx.session.send(f"Spawned {len(spawned)} entities from '{template}'")
return True
3.4 Complete Command Reference¶
| Command | Aliases | Access | Description |
|---|---|---|---|
@create |
@new |
BUILDER | Create a new entity |
@spawn |
@summon |
BUILDER | Spawn from template |
@destroy |
@delete, @del |
BUILDER | Delete an entity |
@purge |
ADMIN | Delete multiple entities by filter | |
@dig |
BUILDER | Create room and exit in direction | |
@tunnel |
BUILDER | Create bidirectional passage | |
@describe |
@desc |
BUILDER | Set entity description |
@name |
@rename |
BUILDER | Set entity name |
@examine |
@exam |
BUILDER | Detailed entity inspection |
@stat |
@stats |
BUILDER | Show entity statistics |
@set |
BUILDER | Set entity attribute | |
@attribute |
@attr |
BUILDER | Manage entity attributes |
@link |
BUILDER | Link exit to destination | |
@unlink |
BUILDER | Remove exit link | |
@teleport |
@tel, @tp |
BUILDER | Move entity to location |
@goto |
@go |
BUILDER | Move self to location |
@find |
BUILDER | Search for entities | |
@search |
BUILDER | Search with complex filters | |
@copy |
@clone |
BUILDER | Duplicate an entity |
@lock |
BUILDER | Set lock on entity | |
@unlock |
BUILDER | Remove lock from entity | |
@zone |
@area |
BUILDER | Manage zones/areas |
@flag |
BUILDER | Set/unset entity flags | |
@component |
@comp |
BUILDER | Add/remove/edit components |
@script |
ADMIN | Attach script to entity | |
@trigger |
BUILDER | Manage entity triggers | |
@reset |
BUILDER | Configure entity reset behavior | |
@wipe |
ADMIN | Clear all non-persistent entities |
3.5 Detailed Command Specifications¶
@dig - Create Room and Exit¶
async def cmd_dig(ctx: CommandContext) -> bool:
"""Dig a new room in a direction.
Usage:
@dig <direction> [= room_name]
@dig north = "Forest Clearing"
@dig down = "Underground Cavern"
@dig portal = "Mystic Realm" # custom exit name
Creates:
1. A new room (with optional name)
2. An exit from current room to new room
3. A return exit from new room back (unless --oneway)
Flags:
--oneway Don't create return exit
--door Create as door (closeable)
--locked Create as locked door (requires key)
--hidden Create as hidden exit
"""
...
@set - Set Attributes¶
async def cmd_set(ctx: CommandContext) -> bool:
"""Set an attribute on an entity.
Usage:
@set <target>/<attribute> = <value>
@set here/description = "A dark and musty cellar."
@set sword/damage = 10
@set guard/patrol_route = ["square", "gate", "tower"]
@set #1234/health.max = 100
Targets:
here - Current room
me - Your character
<name> - Entity by name in room
#<id> - Entity by ID
Value Types:
string - "quoted text" or bare words
number - 42, 3.14
boolean - true, false
list - [1, 2, 3]
dict - {key: value}
null - null, none
Special paths:
component.field - Nested component access
component[index] - List index access
"""
...
@examine - Detailed Inspection¶
async def cmd_examine(ctx: CommandContext) -> bool:
"""Examine an entity in detail.
Usage:
@examine <target> [component]
@examine sword
@examine here
@examine guard CombatComponent
@examine #1234
Output includes:
- Entity ID and type
- All components with full data
- Entity flags and tags
- Related entities (contents, location, etc.)
- Debug information (creation time, last modified)
Flags:
--json Output as JSON
--raw Include internal fields
--history Show modification history
"""
...
@find - Search Entities¶
async def cmd_find(ctx: CommandContext) -> bool:
"""Find entities by criteria.
Usage:
@find <type> [filter]
@find item name:sword
@find npc zone:village
@find room flag:dark
@find * component:CombatComponent
Filters:
name:<pattern> - Match name (supports wildcards)
zone:<name> - In specific zone
flag:<flag> - Has specific flag
component:<type> - Has specific component
owner:<id> - Owned by entity
range:<distance> - Within distance of you
level:<min>-<max> - Level range (for NPCs/items)
created:<date> - Created after date
modified:<date> - Modified after date
Output:
List of matching entities with location and brief info
Flags:
--count Only show count
--verbose Show full details
--limit=N Limit results (default 50)
"""
...
3.6 API/Interface Design¶
# packages/maid-stdlib/src/maid_stdlib/commands/building/__init__.py
from maid_engine.commands.registry import (
CommandRegistry,
AccessLevel,
CommandDefinition
)
def register_building_commands(registry: CommandRegistry) -> None:
"""Register all building commands."""
# Creation commands
registry.register(
name="@create",
handler=cmd_create,
aliases=["@new"],
category="building",
description="Create a new entity",
usage="@create <type> <name> [= template]",
access_level=AccessLevel.BUILDER,
)
registry.register(
name="@spawn",
handler=cmd_spawn,
aliases=["@summon"],
category="building",
description="Spawn entity from template",
usage="@spawn <template> [count] [at <location>]",
access_level=AccessLevel.BUILDER,
)
registry.register(
name="@destroy",
handler=cmd_destroy,
aliases=["@delete", "@del"],
category="building",
description="Delete an entity",
usage="@destroy <target>",
access_level=AccessLevel.BUILDER,
)
# ... additional registrations ...
# Entity factory interface
class EntityFactory(Protocol):
"""Factory for creating game entities."""
async def create(
self,
world: World,
*,
name: str,
template: str | None = None,
location: UUID | None = None,
components: dict | None = None,
) -> Entity:
"""Create a new entity."""
...
def get_templates(self) -> list[str]:
"""List available templates."""
...
def validate_template(self, template: str) -> bool:
"""Check if template exists."""
...
3.7 Configuration¶
class BuildingSettings(BaseSettings):
"""Building command configuration."""
model_config = SettingsConfigDict(env_prefix="MAID_BUILDING_")
# Limits
max_spawn_count: int = 100
max_find_results: int = 500
max_description_length: int = 10000
# Defaults
default_room_template: str = "rooms/empty"
default_npc_template: str = "npcs/basic"
default_item_template: str = "items/basic"
# Permissions
allow_global_teleport: bool = True
allow_cross_zone_dig: bool = False
require_zone_ownership: bool = True
# Undo
enable_undo: bool = True
undo_history_limit: int = 50
Environment Variables:
MAID_BUILDING_MAX_SPAWN_COUNT=100
MAID_BUILDING_MAX_FIND_RESULTS=500
MAID_BUILDING_DEFAULT_ROOM_TEMPLATE=rooms/empty
MAID_BUILDING_ALLOW_GLOBAL_TELEPORT=true
MAID_BUILDING_ENABLE_UNDO=true
3.8 Dependencies¶
3.9 Implementation Tasks¶
- [ ] Phase 1: Core Commands (Week 1)
- [ ] Implement @create with entity factories
- [ ] Implement @destroy with safety checks
- [ ] Implement @examine with component introspection
- [ ] Implement @set with type coercion
-
[ ] Implement @dig with room creation
-
[ ] Phase 2: Navigation & Search (Week 2)
- [ ] Implement @teleport and @goto
- [ ] Implement @find with filters
- [ ] Implement @search with complex queries
-
[ ] Add target resolution (here, me, #id, name)
-
[ ] Phase 3: Advanced Features (Week 3)
- [ ] Implement @copy/@clone
- [ ] Implement @link/@unlink
- [ ] Implement @component management
- [ ] Implement @zone commands
-
[ ] Add undo/redo system
-
[ ] Phase 4: Templates & Scripting (Week 4)
- [ ] Implement template system
- [ ] Add @spawn with templates
- [ ] Implement @script attachment
- [ ] Add @trigger management
- [ ] Documentation and help text
3.10 Testing Requirements¶
@pytest.mark.asyncio
async def test_create_room(building_context):
"""Test @create room command."""
ctx = building_context
ctx.args = ["room", "Test Chamber"]
result = await cmd_create(ctx)
assert result is True
# Verify room was created
rooms = ctx.world.entities.with_component(RoomComponent)
new_room = next(r for r in rooms if r.name == "Test Chamber")
assert new_room is not None
@pytest.mark.asyncio
async def test_dig_creates_bidirectional_exits(building_context):
"""Test @dig creates exit and return exit."""
ctx = building_context
ctx.args = ["north", "=", "Northern Hall"]
# Get initial room
player = ctx.world.entities.get(ctx.player_id)
start_room = player.get_component(PositionComponent).room_id
result = await cmd_dig(ctx)
assert result is True
# Verify exits created
exits = ctx.world.get_exits(start_room)
assert any(e.direction == "north" for e in exits)
# Verify return exit
new_room = next(e.destination for e in exits if e.direction == "north")
return_exits = ctx.world.get_exits(new_room)
assert any(e.direction == "south" and e.destination == start_room for e in return_exits)
@pytest.mark.asyncio
async def test_set_modifies_component(building_context):
"""Test @set modifies entity attributes."""
ctx = building_context
# Create test entity
entity = await create_test_item(ctx.world, "Test Sword")
ctx.args = [f"#{entity.id}/damage", "=", "25"]
result = await cmd_set(ctx)
assert result is True
updated = ctx.world.entities.get(entity.id)
assert updated.get_component(WeaponComponent).damage == 25
@pytest.mark.asyncio
async def test_find_filters_correctly(building_context):
"""Test @find with various filters."""
ctx = building_context
# Create test entities
await create_test_item(ctx.world, "Iron Sword", flags=["weapon"])
await create_test_item(ctx.world, "Steel Sword", flags=["weapon"])
await create_test_item(ctx.world, "Health Potion", flags=["consumable"])
ctx.args = ["item", "name:*Sword"]
result = await cmd_find(ctx)
assert result is True
# Verify output contains only swords
output = ctx.session.get_output()
assert "Iron Sword" in output
assert "Steel Sword" in output
assert "Health Potion" not in output
3.11 Acceptance Criteria¶
- Entity Creation
- [ ] @create works for all entity types (room, item, npc, exit)
- [ ] Templates are loaded and applied correctly
-
[ ] Created entities appear in correct location
-
Entity Modification
- [ ] @set modifies any component attribute
- [ ] @describe updates descriptions
- [ ] @name changes entity names
-
[ ] Changes persist to database
-
Navigation
- [ ] @dig creates room and bidirectional exits
- [ ] @teleport moves entities between rooms
- [ ] @goto moves builder character
-
[ ] @link/@unlink manage exit connections
-
Search & Inspection
- [ ] @examine shows all component data
- [ ] @find searches with wildcards
- [ ] @find filters by component, zone, flag
-
[ ] Results are paginated for large sets
-
Safety
- [ ] Commands respect access levels
- [ ] @destroy requires confirmation for important entities
- [ ] Undo/redo works for all modifications
- [ ] Cross-zone operations respect permissions
4. Hot Reload System¶
4.1 Feature Overview¶
Hot reload enables developers to modify Python code, content packs, and game data while the server is running, without disconnecting players or losing game state. This dramatically accelerates the development cycle.
Why it's needed: - Eliminates server restart cycle during development - Allows live bug fixes on production servers - Enables real-time content updates - Supports A/B testing of game mechanics
4.2 User Stories¶
US-4.1: Code Reload
As a developer, I want to modify Python system code and see changes immediately so that I can iterate quickly on game mechanics.
US-4.2: Content Pack Reload
As a content creator, I want to reload content pack data without restarting the server so that I can test changes instantly.
US-4.3: Template Reload
As a builder, I want to update entity templates and have existing entities optionally updated so that I can refine content.
US-4.4: Safe Rollback
As a server admin, I want the ability to rollback a failed reload so that I can recover from broken code.
4.3 Technical Requirements¶
4.3.1 Architecture¶
packages/maid-engine/src/maid_engine/
├── reload/
│ ├── __init__.py
│ ├── manager.py # ReloadManager orchestrates reloads
│ ├── module_reloader.py # Python module hot reload
│ ├── pack_reloader.py # Content pack reload
│ ├── system_reloader.py # System hot swap
│ ├── watcher.py # File system watcher
│ └── rollback.py # Rollback mechanism
4.3.2 Reload Manager¶
# packages/maid-engine/src/maid_engine/reload/manager.py
from dataclasses import dataclass
from enum import Enum
from typing import Callable, Awaitable
import asyncio
class ReloadScope(Enum):
"""Scope of reload operation."""
MODULE = "module" # Single Python module
PACKAGE = "package" # Entire package
CONTENT_PACK = "content_pack" # Content pack data
SYSTEM = "system" # Specific system
TEMPLATES = "templates" # Entity templates
ALL = "all" # Full reload
@dataclass
class ReloadResult:
"""Result of a reload operation."""
success: bool
scope: ReloadScope
target: str
duration_ms: float
changes: list[str]
errors: list[str]
warnings: list[str]
rollback_available: bool
class ReloadManager:
"""Manages hot reload operations for the game engine."""
def __init__(self, engine: GameEngine) -> None:
self._engine = engine
self._lock = asyncio.Lock()
self._snapshots: dict[str, Any] = {}
self._hooks: dict[str, list[Callable]] = {
"pre_reload": [],
"post_reload": [],
"on_error": [],
}
self._watcher: FileWatcher | None = None
async def reload_module(
self,
module_path: str,
*,
cascade: bool = True,
) -> ReloadResult:
"""Hot reload a Python module.
Args:
module_path: Dotted path to module (e.g., "maid_stdlib.systems.combat")
cascade: Also reload dependent modules
"""
async with self._lock:
start = time.perf_counter()
errors = []
changes = []
try:
# Create snapshot for rollback
snapshot_id = await self._create_snapshot(module_path)
# Fire pre-reload hooks
await self._fire_hooks("pre_reload", module_path)
# Reload the module
reloader = ModuleReloader(self._engine)
changed = await reloader.reload(module_path, cascade=cascade)
changes.extend(changed)
# Re-register systems if needed
if any("systems" in c for c in changes):
await self._reregister_systems(changes)
# Fire post-reload hooks
await self._fire_hooks("post_reload", module_path)
return ReloadResult(
success=True,
scope=ReloadScope.MODULE,
target=module_path,
duration_ms=(time.perf_counter() - start) * 1000,
changes=changes,
errors=[],
warnings=[],
rollback_available=True,
)
except Exception as e:
errors.append(str(e))
await self._fire_hooks("on_error", module_path, e)
return ReloadResult(
success=False,
scope=ReloadScope.MODULE,
target=module_path,
duration_ms=(time.perf_counter() - start) * 1000,
changes=changes,
errors=errors,
warnings=[],
rollback_available=snapshot_id in self._snapshots,
)
async def reload_content_pack(
self,
pack_name: str,
*,
include_templates: bool = True,
) -> ReloadResult:
"""Reload a content pack's data without unloading."""
async with self._lock:
pack = self._engine.get_content_pack(pack_name)
if not pack:
return ReloadResult(
success=False,
scope=ReloadScope.CONTENT_PACK,
target=pack_name,
duration_ms=0,
changes=[],
errors=[f"Content pack '{pack_name}' not found"],
warnings=[],
rollback_available=False,
)
reloader = ContentPackReloader(self._engine)
return await reloader.reload(pack, include_templates=include_templates)
async def reload_system(
self,
system_name: str,
) -> ReloadResult:
"""Hot-swap a specific system."""
async with self._lock:
reloader = SystemReloader(self._engine)
return await reloader.reload(system_name)
async def rollback(self, snapshot_id: str) -> bool:
"""Rollback to a previous snapshot."""
if snapshot_id not in self._snapshots:
return False
snapshot = self._snapshots[snapshot_id]
# Restore module state
# Re-register systems
# Restore content pack state
return True
def enable_auto_reload(
self,
paths: list[str],
*,
debounce_ms: int = 500,
) -> None:
"""Enable automatic reload on file changes."""
self._watcher = FileWatcher(
paths=paths,
callback=self._on_file_change,
debounce_ms=debounce_ms,
)
self._watcher.start()
def disable_auto_reload(self) -> None:
"""Disable automatic reload."""
if self._watcher:
self._watcher.stop()
self._watcher = None
4.3.3 Module Reloader¶
# packages/maid-engine/src/maid_engine/reload/module_reloader.py
import importlib
import sys
from types import ModuleType
class ModuleReloader:
"""Handles Python module hot reloading."""
def __init__(self, engine: GameEngine) -> None:
self._engine = engine
self._dependency_graph: dict[str, set[str]] = {}
async def reload(
self,
module_path: str,
*,
cascade: bool = True,
) -> list[str]:
"""Reload a module and optionally its dependents.
Returns list of reloaded module paths.
"""
changed = []
# Build dependency graph if needed
if not self._dependency_graph:
self._build_dependency_graph()
# Get modules to reload
modules_to_reload = [module_path]
if cascade:
modules_to_reload.extend(self._get_dependents(module_path))
# Sort by dependency order (leaves first)
modules_to_reload = self._topological_sort(modules_to_reload)
for mod_path in modules_to_reload:
if mod_path in sys.modules:
module = sys.modules[mod_path]
# Preserve references to classes/functions
old_refs = self._capture_references(module)
# Reload the module
importlib.reload(module)
# Update references
self._update_references(old_refs, module)
changed.append(mod_path)
return changed
def _capture_references(self, module: ModuleType) -> dict[str, Any]:
"""Capture references to module's classes and functions."""
refs = {}
for name in dir(module):
obj = getattr(module, name)
if isinstance(obj, type) or callable(obj):
refs[name] = obj
return refs
def _update_references(
self,
old_refs: dict[str, Any],
new_module: ModuleType,
) -> None:
"""Update existing references to point to new implementations."""
for name, old_obj in old_refs.items():
if hasattr(new_module, name):
new_obj = getattr(new_module, name)
# Update class instances
if isinstance(old_obj, type):
self._update_class_instances(old_obj, new_obj)
# Update registered handlers
self._update_handlers(old_obj, new_obj)
4.4 API/Interface Design¶
# CLI Commands
@app.command()
def reload(
target: str = typer.Argument(..., help="Module path or content pack name"),
scope: str = typer.Option("auto", help="Scope: module, pack, system, all"),
cascade: bool = typer.Option(True, help="Reload dependent modules"),
watch: bool = typer.Option(False, help="Watch for changes and auto-reload"),
):
"""Hot reload code or content.
Examples:
maid dev reload maid_stdlib.systems.combat
maid dev reload classic-rpg --scope=pack
maid dev reload --watch packages/
"""
...
# In-game commands
async def cmd_reload(ctx: CommandContext) -> bool:
"""@reload - Hot reload code or content.
Usage:
@reload system CombatSystem
@reload pack classic-rpg
@reload module maid_stdlib.systems.combat
@reload templates items/*
@reload all
"""
...
async def cmd_rollback(ctx: CommandContext) -> bool:
"""@rollback - Rollback to previous state.
Usage:
@rollback # Rollback last reload
@rollback <snapshot> # Rollback to specific snapshot
@rollback list # List available snapshots
"""
...
4.5 Configuration¶
class ReloadSettings(BaseSettings):
"""Hot reload configuration."""
model_config = SettingsConfigDict(env_prefix="MAID_RELOAD_")
# Enable/disable
enabled: bool = True
auto_reload: bool = False # Auto-reload on file change
# Watched paths (for auto-reload)
watch_paths: list[str] = ["packages/"]
watch_extensions: list[str] = [".py", ".yaml", ".json"]
debounce_ms: int = 500
# Safety
max_snapshots: int = 10
snapshot_retention_minutes: int = 60
require_confirmation: bool = False
# Restrictions
allow_production_reload: bool = False
protected_modules: list[str] = [
"maid_engine.core.engine",
"maid_engine.core.world",
]
4.6 Dependencies¶
4.7 Implementation Tasks¶
- [ ] Phase 1: Module Reloading (Week 1)
- [ ] Implement ModuleReloader with reference preservation
- [ ] Build module dependency graph
- [ ] Implement cascade reloading
-
[ ] Add snapshot/rollback mechanism
-
[ ] Phase 2: System Reloading (Week 1-2)
- [ ] Implement SystemReloader
- [ ] Handle system state preservation
- [ ] Update system registration
-
[ ] Test with running game
-
[ ] Phase 3: Content Pack Reloading (Week 2)
- [ ] Implement ContentPackReloader
- [ ] Handle template updates
- [ ] Update command registrations
-
[ ] Preserve entity state
-
[ ] Phase 4: Auto-Reload & CLI (Week 3)
- [ ] Implement FileWatcher
- [ ] Add CLI reload commands
- [ ] Add @reload in-game command
- [ ] Integration testing
4.8 Testing Requirements¶
@pytest.mark.asyncio
async def test_module_reload_preserves_instances(reload_manager, test_world):
"""Test that existing instances use new code after reload."""
# Create entity with old system behavior
entity = await test_world.create_entity()
old_result = await entity.process()
# Modify system code (simulated)
# ...
# Reload
result = await reload_manager.reload_module("test_system")
assert result.success
# Entity should now use new behavior
new_result = await entity.process()
assert new_result != old_result
@pytest.mark.asyncio
async def test_rollback_restores_state(reload_manager):
"""Test rollback restores previous state."""
# Initial reload
result1 = await reload_manager.reload_module("test_module")
snapshot_id = result1.changes[0]
# Introduce breaking change
# ...
# Failed reload
result2 = await reload_manager.reload_module("test_module")
assert not result2.success
# Rollback
rollback_success = await reload_manager.rollback(snapshot_id)
assert rollback_success
4.9 Acceptance Criteria¶
- Module Reloading
- [ ] Python modules reload without server restart
- [ ] Existing object instances use new code
- [ ] Dependent modules cascade reload
-
[ ] Errors don't crash server
-
System Reloading
- [ ] Systems hot-swap during runtime
- [ ] System state is preserved
-
[ ] Event handlers update correctly
-
Content Pack Reloading
- [ ] Content packs reload data files
- [ ] Commands re-register correctly
-
[ ] Templates update entities
-
Safety
- [ ] Rollback works for failed reloads
- [ ] Protected modules cannot be reloaded
- [ ] Production reload requires flag
5. Profiling Tools¶
5.1 Feature Overview¶
Profiling tools help developers identify performance bottlenecks in memory usage, database queries, tick processing time, and network I/O. These tools are essential for maintaining server performance as content scales.
Why it's needed: - Identifies memory leaks and excessive allocations - Finds slow database queries - Measures tick processing time - Tracks network latency - Enables data-driven optimization
5.2 User Stories¶
US-5.1: Memory Profiling
As a developer, I want to see memory usage by system and component type so that I can identify memory leaks and optimize allocations.
US-5.2: Query Profiling
As a developer, I want to see all database queries with timing so that I can optimize slow queries and reduce database load.
US-5.3: Tick Profiling
As a developer, I want to see how long each system takes per tick so that I can identify performance bottlenecks.
US-5.4: Profiling Reports
As a developer, I want to generate profiling reports that can be analyzed offline so that I can track performance over time.
5.3 Technical Requirements¶
5.3.1 Architecture¶
packages/maid-engine/src/maid_engine/
├── profiling/
│ ├── __init__.py
│ ├── manager.py # ProfileManager orchestration
│ ├── memory.py # Memory profiling
│ ├── query.py # Database query profiling
│ ├── tick.py # Tick/system timing
│ ├── network.py # Network I/O profiling
│ ├── reports.py # Report generation
│ └── middleware.py # FastAPI profiling middleware
5.3.2 Profile Manager¶
# packages/maid-engine/src/maid_engine/profiling/manager.py
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
import asyncio
class ProfileType(Enum):
"""Types of profiling."""
MEMORY = "memory"
QUERY = "query"
TICK = "tick"
NETWORK = "network"
ALL = "all"
@dataclass
class ProfileSession:
"""Active profiling session."""
id: str
types: list[ProfileType]
start_time: datetime
duration: timedelta | None
samples: int = 0
data: dict = field(default_factory=dict)
@dataclass
class MemorySnapshot:
"""Memory usage snapshot."""
timestamp: datetime
total_mb: float
by_type: dict[str, float] # type name -> MB
by_system: dict[str, float] # system name -> MB
top_objects: list[tuple[str, int, float]] # (type, count, MB)
@dataclass
class QueryProfile:
"""Database query profile."""
query: str
params: dict | None
duration_ms: float
rows_affected: int
timestamp: datetime
stack_trace: list[str]
@dataclass
class TickProfile:
"""Single tick timing profile."""
tick_number: int
total_ms: float
by_system: dict[str, float] # system name -> ms
entity_count: int
timestamp: datetime
@dataclass
class NetworkProfile:
"""Network I/O profile."""
connection_id: str
direction: str # "in" or "out"
bytes_count: int
message_type: str
duration_ms: float
timestamp: datetime
class ProfileManager:
"""Manages profiling sessions and data collection."""
def __init__(self, engine: GameEngine) -> None:
self._engine = engine
self._sessions: dict[str, ProfileSession] = {}
self._active_session: ProfileSession | None = None
# Collectors
self._memory_collector: MemoryCollector | None = None
self._query_collector: QueryCollector | None = None
self._tick_collector: TickCollector | None = None
self._network_collector: NetworkCollector | None = None
async def start_session(
self,
types: list[ProfileType] | None = None,
duration: timedelta | None = None,
) -> ProfileSession:
"""Start a new profiling session."""
if self._active_session:
raise RuntimeError("Profiling session already active")
types = types or [ProfileType.ALL]
session = ProfileSession(
id=str(uuid.uuid4()),
types=types,
start_time=datetime.now(),
duration=duration,
)
# Start collectors
if ProfileType.ALL in types or ProfileType.MEMORY in types:
self._memory_collector = MemoryCollector()
await self._memory_collector.start()
if ProfileType.ALL in types or ProfileType.QUERY in types:
self._query_collector = QueryCollector(self._engine)
await self._query_collector.start()
if ProfileType.ALL in types or ProfileType.TICK in types:
self._tick_collector = TickCollector(self._engine)
await self._tick_collector.start()
if ProfileType.ALL in types or ProfileType.NETWORK in types:
self._network_collector = NetworkCollector(self._engine)
await self._network_collector.start()
self._active_session = session
self._sessions[session.id] = session
# Auto-stop after duration
if duration:
asyncio.create_task(self._auto_stop(session.id, duration))
return session
async def stop_session(self) -> ProfileSession | None:
"""Stop the active profiling session."""
if not self._active_session:
return None
session = self._active_session
# Collect final data
if self._memory_collector:
session.data["memory"] = await self._memory_collector.collect()
await self._memory_collector.stop()
if self._query_collector:
session.data["queries"] = await self._query_collector.collect()
await self._query_collector.stop()
if self._tick_collector:
session.data["ticks"] = await self._tick_collector.collect()
await self._tick_collector.stop()
if self._network_collector:
session.data["network"] = await self._network_collector.collect()
await self._network_collector.stop()
self._active_session = None
return session
async def get_memory_snapshot(self) -> MemorySnapshot:
"""Get current memory snapshot (no active session required)."""
collector = MemoryCollector()
return await collector.snapshot()
async def get_tick_stats(self, last_n: int = 100) -> list[TickProfile]:
"""Get recent tick timing stats."""
if self._tick_collector:
return self._tick_collector.get_recent(last_n)
return []
def generate_report(
self,
session_id: str,
format: str = "html",
) -> str:
"""Generate a profiling report."""
session = self._sessions.get(session_id)
if not session:
raise ValueError(f"Session {session_id} not found")
reporter = ProfileReporter(session)
return reporter.generate(format)
5.3.3 Memory Profiler¶
# packages/maid-engine/src/maid_engine/profiling/memory.py
import tracemalloc
import gc
from collections import defaultdict
class MemoryCollector:
"""Collects memory profiling data."""
def __init__(self) -> None:
self._snapshots: list[MemorySnapshot] = []
self._started = False
async def start(self) -> None:
"""Start memory tracking."""
tracemalloc.start(25) # 25 frames for stack traces
self._started = True
async def stop(self) -> None:
"""Stop memory tracking."""
tracemalloc.stop()
self._started = False
async def snapshot(self) -> MemorySnapshot:
"""Take a memory snapshot."""
gc.collect() # Force GC for accurate measurement
snapshot = tracemalloc.take_snapshot()
stats = snapshot.statistics("lineno")
# Calculate totals
total_bytes = sum(stat.size for stat in stats)
total_mb = total_bytes / (1024 * 1024)
# Group by type
by_type: dict[str, float] = defaultdict(float)
for stat in stats:
# Extract type from traceback
type_name = self._extract_type(stat)
by_type[type_name] += stat.size / (1024 * 1024)
# Top objects
top_objects = []
for stat in stats[:20]:
top_objects.append((
str(stat.traceback),
stat.count,
stat.size / (1024 * 1024),
))
return MemorySnapshot(
timestamp=datetime.now(),
total_mb=total_mb,
by_type=dict(by_type),
by_system={}, # Filled by system analysis
top_objects=top_objects,
)
async def collect(self) -> list[MemorySnapshot]:
"""Get all collected snapshots."""
return self._snapshots
def compare_snapshots(
self,
before: MemorySnapshot,
after: MemorySnapshot,
) -> dict:
"""Compare two snapshots to find leaks."""
diff = {
"total_change_mb": after.total_mb - before.total_mb,
"by_type": {},
}
all_types = set(before.by_type.keys()) | set(after.by_type.keys())
for type_name in all_types:
before_mb = before.by_type.get(type_name, 0)
after_mb = after.by_type.get(type_name, 0)
change = after_mb - before_mb
if abs(change) > 0.1: # Only report >100KB changes
diff["by_type"][type_name] = change
return diff
5.3.4 Query Profiler¶
# packages/maid-engine/src/maid_engine/profiling/query.py
import asyncio
from contextlib import asynccontextmanager
class QueryCollector:
"""Collects database query profiling data."""
def __init__(self, engine: GameEngine) -> None:
self._engine = engine
self._queries: list[QueryProfile] = []
self._original_execute: Callable | None = None
async def start(self) -> None:
"""Start query profiling by wrapping database methods."""
# Monkey-patch the document store's execute method
store = self._engine.document_store
self._original_execute = store._execute
store._execute = self._profiled_execute
async def stop(self) -> None:
"""Stop query profiling."""
if self._original_execute:
store = self._engine.document_store
store._execute = self._original_execute
async def _profiled_execute(
self,
query: str,
params: dict | None = None,
) -> Any:
"""Wrapper that profiles query execution."""
start = time.perf_counter()
# Capture stack trace
stack = traceback.format_stack()[:-2]
try:
result = await self._original_execute(query, params)
rows = len(result) if isinstance(result, list) else 1
finally:
duration_ms = (time.perf_counter() - start) * 1000
self._queries.append(QueryProfile(
query=query,
params=params,
duration_ms=duration_ms,
rows_affected=rows,
timestamp=datetime.now(),
stack_trace=stack,
))
return result
async def collect(self) -> list[QueryProfile]:
"""Get all collected queries."""
return self._queries
def get_slow_queries(self, threshold_ms: float = 100) -> list[QueryProfile]:
"""Get queries slower than threshold."""
return [q for q in self._queries if q.duration_ms > threshold_ms]
def get_query_stats(self) -> dict:
"""Get aggregated query statistics."""
if not self._queries:
return {}
durations = [q.duration_ms for q in self._queries]
return {
"total_queries": len(self._queries),
"total_time_ms": sum(durations),
"avg_time_ms": sum(durations) / len(durations),
"max_time_ms": max(durations),
"min_time_ms": min(durations),
"queries_over_100ms": len([d for d in durations if d > 100]),
}
5.3.5 Tick Profiler¶
# packages/maid-engine/src/maid_engine/profiling/tick.py
class TickCollector:
"""Collects tick timing profiling data."""
def __init__(self, engine: GameEngine) -> None:
self._engine = engine
self._profiles: list[TickProfile] = []
self._max_profiles = 10000
async def start(self) -> None:
"""Start tick profiling."""
# Register as tick observer
self._engine.world.add_tick_observer(self._on_tick)
async def stop(self) -> None:
"""Stop tick profiling."""
self._engine.world.remove_tick_observer(self._on_tick)
async def _on_tick(
self,
tick_number: int,
delta: float,
system_timings: dict[str, float],
) -> None:
"""Called after each tick with timing data."""
profile = TickProfile(
tick_number=tick_number,
total_ms=delta * 1000,
by_system=system_timings,
entity_count=len(self._engine.world.entities),
timestamp=datetime.now(),
)
self._profiles.append(profile)
# Trim old profiles
if len(self._profiles) > self._max_profiles:
self._profiles = self._profiles[-self._max_profiles:]
async def collect(self) -> list[TickProfile]:
"""Get all collected profiles."""
return self._profiles
def get_recent(self, n: int = 100) -> list[TickProfile]:
"""Get last n tick profiles."""
return self._profiles[-n:]
def get_system_stats(self) -> dict[str, dict]:
"""Get aggregated stats per system."""
if not self._profiles:
return {}
system_times: dict[str, list[float]] = defaultdict(list)
for profile in self._profiles:
for system, time_ms in profile.by_system.items():
system_times[system].append(time_ms)
stats = {}
for system, times in system_times.items():
stats[system] = {
"avg_ms": sum(times) / len(times),
"max_ms": max(times),
"min_ms": min(times),
"total_ms": sum(times),
"calls": len(times),
}
return stats
def get_slow_ticks(self, threshold_ms: float = 250) -> list[TickProfile]:
"""Get ticks that exceeded threshold (at 4 TPS, budget is 250ms)."""
return [p for p in self._profiles if p.total_ms > threshold_ms]
5.4 API/Interface Design¶
# CLI Commands
@app.command()
def profile(
types: list[str] = typer.Option(["all"], help="Profile types: memory, query, tick, network"),
duration: int = typer.Option(60, help="Duration in seconds"),
output: str = typer.Option("profile.html", help="Output file"),
):
"""Run a profiling session.
Examples:
maid dev profile --duration=60
maid dev profile --types=memory,query --duration=300
maid dev profile --output=report.html
"""
...
@app.command()
def memory_snapshot():
"""Take an instant memory snapshot.
Examples:
maid dev memory-snapshot
"""
...
# In-game commands
async def cmd_profile(ctx: CommandContext) -> bool:
"""@profile - Manage profiling sessions.
Usage:
@profile start [types] [duration]
@profile stop
@profile status
@profile report [session_id]
Examples:
@profile start memory,tick 60
@profile stop
@profile report
"""
...
async def cmd_memory(ctx: CommandContext) -> bool:
"""@memory - Show memory statistics.
Usage:
@memory # Current snapshot
@memory top 20 # Top 20 allocations
@memory systems # Memory by system
@memory compare # Compare with last snapshot
"""
...
async def cmd_timing(ctx: CommandContext) -> bool:
"""@timing - Show tick timing statistics.
Usage:
@timing # Recent tick stats
@timing systems # Per-system breakdown
@timing slow # Show slow ticks
@timing history 100 # Last 100 ticks
"""
...
5.5 Configuration¶
class ProfilingSettings(BaseSettings):
"""Profiling configuration."""
model_config = SettingsConfigDict(env_prefix="MAID_PROFILING_")
enabled: bool = True
# Memory profiling
memory_snapshot_interval_seconds: int = 60
memory_traceback_depth: int = 25
# Query profiling
slow_query_threshold_ms: float = 100
max_query_history: int = 10000
log_slow_queries: bool = True
# Tick profiling
slow_tick_threshold_ms: float = 250 # 4 TPS = 250ms budget
max_tick_history: int = 10000
# Reports
report_output_dir: str = "profiles/"
report_format: str = "html"
5.6 Dependencies¶
[project.optional-dependencies]
profiling = [
"tracemalloc-utils>=0.1.0", # Memory profiling helpers
"py-spy>=0.3.14", # Sampling profiler (optional)
"memory-profiler>=0.61.0", # Additional memory tools
"jinja2>=3.1.0", # Report templates
]
5.7 Implementation Tasks¶
- [ ] Phase 1: Memory Profiling (Week 1)
- [ ] Implement MemoryCollector
- [ ] Add snapshot comparison
- [ ] Create memory report
-
[ ] Add @memory command
-
[ ] Phase 2: Query & Tick Profiling (Week 1-2)
- [ ] Implement QueryCollector with DB wrapping
- [ ] Implement TickCollector with system timing
- [ ] Add slow query/tick detection
-
[ ] Add @timing command
-
[ ] Phase 3: Integration & Reports (Week 2)
- [ ] Implement ProfileManager
- [ ] Create HTML report templates
- [ ] Add CLI commands
- [ ] Add Web API endpoints
5.8 Testing Requirements¶
@pytest.mark.asyncio
async def test_memory_snapshot(profile_manager):
"""Test memory snapshot captures allocations."""
# Allocate some objects
objects = [{"data": "x" * 1000} for _ in range(1000)]
snapshot = await profile_manager.get_memory_snapshot()
assert snapshot.total_mb > 0
assert len(snapshot.top_objects) > 0
@pytest.mark.asyncio
async def test_query_profiling(profile_manager, test_db):
"""Test query profiling captures queries."""
session = await profile_manager.start_session([ProfileType.QUERY])
# Execute some queries
await test_db.execute("SELECT * FROM entities")
await test_db.execute("SELECT * FROM components WHERE entity_id = $1", {"$1": "123"})
session = await profile_manager.stop_session()
assert len(session.data["queries"]) >= 2
assert all(q.duration_ms >= 0 for q in session.data["queries"])
@pytest.mark.asyncio
async def test_tick_profiling(profile_manager, test_engine):
"""Test tick profiling captures system timings."""
session = await profile_manager.start_session([ProfileType.TICK])
# Run a few ticks
for _ in range(10):
await test_engine.tick()
session = await profile_manager.stop_session()
assert len(session.data["ticks"]) == 10
assert all(t.by_system for t in session.data["ticks"])
5.9 Acceptance Criteria¶
- Memory Profiling
- [ ] Shows total memory usage
- [ ] Breaks down by object type
- [ ] Identifies top allocations with stack traces
-
[ ] Compares snapshots to find leaks
-
Query Profiling
- [ ] Captures all database queries
- [ ] Records execution time
- [ ] Identifies slow queries
-
[ ] Shows query statistics
-
Tick Profiling
- [ ] Records per-tick timing
- [ ] Breaks down by system
- [ ] Identifies slow ticks
-
[ ] Provides historical data
-
Reports
- [ ] Generates HTML reports
- [ ] Includes visualizations (charts)
- [ ] Can be exported for analysis
6. MaidEditor (In-Game Text Editor)¶
6.1 Feature Overview¶
MaidEditor is a VI-like in-game text editor for editing room descriptions, NPC dialogue, help files, and other multi-line text content directly from within the game client.
Why it's needed: - Allows editing long text without external tools - Maintains familiar MUD workflow - Supports syntax highlighting for code blocks - Enables collaborative editing sessions
6.2 User Stories¶
US-6.1: Description Editing
As a builder, I want to edit room descriptions using a full-featured text editor so that I can write detailed, multi-paragraph descriptions.
US-6.2: Code Editing
As a developer, I want to edit script code in-game so that I can make quick fixes without external tools.
US-6.3: Template Editing
As a builder, I want to edit message templates with placeholders so that I can customize game messages.
6.3 Technical Requirements¶
# packages/maid-stdlib/src/maid_stdlib/utils/editor.py
from dataclasses import dataclass, field
from enum import Enum
from typing import Callable, Awaitable
class EditorMode(Enum):
"""Editor modes."""
VIEW = "view" # Read-only viewing
INSERT = "insert" # Text insertion
COMMAND = "command" # Command mode (like VI)
SEARCH = "search" # Search mode
REPLACE = "replace" # Search and replace
@dataclass
class EditorBuffer:
"""Text buffer with undo history."""
lines: list[str] = field(default_factory=list)
cursor_line: int = 0
cursor_col: int = 0
undo_stack: list[tuple[list[str], int, int]] = field(default_factory=list)
redo_stack: list[tuple[list[str], int, int]] = field(default_factory=list)
modified: bool = False
@dataclass
class EditorConfig:
"""Editor configuration."""
max_lines: int = 1000
max_line_length: int = 200
tab_width: int = 4
auto_indent: bool = True
show_line_numbers: bool = True
syntax_highlight: bool = True
syntax_type: str | None = None # "python", "yaml", "markdown", None
class MaidEditor:
"""In-game VI-like text editor."""
def __init__(
self,
session: Any, # Player session
*,
initial_text: str = "",
config: EditorConfig | None = None,
on_save: Callable[[str], Awaitable[bool]] | None = None,
on_quit: Callable[[], Awaitable[None]] | None = None,
) -> None:
self._session = session
self._config = config or EditorConfig()
self._on_save = on_save
self._on_quit = on_quit
self._buffer = EditorBuffer()
self._buffer.lines = initial_text.split("\n") if initial_text else [""]
self._mode = EditorMode.COMMAND
self._status_message = ""
self._search_pattern = ""
self._command_buffer = ""
async def start(self) -> None:
"""Start the editor session."""
await self._render()
await self._input_loop()
async def _input_loop(self) -> None:
"""Main input processing loop."""
while True:
input_char = await self._session.read_char()
if self._mode == EditorMode.COMMAND:
if not await self._handle_command(input_char):
break
elif self._mode == EditorMode.INSERT:
await self._handle_insert(input_char)
elif self._mode == EditorMode.SEARCH:
await self._handle_search(input_char)
await self._render()
async def _handle_command(self, char: str) -> bool:
"""Handle command mode input. Returns False to exit."""
commands = {
"i": self._enter_insert_mode,
"a": self._enter_insert_after,
"o": self._open_line_below,
"O": self._open_line_above,
"x": self._delete_char,
"dd": self._delete_line,
"yy": self._yank_line,
"p": self._paste,
"u": self._undo,
"ctrl+r": self._redo,
"/": self._enter_search_mode,
"n": self._find_next,
"N": self._find_prev,
":w": self._save,
":q": self._quit,
":wq": self._save_and_quit,
":q!": self._force_quit,
"h": self._move_left,
"j": self._move_down,
"k": self._move_up,
"l": self._move_right,
"0": self._move_line_start,
"$": self._move_line_end,
"gg": self._move_file_start,
"G": self._move_file_end,
}
self._command_buffer += char
# Check for complete command
for cmd, handler in commands.items():
if self._command_buffer == cmd:
self._command_buffer = ""
result = await handler()
if result is False:
return False
return True
# Check for partial match
if any(cmd.startswith(self._command_buffer) for cmd in commands):
return True
# No match, reset
self._command_buffer = ""
return True
async def _handle_insert(self, char: str) -> None:
"""Handle insert mode input."""
if char == "\x1b": # Escape
self._mode = EditorMode.COMMAND
return
if char == "\n":
await self._insert_newline()
elif char == "\x7f": # Backspace
await self._backspace()
else:
await self._insert_char(char)
async def _render(self) -> None:
"""Render the editor screen."""
# Clear screen
await self._session.send("\x1b[2J\x1b[H")
# Calculate visible lines
height = 24 # Assume 24 lines
start_line = max(0, self._buffer.cursor_line - height // 2)
end_line = min(len(self._buffer.lines), start_line + height - 2)
# Render lines
for i in range(start_line, end_line):
line = self._buffer.lines[i]
# Line number
if self._config.show_line_numbers:
line_num = f"{i + 1:4} "
await self._session.send(f"\x1b[90m{line_num}\x1b[0m")
# Syntax highlighting
if self._config.syntax_highlight and self._config.syntax_type:
line = self._highlight(line)
# Cursor position
if i == self._buffer.cursor_line:
before = line[:self._buffer.cursor_col]
cursor = line[self._buffer.cursor_col:self._buffer.cursor_col + 1] or " "
after = line[self._buffer.cursor_col + 1:]
await self._session.send(f"{before}\x1b[7m{cursor}\x1b[0m{after}\n")
else:
await self._session.send(f"{line}\n")
# Status line
mode_str = f"-- {self._mode.value.upper()} --"
pos_str = f"{self._buffer.cursor_line + 1},{self._buffer.cursor_col + 1}"
modified = "[+]" if self._buffer.modified else ""
status = f"{mode_str} {modified} {self._status_message} {pos_str}"
await self._session.send(f"\x1b[7m{status:80}\x1b[0m")
async def _save(self) -> bool:
"""Save the buffer."""
if self._on_save:
text = "\n".join(self._buffer.lines)
success = await self._on_save(text)
if success:
self._buffer.modified = False
self._status_message = "Saved"
else:
self._status_message = "Save failed!"
return success
return False
# ... additional methods for movement, editing, etc.
6.4 API/Interface Design¶
# Usage from building commands
async def cmd_edit(ctx: CommandContext) -> bool:
"""@edit - Open the in-game text editor.
Usage:
@edit here/description # Edit room description
@edit sword/description # Edit item description
@edit new help/commands # Create new help file
@edit script guards/patrol # Edit script file
Editor Commands:
i - Enter insert mode
Esc - Return to command mode
:w - Save
:q - Quit
:wq - Save and quit
/text - Search
u - Undo
dd - Delete line
yy - Copy line
p - Paste
"""
if not ctx.args:
await ctx.session.send("Usage: @edit <target>/<field>")
return False
target_path = ctx.args[0]
# Resolve target and field
entity, field = await resolve_edit_target(ctx, target_path)
if not entity:
await ctx.session.send(f"Target not found: {target_path}")
return False
# Get current text
current_text = getattr(entity, field, "")
# Define save callback
async def on_save(text: str) -> bool:
setattr(entity, field, text)
await ctx.world.save_entity(entity)
return True
# Start editor
editor = MaidEditor(
ctx.session,
initial_text=current_text,
config=EditorConfig(
syntax_type="markdown" if field == "description" else None,
),
on_save=on_save,
)
await editor.start()
return True
6.5 Configuration¶
class EditorSettings(BaseSettings):
"""Editor configuration."""
model_config = SettingsConfigDict(env_prefix="MAID_EDITOR_")
max_lines: int = 1000
max_line_length: int = 200
tab_width: int = 4
auto_indent: bool = True
show_line_numbers: bool = True
default_syntax: str | None = None
6.6 Implementation Tasks¶
- [ ] Implement EditorBuffer with undo/redo
- [ ] Implement command mode handlers
- [ ] Implement insert mode
- [ ] Add search and replace
- [ ] Add syntax highlighting
- [ ] Create @edit command
- [ ] Add help overlay
6.7 Acceptance Criteria¶
- [ ] Can edit multi-line text with VI-like commands
- [ ] Supports undo/redo
- [ ] Saves changes to entities
- [ ] Shows line numbers
- [ ] Basic syntax highlighting works
7. MaidMenu (Dynamic Menu System)¶
7.1 Feature Overview¶
MaidMenu provides a flexible system for creating interactive menus in-game, supporting navigation, selection, and input collection. Useful for character creation, shops, dialogue trees, and admin interfaces.
7.2 User Stories¶
US-7.1: Character Creation
As a player, I want to navigate through character creation using a menu system so that the process is intuitive.
US-7.2: Shop Interface
As a player, I want to browse and purchase items through a menu so that shopping is easy.
US-7.3: Admin Menus
As an admin, I want to access admin functions through hierarchical menus so that I can find features quickly.
7.3 Technical Requirements¶
# packages/maid-stdlib/src/maid_stdlib/utils/menu.py
from dataclasses import dataclass, field
from typing import Callable, Awaitable, Any
from enum import Enum
class MenuNodeType(Enum):
"""Types of menu nodes."""
TEXT = "text" # Display text only
OPTION = "option" # Selectable option
INPUT = "input" # Text input field
SUBMENU = "submenu" # Link to submenu
SEPARATOR = "separator" # Visual separator
DYNAMIC = "dynamic" # Dynamically generated
@dataclass
class MenuNode:
"""Single node in a menu."""
key: str # Selection key (1, 2, a, b, etc.)
text: str # Display text
node_type: MenuNodeType = MenuNodeType.OPTION
callback: Callable[..., Awaitable[Any]] | None = None
submenu: "MaidMenu | None" = None
enabled: bool = True
visible: bool = True
data: dict = field(default_factory=dict)
@dataclass
class MenuConfig:
"""Menu configuration."""
title: str = ""
header: str = ""
footer: str = ""
prompt: str = "Select: "
back_key: str = "b"
back_text: str = "Back"
quit_key: str = "q"
quit_text: str = "Quit"
show_back: bool = True
show_quit: bool = True
allow_invalid: bool = False
clear_screen: bool = False
columns: int = 1
auto_number: bool = True
class MaidMenu:
"""Dynamic menu system."""
def __init__(
self,
session: Any,
config: MenuConfig | None = None,
) -> None:
self._session = session
self._config = config or MenuConfig()
self._nodes: list[MenuNode] = []
self._parent: MaidMenu | None = None
self._context: dict[str, Any] = {}
self._result: Any = None
def add(
self,
text: str,
callback: Callable[..., Awaitable[Any]] | None = None,
*,
key: str | None = None,
node_type: MenuNodeType = MenuNodeType.OPTION,
enabled: bool = True,
visible: bool = True,
**data,
) -> "MaidMenu":
"""Add a menu option."""
if key is None and self._config.auto_number:
key = str(len([n for n in self._nodes if n.node_type == MenuNodeType.OPTION]) + 1)
node = MenuNode(
key=key or "",
text=text,
node_type=node_type,
callback=callback,
enabled=enabled,
visible=visible,
data=data,
)
self._nodes.append(node)
return self
def add_submenu(
self,
text: str,
submenu: "MaidMenu",
*,
key: str | None = None,
) -> "MaidMenu":
"""Add a submenu."""
if key is None and self._config.auto_number:
key = str(len([n for n in self._nodes if n.node_type == MenuNodeType.OPTION]) + 1)
submenu._parent = self
node = MenuNode(
key=key or "",
text=text,
node_type=MenuNodeType.SUBMENU,
submenu=submenu,
)
self._nodes.append(node)
return self
def add_input(
self,
prompt: str,
callback: Callable[[str], Awaitable[Any]],
*,
key: str | None = None,
validator: Callable[[str], bool] | None = None,
) -> "MaidMenu":
"""Add a text input option."""
node = MenuNode(
key=key or "",
text=prompt,
node_type=MenuNodeType.INPUT,
callback=callback,
data={"validator": validator},
)
self._nodes.append(node)
return self
def add_separator(self, text: str = "") -> "MaidMenu":
"""Add a visual separator."""
self._nodes.append(MenuNode(
key="",
text=text,
node_type=MenuNodeType.SEPARATOR,
))
return self
def add_dynamic(
self,
generator: Callable[[], Awaitable[list[MenuNode]]],
) -> "MaidMenu":
"""Add dynamically generated options."""
self._nodes.append(MenuNode(
key="",
text="",
node_type=MenuNodeType.DYNAMIC,
callback=generator,
))
return self
async def run(self) -> Any:
"""Run the menu and return result."""
while True:
# Generate dynamic nodes
display_nodes = await self._prepare_nodes()
# Render menu
await self._render(display_nodes)
# Get input
choice = await self._session.readline()
choice = choice.strip().lower()
# Handle back/quit
if self._config.show_back and choice == self._config.back_key:
if self._parent:
return await self._parent.run()
return None
if self._config.show_quit and choice == self._config.quit_key:
return None
# Find matching node
node = self._find_node(display_nodes, choice)
if not node:
if not self._config.allow_invalid:
await self._session.send("Invalid selection.\n")
continue
# Handle selection
result = await self._handle_selection(node)
if result is not None:
self._result = result
return result
async def _render(self, nodes: list[MenuNode]) -> None:
"""Render the menu."""
if self._config.clear_screen:
await self._session.send("\x1b[2J\x1b[H")
# Title and header
if self._config.title:
await self._session.send(f"\n=== {self._config.title} ===\n")
if self._config.header:
await self._session.send(f"{self._config.header}\n")
await self._session.send("\n")
# Options
for node in nodes:
if not node.visible:
continue
if node.node_type == MenuNodeType.SEPARATOR:
if node.text:
await self._session.send(f"\n--- {node.text} ---\n")
else:
await self._session.send("\n")
continue
# Format option
status = "" if node.enabled else " (disabled)"
line = f" [{node.key}] {node.text}{status}\n"
await self._session.send(line)
# Back and quit options
await self._session.send("\n")
if self._config.show_back and self._parent:
await self._session.send(f" [{self._config.back_key}] {self._config.back_text}\n")
if self._config.show_quit:
await self._session.send(f" [{self._config.quit_key}] {self._config.quit_text}\n")
# Footer and prompt
if self._config.footer:
await self._session.send(f"\n{self._config.footer}\n")
await self._session.send(f"\n{self._config.prompt}")
async def _handle_selection(self, node: MenuNode) -> Any:
"""Handle a node selection."""
if not node.enabled:
await self._session.send("That option is not available.\n")
return None
if node.node_type == MenuNodeType.SUBMENU and node.submenu:
return await node.submenu.run()
if node.node_type == MenuNodeType.INPUT:
await self._session.send(f"{node.text}: ")
value = await self._session.readline()
value = value.strip()
validator = node.data.get("validator")
if validator and not validator(value):
await self._session.send("Invalid input.\n")
return None
if node.callback:
return await node.callback(value)
if node.callback:
return await node.callback(self._context, node.data)
return node.data
7.4 Usage Example¶
# Character creation menu
async def character_creation_menu(session) -> dict:
"""Run character creation menu."""
result = {"name": "", "race": "", "class": ""}
# Name input
async def set_name(value: str) -> None:
result["name"] = value
# Race selection
async def set_race(ctx: dict, data: dict) -> None:
result["race"] = data["race"]
race_menu = MaidMenu(session, MenuConfig(title="Select Race"))
race_menu.add("Human", set_race, race="human")
race_menu.add("Elf", set_race, race="elf")
race_menu.add("Dwarf", set_race, race="dwarf")
# Class selection (filtered by race)
async def generate_classes() -> list[MenuNode]:
classes = get_classes_for_race(result["race"])
return [
MenuNode(key=str(i+1), text=c.name, callback=set_class, data={"class": c.id})
for i, c in enumerate(classes)
]
class_menu = MaidMenu(session, MenuConfig(title="Select Class"))
class_menu.add_dynamic(generate_classes)
# Main menu
main = MaidMenu(session, MenuConfig(title="Character Creation"))
main.add_input("Enter your name", set_name)
main.add_submenu("Select Race", race_menu)
main.add_submenu("Select Class", class_menu)
main.add("Confirm and Create", lambda ctx, data: result)
return await main.run()
7.5 Implementation Tasks¶
- [ ] Implement MaidMenu core class
- [ ] Add navigation (back, quit)
- [ ] Implement input handling
- [ ] Add dynamic node generation
- [ ] Create menu builder helpers
- [ ] Add styling options
7.6 Acceptance Criteria¶
- [ ] Menus render with options and prompts
- [ ] Selection triggers callbacks
- [ ] Submenus navigate correctly
- [ ] Back/quit work as expected
- [ ] Dynamic menus generate at runtime
8. MaidTable (Formatted Table Display)¶
8.1 Feature Overview¶
MaidTable provides formatted table output for displaying structured data in the terminal, supporting column alignment, borders, colors, and responsive width.
8.2 Technical Requirements¶
# packages/maid-stdlib/src/maid_stdlib/utils/table.py
from dataclasses import dataclass, field
from enum import Enum
from typing import Any
class Alignment(Enum):
"""Column alignment."""
LEFT = "left"
CENTER = "center"
RIGHT = "right"
class BorderStyle(Enum):
"""Table border styles."""
NONE = "none"
ASCII = "ascii"
UNICODE = "unicode"
DOUBLE = "double"
@dataclass
class Column:
"""Table column definition."""
header: str
key: str | None = None # Key for dict rows
width: int | None = None # Fixed width or None for auto
min_width: int = 3
max_width: int = 50
align: Alignment = Alignment.LEFT
formatter: Callable[[Any], str] | None = None
color: str | None = None
@dataclass
class TableConfig:
"""Table configuration."""
border: BorderStyle = BorderStyle.ASCII
show_header: bool = True
show_row_numbers: bool = False
zebra_stripe: bool = False
max_width: int = 80
padding: int = 1
null_value: str = "-"
truncate_marker: str = "..."
class MaidTable:
"""Formatted table display."""
BORDERS = {
BorderStyle.NONE: {"h": "", "v": " ", "tl": "", "tr": "", "bl": "", "br": "", "t": "", "b": "", "l": "", "r": "", "cross": " "},
BorderStyle.ASCII: {"h": "-", "v": "|", "tl": "+", "tr": "+", "bl": "+", "br": "+", "t": "+", "b": "+", "l": "+", "r": "+", "cross": "+"},
BorderStyle.UNICODE: {"h": "─", "v": "│", "tl": "┌", "tr": "┐", "bl": "└", "br": "┘", "t": "┬", "b": "┴", "l": "├", "r": "┤", "cross": "┼"},
BorderStyle.DOUBLE: {"h": "═", "v": "║", "tl": "╔", "tr": "╗", "bl": "╚", "br": "╝", "t": "╦", "b": "╩", "l": "╠", "r": "╣", "cross": "╬"},
}
def __init__(
self,
columns: list[Column | str] | None = None,
config: TableConfig | None = None,
) -> None:
self._config = config or TableConfig()
self._columns: list[Column] = []
self._rows: list[list[Any]] = []
if columns:
for col in columns:
if isinstance(col, str):
self._columns.append(Column(header=col))
else:
self._columns.append(col)
def add_column(
self,
header: str,
*,
key: str | None = None,
width: int | None = None,
align: Alignment = Alignment.LEFT,
formatter: Callable[[Any], str] | None = None,
) -> "MaidTable":
"""Add a column definition."""
self._columns.append(Column(
header=header,
key=key,
width=width,
align=align,
formatter=formatter,
))
return self
def add_row(self, *values, **kwargs) -> "MaidTable":
"""Add a row of data."""
if kwargs:
# Dict row - extract by column keys
row = [kwargs.get(col.key or col.header, None) for col in self._columns]
else:
row = list(values)
self._rows.append(row)
return self
def add_rows(self, rows: list[list | dict]) -> "MaidTable":
"""Add multiple rows."""
for row in rows:
if isinstance(row, dict):
self.add_row(**row)
else:
self.add_row(*row)
return self
def render(self) -> str:
"""Render the table as a string."""
if not self._columns:
return ""
# Calculate column widths
widths = self._calculate_widths()
# Get border characters
b = self.BORDERS[self._config.border]
pad = " " * self._config.padding
lines = []
# Top border
if self._config.border != BorderStyle.NONE:
top = b["tl"]
for i, w in enumerate(widths):
top += b["h"] * (w + self._config.padding * 2)
top += b["t"] if i < len(widths) - 1 else b["tr"]
lines.append(top)
# Header
if self._config.show_header:
header_row = b["v"]
for i, col in enumerate(self._columns):
cell = self._format_cell(col.header, widths[i], col.align)
header_row += f"{pad}{cell}{pad}{b['v']}"
lines.append(header_row)
# Header separator
if self._config.border != BorderStyle.NONE:
sep = b["l"]
for i, w in enumerate(widths):
sep += b["h"] * (w + self._config.padding * 2)
sep += b["cross"] if i < len(widths) - 1 else b["r"]
lines.append(sep)
# Data rows
for row_idx, row in enumerate(self._rows):
data_row = b["v"]
# Zebra striping
stripe = "\x1b[48;5;236m" if self._config.zebra_stripe and row_idx % 2 else ""
reset = "\x1b[0m" if stripe else ""
for i, col in enumerate(self._columns):
value = row[i] if i < len(row) else None
# Format value
if col.formatter:
formatted = col.formatter(value)
elif value is None:
formatted = self._config.null_value
else:
formatted = str(value)
cell = self._format_cell(formatted, widths[i], col.align)
# Color
color = f"\x1b[{col.color}m" if col.color else ""
color_reset = "\x1b[0m" if col.color else ""
data_row += f"{stripe}{pad}{color}{cell}{color_reset}{pad}{reset}{b['v']}"
lines.append(data_row)
# Bottom border
if self._config.border != BorderStyle.NONE:
bottom = b["bl"]
for i, w in enumerate(widths):
bottom += b["h"] * (w + self._config.padding * 2)
bottom += b["b"] if i < len(widths) - 1 else b["br"]
lines.append(bottom)
return "\n".join(lines)
def _calculate_widths(self) -> list[int]:
"""Calculate column widths."""
widths = []
for i, col in enumerate(self._columns):
if col.width:
widths.append(col.width)
else:
# Auto-calculate
max_len = len(col.header)
for row in self._rows:
if i < len(row):
val = str(row[i]) if row[i] is not None else self._config.null_value
max_len = max(max_len, len(val))
width = max(col.min_width, min(max_len, col.max_width))
widths.append(width)
return widths
def _format_cell(self, value: str, width: int, align: Alignment) -> str:
"""Format a cell value with alignment and truncation."""
# Truncate if needed
if len(value) > width:
value = value[:width - len(self._config.truncate_marker)] + self._config.truncate_marker
# Align
if align == Alignment.LEFT:
return value.ljust(width)
elif align == Alignment.RIGHT:
return value.rjust(width)
else:
return value.center(width)
def __str__(self) -> str:
return self.render()
8.3 Usage Example¶
# Inventory display
table = MaidTable([
Column("Item", width=20),
Column("Qty", width=5, align=Alignment.RIGHT),
Column("Weight", width=8, align=Alignment.RIGHT, formatter=lambda x: f"{x:.1f} kg"),
Column("Value", width=10, align=Alignment.RIGHT, formatter=lambda x: f"{x:,} gold"),
])
table.add_row("Iron Sword", 1, 3.5, 150)
table.add_row("Health Potion", 5, 0.5, 50)
table.add_row("Leather Armor", 1, 8.0, 200)
print(table.render())
# Output:
# +----------------------+-------+----------+------------+
# | Item | Qty | Weight | Value |
# +----------------------+-------+----------+------------+
# | Iron Sword | 1 | 3.5 kg | 150 gold |
# | Health Potion | 5 | 0.5 kg | 50 gold |
# | Leather Armor | 1 | 8.0 kg | 200 gold |
# +----------------------+-------+----------+------------+
8.4 Implementation Tasks¶
- [ ] Implement MaidTable core class
- [ ] Add column width calculation
- [ ] Implement border styles
- [ ] Add alignment and truncation
- [ ] Add color support
- [ ] Add row formatting (zebra, etc.)
8.5 Acceptance Criteria¶
- [ ] Tables render with proper alignment
- [ ] All border styles work
- [ ] Column widths auto-calculate
- [ ] Long values truncate with ellipsis
- [ ] Colors display correctly in terminal
9. Batch Command/Code Processors¶
9.1 Feature Overview¶
Batch processors allow executing multiple commands or code blocks in sequence, useful for world setup, testing, migrations, and automated tasks.
Why it's needed: - Automates repetitive setup tasks - Enables reproducible world creation - Supports data migrations - Facilitates testing scenarios
9.2 User Stories¶
US-9.1: Batch Commands
As a builder, I want to execute a file of building commands so that I can set up areas quickly.
US-9.2: Batch Code
As a developer, I want to run Python code files against the running server so that I can perform complex operations.
US-9.3: Scheduled Batches
As an admin, I want to schedule batch jobs to run at specific times so that maintenance happens automatically.
9.3 Technical Requirements¶
# packages/maid-engine/src/maid_engine/batch/
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
import asyncio
class BatchType(Enum):
"""Type of batch processing."""
COMMAND = "command" # Game commands
CODE = "code" # Python code
MIXED = "mixed" # Both
@dataclass
class BatchResult:
"""Result of batch execution."""
success: bool
total_commands: int
executed: int
failed: int
errors: list[tuple[int, str, str]] # (line, command, error)
duration_seconds: float
output: list[str]
class BatchProcessor:
"""Processes batch command and code files."""
def __init__(self, engine: GameEngine) -> None:
self._engine = engine
self._running = False
self._current_line = 0
async def execute_file(
self,
path: Path | str,
*,
batch_type: BatchType = BatchType.COMMAND,
continue_on_error: bool = True,
dry_run: bool = False,
context: dict | None = None,
) -> BatchResult:
"""Execute a batch file."""
path = Path(path)
if not path.exists():
raise FileNotFoundError(f"Batch file not found: {path}")
content = path.read_text()
return await self.execute(
content,
batch_type=batch_type,
continue_on_error=continue_on_error,
dry_run=dry_run,
context=context,
)
async def execute(
self,
content: str,
*,
batch_type: BatchType = BatchType.COMMAND,
continue_on_error: bool = True,
dry_run: bool = False,
context: dict | None = None,
) -> BatchResult:
"""Execute batch content."""
start_time = time.time()
self._running = True
if batch_type == BatchType.CODE:
return await self._execute_code(content, context, dry_run)
elif batch_type == BatchType.COMMAND:
return await self._execute_commands(content, continue_on_error, dry_run)
else:
return await self._execute_mixed(content, continue_on_error, dry_run, context)
async def _execute_commands(
self,
content: str,
continue_on_error: bool,
dry_run: bool,
) -> BatchResult:
"""Execute command batch."""
lines = content.split("\n")
executed = 0
failed = 0
errors = []
output = []
for i, line in enumerate(lines):
self._current_line = i + 1
line = line.strip()
# Skip empty lines and comments
if not line or line.startswith("#"):
continue
# Handle multi-line commands (ending with \)
while line.endswith("\\"):
line = line[:-1]
i += 1
if i < len(lines):
line += lines[i].strip()
if dry_run:
output.append(f"[DRY RUN] Would execute: {line}")
executed += 1
continue
try:
result = await self._engine.execute_command(line)
output.append(f"[{i+1}] {line}: OK")
executed += 1
except Exception as e:
failed += 1
errors.append((i + 1, line, str(e)))
output.append(f"[{i+1}] {line}: ERROR - {e}")
if not continue_on_error:
break
self._running = False
return BatchResult(
success=failed == 0,
total_commands=executed + failed,
executed=executed,
failed=failed,
errors=errors,
duration_seconds=time.time() - start_time,
output=output,
)
async def _execute_code(
self,
content: str,
context: dict | None,
dry_run: bool,
) -> BatchResult:
"""Execute Python code batch."""
if dry_run:
return BatchResult(
success=True,
total_commands=1,
executed=0,
failed=0,
errors=[],
duration_seconds=0,
output=["[DRY RUN] Would execute code block"],
)
# Create execution namespace
namespace = {
"engine": self._engine,
"world": self._engine.world,
"entities": self._engine.world.entities,
"asyncio": asyncio,
**(context or {}),
}
errors = []
output = []
try:
# Compile and execute
code = compile(content, "<batch>", "exec")
exec(code, namespace)
# If there's an async main, run it
if "main" in namespace and asyncio.iscoroutinefunction(namespace["main"]):
await namespace["main"]()
output.append("Code executed successfully")
except Exception as e:
errors.append((0, "code block", str(e)))
output.append(f"Error: {e}")
self._running = False
return BatchResult(
success=len(errors) == 0,
total_commands=1,
executed=1 if not errors else 0,
failed=1 if errors else 0,
errors=errors,
duration_seconds=time.time() - start_time,
output=output,
)
async def _execute_mixed(
self,
content: str,
continue_on_error: bool,
dry_run: bool,
context: dict | None,
) -> BatchResult:
"""Execute mixed command/code batch."""
# Parse content into blocks
blocks = self._parse_mixed(content)
total_executed = 0
total_failed = 0
all_errors = []
all_output = []
for block_type, block_content, start_line in blocks:
if block_type == "command":
result = await self._execute_commands(block_content, continue_on_error, dry_run)
else:
result = await self._execute_code(block_content, context, dry_run)
total_executed += result.executed
total_failed += result.failed
all_errors.extend([(start_line + e[0], e[1], e[2]) for e in result.errors])
all_output.extend(result.output)
if result.failed and not continue_on_error:
break
return BatchResult(
success=total_failed == 0,
total_commands=total_executed + total_failed,
executed=total_executed,
failed=total_failed,
errors=all_errors,
duration_seconds=time.time() - start_time,
output=all_output,
)
def _parse_mixed(self, content: str) -> list[tuple[str, str, int]]:
"""Parse mixed content into command and code blocks.
Format:
# Regular commands
@create room "Town Square"
#BEGIN CODE
# Python code block
async def main():
pass
#END CODE
# More commands
@dig north
"""
blocks = []
lines = content.split("\n")
current_type = "command"
current_lines = []
current_start = 0
for i, line in enumerate(lines):
if line.strip() == "#BEGIN CODE":
if current_lines:
blocks.append((current_type, "\n".join(current_lines), current_start))
current_type = "code"
current_lines = []
current_start = i + 1
elif line.strip() == "#END CODE":
if current_lines:
blocks.append((current_type, "\n".join(current_lines), current_start))
current_type = "command"
current_lines = []
current_start = i + 1
else:
current_lines.append(line)
if current_lines:
blocks.append((current_type, "\n".join(current_lines), current_start))
return blocks
9.4 Batch File Format¶
# example_area.batch
# This file creates the starting village area
# Create rooms
@create room "Village Square"
@set here/description = "A bustling village square with a fountain in the center."
@set here/terrain = outdoor
@set here/zone = starting_village
@dig north = "North Road"
@set here/description = "A dirt road leading north out of the village."
@dig east = "General Store"
@set here/description = "A small shop with various goods on display."
# Go back to square
@goto "Village Square"
@dig west = "Tavern"
@set here/description = "A cozy tavern with a roaring fireplace."
# Add NPCs
@goto "Village Square"
@spawn npcs/villager 5
@spawn npcs/guard 2
# Add items to store
@goto "General Store"
@spawn items/torch 10
@spawn items/rope 5
@spawn items/rations 20
#BEGIN CODE
# Custom setup code
async def main():
square = await world.find_room("Village Square")
# Set up the fountain as interactable
fountain = await world.create_entity()
fountain.add_component(DescriptionComponent(
name="Fountain",
description="A stone fountain with clear water bubbling up."
))
fountain.add_component(PositionComponent(room_id=square.id))
fountain.add_component(InteractableComponent(
actions=["drink", "wash", "examine"],
))
print(f"Created fountain in {square.id}")
#END CODE
# Final commands
@goto "Village Square"
@describe here "A bustling village square with a stone fountain in the center. \
Villagers mill about, going about their daily business. \
A guard stands watch near the north road."
9.5 API/Interface Design¶
# CLI Commands
@app.command()
def batch(
file: Path = typer.Argument(..., help="Batch file to execute"),
batch_type: str = typer.Option("auto", help="Type: command, code, mixed, auto"),
continue_on_error: bool = typer.Option(True, help="Continue after errors"),
dry_run: bool = typer.Option(False, help="Show what would be executed"),
):
"""Execute a batch file.
Examples:
maid batch setup/village.batch
maid batch scripts/migrate.py --type=code
maid batch mixed_setup.batch --dry-run
"""
...
# In-game commands
async def cmd_batch(ctx: CommandContext) -> bool:
"""@batch - Execute batch commands.
Usage:
@batch <file> # Execute batch file
@batch --code <file> # Execute as code
@batch --dry-run <file> # Preview execution
"""
...
9.6 Configuration¶
class BatchSettings(BaseSettings):
"""Batch processor configuration."""
model_config = SettingsConfigDict(env_prefix="MAID_BATCH_")
enabled: bool = True
batch_dir: str = "batches/"
max_commands_per_batch: int = 10000
execution_timeout_seconds: int = 300
allow_code_batches: bool = True
require_code_approval: bool = True
9.7 Implementation Tasks¶
- [ ] Implement BatchProcessor core class
- [ ] Add command parsing and execution
- [ ] Add code block execution
- [ ] Add mixed mode parsing
- [ ] Create CLI commands
- [ ] Add @batch in-game command
- [ ] Add scheduling support
9.8 Acceptance Criteria¶
- [ ] Command batches execute sequentially
- [ ] Errors are captured and reported
- [ ] Code batches run in server context
- [ ] Mixed batches parse correctly
- [ ] Dry run shows preview
- [ ] Batch files support comments
10. Internationalization (i18n) Support¶
10.1 Feature Overview¶
Internationalization enables MAID to support multiple languages for game text, system messages, and user interfaces. This includes message extraction, translation files, runtime language switching, and fallback handling.
Why it's needed: - Expands player base to non-English speakers - Supports international game development teams - Enables community-contributed translations - Professional-grade localization infrastructure
10.2 User Stories¶
US-10.1: Language Selection
As a player, I want to select my preferred language so that I can play the game in my native language.
US-10.2: Translation Management
As a translator, I want to edit translation files in a standard format so that I can contribute translations.
US-10.3: Missing Translation Handling
As a developer, I want untranslated strings to fall back to English so that players always see something.
US-10.4: Dynamic Content
As a builder, I want to create translatable room descriptions so that my content works in all languages.
10.3 Technical Requirements¶
10.3.1 Architecture¶
packages/maid-engine/src/maid_engine/
├── i18n/
│ ├── __init__.py
│ ├── catalog.py # Translation catalog
│ ├── extractor.py # Message extraction
│ ├── loader.py # Translation file loading
│ ├── translator.py # Runtime translation
│ └── middleware.py # Session language middleware
packages/maid-engine/
├── locales/
│ ├── en/
│ │ └── LC_MESSAGES/
│ │ └── messages.po
│ ├── es/
│ │ └── LC_MESSAGES/
│ │ └── messages.po
│ ├── de/
│ │ └── LC_MESSAGES/
│ │ └── messages.po
│ └── ... (additional languages)
10.3.2 Translation Catalog¶
# packages/maid-engine/src/maid_engine/i18n/catalog.py
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
import gettext
@dataclass
class TranslationEntry:
"""Single translation entry."""
msgid: str
msgstr: str
msgid_plural: str | None = None
msgstr_plural: list[str] | None = None
context: str | None = None
locations: list[tuple[str, int]] = field(default_factory=list)
flags: list[str] = field(default_factory=list)
translator_comments: list[str] = field(default_factory=list)
@dataclass
class TranslationCatalog:
"""Collection of translations for a language."""
locale: str
domain: str = "messages"
entries: dict[str, TranslationEntry] = field(default_factory=dict)
metadata: dict[str, str] = field(default_factory=dict)
_gettext: gettext.GNUTranslations | None = None
@classmethod
def load(cls, path: Path, locale: str, domain: str = "messages") -> "TranslationCatalog":
"""Load catalog from .po or .mo file."""
catalog = cls(locale=locale, domain=domain)
po_path = path / locale / "LC_MESSAGES" / f"{domain}.po"
mo_path = path / locale / "LC_MESSAGES" / f"{domain}.mo"
if mo_path.exists():
with open(mo_path, "rb") as f:
catalog._gettext = gettext.GNUTranslations(f)
elif po_path.exists():
catalog._parse_po(po_path)
return catalog
def _parse_po(self, path: Path) -> None:
"""Parse a .po file into entries."""
content = path.read_text(encoding="utf-8")
# Parse PO file format
# ... implementation details ...
def translate(
self,
msgid: str,
*,
context: str | None = None,
n: int | None = None,
) -> str:
"""Translate a message."""
if self._gettext:
if context:
return self._gettext.pgettext(context, msgid)
elif n is not None:
return self._gettext.ngettext(msgid, msgid, n)
return self._gettext.gettext(msgid)
# Fallback to entries dict
key = f"{context}\x04{msgid}" if context else msgid
entry = self.entries.get(key)
if entry and entry.msgstr:
if n is not None and entry.msgstr_plural:
idx = 0 if n == 1 else 1
return entry.msgstr_plural[idx] if idx < len(entry.msgstr_plural) else entry.msgstr
return entry.msgstr
return msgid # Fallback to original
10.3.3 Translator Service¶
# packages/maid-engine/src/maid_engine/i18n/translator.py
from functools import lru_cache
from contextvars import ContextVar
from typing import Callable
# Current locale context variable
current_locale: ContextVar[str] = ContextVar("current_locale", default="en")
class Translator:
"""Main translation service."""
SUPPORTED_LOCALES = [
"en", # English (default)
"es", # Spanish
"de", # German
"fr", # French
"pt", # Portuguese
"ru", # Russian
"zh", # Chinese (Simplified)
"ja", # Japanese
"ko", # Korean
"it", # Italian
"pl", # Polish
]
def __init__(self, locales_path: Path) -> None:
self._locales_path = locales_path
self._catalogs: dict[str, TranslationCatalog] = {}
self._fallback_locale = "en"
self._missing_handler: Callable[[str, str], None] | None = None
def load_locale(self, locale: str) -> None:
"""Load translations for a locale."""
if locale not in self.SUPPORTED_LOCALES:
raise ValueError(f"Unsupported locale: {locale}")
catalog = TranslationCatalog.load(
self._locales_path,
locale,
domain="messages"
)
self._catalogs[locale] = catalog
def load_all(self) -> None:
"""Load all supported locales."""
for locale in self.SUPPORTED_LOCALES:
locale_path = self._locales_path / locale
if locale_path.exists():
self.load_locale(locale)
def translate(
self,
msgid: str,
*,
locale: str | None = None,
context: str | None = None,
n: int | None = None,
**kwargs,
) -> str:
"""Translate a message.
Args:
msgid: Message ID (original English text)
locale: Target locale (uses context var if not specified)
context: Message context for disambiguation
n: Count for pluralization
**kwargs: Format string arguments
"""
locale = locale or current_locale.get()
# Try target locale
catalog = self._catalogs.get(locale)
if catalog:
result = catalog.translate(msgid, context=context, n=n)
if result != msgid:
return result.format(**kwargs) if kwargs else result
# Fallback to default locale
if locale != self._fallback_locale:
fallback = self._catalogs.get(self._fallback_locale)
if fallback:
result = fallback.translate(msgid, context=context, n=n)
if result != msgid:
return result.format(**kwargs) if kwargs else result
# Report missing translation
if self._missing_handler and locale != self._fallback_locale:
self._missing_handler(locale, msgid)
# Return original with formatting
return msgid.format(**kwargs) if kwargs else msgid
def set_missing_handler(
self,
handler: Callable[[str, str], None],
) -> None:
"""Set handler for missing translations."""
self._missing_handler = handler
def get_available_locales(self) -> list[str]:
"""Get list of loaded locales."""
return list(self._catalogs.keys())
# Global translator instance
_translator: Translator | None = None
def get_translator() -> Translator:
"""Get the global translator instance."""
global _translator
if _translator is None:
raise RuntimeError("Translator not initialized")
return _translator
def init_translator(locales_path: Path) -> Translator:
"""Initialize the global translator."""
global _translator
_translator = Translator(locales_path)
_translator.load_all()
return _translator
# Convenience functions
def _(msgid: str, **kwargs) -> str:
"""Translate a message using current locale."""
return get_translator().translate(msgid, **kwargs)
def _n(msgid: str, msgid_plural: str, n: int, **kwargs) -> str:
"""Translate a pluralized message."""
return get_translator().translate(msgid, n=n, **kwargs)
def _p(context: str, msgid: str, **kwargs) -> str:
"""Translate a message with context."""
return get_translator().translate(msgid, context=context, **kwargs)
10.3.4 Message Extraction¶
# packages/maid-engine/src/maid_engine/i18n/extractor.py
import ast
from dataclasses import dataclass
from pathlib import Path
@dataclass
class ExtractedMessage:
"""Extracted translatable message."""
msgid: str
locations: list[tuple[str, int]]
context: str | None = None
plural: str | None = None
comments: list[str] = None
class MessageExtractor:
"""Extracts translatable strings from Python source."""
# Functions that mark strings as translatable
GETTEXT_FUNCTIONS = {
"_": {"msgid": 0},
"_n": {"msgid": 0, "plural": 1},
"_p": {"context": 0, "msgid": 1},
"gettext": {"msgid": 0},
"ngettext": {"msgid": 0, "plural": 1},
"pgettext": {"context": 0, "msgid": 1},
}
def __init__(self) -> None:
self._messages: dict[str, ExtractedMessage] = {}
def extract_file(self, path: Path) -> list[ExtractedMessage]:
"""Extract messages from a Python file."""
content = path.read_text()
tree = ast.parse(content)
messages = []
for node in ast.walk(tree):
if isinstance(node, ast.Call):
message = self._extract_call(node, str(path))
if message:
messages.append(message)
return messages
def extract_directory(self, path: Path, pattern: str = "**/*.py") -> list[ExtractedMessage]:
"""Extract messages from all Python files in directory."""
for py_file in path.glob(pattern):
for msg in self.extract_file(py_file):
key = f"{msg.context}\x04{msg.msgid}" if msg.context else msg.msgid
if key in self._messages:
self._messages[key].locations.extend(msg.locations)
else:
self._messages[key] = msg
return list(self._messages.values())
def _extract_call(self, node: ast.Call, filename: str) -> ExtractedMessage | None:
"""Extract message from a function call."""
func_name = None
if isinstance(node.func, ast.Name):
func_name = node.func.id
elif isinstance(node.func, ast.Attribute):
func_name = node.func.attr
if func_name not in self.GETTEXT_FUNCTIONS:
return None
spec = self.GETTEXT_FUNCTIONS[func_name]
# Extract arguments
msgid = None
context = None
plural = None
for arg_name, arg_idx in spec.items():
if arg_idx < len(node.args):
arg = node.args[arg_idx]
if isinstance(arg, ast.Constant) and isinstance(arg.value, str):
if arg_name == "msgid":
msgid = arg.value
elif arg_name == "context":
context = arg.value
elif arg_name == "plural":
plural = arg.value
if not msgid:
return None
return ExtractedMessage(
msgid=msgid,
locations=[(filename, node.lineno)],
context=context,
plural=plural,
)
def write_pot(self, messages: list[ExtractedMessage], output: Path) -> None:
"""Write extracted messages to a .pot template file."""
with open(output, "w", encoding="utf-8") as f:
# Header
f.write('# MAID Translation Template\n')
f.write('msgid ""\n')
f.write('msgstr ""\n')
f.write('"Content-Type: text/plain; charset=UTF-8\\n"\n')
f.write('"Content-Transfer-Encoding: 8bit\\n"\n')
f.write('\n')
# Messages
for msg in messages:
# Locations
for loc in msg.locations:
f.write(f'#: {loc[0]}:{loc[1]}\n')
# Context
if msg.context:
f.write(f'msgctxt "{self._escape(msg.context)}"\n')
# Message ID
f.write(f'msgid "{self._escape(msg.msgid)}"\n')
# Plural
if msg.plural:
f.write(f'msgid_plural "{self._escape(msg.plural)}"\n')
f.write('msgstr[0] ""\n')
f.write('msgstr[1] ""\n')
else:
f.write('msgstr ""\n')
f.write('\n')
def _escape(self, s: str) -> str:
"""Escape string for PO file."""
return s.replace('\\', '\\\\').replace('"', '\\"').replace('\n', '\\n')
10.4 Translation File Format¶
# locales/es/LC_MESSAGES/messages.po
# Spanish translation for MAID
# Copyright (C) 2025 MAID Development Team
# This file is distributed under the same license as MAID.
#
msgid ""
msgstr ""
"Project-Id-Version: MAID 0.1.0\n"
"Report-Msgid-Bugs-To: translations@maid.dev\n"
"POT-Creation-Date: 2025-01-15 10:00+0000\n"
"PO-Revision-Date: 2025-01-20 14:30+0000\n"
"Last-Translator: Translator Name <translator@example.com>\n"
"Language-Team: Spanish <es@li.org>\n"
"Language: es\n"
"MIME-Version: 1.0\n"
"Content-Type: text/plain; charset=UTF-8\n"
"Content-Transfer-Encoding: 8bit\n"
"Plural-Forms: nplurals=2; plural=(n != 1);\n"
#: maid_engine/net/telnet/login.py:45
msgid "Welcome to {game_name}!"
msgstr "¡Bienvenido a {game_name}!"
#: maid_engine/net/telnet/login.py:50
msgid "Enter your username:"
msgstr "Ingrese su nombre de usuario:"
#: maid_engine/net/telnet/login.py:55
msgid "Enter your password:"
msgstr "Ingrese su contraseña:"
#: maid_engine/net/telnet/login.py:70
msgid "Invalid username or password."
msgstr "Usuario o contraseña inválidos."
#: maid_stdlib/commands/look.py:30
msgid "You see nothing special."
msgstr "No ves nada especial."
#: maid_stdlib/commands/movement.py:40
msgid "You can't go that way."
msgstr "No puedes ir por ahí."
#: maid_stdlib/systems/combat.py:100
#, python-format
msgid "You hit {target} for {damage} damage!"
msgstr "¡Golpeas a {target} por {damage} puntos de daño!"
#: maid_stdlib/systems/combat.py:110
msgid "{attacker} hits you for {damage} damage!"
msgstr "¡{attacker} te golpea por {damage} puntos de daño!"
#: maid_classic_rpg/systems/leveling.py:50
msgid "Congratulations! You have reached level {level}!"
msgstr "¡Felicitaciones! ¡Has alcanzado el nivel {level}!"
#: maid_stdlib/utils/inventory.py:80
#, python-format
msgid "You have {count} item."
msgid_plural "You have {count} items."
msgstr[0] "Tienes {count} objeto."
msgstr[1] "Tienes {count} objetos."
10.5 API/Interface Design¶
# Usage in code
from maid_engine.i18n import _, _n, _p, current_locale
# Simple translation
message = _("Welcome to the game!")
# With format arguments
message = _("You hit {target} for {damage} damage!", target="Goblin", damage=15)
# Pluralization
items = _n(
"You have {count} item.",
"You have {count} items.",
n=item_count,
count=item_count,
)
# With context (for disambiguation)
# "bark" as in tree bark vs dog bark
message = _p("tree", "The bark is rough.")
# Set locale for session
current_locale.set("es")
# CLI Commands
@app.command()
def extract_messages(
output: Path = typer.Option(Path("locales/messages.pot"), help="Output file"),
packages: list[str] = typer.Option(["maid_engine", "maid_stdlib", "maid_classic_rpg"]),
):
"""Extract translatable messages from source code."""
...
@app.command()
def compile_translations(
locales: list[str] = typer.Option(None, help="Specific locales to compile"),
):
"""Compile .po files to .mo files."""
...
# In-game commands
async def cmd_language(ctx: CommandContext) -> bool:
"""@language - Set your preferred language.
Usage:
@language # Show current language
@language list # List available languages
@language es # Set language to Spanish
"""
...
10.6 Translatable Content¶
For dynamic game content (room descriptions, item names, etc.), MAID uses a content translation system:
# packages/maid-engine/src/maid_engine/i18n/content.py
from dataclasses import dataclass
@dataclass
class TranslatableText:
"""Text with translations for multiple locales."""
default: str
translations: dict[str, str] = field(default_factory=dict)
def get(self, locale: str | None = None) -> str:
"""Get text for locale with fallback."""
locale = locale or current_locale.get()
return self.translations.get(locale, self.default)
def set(self, locale: str, text: str) -> None:
"""Set translation for locale."""
self.translations[locale] = text
# Usage in components
@dataclass
class DescriptionComponent:
name: TranslatableText
description: TranslatableText
Building commands:
# Building command
@dig north = "Forest Path"
@set here/name.es = "Sendero del Bosque"
@set here/description.es = "Un sendero serpenteante a través del bosque."
10.7 Configuration¶
class I18nSettings(BaseSettings):
"""Internationalization configuration."""
model_config = SettingsConfigDict(env_prefix="MAID_I18N_")
enabled: bool = True
default_locale: str = "en"
fallback_locale: str = "en"
# Paths
locales_dir: str = "locales/"
# Behavior
log_missing: bool = True
missing_placeholder: str = "[{locale}:{msgid}]"
auto_detect: bool = True # Detect from client
# Supported locales (empty = all available)
supported_locales: list[str] = []
Environment Variables:
MAID_I18N_ENABLED=true
MAID_I18N_DEFAULT_LOCALE=en
MAID_I18N_FALLBACK_LOCALE=en
MAID_I18N_LOCALES_DIR=locales/
MAID_I18N_LOG_MISSING=true
MAID_I18N_AUTO_DETECT=true
MAID_I18N_SUPPORTED_LOCALES=["en", "es", "de", "fr"]
10.8 Dependencies¶
[project.optional-dependencies]
i18n = [
"babel>=2.14.0", # Message extraction and formatting
"polib>=1.2.0", # PO file manipulation
"lingua>=4.15.0", # Advanced extraction (optional)
]
10.9 Message Extraction Workflow¶
# 1. Extract messages from source
maid i18n extract --output=locales/messages.pot
# 2. Initialize new language
maid i18n init es # Creates locales/es/LC_MESSAGES/messages.po
# 3. Update existing translations
maid i18n update # Merges new messages into existing .po files
# 4. Translate messages (edit .po files)
# Use any PO editor: Poedit, Lokalize, web tools, etc.
# 5. Compile translations
maid i18n compile # Creates .mo files from .po files
# 6. Verify translations
maid i18n check # Reports missing/fuzzy translations
10.10 Implementation Tasks¶
- [ ] Phase 1: Core Infrastructure (Week 1)
- [ ] Implement TranslationCatalog
- [ ] Implement Translator service
- [ ] Create convenience functions (_, _n, _p)
-
[ ] Add current_locale context var
-
[ ] Phase 2: Extraction & Compilation (Week 2)
- [ ] Implement MessageExtractor
- [ ] Add POT file generation
- [ ] Add PO file parsing
- [ ] Add MO file compilation
-
[ ] Create CLI commands
-
[ ] Phase 3: Runtime Integration (Week 2-3)
- [ ] Add session locale middleware
- [ ] Integrate with commands
- [ ] Add @language command
-
[ ] Handle TranslatableText in components
-
[ ] Phase 4: Content & Polish (Week 3-4)
- [ ] Create initial English messages.pot
- [ ] Add Spanish translation
- [ ] Add German translation
- [ ] Documentation
- [ ] Add translation percentage tracking
10.11 Testing Requirements¶
@pytest.fixture
def translator(tmp_path):
"""Create translator with test translations."""
# Create test PO file
es_dir = tmp_path / "es" / "LC_MESSAGES"
es_dir.mkdir(parents=True)
(es_dir / "messages.po").write_text('''
msgid "Hello"
msgstr "Hola"
msgid "You have {count} item."
msgid_plural "You have {count} items."
msgstr[0] "Tienes {count} objeto."
msgstr[1] "Tienes {count} objetos."
''')
t = Translator(tmp_path)
t.load_locale("es")
return t
def test_simple_translation(translator):
"""Test simple message translation."""
assert translator.translate("Hello", locale="es") == "Hola"
def test_fallback_to_english(translator):
"""Test fallback when translation missing."""
assert translator.translate("Unknown message", locale="es") == "Unknown message"
def test_plural_translation(translator):
"""Test pluralization."""
assert translator.translate("You have {count} item.", locale="es", n=1, count=1) == "Tienes 1 objeto."
assert translator.translate("You have {count} item.", locale="es", n=5, count=5) == "Tienes 5 objetos."
def test_format_arguments(translator):
"""Test format string replacement."""
result = translator.translate("Hello {name}!", locale="en", name="World")
assert result == "Hello World!"
@pytest.mark.asyncio
async def test_session_locale(test_session, translator):
"""Test locale follows session."""
test_session.locale = "es"
async with session_locale_context(test_session):
assert _("Hello") == "Hola"
10.12 Acceptance Criteria¶
- Translation Loading
- [ ] PO files load correctly
- [ ] MO files load correctly
-
[ ] Missing files don't crash
-
Message Translation
- [ ] Simple messages translate
- [ ] Format arguments work
- [ ] Pluralization works
-
[ ] Context disambiguation works
-
Fallback Handling
- [ ] Missing translations fall back to default
- [ ] Missing locales fall back to English
-
[ ] Missing translations are logged
-
Runtime Integration
- [ ] Session locale works
- [ ] @language command works
-
[ ] Messages throughout game translate
-
Tooling
- [ ] Message extraction works
- [ ] PO file generation works
- [ ] MO compilation works
- [ ] Update preserves existing translations
11. Implementation Roadmap¶
11.1 Phase Overview¶
| Phase | Duration | Focus | Deliverables |
|---|---|---|---|
| Phase 1 | Weeks 1-4 | Foundation | Web Admin Backend, Building Commands |
| Phase 2 | Weeks 5-8 | Core Tools | Hot Reload, Profiling, i18n |
| Phase 3 | Weeks 9-12 | UX Tools | MaidEditor, MaidMenu, MaidTable, Batch |
| Phase 4 | Weeks 13-14 | Polish | Testing, Documentation, Integration |
11.2 Detailed Timeline¶
Week 1-2: Web Admin Backend Foundation
├── Admin API authentication
├── Dashboard endpoints
├── Entity CRUD API
└── WebSocket infrastructure
Week 3-4: Building Commands
├── @create, @destroy, @dig
├── @set, @describe, @examine
├── @teleport, @goto, @find
└── Target resolution system
Week 5-6: Hot Reload System
├── Module reloader
├── System hot-swap
├── Content pack reload
└── File watcher
Week 7-8: Profiling Tools
├── Memory profiling
├── Query profiling
├── Tick timing
└── Report generation
Week 9-10: i18n Infrastructure
├── Translation catalog
├── Message extraction
├── Runtime translation
└── Initial translations
Week 11: MaidEditor
├── Buffer and undo
├── Command mode
├── Insert mode
└── @edit command
Week 12: MaidMenu & MaidTable
├── Menu navigation
├── Input handling
├── Table formatting
└── Integration
Week 13: Batch Processor
├── Command batches
├── Code batches
├── Mixed batches
└── CLI tools
Week 14: Integration & Polish
├── Web Admin Frontend
├── Documentation
├── Integration tests
└── Performance tuning
11.3 Dependencies Between Features¶
┌─────────────────┐
│ maid-engine │
│ (core APIs) │
└────────┬────────┘
│
┌────────────────────┼────────────────────┐
│ │ │
▼ ▼ ▼
┌───────────────┐ ┌─────────────────┐ ┌──────────────┐
│ Web Admin │ │ Hot Reload │ │ i18n │
│ Backend │ │ System │ │ Support │
└───────┬───────┘ └────────┬────────┘ └──────┬───────┘
│ │ │
│ ┌────────┴────────┐ │
│ │ │ │
▼ ▼ ▼ ▼
┌───────────────┐ ┌─────────────┐ ┌──────────────────┐
│ Profiling │ │ Building │ │ MaidEditor │
│ Tools │ │ Commands │ │ MaidMenu │
└───────────────┘ └─────────────┘ │ MaidTable │
└──────────────────┘
│
▼
┌───────────────┐
│ Batch │
│ Processor │
└───────────────┘
11.4 Resource Requirements¶
| Role | Allocation | Focus Areas |
|---|---|---|
| Backend Developer | 1 FTE | Web Admin API, Hot Reload, Profiling |
| Frontend Developer | 0.5 FTE | Web Admin UI (React) |
| Systems Developer | 1 FTE | Building Commands, Batch Processor |
| Infrastructure | 0.5 FTE | i18n, MaidEditor/Menu/Table |
| QA Engineer | 0.5 FTE | Testing, Documentation |
11.5 Risk Mitigation¶
| Risk | Probability | Impact | Mitigation |
|---|---|---|---|
| Hot reload complexity | High | High | Start with module reload, defer system hot-swap |
| Web admin scope creep | Medium | Medium | Fixed feature set, v2 for enhancements |
| i18n message count | Medium | Low | Extract incrementally, prioritize UI messages |
| Performance impact of profiling | Low | Medium | Make profiling opt-in, use sampling |
11.6 Success Metrics¶
| Metric | Target | Measurement |
|---|---|---|
| Web Admin Uptime | 99.9% | Monitoring |
| Hot Reload Success Rate | >95% | Error logging |
| Building Command Coverage | 25+ commands | Feature count |
| Profiling Overhead | <5% | Benchmarking |
| Translation Coverage | 80% for 3 languages | Automated check |
| Test Coverage | >80% | pytest-cov |
Appendices¶
Appendix A: Complete Command Reference¶
A.1 Building Commands¶
| Command | Syntax | Description |
|---|---|---|
@create |
@create <type> <name> [= template] |
Create entity |
@spawn |
@spawn <template> [count] [at location] |
Spawn from template |
@destroy |
@destroy <target> |
Delete entity |
@purge |
@purge <filter> |
Mass delete |
@dig |
@dig <direction> [= name] |
Create room and exit |
@tunnel |
@tunnel <direction> <to_room> |
Create bidirectional passage |
@describe |
@describe <target> [text] |
Set description |
@name |
@name <target> <new_name> |
Rename entity |
@examine |
@examine <target> [component] |
Inspect entity |
@stat |
@stat <target> |
Show statistics |
@set |
@set <target>/<attr> = <value> |
Set attribute |
@attribute |
@attribute <target> <action> <attr> |
Manage attributes |
@link |
@link <exit> <destination> |
Link exit |
@unlink |
@unlink <exit> |
Remove exit link |
@teleport |
@teleport <target> <destination> |
Move entity |
@goto |
@goto <location> |
Move self |
@find |
@find <type> [filter] |
Search entities |
@search |
@search <query> |
Complex search |
@copy |
@copy <target> [name] |
Clone entity |
@lock |
@lock <target> <lock> |
Set lock |
@unlock |
@unlock <target> |
Remove lock |
@zone |
@zone <action> [args] |
Manage zones |
@flag |
@flag <target> <+/-flag> |
Set/unset flags |
@component |
@component <target> <action> <type> |
Manage components |
@script |
@script <target> <script> |
Attach script |
@trigger |
@trigger <target> <action> |
Manage triggers |
@reset |
@reset <target> [config] |
Configure reset |
@wipe |
@wipe [zone] |
Clear non-persistent |
A.2 Admin Commands¶
| Command | Syntax | Description |
|---|---|---|
@reload |
@reload <module/pack/system/all> |
Hot reload |
@rollback |
@rollback [snapshot] |
Rollback reload |
@profile |
@profile <start/stop/status> |
Profiling |
@memory |
@memory [top/systems/compare] |
Memory stats |
@timing |
@timing [systems/slow/history] |
Tick timing |
@batch |
@batch <file> |
Execute batch |
@language |
@language [locale] |
Set language |
@edit |
@edit <target>/<field> |
Text editor |
Appendix B: Configuration Reference¶
# Complete settings example
class AdminDeveloperSettings(BaseSettings):
"""All admin and developer tool settings."""
# Web Admin
admin: AdminSettings = Field(default_factory=AdminSettings)
# Building
building: BuildingSettings = Field(default_factory=BuildingSettings)
# Hot Reload
reload: ReloadSettings = Field(default_factory=ReloadSettings)
# Profiling
profiling: ProfilingSettings = Field(default_factory=ProfilingSettings)
# i18n
i18n: I18nSettings = Field(default_factory=I18nSettings)
# Editor
editor: EditorSettings = Field(default_factory=EditorSettings)
# Batch
batch: BatchSettings = Field(default_factory=BatchSettings)
Appendix C: API Endpoint Summary¶
| Endpoint | Method | Description |
|---|---|---|
/admin/dashboard/ |
GET | Dashboard data |
/admin/dashboard/ws |
WS | Real-time updates |
/admin/entities/ |
GET | List entities |
/admin/entities/{id} |
GET/PUT/DELETE | Entity CRUD |
/admin/entities/{id}/components/{type} |
PUT/DELETE | Component CRUD |
/admin/players/ |
GET | List players |
/admin/players/{id} |
GET/PUT | Player details |
/admin/players/{id}/ban |
POST | Ban player |
/admin/players/{id}/unban |
POST | Unban player |
/admin/players/{id}/kick |
POST | Kick player |
/admin/world/rooms |
GET/POST | Room management |
/admin/world/rooms/{id} |
GET/PUT/DELETE | Room CRUD |
/admin/world/exits |
POST | Create exit |
/admin/world/graph |
GET | World visualization |
/admin/logs/ |
GET | Search logs |
/admin/logs/stream |
WS | Stream logs |
/admin/config/ |
GET | All config |
/admin/config/{section} |
GET/PUT | Section config |
/admin/packs/ |
GET | Content packs |
/admin/reload/ |
POST | Trigger reload |
/admin/profile/ |
GET/POST | Profiling |
Appendix D: Localization Statistics¶
| Language | Code | Plural Forms | Status |
|---|---|---|---|
| English | en | 2 | Default |
| Spanish | es | 2 | Planned |
| German | de | 2 | Planned |
| French | fr | 2 | Planned |
| Portuguese | pt | 2 | Planned |
| Russian | ru | 3 | Planned |
| Chinese (Simplified) | zh | 1 | Planned |
| Japanese | ja | 1 | Planned |
| Korean | ko | 1 | Planned |
| Italian | it | 2 | Planned |
| Polish | pl | 3 | Planned |
Appendix E: Glossary¶
| Term | Definition |
|---|---|
| Content Pack | Pluggable module providing game content (systems, commands, entities) |
| ECS | Entity Component System - architecture pattern for game objects |
| Hot Reload | Updating code at runtime without server restart |
| i18n | Internationalization - infrastructure for multi-language support |
| L10n | Localization - actual translation to specific languages |
| MO File | Machine Object - compiled binary translation file |
| PO File | Portable Object - human-readable translation file |
| POT File | PO Template - source file for translations |
| Tick | Single iteration of the game loop (default: 4 per second) |
Document History¶
| Version | Date | Author | Changes |
|---|---|---|---|
| 1.0.0 | 2025-01-30 | MAID Team | Initial draft |
This document is part of the MAID Engine specification. For questions or contributions, please refer to the main MAID repository.