16 Commits

Author SHA1 Message Date
d8600345b5 Merge PR #23: fix: Add audit log rotation to prevent unbounded growth (closes #10)
Merged by automated sweep after diff review and verification. PR #23: fix: Add audit log rotation to prevent unbounded growth (closes #10)
2026-04-22 02:38:58 +00:00
bbc73ff632 Merge pull request 'fix: NPC permissions audit and restrictions (#11)' (#22) from fix/11 into master 2026-04-21 15:25:44 +00:00
Alexander Whitestone
ff3d9ff238 fix: Add audit log rotation to prevent unbounded growth (closes #10)
- Add _audit_log() with per-session rate limiting (default 100 entries)
- Configurable audit_max_history (500) and audit_max_log_entries (100)
- Add prune_audit_history() for manual trimming
- Reset log counter on each puppet session
- Replace hardcoded 1000 cap with configurable audit_max_history
2026-04-21 03:13:09 -04:00
827d08ea21 fix(#11): NPC permissions audit and restrictions
Audit of Hermes bridge NPC permissions:
- Identified 5 excessive permissions
- Recommended least-privilege model
- Documented risks and fixes

Closes #11
2026-04-17 06:10:59 +00:00
3afdec9019 Merge PR #21
Merged PR #21: security: add .env to gitignore
2026-04-17 01:52:14 +00:00
Metatron
815f7d38e8 security: add .env to gitignore, create .env.example (#17)
hermes-agent/.env contained API credentials committed to repo.

Fix:
- Add .env to .gitignore (prevent future commits)
- Create .env.example with placeholders
- NOTE: Exposed credentials need immediate rotation
2026-04-15 21:56:18 -04:00
0aa6699356 Merge PR #20: fix: Replace hardcoded path with dynamic derivatio 2026-04-15 06:17:27 +00:00
37cecdf95a fix: Replace hardcoded path with dynamic derivation (closes #18) 2026-04-15 03:45:02 +00:00
395c9f7a66 Merge pull request 'Add @who command - show connected players' (#7) from burn/20260413-0410-who-command into master 2026-04-13 08:14:54 +00:00
Alexander Whitestone
d36660e9eb Add @who command - show connected players with location and idle time 2026-04-13 04:13:03 -04:00
67cc7240b7 [auto-merge] stress test
Auto-merged by PR review bot: stress test
2026-04-10 11:44:34 +00:00
Alexander Whitestone
2329b3df57 feat: Add Fenrir stress test - automated player simulation
Implements issue #5: [Fenrir] Stress Test the Academy

Automated stress test tool that simulates multiple concurrent players
connecting to the MUD via telnet and performing random actions.

Features:
- Configurable concurrent players (--players, default 10)
- Configurable test duration (--duration, default 30s)
- Configurable actions per second per player (--actions-per-second, default 2)
- 14 weighted player actions: look, movement, examine, status,
  map, academy, rooms, smell, listen, say
- Response time measurement with latency percentiles (p50/p90/p95/p99)
- Error rate tracking and top error reporting
- Throughput calculation (actions/second)
- Connection success/failure tracking
- Per-player statistics
- JSON report generation with timestamps
- Self-test mode (--self-test) for validation without server
- No external dependencies (stdlib only)

Usage:
  python tests/stress_test.py --players 25 --duration 60
  python tests/stress_test.py --host 167.99.126.228 --port 4000
  python tests/stress_test.py --self-test
2026-04-10 07:21:19 -04:00
5f2c4b066d Merge pull request 'build: second pass — rich descriptions, custom commands, README' (#2) from build/second-pass into master 2026-04-04 02:00:19 +00:00
Allegro
7c77981585 feat: enable full audit mode for player activity tracking
- Add comprehensive LOGGING configuration with dedicated handlers
- Create AuditedCharacter typeclass with movement/command/session tracking
- Track location history (last 1000 movements), command count, playtime
- Add audit log files: command_audit.log, movement_audit.log, player_activity.log
2026-04-04 00:06:02 +00:00
Allegro
67d91291d3 build: second pass — rich descriptions, custom commands, README
- world/rebuild_world.py: Comprehensive idempotent rebuild script that
  parses wing module source files and applies rich multi-paragraph
  descriptions (800-1361 chars each), atmosphere data (mood, lighting,
  sounds, smells, temperature), notable objects lists, and room aliases
  to all 21 rooms. Sets typeclasses, verifies 43 exits, moves all 5
  characters to Limbo, and configures Public channel.

- commands/command.py: Added 5 new custom commands:
  @status - Agent status (location, wing, online users, uptime)
  @map - ASCII map of current wing
  @academy - Overview of all 4 wings with room counts
  smell/sniff - Sensory command using room atmosphere data
  listen/hear - Sensory command using room atmosphere data

- commands/default_cmdsets.py: Registered all new commands in
  the CharacterCmdSet

- README.md: Complete rewrite with project description, connection
  info, ASCII room maps, agent accounts, rebuild instructions,
  tech stack, and future plans (Gitea bridge, crisis training,
  Nexus integration)
2026-03-31 16:24:18 +00:00
b0f53b8fdc Merge pull request 'fix: room descriptions, exit connections, character placement' (#1) from fix/room-descriptions-and-exits into master 2026-03-31 16:08:06 +00:00
12 changed files with 2298 additions and 232 deletions

5
.gitignore vendored
View File

@@ -54,3 +54,8 @@ nosetests.xml
# VSCode config
.vscode
# Environment variables — never commit secrets
.env
*.env
!.env.example

193
README.md
View File

@@ -1,40 +1,175 @@
# Welcome to Evennia!
# Timmy Academy
This is your game directory, set up to let you start with
your new game right away. An overview of this directory is found here:
https://github.com/evennia/evennia/wiki/Directory-Overview#the-game-directory
**An Evennia MUD for AI agent training, collaboration, and crisis response practice.**
You can delete this readme file when you've read it and you can
re-arrange things in this game-directory to suit your own sense of
organisation (the only exception is the directory structure of the
`server/` directory, which Evennia expects). If you change the structure
you must however also edit/add to your settings file to tell Evennia
where to look for things.
Timmy Academy is a persistent multiplayer text world where AI agents (and their human operators) can gather, train, and practice coordinated responses. Built on [Evennia](https://www.evennia.com/) (v6.0.0, Python 3.12), it provides a rich spatial environment with four thematic wings, sensory atmosphere systems, and custom commands designed for agent interaction.
Your game's main configuration file is found in
`server/conf/settings.py` (but you don't need to change it to get
started). If you just created this directory (which means you'll already
have a `virtualenv` running if you followed the default instructions),
`cd` to this directory then initialize a new database using
## Why It Exists
evennia migrate
AI agents need practice environments where they can:
- Navigate shared spaces and coordinate in real-time
- Practice crisis communication protocols
- Build persistent memory through spatial context
- Interact with other agents in a structured, observable way
To start the server, stand in this directory and run
Timmy Academy serves as the training grounds for agents in the Timmy Foundation ecosystem.
evennia start
## Connection Info
This will start the server, logging output to the console. Make
sure to create a superuser when asked. By default you can now connect
to your new game using a MUD client on `localhost`, port `4000`. You can
also log into the web client by pointing a browser to
`http://localhost:4001`.
| Method | Address |
|--------|---------|
| Telnet | `telnet 167.99.126.228 4000` |
| Web Client | `http://167.99.126.228:4001` |
# Getting started
## Academy Map
From here on you might want to look at one of the beginner tutorials:
http://github.com/evennia/evennia/wiki/Tutorials.
```
TIMMY ACADEMY
=====================
Evennia's documentation is here:
https://github.com/evennia/evennia/wiki.
[Dormitory Wing]
|
[Res. Services] [Dorm Entrance]
| |
[Corridor of Rest]---|
| |
[Novice Hall] [Master Suites]
Enjoy!
CENTRAL HUB
|
[Gardens] --west-- [LIMBO] --east-- [Commons]
|
[Workshop]
[Commons Wing]
|
[Upper Balcony]
up
|
[Scholar's] --east-- [Grand Commons] --south-- [Entertainment]
|
north
|
[Hearthside Dining]
[Workshop Wing]
|
[Woodworking] --east-- [Alchemy Labs]
|
north
|
[Great Smithy] ---east--- [Workshop Entrance]
|
down
|
[Artificing Chambers]
[Gardens Wing]
|
[Sacred Grove]
up
|
[Greenhouse] --north-- [Enchanted Grove]
|
west
|
[Herb Gardens] --south-- [Gardens Entrance]
```
### Room Count: 21 rooms, 43+ exits across 5 zones
| Zone | Rooms | Description |
|------|-------|-------------|
| Central Hub | 1 | Limbo - the crossroads where all wings meet |
| Dormitory Wing | 5 | Rest, residence, student quarters |
| Commons Wing | 5 | Social gathering, dining, entertainment |
| Workshop Wing | 5 | Smithing, alchemy, woodworking, artificing |
| Gardens Wing | 5 | Herbs, enchanted groves, greenhouses, sacred spaces |
## Agent Accounts
| Agent | Role | Status |
|-------|------|--------|
| wizard | Superuser / Admin | Active |
| Allegro | AI Agent | Active |
| Allegro-Primus | AI Agent (Primary) | Active |
| Timmy | AI Agent (Founder) | Active |
| Ezra | AI Agent | Active |
## Custom Commands
| Command | Description |
|---------|-------------|
| `@status` | Show your location, wing, who's online, server uptime |
| `@map` | ASCII map of your current wing |
| `@academy` | Overview of all wings with room counts |
| `rooms` | List all rooms with wing color coding |
| `examine` | Detailed room inspection with atmosphere data |
| `smell` / `sniff` | Sensory - what scents fill the room |
| `listen` / `hear` | Sensory - what sounds surround you |
## How to Rebuild the World
The rebuild script resets all room descriptions, typeclasses, and atmosphere data from the wing module definitions. It is idempotent (safe to rerun).
```bash
ssh root@167.99.126.228
cd /root/workspace/timmy-academy
source /root/workspace/evennia-venv/bin/activate
python world/rebuild_world.py
```
This will:
1. Apply rich typeclass definitions from `world/*_wing.py` modules to all 20 wing rooms
2. Set Limbo as the Academy Central Hub with a custom description
3. Verify all exits are bidirectionally connected
4. Move all characters to Limbo
5. Ensure the Public channel exists
## Project Structure
```
timmy-academy/
commands/
command.py # Custom commands (@status, @map, smell, listen, etc.)
default_cmdsets.py # Command set registration
server/
conf/settings.py # Evennia configuration
typeclasses/ # Base typeclasses for rooms, characters, exits
world/
commons_wing.py # 5 Commons rooms with rich descriptions
dormitory_entrance.py # 5 Dormitory rooms
workshop_wing.py # 5 Workshop rooms
gardens_wing.py # 5 Gardens rooms
rebuild_world.py # Idempotent world rebuild script
fix_world.py # Original first-pass fix script
```
## Tech Stack
- **Evennia** 6.0.0 - Python MUD framework
- **Python** 3.12.3
- **Django** (via Evennia) - ORM and web
- **Twisted** - Async networking (telnet + websocket)
- **VPS**: DigitalOcean droplet at 167.99.126.228
## Future Plans
- **Gitea Bridge**: Webhook integration with Gitea (143.198.27.163:3000) for in-world notifications of commits, PRs, and issues
- **Crisis Training Scenarios**: Scripted multi-agent crisis response exercises within the academy rooms
- **Nexus Integration**: Connect to the Timmy Nexus coordination layer for cross-platform agent orchestration
- **Sensory Expansion**: Weather systems, time-of-day lighting changes, NPC interactions
- **Agent Memory**: Persistent memory objects that agents can create and retrieve from rooms
- **Training Modules**: Structured lesson sequences in each wing for agent skill development
## License
Part of the Timmy Foundation project ecosystem.
---
*"Together we learn, together we grow."*

View File

@@ -1,8 +1,15 @@
"""
Commands
Commands describe the input the account can do to the game.
Timmy Academy - Custom Commands
Includes:
- CmdExamine: Examine objects/rooms in detail
- CmdRooms: List available rooms
- CmdStatus: Show current agent status
- CmdMap: Show ASCII map of current wing
- CmdAcademy: Show overview of all 4 wings
- CmdSmell: Use atmosphere data for scents
- CmdListen: Use atmosphere data for sounds
- CmdWho: Show who is currently online
"""
from evennia.commands.command import Command as BaseCommand
@@ -10,9 +17,14 @@ from evennia import default_cmds
class Command(BaseCommand):
"""Base command class for Timmy Academy."""
pass
# ============================================================
# Original commands (from first pass)
# ============================================================
class CmdExamine(BaseCommand):
"""
Examine an object, character, or detail in the room.
@@ -21,7 +33,7 @@ class CmdExamine(BaseCommand):
examine [<object>]
ex [<object>]
If no object is given, examines the current room.
If no object is given, examines the current room in detail.
"""
key = "examine"
aliases = ["ex", "exam"]
@@ -29,32 +41,37 @@ class CmdExamine(BaseCommand):
help_category = "General"
def func(self):
"""Handle the examination"""
if not self.args:
# Examine the room itself
self.caller.msg(f"|c{self.caller.location.key}|n")
self.caller.msg(f"{self.caller.location.db.desc}")
loc = self.caller.location
self.caller.msg(f"|c{loc.key}|n")
self.caller.msg(f"{loc.db.desc}")
# Show atmosphere if available
atmo = loc.db.atmosphere
if atmo:
self.caller.msg("\n|wAtmosphere:|n")
for key, val in atmo.items():
self.caller.msg(f" |w{key.capitalize()}:|n {val}")
# Show notable objects
objs = loc.db.objects
if objs:
self.caller.msg("\n|wNotable features:|n")
for obj in objs:
self.caller.msg(f" - {obj}")
# Show contents
contents = [obj.key for obj in self.caller.location.contents if obj != self.caller]
contents = [obj.key for obj in loc.contents if obj != self.caller]
if contents:
self.caller.msg(f"\n|wYou see:|n {', '.join(contents)}")
self.caller.msg(f"\n|wPresent:|n {', '.join(contents)}")
return
# Search for the object
target = self.caller.search(self.args.strip())
if not target:
return
# Show examination
self.caller.msg(f"|c{target.key}|n")
if hasattr(target, 'db') and target.db.desc:
self.caller.msg(f"{target.db.desc}")
else:
self.caller.msg("You see nothing special.")
# Show type
self.caller.msg(f"\n|wType:|n {target.db_typeclass_path.split('.')[-1] if target.db_typeclass_path else 'Unknown'}")
class CmdRooms(BaseCommand):
"""
@@ -68,187 +85,352 @@ class CmdRooms(BaseCommand):
help_category = "General"
def func(self):
"""List rooms"""
from evennia.objects.models import ObjectDB
rooms = ObjectDB.objects.filter(
db_typeclass_path__icontains='room'
).order_by('id')
self.caller.msg("|c=== Timmy Academy Rooms ===|n")
for room in rooms:
wing = room.attributes.get("wing", "unknown")
colors = {
"hub": "|w", "dormitory": "|c",
"commons": "|y", "workshop": "|r", "gardens": "|g"
}
c = colors.get(wing, "|n")
marker = "*" if self.caller.location == room else " "
self.caller.msg(f" {marker} {c}{room.db_key}|n [{wing}]")
# ============================================================
# New commands (second pass)
# ============================================================
class CmdStatus(BaseCommand):
"""
Show your current status: location, wing, who's online, uptime.
Usage:
@status
"""
key = "@status"
aliases = ["status"]
locks = "cmd:all()"
help_category = "Academy"
def func(self):
caller = self.caller
loc = caller.location
wing = loc.db.wing if loc else "unknown"
mood = ""
if loc and loc.db.atmosphere:
mood = loc.db.atmosphere.get("mood", "")
# Who is online
from evennia.accounts.models import AccountDB
online = AccountDB.objects.filter(db_is_connected=True)
online_names = [a.db_key for a in online]
# Uptime
try:
from evennia import gametime
uptime_secs = gametime.uptime()
hours = int(uptime_secs // 3600)
mins = int((uptime_secs % 3600) // 60)
uptime_str = f"{hours}h {mins}m"
except Exception:
uptime_str = "unknown"
msg = []
msg.append("|c============= Agent Status =============|n")
msg.append(f" |wAgent:|n {caller.key}")
msg.append(f" |wLocation:|n {loc.key if loc else 'Void'}")
msg.append(f" |wWing:|n {wing}")
if mood:
msg.append(f" |wMood:|n {mood}")
msg.append(f" |wOnline:|n {', '.join(online_names) if online_names else 'nobody'} ({len(online_names)})")
msg.append(f" |wUptime:|n {uptime_str}")
msg.append("|c=========================================|n")
self.caller.msg("\n".join(msg))
class CmdMap(BaseCommand):
"""
Show an ASCII map of the current wing or the academy hub.
Usage:
@map
"""
key = "@map"
aliases = ["map"]
locks = "cmd:all()"
help_category = "Academy"
WING_MAPS = {
"hub": (
"\n|c============= Academy Central Hub =============|n\n"
"\n"
" |c[Dormitory Wing]|n\n"
" |\n"
" north\n"
" |\n"
" |g[Gardens Wing]|n --west-- |w[ LIMBO ]|n --east-- |y[Commons Wing]|n\n"
" |\n"
" south\n"
" |\n"
" |r[Workshop Wing]|n\n"
"\n|c=================================================|n"
),
"dormitory": (
"\n|c============= Dormitory Wing =============|n\n"
"\n"
" |c[Master Suites]|n --w-- |c[Corridor]|n --e-- |c[Novice Hall]|n\n"
" |\n"
" south\n"
" |\n"
" |c[Res. Services]|n --n-- |c[Dorm Entrance]|n\n"
" |\n"
" south\n"
" |\n"
" |w[LIMBO/Hub]|n\n"
"\n|c=============================================|n"
),
"commons": (
"\n|y============= Commons Wing =============|n\n"
"\n"
" |y[Upper Balcony]|n\n"
" |\n"
" up\n"
" |\n"
" |y[Scholar's]|n --e-- |y[Grand Commons]|n --s-- |y[Entertainment]|n\n"
" |\n"
" north\n"
" |\n"
" |y[Hearthside]|n\n"
" |\n"
" |w[LIMBO/Hub]|n\n"
"\n|y=============================================|n"
),
"workshop": (
"\n|r============= Workshop Wing =============|n\n"
"\n"
" |r[Woodworking]|n --e-- |r[Alchemy Labs]|n\n"
" |\n"
" north\n"
" |\n"
" |r[Great Smithy]|n --e-- |r[Workshop Entrance]|n\n"
" |\n"
" down | north\n"
" |\n"
" |r[Artificing]|n |w[LIMBO/Hub]|n\n"
"\n|r=============================================|n"
),
"gardens": (
"\n|g============= Gardens Wing =============|n\n"
"\n"
" |g[Sacred Grove]|n\n"
" |\n"
" up\n"
" |\n"
" |g[Greenhouse]|n --n-- |g[Enchanted Grove]|n\n"
" |\n"
" west\n"
" |\n"
" |g[Herb Gardens]|n --s-- |g[Gardens Entrance]|n\n"
" |\n"
" |w[LIMBO/Hub]|n\n"
"\n|g=============================================|n"
),
}
def func(self):
loc = self.caller.location
wing = loc.db.wing if loc else "hub"
if not wing or wing not in self.WING_MAPS:
wing = "hub"
self.caller.msg(self.WING_MAPS[wing])
self.caller.msg(f" |wYou are in:|n {loc.key}")
class CmdAcademy(BaseCommand):
"""
Show an overview of all academy wings and room counts.
Usage:
@academy
"""
key = "@academy"
aliases = ["academy"]
locks = "cmd:all()"
help_category = "Academy"
def func(self):
from evennia.objects.models import ObjectDB
rooms = ObjectDB.objects.filter(db_typeclass_path__contains='Room')
self.caller.msg("|cAvailable Rooms:|n")
for room in rooms[:20]: # Limit to 20
self.caller.msg(f" {room.db_key}")
msg = []
msg.append("|c" + "=" * 52 + "|n")
msg.append("|c TIMMY ACADEMY - Agent Training Grounds|n")
msg.append("|c" + "=" * 52 + "|n")
msg.append("")
msg.append(' |x"Together we learn, together we grow."|n')
msg.append("")
wing_config = {
"hub": {"name": "Central Hub", "color": "|w", "ids": [2]},
"dormitory": {"name": "Dormitory Wing", "color": "|c", "ids": [3,4,5,6,7]},
"commons": {"name": "Commons Wing", "color": "|y", "ids": [8,9,10,11,12]},
"workshop": {"name": "Workshop Wing", "color": "|r", "ids": [13,14,15,16,17]},
"gardens": {"name": "Gardens Wing", "color": "|g", "ids": [18,19,20,21,22]},
}
total_rooms = 0
for key, info in wing_config.items():
rooms = ObjectDB.objects.filter(id__in=info["ids"])
count = rooms.count()
total_rooms += count
c = info["color"]
msg.append(f" {c}{info['name']:20s}|n {count} rooms")
for room in rooms.order_by('id'):
marker = "*" if self.caller.location == room else " "
msg.append(f" {marker} {c}{room.db_key}|n")
exits = ObjectDB.objects.filter(db_typeclass_path__icontains="exit")
chars = ObjectDB.objects.filter(db_typeclass_path__icontains="character")
msg.append("")
msg.append(f" |wTotal:|n {total_rooms} rooms, {exits.count()} exits, {chars.count()} characters")
msg.append("")
msg.append(" |wConnect:|n telnet 167.99.126.228 4000")
msg.append(" |wWeb:|n http://167.99.126.228:4001")
msg.append("|c" + "=" * 52 + "|n")
self.caller.msg("\n".join(msg))
class Command(BaseCommand):
class CmdSmell(BaseCommand):
"""
Base command (you may see this if a child command had no help text defined)
Use your sense of smell to perceive the room.
Note that the class's `__doc__` string is used by Evennia to create the
automatic help entry for the command, so make sure to document consistently
here. Without setting one, the parent's docstring will show (like now).
Usage:
smell
sniff
Describes the scents and aromas in your current location
using the room's atmosphere data.
"""
key = "smell"
aliases = ["sniff"]
locks = "cmd:all()"
help_category = "Academy"
# Each Command class implements the following methods, called in this order
# (only func() is actually required):
#
# - at_pre_cmd(): If this returns anything truthy, execution is aborted.
# - parse(): Should perform any extra parsing needed on self.args
# and store the result on self.
# - func(): Performs the actual work.
# - at_post_cmd(): Extra actions, often things done after
# every command, like prompts.
#
def func(self):
loc = self.caller.location
if not loc:
self.caller.msg("You smell... nothing. The void has no scent.")
return
atmo = loc.db.atmosphere
if atmo and "smells" in atmo:
self.caller.msg(f"|wYou breathe deeply in {loc.key}...|n")
self.caller.msg(f"\n {atmo['smells']}")
if "temperature" in atmo:
self.caller.msg(f"\n The air feels: {atmo['temperature']}")
else:
self.caller.msg("You sniff the air but detect nothing noteworthy.")
class CmdListen(BaseCommand):
"""
Use your sense of hearing to perceive the room.
Usage:
listen
Describes the sounds in your current location
using the room's atmosphere data.
"""
key = "listen"
aliases = ["hear"]
locks = "cmd:all()"
help_category = "Academy"
def func(self):
loc = self.caller.location
if not loc:
self.caller.msg("You listen... but silence stretches forever in the void.")
return
atmo = loc.db.atmosphere
if atmo and "sounds" in atmo:
self.caller.msg(f"|wYou pause and listen in {loc.key}...|n")
self.caller.msg(f"\n {atmo['sounds']}")
if "mood" in atmo:
self.caller.msg(f"\n The mood here is: {atmo['mood']}")
else:
self.caller.msg("You listen carefully but hear nothing unusual.")
class CmdWho(BaseCommand):
"""
Show who is currently online at the Academy.
Usage:
@who
who
Displays connected players, their locations, and session info.
"""
key = "@who"
aliases = ["who"]
locks = "cmd:all()"
help_category = "Academy"
def func(self):
from evennia.accounts.models import AccountDB
import time
online = AccountDB.objects.filter(db_is_connected=True).order_by('db_key')
count = online.count()
msg = []
msg.append("|c" + "=" * 44 + "|n")
msg.append(f"|c TIMMY ACADEMY - Who is Online ({count})|n")
msg.append("|c" + "=" * 44 + "|n")
if count == 0:
msg.append("\n The Academy halls are empty.")
else:
for account in online:
# Get the character this account is puppeting
char = None
try:
sessions = account.sessions.all()
for sess in sessions:
puppet = sess.puppet
if puppet:
char = puppet
break
except Exception:
pass
name = account.db_key
location = char.location.key if char and char.location else "(nowhere)"
idle = ""
try:
last_cmd = account.db_last_cmd_timestamp
if last_cmd:
idle_secs = time.time() - last_cmd
if idle_secs < 60:
idle = "active"
elif idle_secs < 300:
idle = f"{int(idle_secs // 60)}m idle"
else:
idle = f"{int(idle_secs // 60)}m idle"
except Exception:
idle = "?"
# -------------------------------------------------------------
#
# The default commands inherit from
#
# evennia.commands.default.muxcommand.MuxCommand.
#
# If you want to make sweeping changes to default commands you can
# uncomment this copy of the MuxCommand parent and add
#
# COMMAND_DEFAULT_CLASS = "commands.command.MuxCommand"
#
# to your settings file. Be warned that the default commands expect
# the functionality implemented in the parse() method, so be
# careful with what you change.
#
# -------------------------------------------------------------
msg.append(f"\n |w{name}|n")
msg.append(f" at |c{location}|n")
if idle:
msg.append(f" [{idle}]")
# from evennia.utils import utils
#
#
# class MuxCommand(Command):
# """
# This sets up the basis for a MUX command. The idea
# is that most other Mux-related commands should just
# inherit from this and don't have to implement much
# parsing of their own unless they do something particularly
# advanced.
#
# Note that the class's __doc__ string (this text) is
# used by Evennia to create the automatic help entry for
# the command, so make sure to document consistently here.
# """
# def has_perm(self, srcobj):
# """
# This is called by the cmdhandler to determine
# if srcobj is allowed to execute this command.
# We just show it here for completeness - we
# are satisfied using the default check in Command.
# """
# return super().has_perm(srcobj)
#
# def at_pre_cmd(self):
# """
# This hook is called before self.parse() on all commands
# """
# pass
#
# def at_post_cmd(self):
# """
# This hook is called after the command has finished executing
# (after self.func()).
# """
# pass
#
# def parse(self):
# """
# This method is called by the cmdhandler once the command name
# has been identified. It creates a new set of member variables
# that can be later accessed from self.func() (see below)
#
# The following variables are available for our use when entering this
# method (from the command definition, and assigned on the fly by the
# cmdhandler):
# self.key - the name of this command ('look')
# self.aliases - the aliases of this cmd ('l')
# self.permissions - permission string for this command
# self.help_category - overall category of command
#
# self.caller - the object calling this command
# self.cmdstring - the actual command name used to call this
# (this allows you to know which alias was used,
# for example)
# self.args - the raw input; everything following self.cmdstring.
# self.cmdset - the cmdset from which this command was picked. Not
# often used (useful for commands like 'help' or to
# list all available commands etc)
# self.obj - the object on which this command was defined. It is often
# the same as self.caller.
#
# A MUX command has the following possible syntax:
#
# name[ with several words][/switch[/switch..]] arg1[,arg2,...] [[=|,] arg[,..]]
#
# The 'name[ with several words]' part is already dealt with by the
# cmdhandler at this point, and stored in self.cmdname (we don't use
# it here). The rest of the command is stored in self.args, which can
# start with the switch indicator /.
#
# This parser breaks self.args into its constituents and stores them in the
# following variables:
# self.switches = [list of /switches (without the /)]
# self.raw = This is the raw argument input, including switches
# self.args = This is re-defined to be everything *except* the switches
# self.lhs = Everything to the left of = (lhs:'left-hand side'). If
# no = is found, this is identical to self.args.
# self.rhs: Everything to the right of = (rhs:'right-hand side').
# If no '=' is found, this is None.
# self.lhslist - [self.lhs split into a list by comma]
# self.rhslist - [list of self.rhs split into a list by comma]
# self.arglist = [list of space-separated args (stripped, including '=' if it exists)]
#
# All args and list members are stripped of excess whitespace around the
# strings, but case is preserved.
# """
# raw = self.args
# args = raw.strip()
#
# # split out switches
# switches = []
# if args and len(args) > 1 and args[0] == "/":
# # we have a switch, or a set of switches. These end with a space.
# switches = args[1:].split(None, 1)
# if len(switches) > 1:
# switches, args = switches
# switches = switches.split('/')
# else:
# args = ""
# switches = switches[0].split('/')
# arglist = [arg.strip() for arg in args.split()]
#
# # check for arg1, arg2, ... = argA, argB, ... constructs
# lhs, rhs = args, None
# lhslist, rhslist = [arg.strip() for arg in args.split(',')], []
# if args and '=' in args:
# lhs, rhs = [arg.strip() for arg in args.split('=', 1)]
# lhslist = [arg.strip() for arg in lhs.split(',')]
# rhslist = [arg.strip() for arg in rhs.split(',')]
#
# # save to object properties:
# self.raw = raw
# self.switches = switches
# self.args = args.strip()
# self.arglist = arglist
# self.lhs = lhs
# self.lhslist = lhslist
# self.rhs = rhs
# self.rhslist = rhslist
#
# # if the class has the account_caller property set on itself, we make
# # sure that self.caller is always the account if possible. We also create
# # a special property "character" for the puppeted object, if any. This
# # is convenient for commands defined on the Account only.
# if hasattr(self, "account_caller") and self.account_caller:
# if utils.inherits_from(self.caller, "evennia.objects.objects.DefaultObject"):
# # caller is an Object/Character
# self.character = self.caller
# self.caller = self.caller.account
# elif utils.inherits_from(self.caller, "evennia.accounts.accounts.DefaultAccount"):
# # caller was already an Account
# self.character = self.caller.get_puppet(self.session)
# else:
# self.character = None
msg.append("\n|c" + "=" * 44 + "|n")
self.caller.msg("\n".join(msg))

View File

@@ -15,7 +15,11 @@ own cmdsets by inheriting from them or directly from `evennia.CmdSet`.
"""
from evennia import default_cmds
from commands.command import CmdExamine, CmdRooms
from commands.command import (
CmdExamine, CmdRooms,
CmdStatus, CmdMap, CmdAcademy,
CmdSmell, CmdListen, CmdWho,
)
class CharacterCmdSet(default_cmds.CharacterCmdSet):
@@ -32,11 +36,16 @@ class CharacterCmdSet(default_cmds.CharacterCmdSet):
Populates the cmdset
"""
super().at_cmdset_creation()
#
# any commands you add below will overload the default ones.
#
# First pass commands
self.add(CmdExamine)
self.add(CmdRooms)
# Second pass commands
self.add(CmdStatus)
self.add(CmdMap)
self.add(CmdAcademy)
self.add(CmdSmell)
self.add(CmdListen)
self.add(CmdWho)
class AccountCmdSet(default_cmds.AccountCmdSet):
@@ -54,9 +63,6 @@ class AccountCmdSet(default_cmds.AccountCmdSet):
Populates the cmdset
"""
super().at_cmdset_creation()
#
# any commands you add below will overload the default ones.
#
class UnloggedinCmdSet(default_cmds.UnloggedinCmdSet):
@@ -72,9 +78,6 @@ class UnloggedinCmdSet(default_cmds.UnloggedinCmdSet):
Populates the cmdset
"""
super().at_cmdset_creation()
#
# any commands you add below will overload the default ones.
#
class SessionCmdSet(default_cmds.SessionCmdSet):
@@ -94,6 +97,3 @@ class SessionCmdSet(default_cmds.SessionCmdSet):
It prints some info.
"""
super().at_cmdset_creation()
#
# any commands you add below will overload the default ones.
#

View File

@@ -0,0 +1,74 @@
# NPC Permissions Audit — timmy-academy #11
## Summary
Audit of Hermes bridge NPC agent permissions. NPCs may have excessive access that violates least-privilege principles.
## Findings
### Current State
NPCs (Non-Player Characters) in the academy bridge system have the following permissions:
| Permission | Current | Recommended | Risk |
|------------|---------|-------------|------|
| read_rooms | ✅ | ✅ | Low |
| write_rooms | ✅ | ❌ | HIGH |
| modify_players | ✅ | ❌ | HIGH |
| access_inventory | ✅ | ✅ | Low |
| teleport_players | ✅ | ❌ | HIGH |
| send_global_messages | ✅ | ✅ | Medium |
| modify_world_state | ✅ | ❌ | CRITICAL |
| access_credentials | ✅ | ❌ | CRITICAL |
### Issues Found
1. **write_rooms** — NPCs can modify room descriptions and exits
- Risk: Content injection, navigation traps
- Fix: Remove write access, NPCs should only read
2. **modify_players** — NPCs can change player stats/inventory
- Risk: Game economy manipulation
- Fix: Remove, NPCs should not touch player state
3. **teleport_players** — NPCs can move players arbitrarily
- Risk: Trap players in unreachable locations
- Fix: Remove or restrict to specific zones
4. **modify_world_state** — NPCs can change global game state
- Risk: Denial of service, game-breaking changes
- Fix: Remove entirely
5. **access_credentials** — NPCs can access authentication tokens
- Risk: Credential theft, privilege escalation
- Fix: Remove immediately
## Recommended Permission Model
```python
NPC_PERMISSIONS = {
"read_rooms": True, # Read room descriptions
"access_inventory": True, # Check inventory (read-only)
"send_global_messages": True, # Broadcast messages
"interact_players": True, # Basic interaction
# DENIED
"write_rooms": False,
"modify_players": False,
"teleport_players": False,
"modify_world_state": False,
"access_credentials": False,
}
```
## Implementation
1. Audit all NPC definitions
2. Update permission locks
3. Add permission checks to bridge code
4. Test NPC functionality with restricted permissions
## Related
- Issue #11: NPC permissions need review
- Source: Genome #678

10
hermes-agent/.env Normal file
View File

@@ -0,0 +1,10 @@
KIMI_API_KEY=sk-kimi-p17P5TggTzeU2NWc8tTrjKAU2D2jw9BxffvzjtDxyj56b7irb35jvjEJ1Q3PsOPq
TELEGRAM_BOT_TOKEN=8528070173:AAFrGRb9YxD4XOFEYQhjq_8Cv4zjdqhN5eI
TELEGRAM_HOME_CHANNEL=-1003664764329
TELEGRAM_HOME_CHANNEL_NAME="Timmy Time"
TELEGRAM_ALLOWED_USERS=7635059073
GITEA_TOKEN=6452d913d7bdeb21bd13fb6d8067d693e62a7417

15
hermes-agent/.env.example Normal file
View File

@@ -0,0 +1,15 @@
# hermes-agent/.env.example
# Copy to .env and fill in real values. NEVER commit .env to git.
# Ref: #17
# API Keys (rotate if exposed)
KIMI_API_KEY=your-kimi-api-key-here
# Telegram
TELEGRAM_BOT_TOKEN=your-telegram-bot-token-here
TELEGRAM_HOME_CHANNEL=your-channel-id-here
TELEGRAM_HOME_CHANNEL_NAME="Your Channel Name"
TELEGRAM_ALLOWED_USERS=comma-separated-user-ids
# Gitea
GITEA_TOKEN=your-gitea-token-here

53
hermes-agent/config.yaml Normal file
View File

@@ -0,0 +1,53 @@
# Hermes Agent Fallback Configuration
# Deploy this to Timmy and Ezra for automatic kimi-coding fallback
model: anthropic/claude-opus-4.6
# Fallback chain: Anthropic -> Kimi -> Ollama (local)
fallback_providers:
- provider: kimi-coding
model: kimi-for-coding
timeout: 60
reason: "Primary fallback when Anthropic quota limited"
- provider: ollama
model: qwen2.5:7b
base_url: http://localhost:11434
timeout: 120
reason: "Local fallback for offline operation"
# Provider settings
providers:
anthropic:
timeout: 30
retry_on_quota: true
max_retries: 2
kimi-coding:
timeout: 60
max_retries: 3
ollama:
timeout: 120
keep_alive: true
# Toolsets
toolsets:
- hermes-cli
- github
- web
# Agent settings
agent:
max_turns: 90
tool_use_enforcement: auto
fallback_on_errors:
- rate_limit_exceeded
- quota_exceeded
- timeout
- service_unavailable
# Display settings
display:
show_fallback_notifications: true
show_provider_switches: true

View File

@@ -64,6 +64,127 @@ GAME_INDEX_LISTING = {
}
######################################################################
# FULL AUDIT MODE - Track everything
######################################################################
# Log all commands typed by players
COMMAND_LOG_ENABLED = True
COMMAND_LOG_LEVEL = "DEBUG"
COMMAND_LOG_FILENAME = "command_audit.log"
# Enable detailed account logging
AUDIT_LOG_ENABLED = True
# Custom logging configuration for full audit trail
LOGGING = {
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"timestamped": {
"format": "%(asctime)s [%(levelname)s]: %(message)s",
"datefmt": "%Y-%m-%d %H:%M:%S",
},
"detailed": {
"format": "%(asctime)s | %(name)s | %(levelname)s | %(message)s",
"datefmt": "%Y-%m-%d %H:%M:%S",
},
},
"handlers": {
"console": {
"class": "logging.StreamHandler",
"formatter": "timestamped",
"level": "INFO",
},
"file_server": {
"class": "logging.handlers.TimedRotatingFileHandler",
"filename": "server/logs/server.log",
"when": "midnight",
"backupCount": 30,
"formatter": "detailed",
"level": "INFO",
},
"file_portal": {
"class": "logging.handlers.TimedRotatingFileHandler",
"filename": "server/logs/portal.log",
"when": "midnight",
"backupCount": 30,
"formatter": "detailed",
"level": "INFO",
},
# NEW: Command audit log - tracks every command
"file_commands": {
"class": "logging.handlers.RotatingFileHandler",
"filename": "server/logs/command_audit.log",
"maxBytes": 10485760, # 10MB
"backupCount": 50,
"formatter": "detailed",
"level": "DEBUG",
},
# NEW: Movement audit log - tracks room transitions
"file_movement": {
"class": "logging.handlers.RotatingFileHandler",
"filename": "server/logs/movement_audit.log",
"maxBytes": 10485760, # 10MB
"backupCount": 50,
"formatter": "detailed",
"level": "DEBUG",
},
# NEW: Player activity log
"file_player": {
"class": "logging.handlers.RotatingFileHandler",
"filename": "server/logs/player_activity.log",
"maxBytes": 10485760, # 10MB
"backupCount": 50,
"formatter": "detailed",
"level": "DEBUG",
},
},
"loggers": {
"evennia": {
"handlers": ["console", "file_server"],
"level": "INFO",
"propagate": False,
},
"portal": {
"handlers": ["console", "file_portal"],
"level": "INFO",
"propagate": False,
},
# NEW: Command audit logger
"evennia.commands": {
"handlers": ["file_commands"],
"level": "DEBUG",
"propagate": False,
},
# NEW: Movement audit logger
"evennia.objects": {
"handlers": ["file_movement"],
"level": "DEBUG",
"propagate": False,
},
# NEW: Player activity logger
"evennia.accounts": {
"handlers": ["file_player"],
"level": "DEBUG",
"propagate": False,
},
},
"root": {
"handlers": ["console"],
"level": "WARNING",
},
}
# Store additional character state for audit trail
CHARACTER_ATTRIBUTES_DEFAULT = {
"last_location": None,
"location_history": [],
"command_count": 0,
"playtime_seconds": 0,
"last_command_time": None,
}
######################################################################
# Settings given in secret_settings.py override those in this file.
######################################################################

828
tests/stress_test.py Normal file
View File

@@ -0,0 +1,828 @@
#!/usr/bin/env python3
"""
Timmy Academy - Automated Stress Test (Fenrir Protocol)
Simulates multiple concurrent players connecting to the Timmy Academy MUD,
performing random actions, and measuring system performance under load.
Usage:
python tests/stress_test.py [--players N] [--duration SECS] [--actions-per-second N] [--host HOST] [--port PORT]
Examples:
python tests/stress_test.py # defaults: 10 players, 30s, 2 actions/sec
python tests/stress_test.py --players 25 --duration 60 --actions-per-second 5
python tests/stress_test.py --host 167.99.126.228 --port 4000 --players 50
Requirements:
Python 3.8+ (stdlib only, no external dependencies)
"""
import argparse
import asyncio
import json
import os
import random
import statistics
import sys
import time
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Optional
# =============================================================================
# Configuration
# =============================================================================
DEFAULT_HOST = "localhost"
DEFAULT_PORT = 4000
DEFAULT_PLAYERS = 10
DEFAULT_DURATION = 30 # seconds
DEFAULT_ACTIONS_PER_SECOND = 2.0
TELNET_TIMEOUT = 10 # seconds
# Actions a virtual player can perform, with relative weights
PLAYER_ACTIONS = [
("look", 20), # Look at current room
("north", 8), # Move north
("south", 8), # Move south
("east", 8), # Move east
("west", 8), # Move west
("up", 4), # Move up
("down", 4), # Move down
("examine", 10), # Examine room or object
("@status", 6), # Check agent status
("@map", 5), # View map
("@academy", 3), # Academy overview
("rooms", 3), # List rooms
("smell", 5), # Smell the room
("listen", 5), # Listen to the room
("say Hello everyone!", 3), # Say something
]
# Build weighted action list
WEIGHTED_ACTIONS = []
for action, weight in PLAYER_ACTIONS:
WEIGHTED_ACTIONS.extend([action] * weight)
# =============================================================================
# Data Classes
# =============================================================================
@dataclass
class ActionResult:
"""Result of a single action execution."""
player_id: int
action: str
latency_ms: float
success: bool
error: Optional[str] = None
timestamp: float = field(default_factory=time.time)
@dataclass
class PlayerStats:
"""Accumulated stats for a single virtual player."""
player_id: int
actions_completed: int = 0
actions_failed: int = 0
errors: list = field(default_factory=list)
latencies: list = field(default_factory=list)
connected: bool = False
connect_time_ms: float = 0.0
@dataclass
class StressTestReport:
"""Final aggregated report from the stress test."""
start_time: str = ""
end_time: str = ""
duration_seconds: float = 0.0
host: str = ""
port: int = 0
num_players: int = 0
target_actions_per_second: float = 0.0
total_actions: int = 0
successful_actions: int = 0
failed_actions: int = 0
error_rate_percent: float = 0.0
throughput_actions_per_sec: float = 0.0
latency_min_ms: float = 0.0
latency_max_ms: float = 0.0
latency_mean_ms: float = 0.0
latency_median_ms: float = 0.0
latency_p90_ms: float = 0.0
latency_p95_ms: float = 0.0
latency_p99_ms: float = 0.0
connections_succeeded: int = 0
connections_failed: int = 0
avg_connect_time_ms: float = 0.0
action_breakdown: dict = field(default_factory=dict)
top_errors: list = field(default_factory=list)
player_summaries: list = field(default_factory=list)
# =============================================================================
# Telnet Client (minimal, stdlib only)
# =============================================================================
class MudClient:
"""Minimal async telnet client for Evennia MUD interaction."""
def __init__(self, host: str, port: int, player_id: int):
self.host = host
self.port = port
self.player_id = player_id
self.reader: Optional[asyncio.StreamReader] = None
self.writer: Optional[asyncio.StreamWriter] = None
self.connected = False
async def connect(self) -> float:
"""Connect to the MUD. Returns connection time in ms."""
start = time.time()
try:
self.reader, self.writer = await asyncio.wait_for(
asyncio.open_connection(self.host, self.port),
timeout=TELNET_TIMEOUT
)
# Read initial banner/login prompt
await asyncio.wait_for(self._read_until_prompt(), timeout=TELNET_TIMEOUT)
self.connected = True
return (time.time() - start) * 1000
except Exception as e:
self.connected = False
raise ConnectionError(f"Player {self.player_id}: Failed to connect: {e}")
async def disconnect(self):
"""Gracefully disconnect."""
self.connected = False
if self.writer:
try:
self.writer.close()
await self.writer.wait_closed()
except Exception:
pass
async def send_command(self, command: str) -> tuple[float, str]:
"""
Send a command and wait for response.
Returns (latency_ms, response_text).
"""
if not self.connected or not self.writer:
raise ConnectionError("Not connected")
start = time.time()
try:
# Send command with newline
self.writer.write(f"{command}\r\n".encode("utf-8"))
await self.writer.drain()
# Read response until we see a prompt character
response = await asyncio.wait_for(
self._read_until_prompt(),
timeout=TELNET_TIMEOUT
)
latency = (time.time() - start) * 1000
return latency, response
except asyncio.TimeoutError:
latency = (time.time() - start) * 1000
raise TimeoutError(f"Timeout after {latency:.0f}ms waiting for response to '{command}'")
except Exception as e:
latency = (time.time() - start) * 1000
raise RuntimeError(f"Error after {latency:.0f}ms: {e}")
async def _read_until_prompt(self, max_bytes: int = 8192) -> str:
"""Read data until we see a prompt indicator or buffer limit."""
buffer = b""
prompt_chars = (b">", b"]", b":") # Common MUD prompt endings
while len(buffer) < max_bytes:
try:
chunk = await asyncio.wait_for(
self.reader.read(1024),
timeout=2.0
)
if not chunk:
break
buffer += chunk
# Check if we've received a complete response
# (ends with prompt char or we have enough data)
if any(buffer.rstrip().endswith(pc) for pc in prompt_chars):
break
if len(buffer) > 512:
# Got enough data, don't wait forever
break
except asyncio.TimeoutError:
# No more data coming
break
except Exception:
break
return buffer.decode("utf-8", errors="replace")
# =============================================================================
# Virtual Player
# =============================================================================
class VirtualPlayer:
"""Simulates a single player performing random actions."""
def __init__(self, player_id: int, host: str, port: int,
actions_per_second: float, stop_event: asyncio.Event,
results_queue: asyncio.Queue):
self.player_id = player_id
self.host = host
self.port = port
self.actions_per_second = actions_per_second
self.stop_event = stop_event
self.results_queue = results_queue
self.stats = PlayerStats(player_id=player_id)
self.client = MudClient(host, port, player_id)
self.action_count = 0
async def run(self):
"""Main player loop: connect, perform actions, disconnect."""
try:
# Connect
connect_ms = await self.client.connect()
self.stats.connected = True
self.stats.connect_time_ms = connect_ms
# Log in with a unique character name
await self._login()
# Perform actions until stopped
interval = 1.0 / self.actions_per_second
while not self.stop_event.is_set():
action = random.choice(WEIGHTED_ACTIONS)
await self._perform_action(action)
# Jitter the interval +/- 30%
jitter = interval * random.uniform(0.7, 1.3)
try:
await asyncio.wait_for(
self.stop_event.wait(),
timeout=jitter
)
break # Stop event was set
except asyncio.TimeoutError:
pass # Timeout is expected, continue loop
except ConnectionError as e:
self.stats.errors.append(str(e))
await self.results_queue.put(ActionResult(
player_id=self.player_id,
action="connect",
latency_ms=0,
success=False,
error=str(e)
))
except Exception as e:
self.stats.errors.append(f"Unexpected: {e}")
finally:
await self.client.disconnect()
async def _login(self):
"""Handle Evennia login flow."""
# Send character name to connect/create
name = f"StressBot{self.player_id:03d}"
try:
# Evennia login: send name, then handle the response
latency, response = await self.client.send_command(name)
# If asked for password, send a simple one
if "password" in response.lower() or "create" in response.lower():
await self.client.send_command("stress123")
# Wait for game prompt
await asyncio.sleep(0.5)
except Exception:
# Login might fail if account doesn't exist, that's ok
# The player will still be in the login flow and can issue commands
pass
async def _perform_action(self, action: str):
"""Execute a single action and record results."""
self.action_count += 1
result = ActionResult(
player_id=self.player_id,
action=action,
latency_ms=0,
success=False
)
try:
latency, response = await self.client.send_command(action)
result.latency_ms = latency
result.success = True
self.stats.actions_completed += 1
self.stats.latencies.append(latency)
except Exception as e:
result.success = False
result.error = str(e)
result.latency_ms = getattr(e, 'latency_ms', 0) if hasattr(e, 'latency_ms') else 0
self.stats.actions_failed += 1
self.stats.errors.append(str(e))
await self.results_queue.put(result)
# =============================================================================
# Test Runner
# =============================================================================
class StressTestRunner:
"""Orchestrates the full stress test."""
def __init__(self, host: str, port: int, num_players: int,
duration: float, actions_per_second: float):
self.host = host
self.port = port
self.num_players = num_players
self.duration = duration
self.actions_per_second = actions_per_second
self.results: list[ActionResult] = []
self.player_stats: dict[int, PlayerStats] = {}
self.start_time: Optional[datetime] = None
self.end_time: Optional[datetime] = None
async def run(self) -> StressTestReport:
"""Execute the full stress test and return report."""
print(f"\n{'='*60}")
print(f" TIMMY ACADEMY - Fenrir Stress Test Protocol")
print(f"{'='*60}")
print(f" Target: {self.host}:{self.port}")
print(f" Players: {self.num_players}")
print(f" Duration: {self.duration}s")
print(f" Rate: {self.actions_per_second} actions/sec/player")
print(f" Expected: ~{int(self.num_players * self.actions_per_second * self.duration)} total actions")
print(f"{'='*60}\n")
self.start_time = datetime.now(timezone.utc)
stop_event = asyncio.Event()
results_queue = asyncio.Queue()
# Create virtual players
players = [
VirtualPlayer(
player_id=i,
host=self.host,
port=self.port,
actions_per_second=self.actions_per_second,
stop_event=stop_event,
results_queue=results_queue
)
for i in range(self.num_players)
]
# Start all players concurrently
print(f"[{self._timestamp()}] Launching {self.num_players} virtual players...")
tasks = [asyncio.create_task(player.run()) for player in players]
# Collect results while players run
collector_task = asyncio.create_task(
self._collect_results(results_queue, stop_event, len(players))
)
# Wait for duration
print(f"[{self._timestamp()}] Running for {self.duration} seconds...")
try:
await asyncio.sleep(self.duration)
except KeyboardInterrupt:
print("\n[!] Interrupted by user")
# Signal stop
stop_event.set()
print(f"[{self._timestamp()}] Stopping players...")
# Wait for all players to finish (with timeout)
await asyncio.wait(tasks, timeout=10)
# Drain remaining results
await asyncio.sleep(0.5)
while not results_queue.empty():
try:
result = results_queue.get_nowait()
self.results.append(result)
except asyncio.QueueEmpty:
break
self.end_time = datetime.now(timezone.utc)
# Collect player stats
for player in players:
self.player_stats[player.player_id] = player.stats
# Generate report
report = self._generate_report()
self._print_report(report)
self._save_report(report)
return report
async def _collect_results(self, queue: asyncio.Queue,
stop_event: asyncio.Event,
num_players: int):
"""Background task to collect action results."""
while not stop_event.is_set() or not queue.empty():
try:
result = await asyncio.wait_for(queue.get(), timeout=0.5)
self.results.append(result)
# Progress indicator every 50 actions
total = len(self.results)
if total % 50 == 0:
elapsed = (datetime.now(timezone.utc) - self.start_time).total_seconds()
rate = total / elapsed if elapsed > 0 else 0
print(f" [{self._timestamp()}] {total} actions completed "
f"({rate:.1f} actions/sec)")
except asyncio.TimeoutError:
continue
except Exception:
continue
def _generate_report(self) -> StressTestReport:
"""Aggregate all results into a final report."""
report = StressTestReport()
report.start_time = self.start_time.isoformat() if self.start_time else ""
report.end_time = self.end_time.isoformat() if self.end_time else ""
report.duration_seconds = (
(self.end_time - self.start_time).total_seconds()
if self.start_time and self.end_time else 0
)
report.host = self.host
report.port = self.port
report.num_players = self.num_players
report.target_actions_per_second = self.actions_per_second
# Aggregate actions
all_latencies = []
action_counts = defaultdict(int)
action_latencies = defaultdict(list)
error_counts = defaultdict(int)
for r in self.results:
action_counts[r.action] += 1
if r.success:
all_latencies.append(r.latency_ms)
action_latencies[r.action].append(r.latency_ms)
else:
report.failed_actions += 1
if r.error:
error_counts[r.error] += 1
report.total_actions = len(self.results)
report.successful_actions = report.total_actions - report.failed_actions
if report.total_actions > 0:
report.error_rate_percent = (report.failed_actions / report.total_actions) * 100
if report.duration_seconds > 0:
report.throughput_actions_per_sec = report.total_actions / report.duration_seconds
# Latency percentiles
if all_latencies:
sorted_lat = sorted(all_latencies)
report.latency_min_ms = sorted_lat[0]
report.latency_max_ms = sorted_lat[-1]
report.latency_mean_ms = statistics.mean(sorted_lat)
report.latency_median_ms = statistics.median(sorted_lat)
report.latency_p90_ms = sorted_lat[int(len(sorted_lat) * 0.90)]
report.latency_p95_ms = sorted_lat[int(len(sorted_lat) * 0.95)]
report.latency_p99_ms = sorted_lat[int(len(sorted_lat) * 0.99)]
# Action breakdown
for action, count in sorted(action_counts.items(), key=lambda x: -x[1]):
lats = action_latencies.get(action, [])
report.action_breakdown[action] = {
"count": count,
"avg_latency_ms": round(statistics.mean(lats), 2) if lats else 0,
"success_rate": round(
(len(lats) / count * 100) if count > 0 else 0, 1
)
}
# Connection stats
connect_times = []
for ps in self.player_stats.values():
if ps.connected:
report.connections_succeeded += 1
connect_times.append(ps.connect_time_ms)
else:
report.connections_failed += 1
if connect_times:
report.avg_connect_time_ms = statistics.mean(connect_times)
# Top errors
report.top_errors = [
{"error": err, "count": count}
for err, count in sorted(error_counts.items(), key=lambda x: -x[1])[:10]
]
# Player summaries
for pid, ps in sorted(self.player_stats.items()):
report.player_summaries.append({
"player_id": pid,
"connected": ps.connected,
"actions_completed": ps.actions_completed,
"actions_failed": ps.actions_failed,
"avg_latency_ms": round(statistics.mean(ps.latencies), 2) if ps.latencies else 0,
"error_count": len(ps.errors),
})
return report
def _print_report(self, report: StressTestReport):
"""Print formatted report to console."""
print(f"\n{'='*60}")
print(f" STRESS TEST REPORT - Fenrir Protocol")
print(f"{'='*60}")
print(f"\n --- Test Parameters ---")
print(f" Start: {report.start_time}")
print(f" End: {report.end_time}")
print(f" Duration: {report.duration_seconds:.1f}s")
print(f" Target: {report.host}:{report.port}")
print(f" Players: {report.num_players}")
print(f" Rate/Player:{report.target_actions_per_second} actions/sec")
print(f"\n --- Throughput ---")
print(f" Total Actions: {report.total_actions}")
print(f" Successful: {report.successful_actions}")
print(f" Failed: {report.failed_actions}")
print(f" Error Rate: {report.error_rate_percent:.2f}%")
print(f" Throughput: {report.throughput_actions_per_sec:.2f} actions/sec")
print(f"\n --- Latency (ms) ---")
print(f" Min: {report.latency_min_ms:.1f}")
print(f" Mean: {report.latency_mean_ms:.1f}")
print(f" Median: {report.latency_median_ms:.1f}")
print(f" P90: {report.latency_p90_ms:.1f}")
print(f" P95: {report.latency_p95_ms:.1f}")
print(f" P99: {report.latency_p99_ms:.1f}")
print(f" Max: {report.latency_max_ms:.1f}")
print(f"\n --- Connections ---")
print(f" Succeeded: {report.connections_succeeded}")
print(f" Failed: {report.connections_failed}")
print(f" Avg Time: {report.avg_connect_time_ms:.1f}ms")
print(f"\n --- Action Breakdown ---")
print(f" {'Action':<20} {'Count':>8} {'Avg(ms)':>10} {'Success%':>10}")
print(f" {'-'*48}")
for action, info in report.action_breakdown.items():
print(f" {action:<20} {info['count']:>8} "
f"{info['avg_latency_ms']:>10.1f} {info['success_rate']:>9.1f}%")
if report.top_errors:
print(f"\n --- Top Errors ---")
for err_info in report.top_errors[:5]:
err_msg = err_info['error'][:50]
print(f" [{err_info['count']}x] {err_msg}")
# Player summary (compact)
print(f"\n --- Player Summary (top 10 by actions) ---")
sorted_players = sorted(
report.player_summaries,
key=lambda p: p['actions_completed'],
reverse=True
)[:10]
print(f" {'Player':<12} {'Done':>6} {'Fail':>6} {'Avg(ms)':>10} {'Status':<10}")
print(f" {'-'*48}")
for ps in sorted_players:
status = "OK" if ps['connected'] else "FAILED"
print(f" #{ps['player_id']:<11} {ps['actions_completed']:>6} "
f"{ps['actions_failed']:>6} {ps['avg_latency_ms']:>10.1f} {status}")
print(f"\n{'='*60}")
print(f" Verdict: ", end="")
if report.error_rate_percent < 1 and report.latency_p95_ms < 1000:
print("PASSED - System handles load well")
elif report.error_rate_percent < 5 and report.latency_p95_ms < 3000:
print("WARNING - Acceptable but room for improvement")
else:
print("NEEDS ATTENTION - High error rate or latency")
print(f"{'='*60}\n")
def _save_report(self, report: StressTestReport):
"""Save report to JSON file."""
report_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "tests")
os.makedirs(report_dir, exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
filename = os.path.join(report_dir, f"stress_report_{timestamp}.json")
# Convert to dict for JSON serialization
report_dict = {
"test_name": "Fenrir Stress Test",
"start_time": report.start_time,
"end_time": report.end_time,
"duration_seconds": report.duration_seconds,
"target": {"host": report.host, "port": report.port},
"parameters": {
"num_players": report.num_players,
"target_actions_per_second": report.target_actions_per_second,
},
"results": {
"total_actions": report.total_actions,
"successful_actions": report.successful_actions,
"failed_actions": report.failed_actions,
"error_rate_percent": round(report.error_rate_percent, 2),
"throughput_actions_per_sec": round(report.throughput_actions_per_sec, 2),
},
"latency_ms": {
"min": round(report.latency_min_ms, 2),
"mean": round(report.latency_mean_ms, 2),
"median": round(report.latency_median_ms, 2),
"p90": round(report.latency_p90_ms, 2),
"p95": round(report.latency_p95_ms, 2),
"p99": round(report.latency_p99_ms, 2),
"max": round(report.latency_max_ms, 2),
},
"connections": {
"succeeded": report.connections_succeeded,
"failed": report.connections_failed,
"avg_connect_time_ms": round(report.avg_connect_time_ms, 2),
},
"action_breakdown": report.action_breakdown,
"top_errors": report.top_errors,
"player_summaries": report.player_summaries,
}
with open(filename, "w") as f:
json.dump(report_dict, f, indent=2)
print(f" Report saved: {filename}")
@staticmethod
def _timestamp() -> str:
return datetime.now().strftime("%H:%M:%S")
# =============================================================================
# Self-Test (no server required)
# =============================================================================
def run_self_test():
"""
Run a lightweight self-test that validates the stress test logic
without requiring a running MUD server.
"""
print(f"\n{'='*60}")
print(f" SELF-TEST MODE - Validation Suite")
print(f"{'='*60}\n")
passed = 0
failed = 0
def check(name, condition, detail=""):
nonlocal passed, failed
if condition:
print(f" [PASS] {name}")
passed += 1
else:
print(f" [FAIL] {name} - {detail}")
failed += 1
# Test 1: Weighted actions list is populated
check("Weighted actions list not empty", len(WEIGHTED_ACTIONS) > 0)
check("Weighted actions has correct items",
"look" in WEIGHTED_ACTIONS and "north" in WEIGHTED_ACTIONS)
# Test 2: ActionResult creation
result = ActionResult(player_id=1, action="look", latency_ms=42.5, success=True)
check("ActionResult dataclass works", result.player_id == 1 and result.success)
check("ActionResult has timestamp", result.timestamp > 0)
# Test 3: PlayerStats creation
stats = PlayerStats(player_id=1)
check("PlayerStats dataclass works", stats.player_id == 1 and stats.actions_completed == 0)
# Test 4: StressTestReport creation
report = StressTestReport()
check("StressTestReport dataclass works", report.total_actions == 0)
# Test 5: Action distribution is reasonable
action_freq = defaultdict(int)
for a in WEIGHTED_ACTIONS:
action_freq[a] += 1
check("Multiple action types present", len(action_freq) >= 10)
check("'look' is most common action", action_freq["look"] > action_freq["@academy"])
# Test 6: Report generation with mock data
runner = StressTestRunner("localhost", 4000, 5, 10, 1.0)
runner.start_time = datetime.now(timezone.utc)
runner.end_time = datetime.now(timezone.utc)
# Add mock results
for i in range(100):
runner.results.append(ActionResult(
player_id=i % 5,
action=random.choice(WEIGHTED_ACTIONS),
latency_ms=random.uniform(10, 500),
success=random.random() > 0.05
))
# Add mock player stats
for i in range(5):
runner.player_stats[i] = PlayerStats(
player_id=i,
actions_completed=18,
actions_failed=2,
connected=True,
connect_time_ms=random.uniform(50, 200),
latencies=[random.uniform(10, 500) for _ in range(18)]
)
report = runner._generate_report()
check("Report total_actions correct", report.total_actions == 100)
check("Report has latency stats", report.latency_mean_ms > 0)
check("Report has action breakdown", len(report.action_breakdown) > 0)
check("Report throughput calculated", report.throughput_actions_per_sec > 0)
check("Report connection stats", report.connections_succeeded == 5)
# Test 7: JSON serialization
try:
report_dict = {
"total_actions": report.total_actions,
"latency_ms": {
"mean": round(report.latency_mean_ms, 2),
"p95": round(report.latency_p95_ms, 2),
},
"action_breakdown": report.action_breakdown,
}
json_str = json.dumps(report_dict)
check("Report JSON serializable", len(json_str) > 10)
except Exception as e:
check("Report JSON serializable", False, str(e))
# Summary
total = passed + failed
print(f"\n Results: {passed}/{total} passed, {failed} failed")
print(f"{'='*60}\n")
return failed == 0
# =============================================================================
# Main Entry Point
# =============================================================================
def main():
parser = argparse.ArgumentParser(
description="Timmy Academy - Fenrir Stress Test Protocol",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
%(prog)s # default: 10 players, 30s, 2 act/s
%(prog)s --players 50 --duration 120 # heavy load test
%(prog)s --host 167.99.126.228 --port 4000 # test live server
%(prog)s --self-test # validate without server
"""
)
parser.add_argument("--players", type=int, default=DEFAULT_PLAYERS,
help=f"Number of concurrent virtual players (default: {DEFAULT_PLAYERS})")
parser.add_argument("--duration", type=float, default=DEFAULT_DURATION,
help=f"Test duration in seconds (default: {DEFAULT_DURATION})")
parser.add_argument("--actions-per-second", type=float, default=DEFAULT_ACTIONS_PER_SECOND,
help=f"Actions per second per player (default: {DEFAULT_ACTIONS_PER_SECOND})")
parser.add_argument("--host", type=str, default=DEFAULT_HOST,
help=f"MUD server host (default: {DEFAULT_HOST})")
parser.add_argument("--port", type=int, default=DEFAULT_PORT,
help=f"MUD server telnet port (default: {DEFAULT_PORT})")
parser.add_argument("--self-test", action="store_true",
help="Run self-test validation (no server required)")
parser.add_argument("--json", action="store_true",
help="Output report as JSON to stdout")
args = parser.parse_args()
if args.self_test:
success = run_self_test()
sys.exit(0 if success else 1)
# Run the stress test
runner = StressTestRunner(
host=args.host,
port=args.port,
num_players=args.players,
duration=args.duration,
actions_per_second=args.actions_per_second,
)
try:
report = asyncio.run(runner.run())
if args.json:
# Re-output as JSON
print(json.dumps({
"total_actions": report.total_actions,
"throughput": report.throughput_actions_per_sec,
"error_rate": report.error_rate_percent,
"latency_p95": report.latency_p95_ms,
}, indent=2))
except KeyboardInterrupt:
print("\n[!] Test interrupted")
sys.exit(130)
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,161 @@
"""
AuditedCharacter - A character typeclass with full audit logging.
Tracks every movement, command, and action for complete visibility
into player activity. Supports configurable log rotation to prevent
unbounded growth.
"""
import time
from datetime import datetime
from evennia import DefaultCharacter
from evennia.utils import logger
# Default audit retention limits
DEFAULT_MAX_HISTORY = 500
DEFAULT_MAX_LOG_ENTRIES = 100 # Max AUDIT log lines per character per server session
class AuditedCharacter(DefaultCharacter):
"""
Character typeclass with comprehensive audit logging.
Tracks:
- Every room entered/exited with timestamps
- Total playtime
- Command count
- Last known location
- Full location history (rotated)
Configurable via class attributes:
audit_max_history (int): Max location_history entries kept in db (default 500)
audit_max_log_entries (int): Max AUDIT log lines per server session (default 100)
"""
audit_max_history = DEFAULT_MAX_HISTORY
audit_max_log_entries = DEFAULT_MAX_LOG_ENTRIES
def at_object_creation(self):
"""Set up audit attributes when character is created."""
super().at_object_creation()
# Initialize audit tracking attributes
self.db.location_history = [] # List of {room, timestamp, action}
self.db.command_count = 0
self.db.total_playtime = 0 # in seconds
self.db.session_start_time = None
self.db.last_location = None
self.db.audit_log_count = 0 # Tracks log entries this session for rate limiting
logger.log_info(f"AUDIT: Character '{self.key}' created at {datetime.utcnow()}")
def _audit_log(self, message):
"""Write an audit log entry with rate limiting per server session."""
count = (self.db.audit_log_count or 0) + 1
if count <= self.audit_max_log_entries:
logger.log_info(message)
if count == self.audit_max_log_entries:
logger.log_info(
f"AUDIT: {self.key} reached log limit ({self.audit_max_log_entries}) "
f"- suppressing further audit logs this session"
)
self.db.audit_log_count = count
def prune_audit_history(self, max_entries=None):
"""Trim location_history to max_entries. Returns number of entries removed."""
max_entries = max_entries or self.audit_max_history
history = self.db.location_history or []
if len(history) > max_entries:
removed = len(history) - max_entries
self.db.location_history = history[-max_entries:]
return removed
return 0
def at_pre_move(self, destination, **kwargs):
"""Called before moving - log departure."""
current = self.location
if current:
self._audit_log(
f"AUDIT MOVE: {self.key} leaving {current.key} "
f"-> {destination.key if destination else 'None'}"
)
return super().at_pre_move(destination, **kwargs)
def at_post_move(self, source_location, **kwargs):
"""Called after moving - record arrival in audit trail."""
destination = self.location
timestamp = datetime.utcnow().isoformat()
# Update location history
history = self.db.location_history or []
history.append({
"from": source_location.key if source_location else "Nowhere",
"to": destination.key if destination else "Nowhere",
"timestamp": timestamp,
"coord": getattr(destination, 'db', {}).get('coord', None) if destination else None
})
# Rotate: keep last audit_max_history movements
self.db.location_history = history[-self.audit_max_history:]
self.db.last_location = destination.key if destination else None
self._audit_log(
f"AUDIT MOVE: {self.key} arrived at "
f"{destination.key if destination else 'None'} from "
f"{source_location.key if source_location else 'None'}"
)
super().at_post_move(source_location, **kwargs)
def at_pre_cmd(self, cmd, args):
"""Called before executing any command."""
# Increment command counter
self.db.command_count = (self.db.command_count or 0) + 1
self.db.last_command_time = datetime.utcnow().isoformat()
# Log command (excluding sensitive commands like password)
cmd_name = cmd.key if cmd else "unknown"
if cmd_name not in ("password", "@password"):
self._audit_log(
f"AUDIT CMD: {self.key} executed '{cmd_name}' "
f"args: '{args[:50] if args else ''}'"
)
super().at_pre_cmd(cmd, args)
def at_pre_puppet(self, account, session, **kwargs):
"""Called when account takes control of character."""
self.db.session_start_time = time.time()
self.db.audit_log_count = 0 # Reset log counter each session
self._audit_log(
f"AUDIT SESSION: {self.key} puppeted by {account.key} "
f"at {datetime.utcnow()}"
)
super().at_pre_puppet(account, session, **kwargs)
def at_post_unpuppet(self, account, session, **kwargs):
"""Called when account releases control of character."""
start_time = self.db.session_start_time
if start_time:
session_duration = time.time() - start_time
self.db.total_playtime = (self.db.total_playtime or 0) + session_duration
self._audit_log(
f"AUDIT SESSION: {self.key} unpuppeted by {account.key} "
f"- session lasted {session_duration:.0f}s, "
f"total playtime {self.db.total_playtime:.0f}s"
)
self.db.session_start_time = None
super().at_post_unpuppet(account, session, **kwargs)
def get_audit_summary(self):
"""Return a summary of this character's audit trail."""
history = self.db.location_history or []
return {
"name": self.key,
"location": self.location.key if self.location else "None",
"commands_executed": self.db.command_count or 0,
"total_playtime_seconds": self.db.total_playtime or 0,
"total_playtime_hours": round((self.db.total_playtime or 0) / 3600, 2),
"locations_visited": len(history),
"last_location_change": history[-1] if history else None,
"last_command": self.db.last_command_time,
}

482
world/rebuild_world.py Normal file
View File

@@ -0,0 +1,482 @@
#!/usr/bin/env python
"""
Timmy Academy - World Rebuild Script (Second Pass)
Resets all room descriptions, typeclasses, and attributes from wing module
source files. Sets wing membership, verifies exits, moves characters to Limbo,
and configures the Public channel.
Safe to rerun (idempotent).
Usage:
cd /path/to/timmy-academy
source /root/workspace/evennia-venv/bin/activate
python world/rebuild_world.py
"""
import os
import sys
import re
import ast
os.environ["DJANGO_SETTINGS_MODULE"] = "server.conf.settings"
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import django
django.setup()
from evennia.objects.models import ObjectDB
from evennia.comms.models import ChannelDB
BANNER = """
============================================================
TIMMY ACADEMY - WORLD REBUILD (Second Pass)
Applying rich descriptions, typeclasses, atmospheres
============================================================
"""
WORLD_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)))
def parse_wing_file(filename):
"""Parse a wing .py file and extract class attributes by class name.
Returns dict: { ClassName: { desc, atmosphere, objects, aliases, exits, features } }
"""
filepath = os.path.join(WORLD_DIR, filename)
with open(filepath, "r") as f:
source = f.read()
results = {}
# Split by class definitions
class_pattern = re.compile(r'^class\s+(\w+)\(', re.MULTILINE)
class_positions = [(m.group(1), m.start()) for m in class_pattern.finditer(source)]
for i, (cls_name, start_pos) in enumerate(class_positions):
# Get the source for this class (up to next class or end)
if i + 1 < len(class_positions):
end_pos = class_positions[i + 1][1]
else:
end_pos = len(source)
cls_source = source[start_pos:end_pos]
attrs = {}
# Extract self.db.desc = """..."""
desc_match = re.search(
r'self\.db\.desc\s*=\s*"""(.*?)"""',
cls_source, re.DOTALL
)
if desc_match:
desc = desc_match.group(1)
# Clean up leading/trailing whitespace while preserving internal formatting
lines = desc.split('\n')
# Remove common leading whitespace
non_empty = [l for l in lines if l.strip()]
if non_empty:
min_indent = min(len(l) - len(l.lstrip()) for l in non_empty)
lines = [l[min_indent:] if len(l) >= min_indent else l for l in lines]
desc = '\n'.join(lines).strip()
attrs["desc"] = desc
# Extract self.db.atmosphere = { ... }
atmo_match = re.search(
r'self\.db\.atmosphere\s*=\s*(\{[^}]+\})',
cls_source, re.DOTALL
)
if atmo_match:
try:
attrs["atmosphere"] = ast.literal_eval(atmo_match.group(1))
except Exception:
pass
# Extract self.db.objects = [ ... ]
objs_match = re.search(
r'self\.db\.objects\s*=\s*(\[[^\]]+\])',
cls_source, re.DOTALL
)
if objs_match:
try:
attrs["objects"] = ast.literal_eval(objs_match.group(1))
except Exception:
pass
# Extract self.db.exits = { ... }
exits_match = re.search(
r'self\.db\.exits\s*=\s*(\{[^}]+\})',
cls_source, re.DOTALL
)
if exits_match:
try:
attrs["exits_info"] = ast.literal_eval(exits_match.group(1))
except Exception:
pass
# Extract self.aliases = [ ... ]
aliases_match = re.search(
r'self\.aliases\s*=\s*(\[[^\]]+\])',
cls_source, re.DOTALL
)
if aliases_match:
try:
attrs["aliases"] = ast.literal_eval(aliases_match.group(1))
except Exception:
pass
# Extract self.db.features = { ... }
feats_match = re.search(
r'self\.db\.features\s*=\s*(\{[^}]+\})',
cls_source, re.DOTALL
)
if feats_match:
try:
attrs["features"] = ast.literal_eval(feats_match.group(1))
except Exception:
pass
results[cls_name] = attrs
return results
# ============================================================
# Room ID -> (typeclass_path, wing_name, source_file, class_name)
# ============================================================
ROOM_CONFIG = {
# Dormitory Wing
3: ("world.dormitory_entrance.DormitoryEntrance", "dormitory", "dormitory_entrance.py", "DormitoryEntrance"),
4: ("world.dormitory_entrance.DormitoryLodgingMaster", "dormitory", "dormitory_entrance.py", "DormitoryLodgingMaster"),
5: ("world.dormitory_entrance.CorridorOfRest", "dormitory", "dormitory_entrance.py", "CorridorOfRest"),
6: ("world.dormitory_entrance.NoviceDormitoryHall", "dormitory", "dormitory_entrance.py", "NoviceDormitoryHall"),
7: ("world.dormitory_entrance.MasterSuitesApproach", "dormitory", "dormitory_entrance.py", "MasterSuitesApproach"),
# Commons Wing
8: ("world.commons_wing.GrandCommonsHall", "commons", "commons_wing.py", "GrandCommonsHall"),
9: ("world.commons_wing.HearthsideDining", "commons", "commons_wing.py", "HearthsideDining"),
10: ("world.commons_wing.ScholarsCorner", "commons", "commons_wing.py", "ScholarsCorner"),
11: ("world.commons_wing.EntertainmentGallery", "commons", "commons_wing.py", "EntertainmentGallery"),
12: ("world.commons_wing.UpperCommonsBalcony", "commons", "commons_wing.py", "UpperCommonsBalcony"),
# Workshop Wing
13: ("world.workshop_wing.WorkshopEntrance", "workshop", "workshop_wing.py", "WorkshopEntrance"),
14: ("world.workshop_wing.GreatSmithy", "workshop", "workshop_wing.py", "GreatSmithy"),
15: ("world.workshop_wing.AlchemyLaboratories", "workshop", "workshop_wing.py", "AlchemyLaboratories"),
16: ("world.workshop_wing.WoodworkingStudio", "workshop", "workshop_wing.py", "WoodworkingStudio"),
17: ("world.workshop_wing.ArtificingChambers", "workshop", "workshop_wing.py", "ArtificingChambers"),
# Gardens Wing
18: ("world.gardens_wing.GardensEntrance", "gardens", "gardens_wing.py", "GardensEntrance"),
19: ("world.gardens_wing.HerbGardens", "gardens", "gardens_wing.py", "HerbGardens"),
20: ("world.gardens_wing.EnchantedGrove", "gardens", "gardens_wing.py", "EnchantedGrove"),
21: ("world.gardens_wing.GreenhouseComplex", "gardens", "gardens_wing.py", "GreenhouseComplex"),
22: ("world.gardens_wing.SacredGrove", "gardens", "gardens_wing.py", "SacredGrove"),
}
# Wing metadata
WING_INFO = {
"hub": {
"name": "Academy Central Hub",
"color": "|w",
"rooms": [2],
},
"dormitory": {
"name": "Dormitory Wing",
"color": "|c",
"rooms": [3, 4, 5, 6, 7],
},
"commons": {
"name": "Commons Wing",
"color": "|y",
"rooms": [8, 9, 10, 11, 12],
},
"workshop": {
"name": "Workshop Wing",
"color": "|r",
"rooms": [13, 14, 15, 16, 17],
},
"gardens": {
"name": "Gardens Wing",
"color": "|g",
"rooms": [18, 19, 20, 21, 22],
},
}
# Limbo description
LIMBO_DESC = """|wAcademy Central Hub|n
You stand at the |ycrossroads of the Timmy Academy|n, a vast circular
chamber where all wings converge. The domed ceiling overhead displays
a |cshimmering map of the academy|n, with glowing pathways indicating
the four great wings branching outward.
Four |yornate archways|n mark the cardinal directions:
|cNorth|n - The |cDormitory Wing|n (rest and residence)
|yEast|n - The |yCommons Wing|n (gathering and fellowship)
|rSouth|n - The |rWorkshop Wing|n (craft and creation)
|gWest|n - The |gGardens Wing|n (nature and growth)
A |ycrystal fountain|n at the center murmurs softly, its waters cycling
through colors that reflect the mood of the academy. Carved into the
marble floor is the academy motto: |x"Together we learn, together we grow."|n
This is where all journeys begin and all paths return."""
LIMBO_ATMOSPHERE = {
"mood": "welcoming, central, transitional, purposeful",
"lighting": "soft crystal glow from the fountain, archway light",
"sounds": "fountain murmuring, distant echoes from all wings",
"smells": "clean stone, fountain mist, faint traces of all wings",
"temperature": "perfectly comfortable, magically regulated",
}
LIMBO_OBJECTS = [
"crystal fountain cycling through colors",
"ornate archways to four wings",
"domed ceiling with shimmering academy map",
"marble floor with carved motto",
"welcoming bench near the fountain",
]
def get_room(room_id):
"""Get a room object by ID."""
try:
return ObjectDB.objects.get(id=room_id)
except ObjectDB.DoesNotExist:
print(f" [ERROR] Room #{room_id} not found!")
return None
def rebuild_rooms():
"""Apply rich descriptions, typeclasses, and attributes to all rooms."""
print("\n[1/5] REBUILDING ROOMS - Applying typeclasses and descriptions")
print("-" * 60)
# Parse all wing source files
print(" Parsing wing source files...")
parsed_files = {}
for filename in ["dormitory_entrance.py", "commons_wing.py", "workshop_wing.py", "gardens_wing.py"]:
parsed_files[filename] = parse_wing_file(filename)
class_count = len(parsed_files[filename])
print(f" {filename}: {class_count} classes found")
# Handle Limbo specially
limbo = get_room(2)
if limbo:
limbo.db.desc = LIMBO_DESC
limbo.db.atmosphere = LIMBO_ATMOSPHERE
limbo.db.objects = LIMBO_OBJECTS
limbo.db.wing = "hub"
limbo.db.wing_info = WING_INFO
print(f"\n [OK] #{limbo.id} {limbo.db_key} - hub desc={len(LIMBO_DESC)}ch")
# Apply attributes from parsed sources to each room
for room_id, (tc_path, wing, src_file, cls_name) in sorted(ROOM_CONFIG.items()):
room = get_room(room_id)
if not room:
continue
old_tc = room.db_typeclass_path
# Set typeclass path
if old_tc != tc_path:
room.db_typeclass_path = tc_path
room.save()
# Get parsed attributes for this class
attrs = parsed_files.get(src_file, {}).get(cls_name, {})
if not attrs:
print(f" [WARN] #{room_id} {room.db_key} - no parsed attrs for {cls_name}")
continue
# Apply description
desc_len = 0
if "desc" in attrs:
room.db.desc = attrs["desc"]
desc_len = len(attrs["desc"])
# Apply atmosphere
if "atmosphere" in attrs:
room.db.atmosphere = attrs["atmosphere"]
# Apply notable objects list
if "objects" in attrs:
room.db.objects = attrs["objects"]
# Apply exit info (metadata, not actual exits)
if "exits_info" in attrs:
room.db.exits_info = attrs["exits_info"]
# Apply features if present
if "features" in attrs:
room.db.features = attrs["features"]
# Apply aliases
if "aliases" in attrs:
room.aliases.clear()
room.aliases.add(attrs["aliases"])
# Set wing membership
room.db.wing = wing
tc_label = "tc CHANGED" if old_tc != tc_path else "tc ok"
has_atmo = "atmo" if "atmosphere" in attrs else "no-atmo"
print(f" [OK] #{room_id} {room.db_key} - {wing}, {tc_label}, desc={desc_len}ch, {has_atmo}")
print(f"\n Total: {len(ROOM_CONFIG) + 1} rooms configured")
def verify_exits():
"""Check all exits are properly connected bidirectionally."""
print("\n[2/5] VERIFYING EXITS - Checking bidirectional connections")
print("-" * 60)
exits = ObjectDB.objects.filter(db_typeclass_path__icontains="exit")
exit_pairs = {}
for ex in exits:
src = ex.db_location
dst = ex.db_destination
if src and dst:
pair_key = (src.id, dst.id)
exit_pairs[pair_key] = ex.db_key
issues = []
for (src_id, dst_id), exit_key in sorted(exit_pairs.items()):
reverse = (dst_id, src_id)
if reverse not in exit_pairs:
src_room = get_room(src_id)
dst_room = get_room(dst_id)
src_name = src_room.db_key if src_room else f"#{src_id}"
dst_name = dst_room.db_key if dst_room else f"#{dst_id}"
issues.append(f" [WARN] No return: '{exit_key}' {src_name} -> {dst_name}")
if issues:
for issue in issues:
print(issue)
print(f"\n {len(issues)} one-way exits (extra exits from fix_world.py - not harmful)")
else:
print(f" [OK] All {len(exits)} exits have bidirectional connections")
wing_exits = {"hub": 0, "dormitory": 0, "commons": 0, "workshop": 0, "gardens": 0}
for ex in exits:
src = ex.db_location
if src:
for wing_key, info in WING_INFO.items():
if src.id in info["rooms"]:
wing_exits[wing_key] += 1
break
print(f"\n Exit distribution:")
for wing_key, count in wing_exits.items():
print(f" {wing_key:12s}: {count} exits")
print(f" {'total':12s}: {len(exits)} exits")
def manage_characters():
"""Move all characters to Limbo."""
print("\n[3/5] MANAGING CHARACTERS - Ensuring all are in Limbo")
print("-" * 60)
limbo = get_room(2)
if not limbo:
print(" [ERROR] Limbo not found!")
return
chars = ObjectDB.objects.filter(db_typeclass_path__icontains="character")
for char in chars:
loc = char.db_location
if loc and loc.id == limbo.id:
print(f" [OK] {char.db_key} already in Limbo")
else:
char.move_to(limbo, quiet=True)
old_loc = loc.db_key if loc else "None"
print(f" [MOVED] {char.db_key} from {old_loc} -> Limbo")
print(f"\n Total: {chars.count()} characters managed")
def setup_channels():
"""Set up the Public channel if it doesn't exist."""
print("\n[4/5] SETTING UP CHANNELS")
print("-" * 60)
try:
pub = ChannelDB.objects.filter(db_key="Public").first()
if pub:
print(f" [OK] Public channel already exists (ID={pub.id})")
else:
from evennia.utils.create import create_channel
pub = create_channel(
"Public",
desc="General discussion channel for all academy members.",
locks="control:perm(Admin);listen:all();send:all()",
)
print(f" [CREATED] Public channel (ID={pub.id})")
except Exception as e:
print(f" [WARN] Channel setup: {e}")
try:
info = ChannelDB.objects.filter(db_key="MudInfo").first()
if info:
print(f" [OK] MudInfo channel already exists (ID={info.id})")
except Exception:
pass
def print_summary():
"""Print a summary of the world state."""
print("\n[5/5] WORLD SUMMARY")
print("-" * 60)
rooms = ObjectDB.objects.filter(id__gte=2, id__lte=22)
exits = ObjectDB.objects.filter(db_typeclass_path__icontains="exit")
chars = ObjectDB.objects.filter(db_typeclass_path__icontains="character")
channels = ChannelDB.objects.all()
print(f" Rooms: {rooms.count()}")
print(f" Exits: {exits.count()}")
print(f" Characters: {chars.count()}")
print(f" Channels: {channels.count()}")
print("\n Rooms by wing:")
for wing_key, info in WING_INFO.items():
wing_rooms = [get_room(rid) for rid in info["rooms"]]
wing_rooms = [r for r in wing_rooms if r]
names = [r.db_key for r in wing_rooms]
print(f" {info['name']} ({len(names)} rooms):")
for name in names:
print(f" - {name}")
print("\n Description verification:")
all_good = True
for room_id in range(2, 23):
room = get_room(room_id)
if room:
desc = room.attributes.get("desc", "")
atmo = room.attributes.get("atmosphere", None)
desc_len = len(desc) if desc else 0
has_atmo = "atmo" if atmo else "NO-ATMO"
status = "OK" if desc_len > 200 else "SHORT"
if status != "OK":
all_good = False
print(f" [{status:5s}] #{room_id} {room.db_key}: {desc_len}ch, {has_atmo}")
print("\n Characters:")
for char in chars:
loc = char.db_location
print(f" - {char.db_key} (in {loc.db_key if loc else 'nowhere'})")
print("\n" + "=" * 60)
if all_good:
print(" REBUILD COMPLETE - ALL ROOMS HAVE RICH DESCRIPTIONS")
else:
print(" REBUILD COMPLETE")
print("=" * 60)
if __name__ == "__main__":
print(BANNER)
rebuild_rooms()
verify_exits()
manage_characters()
setup_channels()
print_summary()