"""
Main game engine with game logic.
This module contains the GameEngine class which implements all core game
mechanics and business logic for the MUD server. It acts as the interface
between the API routes and the underlying world/database systems.
The engine handles:
- Player authentication and sessions
- Movement between rooms
- Inventory management (pickup/drop items)
- Chat systems (say, yell, whisper)
- Room observation and interaction
- Player presence and status
Design Pattern:
The GameEngine uses the Facade pattern, providing a simplified interface
to the complex subsystems (World, Database). API routes call engine methods
which coordinate between the world data and database operations.
Architecture:
API Routes → GameEngine → World + Database
- Routes validate sessions and call engine methods
- Engine implements game logic and coordinates subsystems
- World provides static game data (rooms, items)
- Database provides persistent player state
"""
import html
import logging
from typing import Any, cast
from mud_server.core.bus import MudBus
from mud_server.core.events import Events
from mud_server.core.world import World
from mud_server.core.world_registry import WorldRegistry
from mud_server.db import facade as database
from mud_server.services.policy.types import AxisPolicyValidationReport
logger = logging.getLogger(__name__)
def _get_bus() -> MudBus:
"""
Get the current event bus singleton.
This function exists because of Python's module import system. If we imported
`bus` at module level, tests that call `reset_for_testing()` would orphan that
reference - the engine would emit to the old bus while tests check the new one.
By calling `MudBus()` at runtime, we always get the current singleton, even
after test resets.
Returns:
The current MudBus singleton instance
"""
return MudBus()
[docs]
def sanitize_chat_message(message: str) -> str:
"""
Sanitize a chat message to prevent XSS attacks.
Escapes HTML special characters to prevent injection of malicious
scripts through chat messages. This is critical for security when
messages are displayed in web interfaces.
Args:
message: Raw message text from user input
Returns:
Sanitized message with HTML entities escaped
Example:
>>> sanitize_chat_message("Hello <script>alert('xss')</script>")
"Hello <script>alert('xss')</script>"
"""
return html.escape(message)
[docs]
class GameEngine:
"""
Main game engine managing all game logic and mechanics.
This class is instantiated once at server startup and handles all game
operations. It coordinates between the World (static game data) and the
Database (dynamic player state) to implement game mechanics.
Attributes:
world_registry: WorldRegistry instance for loading worlds on demand
Responsibilities:
- Player login/logout with authentication
- Movement validation and execution
- Inventory operations (pickup, drop, view)
- Chat message handling (say, yell, whisper)
- Room descriptions and observation
- Player status and presence queries
Design Notes:
- All methods return (success: bool, message: str) tuples for API responses
- Database operations are called directly (no repository pattern)
- Broadcasting to other players is stubbed (not yet implemented)
- All game state is persisted to database immediately
"""
def __init__(self):
"""
Initialize the game engine.
Loads the world data from JSON and initializes the database schema.
This is called once when the server starts.
Side Effects:
- Loads world_data.json into memory
- Creates database tables if they don't exist
- Creates default superuser if no players exist
"""
# Load world registry (worlds are lazy-loaded per request)
self.world_registry = WorldRegistry()
# Initialize database schema (creates tables, default admin)
database.init_database()
# Load and validate axis policies, then seed registry tables.
# This ensures policy readiness is surfaced at startup, and the
# database mirrors the world-defined axis vocabulary.
self._bootstrap_axis_policies()
def _bootstrap_axis_policies(self) -> None:
"""
Load world axis policies, emit validation reports, and seed registry tables.
This is a startup-only routine. It validates canonical activation-driven
axis policy readiness and keeps the axis registry tables in sync with
the effective canonical bundle.
"""
import logging
from mud_server.services import policy_service
logger = logging.getLogger(__name__)
# Gather all worlds (including inactive) so policy issues are visible.
worlds = self.world_registry.list_worlds(include_inactive=True)
if not worlds:
logger.warning("Axis policy bootstrap skipped: no worlds registered.")
return
for world in worlds:
world_id = world.get("id")
if not world_id:
# Defensive guard: malformed world rows should not block startup.
logger.warning("Axis policy bootstrap skipped malformed world row: %s", world)
continue
try:
resolved_bundle = policy_service.resolve_effective_axis_bundle(
scope=policy_service.ActivationScope(world_id=world_id, client_profile="")
)
axes_payload = resolved_bundle.axes_payload
thresholds_payload = resolved_bundle.thresholds_payload
policy_hash = resolved_bundle.policy_hash
bundle_version = resolved_bundle.bundle_version
except policy_service.PolicyServiceError as exc:
if exc.code in {
"POLICY_EFFECTIVE_MANIFEST_NOT_FOUND",
"POLICY_EFFECTIVE_AXIS_BUNDLE_NOT_FOUND",
}:
logger.warning(
"Axis policy bootstrap skipped for %s: no effective canonical "
"axis/manifest activation was found. Runtime startup continues, but "
"policy-backed axis features stay unavailable until canonical state "
"is imported. Run "
"'mud-server import-policy-artifact --artifact-path <publish_manifest.json>' "
"or ensure init-db artifact bootstrap has completed for world %s "
"before runtime startup.",
world_id,
world_id,
)
continue
else:
logger.warning(
"Axis policy bootstrap skipped for %s: %s (%s)",
world_id,
exc.detail,
exc.code,
)
continue
payload, report = self._build_axis_policy_report_from_canonical_bundle(
world_id=world_id,
axes_payload=axes_payload,
thresholds_payload=thresholds_payload,
policy_hash=policy_hash,
version=bundle_version,
)
self._log_axis_policy_report(logger, report)
if not report.axes:
logger.warning("Axis registry seeding skipped for %s: no axes defined.", world_id)
continue
stats = database.seed_axis_registry(
world_id=world_id,
axes_payload=payload.get("axes") or {},
thresholds_payload=payload.get("thresholds") or {},
)
logger.info("Axis registry seeded for %s: %s", world_id, stats)
@staticmethod
def _build_axis_policy_report_from_canonical_bundle(
*,
world_id: str,
axes_payload: dict[str, Any],
thresholds_payload: dict[str, Any],
policy_hash: str,
version: str | None,
) -> tuple[dict[str, Any], AxisPolicyValidationReport]:
"""Build axis bootstrap payload/report from canonical DB policy content."""
axes_definitions = (
(axes_payload.get("axes") or {}) if isinstance(axes_payload, dict) else {}
)
threshold_definitions = (
(thresholds_payload.get("axes") or {}) if isinstance(thresholds_payload, dict) else {}
)
axes = [str(axis_name) for axis_name in axes_definitions.keys()]
ordering_definitions = {
str(axis_name): ordering
for axis_name, axis_data in axes_definitions.items()
if isinstance(axis_data, dict)
and isinstance((axis_data.get("ordering")), dict)
and (ordering := axis_data.get("ordering"))
}
ordering_present = list(ordering_definitions.keys())
thresholds_present = [str(axis_name) for axis_name in threshold_definitions.keys()]
missing_components: list[str] = []
if not axes:
missing_components.append("axes list missing or empty")
for axis_name in axes:
if axis_name not in ordering_present:
missing_components.append(f"ordering missing for axis: {axis_name}")
if axis_name not in thresholds_present:
missing_components.append(f"thresholds missing for axis: {axis_name}")
report = AxisPolicyValidationReport(
world_id=world_id,
axes=axes,
ordering_present=ordering_present,
ordering_definitions=ordering_definitions,
thresholds_present=thresholds_present,
thresholds_definitions=threshold_definitions,
missing_components=missing_components,
policy_hash=policy_hash,
version=version,
)
payload = {
"axes": axes_payload,
"thresholds": thresholds_payload,
}
return payload, report
@staticmethod
def _log_axis_policy_report(logger, report: AxisPolicyValidationReport) -> None:
"""
Log the axis policy validation report for a world.
Args:
logger: Logger instance for emission.
report: AxisPolicyValidationReport instance.
"""
version = report.version or "unversioned"
logger.info(
"Axis policy report for %s (version=%s hash=%s)",
report.world_id,
version,
report.policy_hash,
)
logger.info(
"Axis policy axes=%s ordering=%s thresholds=%s",
report.axes,
report.ordering_present,
report.thresholds_present,
)
if report.missing_components:
logger.warning(
"Axis policy missing components for %s: %s",
report.world_id,
"; ".join(report.missing_components),
)
def _get_world(self, world_id: str) -> "World":
"""
Resolve a world_id to a loaded World instance.
Raises:
ValueError: If world_id is missing or invalid.
"""
if not world_id:
raise ValueError("world_id is required for game operations")
return self.world_registry.get_world(world_id)
[docs]
def login(
self, username: str, password: str, session_id: str, client_type: str = "unknown"
) -> tuple[bool, str, str | None]:
"""
Handle account login with authentication and session creation.
This method validates the account and creates a session, but does not
select a character. Character selection happens separately.
"""
if not database.verify_password_for_user(username, password):
return False, "Invalid username or password.", None
if not database.is_user_active(username):
return (
False,
"This account has been deactivated. Please contact an administrator.",
None,
)
role = database.get_user_role(username)
if not role:
return False, "Failed to retrieve account information.", None
user_id = database.get_user_id(username)
if not user_id:
return False, "Failed to retrieve account information.", None
if not database.create_session(user_id, session_id, client_type=client_type):
return False, "Failed to create session.", None
message = "Login successful. Select a character to enter the world."
return True, message, role
[docs]
def logout(self, username: str) -> bool:
"""
Handle player logout by removing their session.
Removes the player's sessions from the database, effectively logging
them out on all devices.
Args:
username: Username of the player logging out
Returns:
True if session removed successfully, False otherwise
Side Effects:
- Removes session record from database sessions table
- Player will no longer appear in active players list
Note:
This removes all sessions for the user. If you want to remove a
single session (multi-device support), use database.remove_session_by_id().
"""
user_id = database.get_user_id(username)
if not user_id:
return False
return database.remove_sessions_for_user(user_id)
[docs]
def move(self, username: str, direction: str, *, world_id: str) -> tuple[bool, str]:
"""
Handle player movement between rooms.
Validates the move, updates player location in database, and generates
appropriate response messages. Also broadcasts movement notifications
to other players in the affected rooms (currently stubbed).
Movement Process:
1. Get player's current room from database
2. Check if move is valid (exit exists, destination valid)
3. Update player's room in database
4. Emit PLAYER_MOVED event (the move is now a fact)
5. Generate room description for new location
6. Broadcast departure message to old room
7. Broadcast arrival message to new room
Args:
username: Player attempting to move
direction: Direction to move ("north", "south", "east", "west")
Returns:
Tuple of (success, message)
- success: True if move succeeded, False otherwise
- message: New room description OR error message
Failure Cases:
- Player not in a valid room
- No exit in that direction
- Database update failed
Events Emitted:
- PLAYER_MOVED: When movement succeeds
- PLAYER_MOVE_FAILED: When movement fails
Example:
>>> engine.move("player1", "north")
(True, "You move north.\\n=== Enchanted Forest ===...")
>>> engine.move("player1", "west")
(False, "You cannot move west from here.")
"""
current_room = database.get_character_room(username, world_id=world_id)
if not current_room:
self._emit_move_failed(
username=username,
room_id=None,
direction=direction,
reason="Player not in a valid room",
)
return False, "You are not in a valid room."
world = self._get_world(world_id)
# Check if current room exists in the world
if not world.get_room(current_room):
self._emit_move_failed(
username=username,
room_id=current_room,
direction=direction,
reason="Room not in world data",
)
return False, "You are not in a valid room."
can_move, destination = world.can_move(current_room, direction)
if not can_move or destination is None:
self._emit_move_failed(
username=username,
room_id=current_room,
direction=direction,
reason=f"No exit {direction}",
)
return False, f"You cannot move {direction} from here."
# Update player room in database
if not database.set_character_room(username, destination, world_id=world_id):
self._emit_move_failed(
username=username,
room_id=current_room,
direction=direction,
reason="Database update failed",
)
return False, "Failed to move."
# =====================================================================
# MOVEMENT SUCCEEDED - Emit the event
# =====================================================================
# This is the point of no return. The player has moved.
# The event records this fact for any listeners.
_get_bus().emit(
Events.PLAYER_MOVED,
{
"username": username,
"from_room": current_room,
"to_room": destination,
"direction": direction,
},
)
# Get room description
room_desc = world.get_room_description(destination, username, world_id=world_id)
message = f"You move {direction}.\n{room_desc}"
# Movement state changes are authoritative on the event bus.
# This broadcast path remains only as a retained no-op compatibility seam.
self._emit_deprecated_room_broadcasts(
username=username,
departure_room=current_room,
departure_message=f"{username} leaves {direction}.",
destination_room=destination,
arrival_message=f"{username} arrives from {self._opposite_direction(direction)}.",
)
return True, message
[docs]
def recall(self, username: str, *, world_id: str) -> tuple[bool, str]:
"""
Recall player to their current zone's spawn point.
This is the "hearthstone" equivalent - a way for players to return
to a known safe location. The destination is the spawn_room defined
in the player's current zone.
If the player is not in a known zone, they are sent to the world's
default spawn point instead.
Args:
username: Player recalling
Returns:
Tuple of (success, message)
- success: True if recall succeeded
- message: Description of new location
Side Effects:
- Updates player's current_room in database
- Broadcasts departure/arrival messages (when implemented)
"""
current_room = database.get_character_room(username, world_id=world_id)
world = self._get_world(world_id)
# Find which zone the player is in
current_zone = None
for _zone_id, zone in world.zones.items():
if current_room in zone.rooms:
current_zone = zone
break
# Determine destination
if current_zone:
destination = current_zone.spawn_room
zone_name = current_zone.name
else:
# Not in a known zone - use world default
_zone_id, destination = world.default_spawn
zone_name = "the world"
# Check if already at spawn
if current_room == destination:
return True, "You are already at the spawn point."
# Update player location
if not database.set_character_room(username, destination, world_id=world_id):
return False, "Failed to recall."
self._emit_deprecated_room_broadcasts(
username=username,
departure_room=current_room,
departure_message=f"{username} vanishes in a puff of smoke.",
destination_room=destination,
arrival_message=f"{username} appears in a puff of smoke.",
)
# Generate response
room_desc = world.get_room_description(destination, username, world_id=world_id)
message = f"You recall to {zone_name}'s spawn point.\n{room_desc}"
return True, message
def _resolve_channel_ipc_hash(
self,
*,
world: World,
speaker_name: str,
channel: str,
world_id: str,
room_id: str | None = None,
listener_name: str | None = None,
) -> str | None:
"""Resolve one chat interaction to an ipc_hash when an axis engine is available."""
axis_engine = world.get_axis_engine()
if axis_engine is None:
return None
resolved_listener = listener_name
if resolved_listener is None and room_id is not None:
co_present = [
name
for name in database.get_characters_in_room(room_id, world_id=world_id)
if name != speaker_name
]
if co_present:
resolved_listener = co_present[0]
if resolved_listener is None:
return None
try:
resolution = axis_engine.resolve_chat_interaction(
speaker_name=speaker_name,
listener_name=resolved_listener,
channel=channel,
world_id=world_id,
)
return cast(str | None, resolution.ipc_hash)
except Exception:
logger.warning(
"Axis resolution failed for %s (speaker=%r, listener=%r) - "
"continuing without ipc_hash.",
channel,
speaker_name,
resolved_listener,
exc_info=True,
)
return None
def _translate_and_sanitize_chat(
self,
*,
world: World,
character_name: str,
message: str,
channel: str,
ipc_hash: str | None,
) -> str:
"""Translate one OOC message when possible, then sanitize the final text."""
translation_service = world.get_translation_service()
final_message = message
if translation_service is not None:
ic_text = translation_service.translate(
character_name=character_name,
ooc_message=message,
channel=channel,
ipc_hash=ipc_hash,
)
if ic_text is not None:
final_message = ic_text
return sanitize_chat_message(final_message)
def _emit_deprecated_room_broadcasts(
self,
*,
username: str,
destination_room: str,
arrival_message: str,
departure_room: str | None = None,
departure_message: str | None = None,
) -> None:
"""Route movement compatibility calls through the deprecated broadcast seam."""
if departure_room and departure_message:
self._broadcast_to_room(departure_room, departure_message, exclude=username)
self._broadcast_to_room(destination_room, arrival_message, exclude=username)
def _emit_move_failed(
self,
*,
username: str,
direction: str,
reason: str,
room_id: str | None,
) -> None:
"""Emit the canonical failed-movement event payload."""
_get_bus().emit(
Events.PLAYER_MOVE_FAILED,
{
"username": username,
"room": room_id,
"direction": direction,
"reason": reason,
},
)
[docs]
def chat(self, username: str, message: str, *, world_id: str) -> tuple[bool, str]:
"""
Handle player chat messages within their current room.
Sends a chat message to all players in the same room. The message is
stored in the database and can be retrieved by other players in the room.
Args:
username: Player sending the message
message: Chat message text
Returns:
Tuple of (success, message)
- success: True if message sent, False otherwise
- message: Confirmation message OR error message
Failure Cases:
- Player not in a valid room
- Database insert failed
Side Effects:
- Adds message to chat_messages table with room association
- Message will appear in other players' chat history
Example:
>>> engine.chat("player1", "Hello everyone!")
(True, "You say: Hello everyone!")
Security Note:
Messages are sanitized to prevent XSS attacks before storage.
"""
room = database.get_character_room(username, world_id=world_id)
if not room:
return False, "You are not in a valid room."
world = self._get_world(world_id)
# ── Axis resolution ───────────────────────────────────────────────────
# Compute demeanor/health deltas before translation so the ipc_hash
# produced here can be forwarded to the translation service (enabling
# deterministic rendering and ledger linkage).
#
# Listener selection for say: first co-present character who is not
# the speaker. If the room is empty (solo), resolution is skipped.
# TODO(axis-engine): resolve against each listener individually once
# the tick system exists to batch multi-listener interactions.
ipc_hash = self._resolve_channel_ipc_hash(
world=world,
speaker_name=username,
room_id=room,
channel="say",
world_id=world_id,
)
# ── OOC → IC translation ─────────────────────────────────────────────
# Translate the raw player message to in-character dialogue. The
# service is non-authoritative and gracefully degrading — see its
# docstring. The ipc_hash (if set above) is forwarded so the service
# can arm deterministic rendering and link this event in the ledger.
# XSS escaping is applied *after* translation so that the IC output
# is cleaned by the same sanitiser as the OOC fallback.
safe_message = self._translate_and_sanitize_chat(
world=world,
character_name=username,
message=message,
channel="say",
ipc_hash=ipc_hash,
)
if not database.add_chat_message(username, safe_message, room, world_id=world_id):
return False, "Failed to send message."
return True, f"You say: {safe_message}"
[docs]
def yell(self, username: str, message: str, *, world_id: str) -> tuple[bool, str]:
"""
Yell a message to current room and all adjoining rooms.
Unlike regular chat which only reaches the current room, yell sends
the message to:
1. The player's current room
2. All rooms directly connected via exits
The message is prefixed with [YELL] to distinguish it from normal chat.
Args:
username: Player yelling the message
message: Message text to yell
Returns:
Tuple of (success, message)
- success: True if yell sent, False otherwise
- message: Confirmation message OR error message
Failure Cases:
- Player not in a valid room
- Room data invalid
- Database insert failed
Side Effects:
- Adds [YELL] message to current room's chat
- Adds [YELL] message to all adjoining rooms' chat
- Players in multiple affected rooms will see the message
Example:
If player in "spawn" with exits to "forest" and "desert":
>>> engine.yell("player1", "Can anyone hear me?")
(True, "You yell: Can anyone hear me?")
# Message appears in spawn, forest, and desert rooms
Security Note:
Messages are sanitized to prevent XSS attacks before storage.
"""
current_room_id = database.get_character_room(username, world_id=world_id)
if not current_room_id:
return False, "You are not in a valid room."
world = self._get_world(world_id)
# Get current room to find adjoining rooms
current_room = world.get_room(current_room_id)
if not current_room:
return False, "Invalid room."
# ── Axis resolution ───────────────────────────────────────────────────
# Same pattern as chat() — first co-present, non-speaker character.
ipc_hash = self._resolve_channel_ipc_hash(
world=world,
speaker_name=username,
room_id=current_room_id,
channel="yell",
world_id=world_id,
)
# ── OOC → IC translation ─────────────────────────────────────────────
# Translation occurs before the [YELL] prefix is applied so that the
# rendered IC dialogue is wrapped naturally.
safe_message = self._translate_and_sanitize_chat(
world=world,
character_name=username,
message=message,
channel="yell",
ipc_hash=ipc_hash,
)
# Add [YELL] prefix to sanitized message
yell_message = f"[YELL] {safe_message}"
# Send to current room
if not database.add_chat_message(
username, yell_message, current_room_id, world_id=world_id
):
return False, "Failed to send message."
# Send to all adjoining rooms; collect any failures for diagnostic note
failed_rooms: list[str] = []
for _direction, room_id in current_room.exits.items():
if not database.add_chat_message(username, yell_message, room_id, world_id=world_id):
logger.warning("yell: failed to insert into adjacent room %r", room_id)
failed_rooms.append(room_id)
note = f" (failed to reach {len(failed_rooms)} room(s))" if failed_rooms else ""
return True, f"You yell: {safe_message}{note}"
[docs]
def whisper(
self,
username: str,
target: str,
message: str,
*,
world_id: str,
) -> tuple[bool, str]:
"""
Send a private whisper to a specific player in the same room.
Whispers are private messages that only the sender and recipient can see.
They are filtered by recipient when retrieving room chat messages.
Validation Checks:
1. Sender must be in a valid room
2. Target player must exist in database
3. Target must be online (have an active session)
4. Target must be in the same room as sender
The message is prefixed with [WHISPER: sender → target] to clearly
indicate it's a private message and show the direction.
Args:
username: Character sending the whisper
target: Character name to whisper to (case-sensitive)
message: Private message text
Returns:
Tuple of (success, message)
- success: True if whisper sent, False otherwise
- message: Confirmation message OR error message explaining failure
Failure Cases:
- Sender not in a valid room
- Target player doesn't exist
- Target player is not online
- Target player is in a different room
- Database insert failed
Side Effects:
- Adds message to chat_messages with recipient field set
- Only sender and target can see this message in their chat
- Extensive logging for debugging whisper issues
Security Note:
Character names are case-sensitive. Command parser preserves case
for arguments to ensure exact identity targeting.
Example:
>>> engine.whisper("player1", "Admin", "Help me please")
(True, "You whisper to Admin: Help me please")
# Only player1 and Admin see: [WHISPER: player1 → Admin] Help me please
>>> engine.whisper("player1", "Player2", "Hi")
(False, "Player 'Player2' is not in this room.")
"""
resolved_sender = database.resolve_character_name(username, world_id=world_id)
sender_name = resolved_sender or username
sender_room = database.get_character_room(sender_name, world_id=world_id)
logger.info(f"Whisper: {username} in room {sender_room} attempting to whisper to {target}")
if not sender_room:
logger.warning(f"Whisper failed: {username} not in valid room")
return False, "You are not in a valid room."
resolved_target = database.resolve_character_name(target, world_id=world_id)
# Check if target player exists
if not resolved_target or not database.character_exists(resolved_target):
logger.warning(f"Whisper failed: target {target} does not exist")
return False, f"Player '{target}' does not exist."
# Check if target is online (has an active session)
active_players = database.get_active_characters(world_id=world_id)
logger.info(f"Active players: {active_players}")
if resolved_target not in active_players:
logger.warning(f"Whisper failed: target {target} not online")
return False, f"Player '{target}' is not online."
# Check if target is in the same room
target_room = database.get_character_room(resolved_target, world_id=world_id)
logger.info(f"Target {target} is in room {target_room}")
if target_room != sender_room:
logger.warning(f"Whisper failed: {target} in {target_room}, sender in {sender_room}")
return False, f"Player '{target}' is not in this room."
world = self._get_world(world_id)
# ── Axis resolution ───────────────────────────────────────────────────
# For whisper the listener is unambiguous: the resolved target character.
ipc_hash = self._resolve_channel_ipc_hash(
world=world,
speaker_name=sender_name,
listener_name=resolved_target,
channel="whisper",
world_id=world_id,
)
# ── OOC → IC translation ─────────────────────────────────────────────
# Translation occurs before the [WHISPER: ...] prefix is applied so
# that the IC dialogue is wrapped naturally. Whispers are rendered
# with channel="whisper" so that the prompt template can lower the
# volume/intensity of the voice appropriately.
safe_message = self._translate_and_sanitize_chat(
world=world,
character_name=sender_name,
message=message,
channel="whisper",
ipc_hash=ipc_hash,
)
# Add whisper message with recipient (include both sender and target for clarity)
whisper_message = f"[WHISPER: {sender_name} → {target}] {safe_message}"
result = database.add_chat_message(
sender_name,
whisper_message,
sender_room,
recipient=resolved_target,
world_id=world_id,
)
logger.info(f"Whisper message save result: {result}")
if not result:
logger.error("Failed to save whisper to database")
return False, "Failed to send whisper."
logger.info(f"Whisper successful: {username} -> {target}: {safe_message}")
return True, f"You whisper to {target}: {safe_message}"
[docs]
def kick_character(self, actor_name: str, target: str, *, world_id: str) -> tuple[bool, str]:
"""
Disconnect all active sessions for a target character.
This is a moderation primitive used by admin/superuser command flows.
Permission checks are intentionally handled at the API route layer so
this method focuses on deterministic target resolution + session revoke.
Resolution behavior:
1. Resolve ``target`` as a character name in the current world.
2. Remove all sessions for the resolved character id.
Args:
actor_name: Character issuing the kick command.
target: Character name to disconnect.
world_id: Active world context for resolution.
Returns:
Tuple of ``(success, message)`` suitable for command responses.
"""
target_name = target.strip()
if not target_name:
return False, "Kick whom? Usage: /kick <character>"
resolved_actor = (
database.resolve_character_name(actor_name, world_id=world_id) or actor_name
)
resolved_target = database.resolve_character_name(target_name, world_id=world_id)
if not resolved_target:
return False, f"Player '{target_name}' was not found in this world."
if resolved_target == resolved_actor:
return False, "You cannot kick your own session."
character = database.get_character_by_name(resolved_target)
if not character:
return False, f"Player '{target_name}' was not found in this world."
removed_sessions = database.remove_sessions_for_character_count(int(character["id"]))
if removed_sessions <= 0:
return False, f"Player '{resolved_target}' is not online."
return True, f"Kicked {resolved_target} ({removed_sessions} session(s) disconnected)."
[docs]
def get_room_chat(self, username: str, limit: int = 20, *, world_id: str) -> str:
"""
Get recent chat messages from the player's current room.
Retrieves and formats chat messages, including regular chat, yells,
and whispers. Whispers are filtered so each player only sees:
- Public messages (no recipient)
- Whispers sent by them
- Whispers sent to them
Args:
username: Player requesting chat history
limit: Maximum number of messages to retrieve (default 20)
Returns:
Formatted string with recent messages
Format: "[Recent messages]:\nusername: message\n..."
Returns "[No messages in this room yet]" if empty
Returns "No messages." if player not in valid room
Example:
>>> engine.get_room_chat("player1", limit=5)
'''[Recent messages]:
player2: Hello!
player1: [WHISPER: player1 → player2] Hi there
player3: [YELL] Can anyone help?
'''
"""
room = database.get_character_room(username, world_id=world_id)
if not room:
return "No messages."
messages = database.get_room_messages(
room, limit=limit, username=username, world_id=world_id
)
if not messages:
return "[No messages in this room yet]"
chat_text = "[Recent messages]:\n"
for msg in messages:
chat_text += f"{msg['username']}: {msg['message']}\n"
return chat_text
[docs]
def get_inventory(self, username: str, *, world_id: str) -> str:
"""
Get formatted player inventory listing.
Retrieves the player's inventory from the database and formats it
as a readable list with item names.
Args:
username: Player whose inventory to retrieve
Returns:
Formatted inventory string
- "Your inventory:\n - Item1\n - Item2..." if items present
- "Your inventory is empty." if no items
Example:
>>> engine.get_inventory("player1")
"Your inventory:\n - Torch\n - Rope\n"
>>> engine.get_inventory("new_player")
"Your inventory is empty."
"""
inventory = database.get_character_inventory(username, world_id=world_id)
if not inventory:
return "Your inventory is empty."
world = self._get_world(world_id)
inv_text = "Your inventory:\n"
for item_id in inventory:
item = world.get_item(item_id)
if item:
inv_text += f" - {item.name}\n"
return inv_text
[docs]
def pickup_item(self, username: str, item_name: str, *, world_id: str) -> tuple[bool, str]:
"""
Pick up an item from the current room and add to inventory.
Searches for an item with matching name (case-insensitive) in the
current room. If found, adds it to the player's inventory.
Design Note:
Items are NOT removed from the room when picked up. This allows
multiple players to pick up the same item. This is intentional
for the current proof-of-concept design.
Args:
username: Player picking up the item
item_name: Name of item to pick up (case-insensitive match)
Returns:
Tuple of (success, message)
- success: True if item picked up, False otherwise
- message: Success confirmation OR error message
Failure Cases:
- Player not in a valid room
- Room doesn't exist in world data
- No item with that name in the room
Example:
>>> engine.pickup_item("player1", "torch")
(True, "You picked up the Torch.")
>>> engine.pickup_item("player1", "sword")
(False, "There is no 'sword' here.")
"""
room_id = database.get_character_room(username, world_id=world_id)
if not room_id:
return False, "You are not in a valid room."
world = self._get_world(world_id)
room = world.get_room(room_id)
if not room:
return False, "Invalid room."
# Find matching item
matching_item = None
for item_id in room.items:
item = world.get_item(item_id)
if item and item.name.lower() == item_name.lower():
matching_item = item_id
break
if not matching_item:
return False, f"There is no '{item_name}' here."
# Add to inventory
inventory = database.get_character_inventory(username, world_id=world_id)
if matching_item not in inventory:
inventory.append(matching_item)
database.set_character_inventory(username, inventory, world_id=world_id)
item = world.get_item(matching_item)
item_name_display = item.name if item else matching_item
return True, f"You picked up the {item_name_display}."
[docs]
def drop_item(self, username: str, item_name: str, *, world_id: str) -> tuple[bool, str]:
"""
Drop an item from player's inventory.
Searches inventory for an item with matching name (case-insensitive)
and removes it from the player's inventory.
Design Note:
Dropped items are NOT added back to the room. They simply disappear
from the player's inventory. This is intentional for the current
proof-of-concept design.
Args:
username: Player dropping the item
item_name: Name of item to drop (case-insensitive match)
Returns:
Tuple of (success, message)
- success: True if item dropped, False otherwise
- message: Success confirmation OR error message
Failure Cases:
- Player doesn't have an item with that name
Example:
>>> engine.drop_item("player1", "torch")
(True, "You dropped the Torch.")
>>> engine.drop_item("player1", "sword")
(False, "You don't have a 'sword'.")
"""
inventory = database.get_character_inventory(username, world_id=world_id)
world = self._get_world(world_id)
# Find matching item in inventory
matching_item = None
for item_id in inventory:
item = world.get_item(item_id)
if item and item.name.lower() == item_name.lower():
matching_item = item_id
break
if not matching_item:
return False, f"You don't have a '{item_name}'."
# Remove from inventory
inventory.remove(matching_item)
database.set_character_inventory(username, inventory, world_id=world_id)
item = world.get_item(matching_item)
item_name_display = item.name if item else matching_item
return True, f"You dropped the {item_name_display}."
[docs]
def look(self, username: str, *, world_id: str) -> str:
"""
Look around the current room to get a full description.
Generates a detailed description of the player's current room including
room name, description, items, other players, and available exits.
Args:
username: Player looking around
Returns:
Formatted room description string
Returns "You are not in a valid room." if player location invalid
Example:
>>> engine.look("player1")
'''
=== Spawn Zone ===
You stand in a peaceful plaza...
[Items here]:
- Torch
- Rope
[Players here]:
- player2
[Exits]:
- north: Enchanted Forest
- south: Golden Desert
'''
"""
room_id = database.get_character_room(username, world_id=world_id)
if not room_id:
return "You are not in a valid room."
# Check if room exists in the world
world = self._get_world(world_id)
if not world.get_room(room_id):
return "You are not in a valid room."
return world.get_room_description(room_id, username, world_id=world_id)
[docs]
def get_active_players(self, *, world_id: str) -> list[str]:
"""
Get list of all currently active (logged in) characters.
Queries the database sessions table to get all characters with active
sessions. These are characters currently logged into the server.
Returns:
List of character names for all active players
Example:
>>> engine.get_active_players(world_id="pipeworks_web")
['player1', 'Admin', 'Mendit']
"""
return database.get_active_characters(world_id=world_id)
def _broadcast_to_room(self, room_id: str, message: str, exclude: str | None = None):
"""
Deprecated compatibility hook for room-broadcast notifications.
Movement and recall are authoritative through emitted events, not this
method. The hook remains as a deliberate no-op seam so older call sites
can be isolated without implying that room broadcasts are an active
delivery mechanism.
A real implementation would require:
- WebSocket connections or message queue system
- Per-player message buffers
- Push notification mechanism
Args:
room_id: Room to broadcast to
message: Message to send
exclude: Optional username to exclude from broadcast (usually sender)
Current Status:
Not implemented; retained only as a compatibility stub.
"""
# This would be handled by the server's message queue
# TODO: Remove once movement/recall call sites no longer reference it.
pass
@staticmethod
def _opposite_direction(direction: str) -> str:
"""
Get the opposite direction for movement notifications.
Used when broadcasting arrival messages. If a player moves north,
players in the destination room see them arrive from the south.
Args:
direction: Direction of movement (north, south, east, west)
Returns:
Opposite direction string
Returns "somewhere" for unrecognized directions
Example:
>>> engine._opposite_direction("north")
"south"
>>> engine._opposite_direction("east")
"west"
>>> engine._opposite_direction("up")
"down"
"""
opposites = {
"north": "south",
"south": "north",
"east": "west",
"west": "east",
"up": "down",
"down": "up",
}
return opposites.get(direction.lower(), "somewhere")