268 lines
8.5 KiB
Python
268 lines
8.5 KiB
Python
"""Evennia commands for querying the MemPalace.
|
|
|
|
CmdRecall — semantic search across the caller's wing (or fleet)
|
|
CmdEnterRoom — teleport to the palace room matching a topic
|
|
CmdAsk — ask a steward NPC a question about their wing's memory
|
|
|
|
These commands are designed to work inside a live Evennia server.
|
|
They import ``evennia`` at class-definition time only to set up the
|
|
command skeleton; the actual search logic lives in ``nexus.mempalace``
|
|
and is fully testable without a running Evennia instance.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from nexus.mempalace.searcher import (
|
|
MemPalaceUnavailable,
|
|
MemPalaceResult,
|
|
search_memories,
|
|
search_fleet,
|
|
)
|
|
from nexus.mempalace.config import FLEET_WING, CORE_ROOMS
|
|
|
|
try:
|
|
from evennia import Command as _EvCommand # type: ignore
|
|
if _EvCommand is None:
|
|
raise ImportError("evennia.Command is None (Django not configured)")
|
|
Command = _EvCommand
|
|
except (ImportError, Exception): # outside a live Evennia environment
|
|
class Command: # type: ignore # minimal stub for import/testing
|
|
key = ""
|
|
aliases: list = []
|
|
locks = "cmd:all()"
|
|
help_category = "MemPalace"
|
|
|
|
def __init__(self):
|
|
self.caller = None
|
|
self.args = ""
|
|
self.switches: list[str] = []
|
|
|
|
def func(self):
|
|
pass
|
|
|
|
|
|
class CmdRecall(Command):
|
|
"""Search the mind palace for memories matching a query.
|
|
|
|
Usage:
|
|
recall <query>
|
|
recall <query> --fleet
|
|
recall <query> --room <room>
|
|
|
|
Examples:
|
|
recall nightly watch failures
|
|
recall GraphQL --fleet
|
|
recall CI pipeline --room forge
|
|
|
|
The ``--fleet`` switch searches the shared fleet wing (closets only).
|
|
Without it, only the caller's private wing is searched.
|
|
"""
|
|
|
|
key = "recall"
|
|
aliases = ["mem", "remember"]
|
|
locks = "cmd:all()"
|
|
help_category = "MemPalace"
|
|
|
|
def func(self):
|
|
raw = self.args.strip()
|
|
if not raw:
|
|
self.caller.msg("Usage: recall <query> [--fleet] [--room <room>]")
|
|
return
|
|
|
|
fleet_mode = "--fleet" in self.switches
|
|
room_filter = None
|
|
if "--room" in self.switches:
|
|
# Grab the word after --room
|
|
parts = raw.split()
|
|
try:
|
|
room_filter = parts[parts.index("--room") + 1]
|
|
parts = [p for p in parts if p not in ("--room", room_filter)]
|
|
raw = " ".join(parts)
|
|
except (ValueError, IndexError):
|
|
pass
|
|
|
|
# Strip inline switch tokens from query text
|
|
query = raw.replace("--fleet", "").strip()
|
|
if not query:
|
|
self.caller.msg("Please provide a search query.")
|
|
return
|
|
|
|
wing = getattr(self.caller.db, "wing", None) or FLEET_WING
|
|
|
|
try:
|
|
if fleet_mode:
|
|
results = search_fleet(query, room=room_filter)
|
|
header = f"|cFleet palace|n — searching all wings for: |w{query}|n"
|
|
else:
|
|
results = search_memories(
|
|
query, wing=wing, room=room_filter
|
|
)
|
|
header = (
|
|
f"|cPalace|n [{wing}] — searching for: |w{query}|n"
|
|
+ (f" in room |y{room_filter}|n" if room_filter else "")
|
|
)
|
|
except MemPalaceUnavailable as exc:
|
|
self.caller.msg(f"|rPalace unavailable:|n {exc}")
|
|
return
|
|
|
|
if not results:
|
|
self.caller.msg(f"{header}\n|yNo memories found.|n")
|
|
return
|
|
|
|
self.caller.msg(header)
|
|
for i, r in enumerate(results[:5], start=1):
|
|
wing_tag = f" |x[{r.wing}]|n" if fleet_mode and r.wing else ""
|
|
self.caller.msg(
|
|
f"|c{i}. {r.room}{wing_tag}|n (score {r.score:.2f})\n"
|
|
f" {r.short(240)}"
|
|
)
|
|
|
|
|
|
class CmdEnterRoom(Command):
|
|
"""Teleport to the palace room that best matches a topic.
|
|
|
|
Usage:
|
|
enter room <topic>
|
|
|
|
Examples:
|
|
enter room forge
|
|
enter room CI failures
|
|
enter room agent architecture
|
|
|
|
If the topic matches a canonical room name exactly, you are
|
|
teleported there directly. Otherwise a semantic search finds
|
|
the closest room and you are taken there.
|
|
"""
|
|
|
|
key = "enter room"
|
|
aliases = ["go to room", "palace room"]
|
|
locks = "cmd:all()"
|
|
help_category = "MemPalace"
|
|
|
|
def func(self):
|
|
topic = self.args.strip()
|
|
if not topic:
|
|
self.caller.msg("Usage: enter room <topic>")
|
|
rooms = ", ".join(f"|c{r}|n" for r in CORE_ROOMS)
|
|
self.caller.msg(f"Core palace rooms: {rooms}")
|
|
return
|
|
|
|
# Resolve room name — exact match first, then semantic
|
|
if topic.lower() in CORE_ROOMS:
|
|
room_name = topic.lower()
|
|
else:
|
|
# Fuzzy: pick the room whose name is most similar
|
|
room_name = _closest_room(topic)
|
|
|
|
# Try to find the in-game room object by key/alias
|
|
try:
|
|
from evennia.utils.search import search_object # type: ignore
|
|
matches = search_object(
|
|
room_name,
|
|
typeclass="nexus.evennia_mempalace.typeclasses.rooms.MemPalaceRoom",
|
|
)
|
|
except Exception:
|
|
matches = []
|
|
|
|
if matches:
|
|
destination = matches[0]
|
|
self.caller.move_to(destination, quiet=False)
|
|
else:
|
|
self.caller.msg(
|
|
f"|yNo palace room found for '|w{room_name}|y'.|n\n"
|
|
"Ask the world administrator to create the room with the "
|
|
"|cMemPalaceRoom|n typeclass."
|
|
)
|
|
|
|
|
|
_ROOM_KEYWORDS: dict[str, list[str]] = {
|
|
"forge": ["ci", "build", "pipeline", "deploy", "docker", "infra", "cron", "runner"],
|
|
"hermes": ["hermes", "agent", "gateway", "cli", "harness", "mcp", "session"],
|
|
"nexus": ["nexus", "report", "doc", "sitrep", "knowledge", "kt", "handoff"],
|
|
"issues": ["issue", "ticket", "bug", "pr", "backlog", "triage", "milestone"],
|
|
"experiments": ["experiment", "spike", "prototype", "bench", "research", "proof"],
|
|
}
|
|
|
|
|
|
def _closest_room(topic: str) -> str:
|
|
"""Return the CORE_ROOMS name most similar to *topic*.
|
|
|
|
Checks in order:
|
|
1. Exact name match.
|
|
2. Name substring in topic (or vice versa).
|
|
3. Keyword synonym lookup.
|
|
"""
|
|
topic_lower = topic.lower()
|
|
topic_words = set(topic_lower.split())
|
|
|
|
for room in CORE_ROOMS:
|
|
if room == topic_lower or room in topic_lower or topic_lower in room:
|
|
return room
|
|
|
|
for room, keywords in _ROOM_KEYWORDS.items():
|
|
for kw in keywords:
|
|
if kw in topic_words or any(kw in w for w in topic_words):
|
|
return room
|
|
|
|
return "general"
|
|
|
|
|
|
class CmdAsk(Command):
|
|
"""Ask a steward NPC a question about their wing's memory.
|
|
|
|
Usage:
|
|
ask <npc-name> about <topic>
|
|
ask steward about CI pipeline
|
|
ask bezalel-steward about nightly watch failures
|
|
|
|
The NPC must be in the current room and must use the StewardNPC
|
|
typeclass. Their response is drawn from a live palace search.
|
|
"""
|
|
|
|
key = "ask"
|
|
locks = "cmd:all()"
|
|
help_category = "MemPalace"
|
|
|
|
def func(self):
|
|
raw = self.args.strip()
|
|
if " about " not in raw:
|
|
self.caller.msg("Usage: ask <npc-name> about <topic>")
|
|
return
|
|
|
|
npc_name, _, topic = raw.partition(" about ")
|
|
npc_name = npc_name.strip()
|
|
topic = topic.strip()
|
|
|
|
if not npc_name or not topic:
|
|
self.caller.msg("Usage: ask <npc-name> about <topic>")
|
|
return
|
|
|
|
# Find the NPC in the current room
|
|
try:
|
|
from evennia.utils.search import search_object # type: ignore
|
|
candidates = search_object(
|
|
npc_name,
|
|
typeclass="nexus.evennia_mempalace.typeclasses.npcs.StewardNPC",
|
|
)
|
|
except Exception:
|
|
candidates = []
|
|
|
|
if not candidates:
|
|
# Fallback: search contents of the current room by name
|
|
location = getattr(self.caller, "location", None)
|
|
candidates = [
|
|
obj for obj in (getattr(location, "contents", []) or [])
|
|
if npc_name.lower() in obj.key.lower()
|
|
]
|
|
|
|
if not candidates:
|
|
self.caller.msg(
|
|
f"|yNo steward named '|w{npc_name}|y' found here.|n\n"
|
|
"Stewards are created with the |cStewardNPC|n typeclass."
|
|
)
|
|
return
|
|
|
|
npc = candidates[0]
|
|
response = npc.respond_to_question(topic, asker=self.caller)
|
|
self.caller.msg(response)
|