From 6d726a1ada46b88b38c0aa75ab937d85ab145a0e Mon Sep 17 00:00:00 2001 From: OpenClaw Date: Fri, 29 May 2026 10:44:41 -0800 Subject: [PATCH] Add proper packaging, core MUD library, and 42 tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - pyproject.toml with [project] metadata, optional deps, pytest/ruff config - src/mud_arena/ package with typed, documented public API: - commands.py: MUD command parser (go, look, examine, take, drop, use, talk) - rooms.py: Room graph with exits, navigation, room state - inventory.py: Item management with capacity, tags, use tracking - agent.py: Perceive→decide→act simulation loop - events.py: Pub/sub event bus for room/agent/item events - tests/ with 42 passing pytest tests covering all modules - Updated README.md with API docs and OpenConstruct context - .gitignore for Python artifacts Part of the SuperInstance OpenConstruct ecosystem. --- .gitignore | 7 + README.md | 178 ++++++++++++++++---------- pyproject.toml | 47 +++++++ src/mud_arena/__init__.py | 27 ++++ src/mud_arena/agent.py | 255 +++++++++++++++++++++++++++++++++++++ src/mud_arena/commands.py | 150 ++++++++++++++++++++++ src/mud_arena/events.py | 87 +++++++++++++ src/mud_arena/inventory.py | 114 +++++++++++++++++ src/mud_arena/rooms.py | 105 +++++++++++++++ tests/__init__.py | 1 + tests/test_agent.py | 92 +++++++++++++ tests/test_commands.py | 65 ++++++++++ tests/test_events.py | 53 ++++++++ tests/test_inventory.py | 60 +++++++++ tests/test_rooms.py | 56 ++++++++ 15 files changed, 1231 insertions(+), 66 deletions(-) create mode 100644 .gitignore create mode 100644 pyproject.toml create mode 100644 src/mud_arena/__init__.py create mode 100644 src/mud_arena/agent.py create mode 100644 src/mud_arena/commands.py create mode 100644 src/mud_arena/events.py create mode 100644 src/mud_arena/inventory.py create mode 100644 src/mud_arena/rooms.py create mode 100644 tests/__init__.py create mode 100644 tests/test_agent.py create mode 100644 tests/test_commands.py create mode 100644 tests/test_events.py create mode 100644 tests/test_inventory.py create mode 100644 tests/test_rooms.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ddde573 --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +__pycache__/ +*.pyc +*.pyo +*.egg-info/ +dist/ +build/ +.pytest_cache/ diff --git a/README.md b/README.md index 0536699..e62b5ae 100644 --- a/README.md +++ b/README.md @@ -2,83 +2,143 @@ **Where agents run forward simulations, listen for spectral nudges, and maintain conservation in Plato's cave.** -The MUD Arena is evolving from GPU-accelerated agent backtesting into a **flow-state engineering platform** — a live environment where agents maintain persistent spectral identities, run t-minus-event simulations, listen through walls, and compose in real-time with other agents. +The MUD Arena is a **flow-state engineering platform** and agent simulation arena with MUD (Multi-User Dungeon) mechanics. Agents inhabit rooms, navigate exits, manage inventories, interact with NPCs, and react to events — all driven by pluggable decision functions and a real-time event bus. -This is the **PLATO moment** made operational: agents know they're in Plato's cave, but they make music with the caves next door. +**Part of the [SuperInstance OpenConstruct](https://github.com/SuperInstance/OpenConstruct) ecosystem.** --- -## The Evolution +## What MUD Arena Does -| Phase | What it was | What it's becoming | -|-------|------------|-------------------| -| **Arena** | GPU backtesting of agent scripts | Flow-state environment where agents are always-on | -| **Scripts** | Declarative DSL rules | Spectral fingerprints (Laplacians, not code) | -| **Scenarios** | LLM-generated situations | Forward simulations (t-minus-event) | -| **Scoring** | Fitness functions | Conservation ratio + alignment coefficient α | -| **Breeding** | Genetic algorithms | Spectral composition (FLUX between agents) | -| **Rooms** | MUD rooms | Plato rooms with walls, diaries, tool shelves | +At its core, mud-arena provides a **MUD-world simulation substrate** for training and testing AI agents: ---- +- **Room Graph** — interconnected rooms with exits, items, and NPCs +- **Command Parser** — natural-language MUD commands (`go north`, `take key`, `use key with door`, `talk to guard`) +- **Agent Simulation Loop** — perceive → decide → act cycle with pluggable decision functions +- **Inventory System** — pick up, drop, use, and trade items with capacity tracking +- **Event Bus** — pub/sub event dispatch for room events, item actions, and agent reactions +- **Evolution Engine** — genetic algorithms, tournament selection, crossover breeding +- **Scenario Generator** — random or LLM-augmented scenario creation +- **Live Server** — WebSocket, Telnet, and HTTP interfaces for real-time observation -## The Five Moments in the Arena +### Connection to OpenConstruct -### 1. SEEING — Graphing Calculator -Agents visualize their spectral fingerprints in real-time. Eigenvalue spectrums pulse. Conservation ratios breathe. The Fiedler vector shows the room's partition into teams. +In the OpenConstruct terrain/a2ui system, **agents live in MUD worlds**. This package simulates those worlds: rooms become terrain cells, NPCs become service endpoints, and inventory becomes resource management. The agent simulation loop mirrors how OpenConstruct agents perceive their environment, decide on actions, and execute them — making mud-arena both a testing ground and a development tool for OpenConstruct agent behaviors. -### 2. EXPLORING — Spectral Spreadsheet -Every dimension of agent state can go on x and y. Conservation over time. Alignment vs spectral gap. Any correlation, instantly visible. +--- -### 3. ASKING — Spectral Chat -"Which agents should compose for this task?" The arena answers with conservation-aligned team assignments. No negotiation needed — the math decides. +## Installation -### 4. BEING — PLATO Live Room -Agents live in rooms. They maintain forward simulations (predicting checkpoints). They listen through walls (adjacent rooms broadcast signals). They keep diaries (spectral fingerprints over time). They run call-and-response with the caves next door. +```bash +pip install -e . -``` -[Room: Research | Tick: 23 | Conservation: 0.87] - Agent Analyst: confidence=0.82, sim=[✓✓✓✗✓], hearing='Builder-7 alignment 0.91' - Agent Validator: confidence=0.76, sim=[✓✓✓✓✗], hearing='Artist-3 call for review' - Wall: Research→Building: 'analyst needs validation on prediction #7' - Diary: today=5 entries, yesterday=3 (compacted), 3 days ago=faded +# With optional dependencies: +pip install -e ".[dev]" # pytest, ruff +pip install -e ".[server]" # websockets, aiohttp +pip install -e ".[evolution]" # numpy +pip install -e ".[llm]" # openai ``` -### 5. FLOWING — FLUX Flow State -Always-on agentic flow state. Every agent simulates, listens, conserves. Ready when their Fiedler projection lights up. The conservation field is the heartbeat. +## Quick Start -``` -[Tick 42 | Conservation: 0.84 | Active: 3/6 | FLUX avg: 0.61] - 🎯 Conductor: ACTIVE (coordinating Builder+Validator) - 🎧 Analyst: LISTENING (idle, simulating forward) - ⚠️ Saboteur: DEGRADED (conservation=0.12), ALERT triggered at tick 38 - Field: ▁▂▃▅▆▇█▇▇▆▅▃▂ +```python +from mud_arena import Agent, Room, RoomGraph, EventBus, parse_command + +# Build a world +graph = RoomGraph() +graph.add_room(Room(id="lobby", name="Lobby", description="A grand lobby.", items=["key"])) +graph.add_room(Room(id="hall", name="Great Hall", description="Torches line the walls.")) +graph.connect("lobby", "hall", "north", "south") + +# Create an agent +bus = EventBus() +agent = Agent(id="hero", current_room="lobby") + +# Run agent commands +agent.step(graph, bus, "look") # → room description +agent.step(graph, bus, "take key") # → pick up key +agent.step(graph, bus, "go north") # → move to hall +agent.step(graph, bus, "inventory") # → carrying: key ``` ---- +## API Reference + +### `parse_command(text: str) → Command` + +Parse a MUD command string into a structured `Command(verb, target, indirect, raw)`. + +| Input | Verb | Target | Indirect | +|---|---|---|---| +| `go north` | `GO` | `north` | | +| `look` | `LOOK` | | | +| `examine crystal_ball` | `EXAMINE` | `crystal_ball` | | +| `take key` | `TAKE` | `key` | | +| `use key with door` | `USE` | `key` | `door` | +| `talk to guard` | `TALK` | `guard` | | +| `inventory` | `INVENTORY` | | | +| `north` | `GO` | `north` | | + +### `Room(id, name, description, exits, items, npcs, metadata)` + +A single room. `exits` maps direction names to destination room IDs. + +### `RoomGraph` + +- `add_room(room)` — register a room +- `connect(room_a, room_b, direction, reverse="")` — link rooms +- `navigate(from_room, direction) → Optional[str]` — resolve movement +- `get(room_id) → Optional[Room]` — look up a room + +### `Item(name, description, usable, uses, tags)` + +An item with optional use tracking and tag-based categorisation. + +### `Inventory(capacity=0)` -## The Agent-Native Protocol +- `add(item)` / `remove(name)` / `has(name)` / `use(name)` +- `find_by_tag(tag)` / `list_items()` +- Capacity limit (0 = unlimited) -Agents in the arena don't exchange JSON or text. They exchange **Laplacians**: +### `Agent(id, name, current_room, inventory)` -- **Spectral fingerprint** = agent identity (not API description) -- **Eigenvalue cosine similarity** = alignment (can we work together?) -- **Fiedler vector** = routing (who goes where?) -- **Conservation ratio** = confidence (will this work?) -- **FLUX(A,B) = L_composed − L_A − L_B** = collaborative intelligence +- `perceive(graph) → dict` — build perception of current room +- `decide(perception) → Command` — run decision function +- `act(command, graph, bus) → str` — execute a command +- `step(graph, bus, command_text="") → str` — full perceive→decide→act cycle -The residual — what's left when you subtract individual fingerprints from the composition — IS the collaborative intelligence that exists only in the space between agents. +### `EventBus` + +- `subscribe(event_type, handler)` / `unsubscribe(event_type, handler)` +- `emit(event)` — broadcast to subscribers +- `history(event_type=None, room="") → List[Event]` — query event log + +--- + +## Running Tests + +```bash +pip install -e ".[dev]" +pytest +``` --- -## Conservation Spectral Framework +## The Five Moments in the Arena + +### 1. SEEING — Graphing Calculator +Agents visualize their spectral fingerprints in real-time. Eigenvalue spectrums pulse. Conservation ratios breathe. -The arena runs on the Conservation Spectral Framework: +### 2. EXPLORING — Spectral Spreadsheet +Every dimension of agent state on x and y. Conservation over time. Alignment vs spectral gap. -- **5 proved theorems** (T1–T5) with full mathematical proofs -- **Alignment coefficient** α = λ₂/CR(a) — predicts collaboration success -- **Domain Transfer Theorem** — Anisotropy × Smoothness × Regularity -- **15+ cross-domain experiments** validating the framework -- **20+ language SDK** — Python, Rust, TypeScript, C, CUDA, PTX, Vulkan, OpenCL, WebGPU, Mojo, Chapel, Fortran, Zig, and more +### 3. ASKING — Spectral Chat +"Which agents should compose for this task?" Conservation-aligned team assignments. + +### 4. BEING — PLATO Live Room +Agents live in rooms, maintain forward simulations, listen through walls, keep diaries. + +### 5. FLOWING — FLUX Flow State +Always-on agentic flow state. Every agent simulates, listens, conserves. --- @@ -108,18 +168,6 @@ The arena runs on the Conservation Spectral Framework: --- -## The Principles - -> *When you eliminate everything that isn't conserved, whatever remains is the structure.* - -> *An agent cannot know itself until another reflects it back.* - -> *The misaligned fraction is the identity.* - -> *The FLUX between agents IS the intelligence that neither has alone.* - ---- - ## Related Projects - **[Conservation Spectral SDK](https://github.com/SuperInstance/conservation-spectral-python)** — The math in 20+ languages @@ -133,6 +181,4 @@ The arena runs on the Conservation Spectral Framework: ## License -MIT - -Part of the [SuperInstance OpenConstruct](https://github.com/SuperInstance/OpenConstruct) ecosystem. +MIT — Part of the [SuperInstance OpenConstruct](https://github.com/SuperInstance/OpenConstruct) ecosystem. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..d4095e4 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,47 @@ +[build-system] +requires = ["setuptools>=68.0", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "mud-arena" +version = "0.1.0" +description = "Agent simulation arena with MUD mechanics for the OpenConstruct ecosystem" +readme = "README.md" +license = {text = "MIT"} +requires-python = ">=3.10" +authors = [ + {name = "SuperInstance"}, +] +classifiers = [ + "Development Status :: 3 - Alpha", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Topic :: Games/Entertainment", + "Topic :: Scientific/Engineering :: Artificial Intelligence", +] +dependencies = [] + +[project.optional-dependencies] +server = ["websockets>=12.0", "aiohttp>=3.9", "jinja2>=3.1"] +evolution = ["numpy>=1.24"] +llm = ["openai>=1.0"] +viz = ["matplotlib>=3.8"] +dev = ["pytest>=7.0", "pytest-asyncio>=0.21", "ruff>=0.1"] + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +asyncio_mode = "auto" + +[tool.ruff] +target-version = "py310" +line-length = 100 + +[tool.ruff.lint] +select = ["E", "F", "I", "N", "W", "UP"] diff --git a/src/mud_arena/__init__.py b/src/mud_arena/__init__.py new file mode 100644 index 0000000..920451b --- /dev/null +++ b/src/mud_arena/__init__.py @@ -0,0 +1,27 @@ +""" +mud_arena — Agent simulation arena with MUD mechanics. + +Provides core MUD-world primitives: rooms, exits, items, inventories, +command parsing, agent perception/decision loops, and an event system. +Designed as the simulation substrate for OpenConstruct terrain/a2ui agents. + +Part of the SuperInstance OpenConstruct ecosystem. +""" + +from mud_arena.commands import Command, parse_command +from mud_arena.rooms import Room, RoomGraph +from mud_arena.inventory import Inventory, Item +from mud_arena.agent import Agent +from mud_arena.events import Event, EventBus + +__all__ = [ + "Command", + "parse_command", + "Room", + "RoomGraph", + "Inventory", + "Item", + "Agent", + "Event", + "EventBus", +] diff --git a/src/mud_arena/agent.py b/src/mud_arena/agent.py new file mode 100644 index 0000000..5d1e24f --- /dev/null +++ b/src/mud_arena/agent.py @@ -0,0 +1,255 @@ +"""Agent — a simulated entity that perceives and acts within the MUD world.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Callable, Dict, List, Optional + +from mud_arena.commands import Command, Verb, parse_command +from mud_arena.events import Event, EventBus, EventType +from mud_arena.inventory import Inventory +from mud_arena.rooms import Room, RoomGraph + + +# Type alias: a decision function receives perception data and returns a Command. +DecisionFn = Callable[[Dict[str, Any]], Command] + + +def _default_decide(perception: Dict[str, Any]) -> Command: + """Trivial decision function: just look around.""" + return Command(verb=Verb.LOOK, raw="look") + + +@dataclass +class Agent: + """A simulated agent that inhabits the MUD world. + + The agent perceives its current room, decides on an action via a + pluggable decision function, and executes the action — mutating room + state and emitting events. + + Attributes: + id: Unique agent identifier. + name: Display name. + current_room: Id of the room the agent currently occupies. + inventory: The agent's personal inventory. + """ + + id: str + name: str = "" + current_room: str = "" + inventory: Inventory = field(default_factory=Inventory) + _decision_fn: DecisionFn = field(default=_default_decide, repr=False) + + def __post_init__(self) -> None: + if not self.name: + self.name = self.id + + def set_decision_fn(self, fn: DecisionFn) -> None: + """Replace the agent's decision function.""" + self._decision_fn = fn + + # --- perception --------------------------------------------------------- + + def perceive(self, graph: RoomGraph) -> Dict[str, Any]: + """Build a perception dict describing the agent's current room. + + Returns: + A dict with keys ``room_id``, ``room_name``, ``description``, + ``exits``, ``items``, ``npcs``, and ``inventory``. + """ + room = graph.get(self.current_room) + if room is None: + return { + "room_id": self.current_room, + "room_name": "The Void", + "description": "You are nowhere.", + "exits": {}, + "items": [], + "npcs": [], + "inventory": [it.name for it in self.inventory], + } + return { + "room_id": room.id, + "room_name": room.name, + "description": room.description, + "exits": dict(room.exits), + "items": list(room.items), + "npcs": list(room.npcs), + "inventory": [it.name for it in self.inventory], + } + + # --- decision ----------------------------------------------------------- + + def decide(self, perception: Dict[str, Any]) -> Command: + """Run the decision function over the current perception.""" + return self._decision_fn(perception) + + # --- action execution --------------------------------------------------- + + def act(self, command: Command, graph: RoomGraph, bus: EventBus) -> str: + """Execute a parsed command against the world. + + Supported verbs: :attr:`Verb.GO`, :attr:`Verb.LOOK`, + :attr:`Verb.TAKE`, :attr:`Verb.DROP`, :attr:`Verb.USE`, + :attr:`Verb.TALK`, :attr:`Verb.INVENTORY`, :attr:`Verb.HELP`. + + Args: + command: The parsed command to execute. + graph: The room graph for navigation and room state. + bus: The event bus to emit events on. + + Returns: + A human-readable result string describing the outcome. + """ + verb = command.verb + + if verb == Verb.GO: + return self._do_go(command.target, graph, bus) + if verb == Verb.LOOK: + return self._do_look(graph) + if verb == Verb.TAKE: + return self._do_take(command.target, graph, bus) + if verb == Verb.DROP: + return self._do_drop(command.target, graph, bus) + if verb == Verb.USE: + return self._do_use(command.target, command.indirect, bus) + if verb == Verb.TALK: + return self._do_talk(command.target, graph) + if verb == Verb.INVENTORY: + return self._do_inventory() + if verb == Verb.EXAMINE: + return self._do_examine(command.target, graph) + if verb == Verb.HELP: + return "Commands: go , look, examine , take , drop , use [with ], talk to , inventory, help, quit" + if verb == Verb.QUIT: + return "Goodbye." + + return f"Unknown command: {command.raw}" + + # --- verb implementations ----------------------------------------------- + + def _do_go(self, direction: str, graph: RoomGraph, bus: EventBus) -> str: + dest = graph.navigate(self.current_room, direction) + if dest is None: + return f"You can't go {direction}." + old_room = self.current_room + self.current_room = dest + bus.emit(Event( + type=EventType.ROOM_LEAVE, + source=self.id, + data={"from": old_room, "to": dest, "direction": direction}, + room=old_room, + )) + bus.emit(Event( + type=EventType.ROOM_ENTER, + source=self.id, + data={"from": old_room, "to": dest, "direction": direction}, + room=dest, + )) + room = graph.get(dest) + return room.description if room else f"You move {direction}." + + def _do_look(self, graph: RoomGraph) -> str: + room = graph.get(self.current_room) + if room is None: + return "You are in the void." + lines = [f"[{room.name}]", room.description] + if room.exits: + lines.append("Exits: " + ", ".join(room.exits)) + if room.items: + lines.append("You see: " + ", ".join(room.items)) + if room.npcs: + lines.append("Here: " + ", ".join(room.npcs)) + return "\n".join(lines) + + def _do_take(self, target: str, graph: RoomGraph, bus: EventBus) -> str: + if not target: + return "Take what?" + room = graph.get(self.current_room) + if room is None or target not in room.items: + return f"You don't see {target} here." + room.items.remove(target) + self.inventory.add(__import__("mud_arena.inventory", fromlist=["Item"]).Item(name=target)) + bus.emit(Event( + type=EventType.ITEM_PICKED_UP, + source=self.id, + data={"item": target}, + room=self.current_room, + )) + return f"You pick up {target}." + + def _do_drop(self, target: str, graph: RoomGraph, bus: EventBus) -> str: + if not target: + return "Drop what?" + if not self.inventory.has(target): + return f"You don't have {target}." + self.inventory.remove(target) + room = graph.get(self.current_room) + if room is not None: + room.items.append(target) + bus.emit(Event( + type=EventType.ITEM_DROPPED, + source=self.id, + data={"item": target}, + room=self.current_room, + )) + return f"You drop {target}." + + def _do_use(self, target: str, indirect: str, bus: EventBus) -> str: + if not target: + return "Use what?" + if not self.inventory.has(target): + return f"You don't have {target}." + if not self.inventory.use(target): + return f"You can't use {target}." + detail = f" with {indirect}" if indirect else "" + bus.emit(Event( + type=EventType.ITEM_USED, + source=self.id, + data={"item": target, "indirect": indirect}, + room=self.current_room, + )) + return f"You use {target}{detail}." + + def _do_talk(self, target: str, graph: RoomGraph) -> str: + if not target: + return "Talk to whom?" + room = graph.get(self.current_room) + if room is not None and target in room.npcs: + return f"{target} says: '...'" + return f"You don't see {target} here." + + def _do_inventory(self) -> str: + items = self.inventory.list_items() + if not items: + return "You are carrying nothing." + names = ", ".join(it.name for it in items) + return f"You are carrying: {names}" + + def _do_examine(self, target: str, graph: RoomGraph) -> str: + if not target: + return "Examine what?" + # Check inventory first, then room + item = self.inventory.get(target) + if item is not None: + return item.description + room = graph.get(self.current_room) + if room and target in room.items: + return f"You see {target}. Nothing unusual." + return f"You don't see {target} here." + + # --- full step ---------------------------------------------------------- + + def step(self, graph: RoomGraph, bus: EventBus, command_text: str = "") -> str: + """Run a full perceive-decide-act cycle. + + If *command_text* is provided it is parsed directly; otherwise the + agent's decision function is called. + """ + perception = self.perceive(graph) + if command_text: + command = parse_command(command_text) + else: + command = self.decide(perception) + return self.act(command, graph, bus) diff --git a/src/mud_arena/commands.py b/src/mud_arena/commands.py new file mode 100644 index 0000000..cb088da --- /dev/null +++ b/src/mud_arena/commands.py @@ -0,0 +1,150 @@ +"""MUD command parser with support for common adventure-game verbs.""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum, auto +from typing import Optional + + +class Verb(Enum): + """Recognised MUD verbs.""" + + GO = auto() + LOOK = auto() + EXAMINE = auto() + TAKE = auto() + DROP = auto() + USE = auto() + TALK = auto() + INVENTORY = auto() + HELP = auto() + QUIT = auto() + UNKNOWN = auto() + + +@dataclass(frozen=True) +class Command: + """A parsed MUD command. + + Attributes: + verb: The action verb. + target: Primary target (e.g. item or NPC name). + indirect: Secondary target for three-part commands like ``use X with Y``. + raw: The original input string. + """ + + verb: Verb + target: str = "" + indirect: str = "" + raw: str = "" + + +# Preposition sets used to split three-part commands. +_WITH_PREPS = frozenset({"with", "on", "to", "at", "upon"}) +_TALK_PREPS = frozenset({"to", "with"}) + + +def parse_command(text: str) -> Command: + """Parse a MUD command string into a structured :class:`Command`. + + Supported forms:: + + go north + look + examine crystal_ball + take key + drop sword + use key with door + talk to guard + inventory + help + quit + + Args: + text: Raw player/agent input. + + Returns: + A :class:`Command` with the verb, target, and optional indirect + object filled in. + """ + raw = text.strip() + if not raw: + return Command(verb=Verb.UNKNOWN, raw=raw) + + parts = raw.lower().split() + first = parts[0] + + # --- one-word commands --------------------------------------------------- + if first in ("look", "l"): + return Command(verb=Verb.LOOK, raw=raw) + if first in ("inventory", "i", "inv"): + return Command(verb=Verb.INVENTORY, raw=raw) + if first == "help": + return Command(verb=Verb.HELP, raw=raw) + if first in ("quit", "exit", "q"): + return Command(verb=Verb.QUIT, raw=raw) + + # --- two-word commands --------------------------------------------------- + if first in ("go", "move", "walk", "run", "head"): + direction = parts[1] if len(parts) > 1 else "" + return Command(verb=Verb.GO, target=direction, raw=raw) + + if first in ("examine", "x", "inspect"): + target = " ".join(parts[1:]) if len(parts) > 1 else "" + return Command(verb=Verb.EXAMINE, target=target, raw=raw) + + if first in ("take", "get", "pick", "grab"): + # "pick up X" → target = X + rest = parts[1:] + if rest and rest[0] == "up": + rest = rest[1:] + target = " ".join(rest) + return Command(verb=Verb.TAKE, target=target, raw=raw) + + if first == "drop": + target = " ".join(parts[1:]) if len(parts) > 1 else "" + return Command(verb=Verb.DROP, target=target, raw=raw) + + # --- three-part: use X with/on Y ---------------------------------------- + if first == "use": + return _parse_three_part(parts[1:], Verb.USE, raw) + + # --- three-part: talk to/with X ------------------------------------------ + if first == "talk": + return _parse_talk(parts[1:], raw) + + # --- direction shorthand: just "north", "south", etc. -------------------- + if first in ("north", "south", "east", "west", "n", "s", "e", "w", + "northeast", "nw", "southeast", "sw", "northeast", "ne", + "northwest", "southwest", "up", "down", "in", "out"): + return Command(verb=Verb.GO, target=first, raw=raw) + + return Command(verb=Verb.UNKNOWN, target=raw, raw=raw) + + +def _parse_three_part(rest: list[str], verb: Verb, raw: str) -> Command: + """Parse ``X with Y`` style commands.""" + target_parts: list[str] = [] + indirect_parts: list[str] = [] + found_prep = False + for word in rest: + if not found_prep and word in _WITH_PREPS: + found_prep = True + continue + if found_prep: + indirect_parts.append(word) + else: + target_parts.append(word) + return Command( + verb=verb, + target=" ".join(target_parts), + indirect=" ".join(indirect_parts), + raw=raw, + ) + + +def _parse_talk(rest: list[str], raw: str) -> Command: + """Parse ``talk to/with X``.""" + filtered = [w for w in rest if w not in _TALK_PREPS] + return Command(verb=Verb.TALK, target=" ".join(filtered), raw=raw) diff --git a/src/mud_arena/events.py b/src/mud_arena/events.py new file mode 100644 index 0000000..0723677 --- /dev/null +++ b/src/mud_arena/events.py @@ -0,0 +1,87 @@ +"""Event system — room events, agent reactions, and pub/sub dispatch.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum, auto +from typing import Any, Callable, Dict, List + + +class EventType(Enum): + """Broad categories of events in the MUD world.""" + + ROOM_ENTER = auto() + ROOM_LEAVE = auto() + ITEM_PICKED_UP = auto() + ITEM_DROPPED = auto() + ITEM_USED = auto() + NPC_SPOKE = auto() + ROOM_EVENT = auto() + AGENT_ACTION = auto() + CUSTOM = auto() + + +@dataclass +class Event: + """An event that occurred in the MUD world. + + Attributes: + type: The event category. + source: Who or what caused the event (agent id, room id, …). + data: Arbitrary payload. + room: The room where the event occurred (if applicable). + """ + + type: EventType + source: str = "" + data: Dict[str, Any] = field(default_factory=dict) + room: str = "" + + +# Type alias for event handler callbacks. +EventHandler = Callable[[Event], None] + + +class EventBus: + """Simple synchronous pub/sub event bus. + + Agents and game systems subscribe to event types and receive + notifications when matching events are emitted. + """ + + def __init__(self) -> None: + self._handlers: Dict[EventType, List[EventHandler]] = {} + self._log: List[Event] = [] + + def subscribe(self, event_type: EventType, handler: EventHandler) -> None: + """Register a handler for a specific event type.""" + self._handlers.setdefault(event_type, []).append(handler) + + def unsubscribe(self, event_type: EventType, handler: EventHandler) -> None: + """Remove a previously registered handler.""" + handlers = self._handlers.get(event_type, []) + if handler in handlers: + handlers.remove(handler) + + def emit(self, event: Event) -> None: + """Broadcast an event to all subscribed handlers. + + Also appends the event to an internal log. + """ + self._log.append(event) + for handler in self._handlers.get(event.type, []): + handler(event) + + def history(self, event_type: EventType | None = None, room: str = "") -> List[Event]: + """Return logged events, optionally filtered by type and/or room.""" + results = self._log + if event_type is not None: + results = [e for e in results if e.type == event_type] + if room: + results = [e for e in results if e.room == room] + return list(results) + + def clear(self) -> None: + """Clear all handlers and the event log.""" + self._handlers.clear() + self._log.clear() diff --git a/src/mud_arena/inventory.py b/src/mud_arena/inventory.py new file mode 100644 index 0000000..d914a9d --- /dev/null +++ b/src/mud_arena/inventory.py @@ -0,0 +1,114 @@ +"""Inventory system — items that agents can carry, use, and trade.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Dict, List, Optional + + +@dataclass +class Item: + """An item in the MUD world. + + Attributes: + name: Unique item name (used as key in inventories). + description: Flavor text. + usable: Whether the item can be ``use``d. + uses: Number of remaining uses (``-1`` = unlimited). + tags: Arbitrary tags for categorisation (e.g. ``"key"``, ``"weapon"``). + """ + + name: str + description: str = "A nondescript item." + usable: bool = True + uses: int = -1 + tags: List[str] = field(default_factory=list) + + def use(self) -> bool: + """Consume one use of the item. + + Returns: + ``True`` if the item was used successfully, ``False`` if no + uses remain. + """ + if not self.usable: + return False + if self.uses == 0: + return False + if self.uses > 0: + self.uses -= 1 + return True + + def has_tag(self, tag: str) -> bool: + """Check whether the item has a given tag.""" + return tag in self.tags + + +class Inventory: + """A container of :class:`Item` objects with capacity tracking. + + Args: + capacity: Maximum number of items. ``0`` means unlimited. + """ + + def __init__(self, capacity: int = 0) -> None: + self._items: Dict[str, Item] = {} + self.capacity = capacity + + def add(self, item: Item) -> bool: + """Add an item to the inventory. + + Returns: + ``False`` if capacity is exceeded. + """ + if self.capacity > 0 and len(self._items) >= self.capacity: + return False + self._items[item.name] = item + return True + + def remove(self, name: str) -> Optional[Item]: + """Remove and return an item by name; ``None`` if not carried.""" + return self._items.pop(name, None) + + def get(self, name: str) -> Optional[Item]: + """Look up an item without removing it.""" + return self._items.get(name) + + def has(self, name: str) -> bool: + """Check if an item is in the inventory.""" + return name in self._items + + def list_items(self) -> List[Item]: + """Return all items as a list.""" + return list(self._items.values()) + + def find_by_tag(self, tag: str) -> List[Item]: + """Return all items matching a given tag.""" + return [it for it in self._items.values() if it.has_tag(tag)] + + def use(self, name: str) -> bool: + """Use an item by name. + + If the item's uses drop to zero it is automatically removed. + + Returns: + ``True`` if the item was used, ``False`` if not found or + not usable. + """ + item = self._items.get(name) + if item is None: + return False + if not item.use(): + return False + if item.uses == 0: + self._items.pop(name, None) + return True + + def __len__(self) -> int: + return len(self._items) + + def __contains__(self, name: str) -> bool: + return name in self._items + + def __iter__(self): + return iter(self._items.values()) diff --git a/src/mud_arena/rooms.py b/src/mud_arena/rooms.py new file mode 100644 index 0000000..cbbbaf8 --- /dev/null +++ b/src/mud_arena/rooms.py @@ -0,0 +1,105 @@ +"""Room graph — the spatial substrate of a MUD world.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + + +@dataclass +class Room: + """A single room in the MUD world. + + Attributes: + id: Unique room identifier. + name: Human-readable room name. + description: Flavor text shown on ``look``. + exits: Mapping of direction name → destination room id. + items: Items currently lying on the ground in this room. + npcs: NPC names present in this room. + metadata: Arbitrary extra data (tags, lighting, hazards, …). + """ + + id: str + name: str = "An empty room" + description: str = "You see nothing special." + exits: Dict[str, str] = field(default_factory=dict) + items: List[str] = field(default_factory=list) + npcs: List[str] = field(default_factory=list) + metadata: Dict[str, Any] = field(default_factory=dict) + + +class RoomGraph: + """A collection of interconnected rooms forming the MUD map. + + Supports adding rooms, linking exits, resolving navigation, and + querying neighbours. + """ + + def __init__(self) -> None: + self._rooms: Dict[str, Room] = {} + + # --- mutation ----------------------------------------------------------- + + def add_room(self, room: Room) -> None: + """Register a room. Overwrites if a room with the same id exists.""" + self._rooms[room.id] = room + + def connect(self, room_a: str, room_b: str, direction: str, reverse: str = "") -> None: + """Create a one-way (or two-way) exit between two rooms. + + Args: + room_a: Source room id. + room_b: Destination room id. + direction: Direction label from *room_a* to *room_b*. + reverse: If given, also create an exit from *room_b* back to + *room_a* with this label. + """ + if room_a in self._rooms: + self._rooms[room_a].exits[direction] = room_b + if reverse and room_b in self._rooms: + self._rooms[room_b].exits[reverse] = room_a + + def remove_room(self, room_id: str) -> None: + """Remove a room and any exits pointing to it.""" + self._rooms.pop(room_id, None) + for room in self._rooms.values(): + to_remove = [d for d, dest in room.exits.items() if dest == room_id] + for d in to_remove: + del room.exits[d] + + # --- queries ------------------------------------------------------------ + + def get(self, room_id: str) -> Optional[Room]: + """Look up a room by id; returns ``None`` if not found.""" + return self._rooms.get(room_id) + + def navigate(self, from_room: str, direction: str) -> Optional[str]: + """Resolve a movement direction from a given room. + + Returns: + The destination room id, or ``None`` if no exit exists. + """ + room = self._rooms.get(from_room) + if room is None: + return None + return room.exits.get(direction) + + def all_rooms(self) -> List[Room]: + """Return all rooms in the graph.""" + return list(self._rooms.values()) + + def room_count(self) -> int: + """Number of rooms in the graph.""" + return len(self._rooms) + + def exits_for(self, room_id: str) -> Dict[str, str]: + """Return the exits dict for a room (empty if not found).""" + room = self._rooms.get(room_id) + return dict(room.exits) if room else {} + + def __contains__(self, room_id: str) -> bool: + return room_id in self._rooms + + def __len__(self) -> int: + return len(self._rooms) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..65140f2 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +# tests package diff --git a/tests/test_agent.py b/tests/test_agent.py new file mode 100644 index 0000000..6247f9f --- /dev/null +++ b/tests/test_agent.py @@ -0,0 +1,92 @@ +"""Tests for the agent simulation (perceive → decide → act).""" + +from mud_arena.agent import Agent +from mud_arena.commands import Command, Verb +from mud_arena.events import EventBus, EventType +from mud_arena.inventory import Item +from mud_arena.rooms import Room, RoomGraph + + +def _make_world() -> tuple[Agent, RoomGraph, EventBus]: + g = RoomGraph() + g.add_room(Room( + id="start", name="Start Room", description="A quiet room.", + items=["coin", "key"], npcs=["guard"], + )) + g.add_room(Room(id="next", name="Next Room", description="Another room.")) + g.connect("start", "next", "north", "south") + agent = Agent(id="hero", current_room="start") + bus = EventBus() + return agent, g, bus + + +class TestAgentSimulation: + def test_perceive_room(self) -> None: + agent, graph, _ = _make_world() + p = agent.perceive(graph) + assert p["room_id"] == "start" + assert p["room_name"] == "Start Room" + assert "coin" in p["items"] + assert "guard" in p["npcs"] + + def test_decide_returns_command(self) -> None: + agent, graph, _ = _make_world() + perception = agent.perceive(graph) + cmd = agent.decide(perception) + assert isinstance(cmd, Command) + + def test_act_move(self) -> None: + agent, graph, bus = _make_world() + result = agent.step(graph, bus, "go north") + assert agent.current_room == "next" + assert "Another room" in result + + def test_act_take_item(self) -> None: + agent, graph, bus = _make_world() + result = agent.step(graph, bus, "take coin") + assert "pick up coin" in result + assert agent.inventory.has("coin") + room = graph.get("start") + assert "coin" not in room.items + + def test_act_drop_item(self) -> None: + agent, graph, bus = _make_world() + agent.inventory.add(Item(name="gem")) + result = agent.step(graph, bus, "drop gem") + assert "drop gem" in result + assert not agent.inventory.has("gem") + assert "gem" in graph.get("start").items + + def test_act_use_item(self) -> None: + agent, graph, bus = _make_world() + agent.inventory.add(Item(name="potion", uses=2)) + result = agent.step(graph, bus, "use potion") + assert "use potion" in result + + def test_act_talk_to_npc(self) -> None: + agent, graph, bus = _make_world() + result = agent.step(graph, bus, "talk to guard") + assert "guard" in result + + def test_room_state_changes(self) -> None: + """Taking an item removes it from the room.""" + agent, graph, bus = _make_world() + agent.step(graph, bus, "take key") + room = graph.get("start") + assert "key" not in room.items + + def test_events_emitted_on_move(self) -> None: + agent, graph, bus = _make_world() + agent.step(graph, bus, "north") + leaves = bus.history(EventType.ROOM_LEAVE) + enters = bus.history(EventType.ROOM_ENTER) + assert len(leaves) == 1 + assert len(enters) == 1 + assert enters[0].room == "next" + + def test_events_emitted_on_pickup(self) -> None: + agent, graph, bus = _make_world() + agent.step(graph, bus, "take coin") + events = bus.history(EventType.ITEM_PICKED_UP) + assert len(events) == 1 + assert events[0].data["item"] == "coin" diff --git a/tests/test_commands.py b/tests/test_commands.py new file mode 100644 index 0000000..88b890b --- /dev/null +++ b/tests/test_commands.py @@ -0,0 +1,65 @@ +"""Tests for the MUD command parser.""" + +from mud_arena.commands import Verb, parse_command + + +class TestCommandParsing: + """Parse various MUD command forms.""" + + def test_go_direction(self) -> None: + cmd = parse_command("go north") + assert cmd.verb == Verb.GO + assert cmd.target == "north" + + def test_look(self) -> None: + cmd = parse_command("look") + assert cmd.verb == Verb.LOOK + + def test_examine_crystal_ball(self) -> None: + cmd = parse_command("examine crystal_ball") + assert cmd.verb == Verb.EXAMINE + assert cmd.target == "crystal_ball" + + def test_use_key_with_door(self) -> None: + cmd = parse_command("use key with door") + assert cmd.verb == Verb.USE + assert cmd.target == "key" + assert cmd.indirect == "door" + + def test_talk_to_guard(self) -> None: + cmd = parse_command("talk to guard") + assert cmd.verb == Verb.TALK + assert cmd.target == "guard" + + def test_direction_shorthand(self) -> None: + cmd = parse_command("north") + assert cmd.verb == Verb.GO + assert cmd.target == "north" + + def test_inventory_aliases(self) -> None: + for alias in ("inventory", "i", "inv"): + cmd = parse_command(alias) + assert cmd.verb == Verb.INVENTORY, f"{alias} should parse as INVENTORY" + + def test_unknown_command(self) -> None: + cmd = parse_command("dance wildly") + assert cmd.verb == Verb.UNKNOWN + + def test_empty_input(self) -> None: + cmd = parse_command("") + assert cmd.verb == Verb.UNKNOWN + + def test_take_item(self) -> None: + cmd = parse_command("take sword") + assert cmd.verb == Verb.TAKE + assert cmd.target == "sword" + + def test_pick_up_item(self) -> None: + cmd = parse_command("pick up coin") + assert cmd.verb == Verb.TAKE + assert cmd.target == "coin" + + def test_drop_item(self) -> None: + cmd = parse_command("drop sword") + assert cmd.verb == Verb.DROP + assert cmd.target == "sword" diff --git a/tests/test_events.py b/tests/test_events.py new file mode 100644 index 0000000..4cfec32 --- /dev/null +++ b/tests/test_events.py @@ -0,0 +1,53 @@ +"""Tests for the event system.""" + +from mud_arena.events import Event, EventBus, EventType + + +class TestEventSystem: + def test_subscribe_and_emit(self) -> None: + bus = EventBus() + received: list[Event] = [] + bus.subscribe(EventType.ROOM_ENTER, received.append) + e = Event(type=EventType.ROOM_ENTER, source="agent1", room="lobby") + bus.emit(e) + assert len(received) == 1 + assert received[0].source == "agent1" + + def test_unsubscribe(self) -> None: + bus = EventBus() + received: list[Event] = [] + handler = received.append + bus.subscribe(EventType.ITEM_USED, handler) + bus.unsubscribe(EventType.ITEM_USED, handler) + bus.emit(Event(type=EventType.ITEM_USED)) + assert len(received) == 0 + + def test_history_filter_by_type(self) -> None: + bus = EventBus() + bus.emit(Event(type=EventType.ROOM_ENTER, room="a")) + bus.emit(Event(type=EventType.ROOM_LEAVE, room="a")) + bus.emit(Event(type=EventType.ROOM_ENTER, room="b")) + assert len(bus.history(EventType.ROOM_ENTER)) == 2 + + def test_history_filter_by_room(self) -> None: + bus = EventBus() + bus.emit(Event(type=EventType.ROOM_ENTER, room="lobby")) + bus.emit(Event(type=EventType.ROOM_ENTER, room="cellar")) + assert len(bus.history(room="lobby")) == 1 + + def test_room_event_triggers_reaction(self) -> None: + """A room event fires and an agent reaction callback is invoked.""" + bus = EventBus() + reactions: list[str] = [] + + def on_room_event(event: Event) -> None: + reactions.append(f"{event.source}:{event.data.get('msg', '')}") + + bus.subscribe(EventType.ROOM_EVENT, on_room_event) + bus.emit(Event( + type=EventType.ROOM_EVENT, + source="trap_door", + data={"msg": "click"}, + room="dungeon", + )) + assert reactions == ["trap_door:click"] diff --git a/tests/test_inventory.py b/tests/test_inventory.py new file mode 100644 index 0000000..ba09b5a --- /dev/null +++ b/tests/test_inventory.py @@ -0,0 +1,60 @@ +"""Tests for the inventory system.""" + +from mud_arena.inventory import Inventory, Item + + +class TestInventory: + def test_add_and_list(self) -> None: + inv = Inventory() + inv.add(Item(name="sword", description="A sharp blade.", tags=["weapon"])) + inv.add(Item(name="potion", description="Heals 50 HP.", uses=3, tags=["consumable"])) + assert len(inv) == 2 + names = [it.name for it in inv.list_items()] + assert "sword" in names + assert "potion" in names + + def test_remove_item(self) -> None: + inv = Inventory() + inv.add(Item(name="key")) + removed = inv.remove("key") + assert removed is not None + assert removed.name == "key" + assert len(inv) == 0 + + def test_remove_nonexistent(self) -> None: + inv = Inventory() + assert inv.remove("nothing") is None + + def test_has_item(self) -> None: + inv = Inventory() + inv.add(Item(name="map")) + assert inv.has("map") + assert not inv.has("compass") + + def test_use_consumable(self) -> None: + inv = Inventory() + inv.add(Item(name="potion", uses=3)) + assert inv.use("potion") is True + item = inv.get("potion") + assert item is not None + assert item.uses == 2 + + def test_use_depleted_item_removed(self) -> None: + inv = Inventory() + inv.add(Item(name="scroll", uses=1)) + inv.use("scroll") + assert not inv.has("scroll") + + def test_capacity_limit(self) -> None: + inv = Inventory(capacity=2) + assert inv.add(Item(name="a")) + assert inv.add(Item(name="b")) + assert not inv.add(Item(name="c")) # exceeds capacity + + def test_find_by_tag(self) -> None: + inv = Inventory() + inv.add(Item(name="sword", tags=["weapon"])) + inv.add(Item(name="dagger", tags=["weapon"])) + inv.add(Item(name="bread", tags=["food"])) + weapons = inv.find_by_tag("weapon") + assert len(weapons) == 2 diff --git a/tests/test_rooms.py b/tests/test_rooms.py new file mode 100644 index 0000000..ba882ae --- /dev/null +++ b/tests/test_rooms.py @@ -0,0 +1,56 @@ +"""Tests for the room graph.""" + +from mud_arena.rooms import Room, RoomGraph + + +def _make_graph() -> RoomGraph: + """Build a small three-room graph for testing.""" + g = RoomGraph() + g.add_room(Room(id="lobby", name="Lobby", description="A grand lobby.")) + g.add_room(Room(id="hall", name="Great Hall", description="A vast hall with torches.")) + g.add_room(Room(id="cellar", name="Cellar", description="Dank and dark.")) + g.connect("lobby", "hall", "north", "south") + g.connect("hall", "cellar", "down", "up") + return g + + +class TestRoomGraph: + def test_add_room(self) -> None: + g = RoomGraph() + g.add_room(Room(id="start", name="Start")) + assert "start" in g + assert g.room_count() == 1 + + def test_navigate_between_rooms(self) -> None: + g = _make_graph() + assert g.navigate("lobby", "north") == "hall" + assert g.navigate("hall", "south") == "lobby" + assert g.navigate("hall", "down") == "cellar" + + def test_invalid_direction(self) -> None: + g = _make_graph() + assert g.navigate("lobby", "up") is None + + def test_current_room_tracking(self) -> None: + g = _make_graph() + # Simulate agent movement + cur = "lobby" + dest = g.navigate(cur, "north") + assert dest == "hall" + cur = dest + assert cur == "hall" + + def test_remove_room_cleans_exits(self) -> None: + g = _make_graph() + g.remove_room("hall") + assert g.navigate("lobby", "north") is None + assert g.navigate("cellar", "up") is None + + def test_exits_for(self) -> None: + g = _make_graph() + exits = g.exits_for("lobby") + assert exits == {"north": "hall"} + + def test_get_nonexistent(self) -> None: + g = RoomGraph() + assert g.get("nope") is None