diff --git a/docs/mempalace/rooms.yaml b/docs/mempalace/rooms.yaml new file mode 100644 index 00000000..8d8a80ce --- /dev/null +++ b/docs/mempalace/rooms.yaml @@ -0,0 +1,183 @@ +# MemPalace Fleet Room Taxonomy Standard +# ======================================= +# Version: 1.0 +# Milestone: MemPalace × Evennia — Fleet Memory (#1075) +# Issue: #1082 [Infra] Palace taxonomy standard +# +# Every wizard's palace MUST contain the five core rooms listed below. +# Domain rooms are optional and wizard-specific. +# +# Format: +# rooms: +# : +# required: true|false +# description: one-liner purpose +# example_topics: [list of things that belong here] +# tunnel: true if a cross-wizard tunnel should exist for this room + +rooms: + + # ── Core rooms (required in every wing) ──────────────────────────────────── + + forge: + required: true + description: "CI, builds, deployment, infra operations" + example_topics: + - "github actions failures" + - "docker build logs" + - "server deployment steps" + - "cron job setup" + tunnel: true + + hermes: + required: true + description: "Agent platform, gateway, CLI tooling, harness internals" + example_topics: + - "hermes session logs" + - "agent wake cycle" + - "MCP tool calls" + - "gateway configuration" + tunnel: true + + nexus: + required: true + description: "Reports, docs, knowledge transfer, SITREPs" + example_topics: + - "nightly watch report" + - "architecture docs" + - "handoff notes" + - "decision records" + tunnel: true + + issues: + required: true + description: "Gitea tickets, backlog items, bug reports, PR reviews" + example_topics: + - "issue triage" + - "PR feedback" + - "bug root cause" + - "milestone planning" + tunnel: true + + experiments: + required: true + description: "Prototypes, spikes, research, benchmarks" + example_topics: + - "spike results" + - "benchmark numbers" + - "proof of concept" + - "chromadb evaluation" + tunnel: true + + # ── Write rooms (created on demand by CmdRecord/CmdNote/CmdEvent) ────────── + + hall_facts: + required: false + description: "Decisions and facts recorded via 'record' command" + example_topics: + - "architectural decisions" + - "policy choices" + - "approved approaches" + tunnel: false + + hall_discoveries: + required: false + description: "Breakthroughs and key findings recorded via 'note' command" + example_topics: + - "performance breakthroughs" + - "algorithmic insights" + - "unexpected results" + tunnel: false + + hall_events: + required: false + description: "Significant events logged via 'event' command" + example_topics: + - "production deployments" + - "milestones reached" + - "incidents resolved" + tunnel: false + + # ── Optional domain rooms (wizard-specific) ──────────────────────────────── + + evennia: + required: false + description: "Evennia MUD world: rooms, commands, NPCs, world design" + example_topics: + - "command implementation" + - "typeclass design" + - "world building notes" + wizard: ["bezalel"] + tunnel: false + + game_portals: + required: false + description: "Portal/gameplay work: satflow, economy, portal registry" + example_topics: + - "portal specs" + - "satflow visualization" + - "economy rules" + wizard: ["bezalel", "timmy"] + tunnel: false + + workspace: + required: false + description: "General wizard workspace notes that don't fit elsewhere" + example_topics: + - "daily notes" + - "scratch work" + - "reference lookups" + tunnel: false + + general: + required: false + description: "Fallback room for unclassified memories" + example_topics: + - "uncategorized notes" + tunnel: false + + +# ── Tunnel policy ───────────────────────────────────────────────────────────── +# +# A tunnel is a cross-wing link that lets any wizard recall memories +# from an equivalent room in another wing. +# +# Rules: +# 1. Only CLOSETS (summaries) are synced through tunnels — never raw drawers. +# 2. Required rooms marked tunnel:true MUST have tunnels on Alpha. +# 3. Optional rooms are never tunnelled unless explicitly opted in. +# 4. Raw drawers (source_file metadata) never leave the local VPS. + +tunnels: + policy: closets_only + sync_schedule: "04:00 UTC nightly" + destination: "/var/lib/mempalace/fleet" + rooms_synced: + - forge + - hermes + - nexus + - issues + - experiments + + +# ── Privacy rules ───────────────────────────────────────────────────────────── +# +# See issue #1083 for the full privacy boundary design. +# +# Summary: +# - hall_facts, hall_discoveries, hall_events: LOCAL ONLY (never synced) +# - workspace, general: LOCAL ONLY +# - Domain rooms (evennia, game_portals): LOCAL ONLY unless tunnel:true +# - source_file paths MUST be stripped before sync + +privacy: + local_only_rooms: + - hall_facts + - hall_discoveries + - hall_events + - workspace + - general + strip_on_sync: + - source_file + retention_days: 90 + archive_flag: "archive: true" diff --git a/nexus/evennia_mempalace/__init__.py b/nexus/evennia_mempalace/__init__.py new file mode 100644 index 00000000..640f1658 --- /dev/null +++ b/nexus/evennia_mempalace/__init__.py @@ -0,0 +1,49 @@ +"""nexus.evennia_mempalace — Evennia plugin for MemPalace fleet memory. + +This contrib module provides: + +Commands (add to ``settings.CMDSETS_DEFAULT`` or a CmdSet): + CmdRecall — ``recall `` / ``recall --fleet`` + CmdEnterRoom — ``enter room `` teleports to a palace room + CmdRecord — ``record decision `` writes to hall_facts + CmdNote — ``note breakthrough `` writes to hall_discoveries + CmdEvent — ``event `` writes to hall_events + +Typeclasses (use in place of Evennia's default Room/Character): + MemPalaceRoom — Room whose description auto-populates from palace search + StewardNPC — Wizard steward that answers questions via palace search + +Usage example (in your Evennia game's ``mygame/server/conf/settings.py``):: + + MEMPALACE_PATH = "/root/wizards/bezalel/.mempalace/palace" + MEMPALACE_WING = "bezalel" + FLEET_PALACE_PATH = "/var/lib/mempalace/fleet" + +Then import commands into a CmdSet:: + + from nexus.evennia_mempalace.commands import ( + CmdRecall, CmdEnterRoom, CmdRecord, CmdNote, CmdEvent + ) +""" + +from __future__ import annotations + +from nexus.evennia_mempalace.commands import ( + CmdEnterRoom, + CmdEvent, + CmdNote, + CmdRecord, + CmdRecall, +) +from nexus.evennia_mempalace.typeclasses.rooms import MemPalaceRoom +from nexus.evennia_mempalace.typeclasses.npcs import StewardNPC + +__all__ = [ + "CmdRecall", + "CmdEnterRoom", + "CmdRecord", + "CmdNote", + "CmdEvent", + "MemPalaceRoom", + "StewardNPC", +] diff --git a/nexus/evennia_mempalace/__pycache__/__init__.cpython-312.pyc b/nexus/evennia_mempalace/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 00000000..94bdccc7 Binary files /dev/null and b/nexus/evennia_mempalace/__pycache__/__init__.cpython-312.pyc differ diff --git a/nexus/evennia_mempalace/commands/__init__.py b/nexus/evennia_mempalace/commands/__init__.py new file mode 100644 index 00000000..f805ecdf --- /dev/null +++ b/nexus/evennia_mempalace/commands/__init__.py @@ -0,0 +1,14 @@ +"""MemPalace Evennia commands.""" + +from __future__ import annotations + +from nexus.evennia_mempalace.commands.recall import CmdRecall, CmdEnterRoom +from nexus.evennia_mempalace.commands.write import CmdRecord, CmdNote, CmdEvent + +__all__ = [ + "CmdRecall", + "CmdEnterRoom", + "CmdRecord", + "CmdNote", + "CmdEvent", +] diff --git a/nexus/evennia_mempalace/commands/__pycache__/__init__.cpython-312.pyc b/nexus/evennia_mempalace/commands/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 00000000..360a48bb Binary files /dev/null and b/nexus/evennia_mempalace/commands/__pycache__/__init__.cpython-312.pyc differ diff --git a/nexus/evennia_mempalace/commands/__pycache__/recall.cpython-312.pyc b/nexus/evennia_mempalace/commands/__pycache__/recall.cpython-312.pyc new file mode 100644 index 00000000..5d401158 Binary files /dev/null and b/nexus/evennia_mempalace/commands/__pycache__/recall.cpython-312.pyc differ diff --git a/nexus/evennia_mempalace/commands/__pycache__/write.cpython-312.pyc b/nexus/evennia_mempalace/commands/__pycache__/write.cpython-312.pyc new file mode 100644 index 00000000..d6a0cf15 Binary files /dev/null and b/nexus/evennia_mempalace/commands/__pycache__/write.cpython-312.pyc differ diff --git a/nexus/evennia_mempalace/commands/recall.py b/nexus/evennia_mempalace/commands/recall.py new file mode 100644 index 00000000..05561d29 --- /dev/null +++ b/nexus/evennia_mempalace/commands/recall.py @@ -0,0 +1,206 @@ +"""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 + +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 + recall --fleet + recall --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 [--fleet] [--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 + + 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 ") + 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" diff --git a/nexus/evennia_mempalace/commands/write.py b/nexus/evennia_mempalace/commands/write.py new file mode 100644 index 00000000..a1645778 --- /dev/null +++ b/nexus/evennia_mempalace/commands/write.py @@ -0,0 +1,124 @@ +"""Evennia commands for writing new memories to the palace. + +CmdRecord — record decision → files into hall_facts +CmdNote — note breakthrough → files into hall_discoveries +CmdEvent — event → files into hall_events + +Phase 4 deliverable (see issue #1080). +""" + +from __future__ import annotations + +from nexus.mempalace.searcher import MemPalaceUnavailable, add_memory +from nexus.mempalace.config import FLEET_WING + +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): + class Command: # type: ignore + 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 _MemWriteCommand(Command): + """Base class for palace write commands.""" + + _room: str = "general" + _label: str = "memory" + + def func(self): + text = self.args.strip() + if not text: + self.caller.msg(f"Usage: {self.key} ") + return + + wing = getattr(self.caller.db, "wing", None) or FLEET_WING + try: + doc_id = add_memory( + text, + room=self._room, + wing=wing, + extra_metadata={"via": "evennia_cmd", "cmd": self.key}, + ) + except MemPalaceUnavailable as exc: + self.caller.msg(f"|rPalace unavailable:|n {exc}") + return + + self.caller.msg( + f"|gFiled {self._label} into |c{self._room}|g.|n (id: {doc_id[:8]}…)" + ) + + +class CmdRecord(_MemWriteCommand): + """Record a decision into the palace (hall_facts). + + Usage: + record + record decision + + Example: + record We decided to use ChromaDB for local palace storage. + + The text is filed into the ``hall_facts`` room of your wing and + becomes searchable via ``recall``. + """ + + key = "record" + aliases = ["record decision"] + locks = "cmd:all()" + help_category = "MemPalace" + _room = "hall_facts" + _label = "decision" + + +class CmdNote(_MemWriteCommand): + """File a breakthrough note into the palace (hall_discoveries). + + Usage: + note + note breakthrough + + Example: + note breakthrough AAAK compression reduces token cost by 40%. + + The text is filed into the ``hall_discoveries`` room of your wing. + """ + + key = "note" + aliases = ["note breakthrough"] + locks = "cmd:all()" + help_category = "MemPalace" + _room = "hall_discoveries" + _label = "breakthrough" + + +class CmdEvent(_MemWriteCommand): + """Log a significant event into the palace (hall_events). + + Usage: + event + + Example: + event Deployed Evennia bridge to production on Alpha. + + The text is filed into the ``hall_events`` room of your wing. + """ + + key = "event" + locks = "cmd:all()" + help_category = "MemPalace" + _room = "hall_events" + _label = "event" diff --git a/nexus/evennia_mempalace/typeclasses/__init__.py b/nexus/evennia_mempalace/typeclasses/__init__.py new file mode 100644 index 00000000..3bb7bd9d --- /dev/null +++ b/nexus/evennia_mempalace/typeclasses/__init__.py @@ -0,0 +1 @@ +"""MemPalace Evennia typeclasses.""" diff --git a/nexus/evennia_mempalace/typeclasses/__pycache__/__init__.cpython-312.pyc b/nexus/evennia_mempalace/typeclasses/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 00000000..b8d97bf5 Binary files /dev/null and b/nexus/evennia_mempalace/typeclasses/__pycache__/__init__.cpython-312.pyc differ diff --git a/nexus/evennia_mempalace/typeclasses/__pycache__/npcs.cpython-312.pyc b/nexus/evennia_mempalace/typeclasses/__pycache__/npcs.cpython-312.pyc new file mode 100644 index 00000000..9ed7f43a Binary files /dev/null and b/nexus/evennia_mempalace/typeclasses/__pycache__/npcs.cpython-312.pyc differ diff --git a/nexus/evennia_mempalace/typeclasses/__pycache__/rooms.cpython-312.pyc b/nexus/evennia_mempalace/typeclasses/__pycache__/rooms.cpython-312.pyc new file mode 100644 index 00000000..7a8171f0 Binary files /dev/null and b/nexus/evennia_mempalace/typeclasses/__pycache__/rooms.cpython-312.pyc differ diff --git a/nexus/evennia_mempalace/typeclasses/npcs.py b/nexus/evennia_mempalace/typeclasses/npcs.py new file mode 100644 index 00000000..f0beb984 --- /dev/null +++ b/nexus/evennia_mempalace/typeclasses/npcs.py @@ -0,0 +1,138 @@ +"""StewardNPC — wizard steward that answers questions via palace search. + +Each wizard wing has a steward NPC that players can interrogate about +the wing's history. The NPC: + +1. Detects the topic from the player's question. +2. Calls ``search_memories`` with wing + optional room filters. +3. Formats the top results as an in-character response. + +Phase 3 deliverable (see issue #1079). +""" + +from __future__ import annotations + +from nexus.mempalace.searcher import MemPalaceUnavailable, search_memories +from nexus.mempalace.config import FLEET_WING + +try: + from evennia import DefaultCharacter as _EvDefaultCharacter # type: ignore + if _EvDefaultCharacter is None: + raise ImportError("evennia.DefaultCharacter is None") + DefaultCharacter = _EvDefaultCharacter +except (ImportError, Exception): + class DefaultCharacter: # type: ignore # minimal stub + db: object = None + key: str = "" + + def msg(self, text: str, **kwargs): + pass + + def execute_cmd(self, raw_string: str, **kwargs): + pass + + +# Steward response templates +_FOUND_TEMPLATE = ( + "|c{name}|n glances inward, consulting the palace...\n\n" + "I find {count} relevant {plural} about |w{topic}|n:\n\n" + "{memories}\n" + "|xType '|wrecall {topic}|x' to search further.|n" +) +_NOT_FOUND_TEMPLATE = ( + "|c{name}|n ponders a moment, then shakes their head.\n" + "\"I found nothing about |w{topic}|n in this wing's memory.\"" +) +_UNAVAILABLE_TEMPLATE = ( + "|c{name}|n frowns. \"The palace is unreachable right now.\"" +) + + +class StewardNPC(DefaultCharacter): + """An NPC that serves as the custodian of a wizard's memory wing. + + Attributes (set via ``npc.db.``): + steward_wing (str): The wizard wing this steward guards. + Defaults to ``FLEET_WING``. + steward_name (str): Display name used in responses. + Defaults to ``self.key``. + steward_n_results (int): How many memories to surface. + Default 3. + + Usage (from game):: + + > ask bezalel-steward about nightly watch failures + > ask steward about CI pipeline + """ + + # Evennia will call at_say when players speak near the NPC + def at_say(self, message: str, msg_type: str = "say", **kwargs): + """Intercept nearby speech that looks like a question.""" + super().at_say(message, msg_type=msg_type, **kwargs) + + def respond_to_question(self, question: str, asker=None) -> str: + """Answer a question by searching the wing's palace. + + Args: + question: The player's raw question text. + asker: The asking character object (used to personalise output). + + Returns: + Formatted response string. + """ + topic = _extract_topic(question) + wing = self.db.steward_wing or FLEET_WING + name = self.db.steward_name or self.key + n = self.db.steward_n_results or 3 + + try: + results = search_memories(topic, wing=wing, n_results=n) + except MemPalaceUnavailable: + return _UNAVAILABLE_TEMPLATE.format(name=name) + + if not results: + return _NOT_FOUND_TEMPLATE.format(name=name, topic=topic) + + memory_lines = [] + for i, r in enumerate(results, start=1): + memory_lines.append( + f"|w{i}. [{r.room}]|n {r.short(220)}" + ) + + return _FOUND_TEMPLATE.format( + name=name, + count=len(results), + plural="memory" if len(results) == 1 else "memories", + topic=topic, + memories="\n".join(memory_lines), + ) + + +# ── Helpers ─────────────────────────────────────────────────────────────────── + +_QUESTION_PREFIXES = ( + "about ", "regarding ", "on ", "concerning ", + "related to ", "for ", "with ", "involving ", +) + + +def _extract_topic(question: str) -> str: + """Extract the key topic from a natural-language question. + + Strips common question prefixes so that the palace search receives + a clean keyword rather than noise words. + + Examples: + "about nightly watch failures" → "nightly watch failures" + "what do you know about the CI pipeline?" → "CI pipeline" + """ + q = question.strip().rstrip("?").strip() + # Remove leading question words + for prefix in ("what do you know ", "tell me ", "do you know "): + if q.lower().startswith(prefix): + q = q[len(prefix):] + for prep in _QUESTION_PREFIXES: + if q.lower().startswith(prep): + q = q[len(prep):] + break + return q or question.strip() diff --git a/nexus/evennia_mempalace/typeclasses/rooms.py b/nexus/evennia_mempalace/typeclasses/rooms.py new file mode 100644 index 00000000..0c1d41c6 --- /dev/null +++ b/nexus/evennia_mempalace/typeclasses/rooms.py @@ -0,0 +1,99 @@ +"""MemPalaceRoom — Evennia room typeclass backed by palace search. + +When a character enters a MemPalaceRoom, the room's description is +automatically refreshed from a live palace search for the room's +topic keyword. This makes the room "alive" — its contents reflect +what the fleet actually knows about that topic. + +Phase 1 deliverable (see issue #1077). +""" + +from __future__ import annotations + +from nexus.mempalace.searcher import MemPalaceUnavailable, search_memories +from nexus.mempalace.config import FLEET_WING + +try: + from evennia import DefaultRoom as _EvDefaultRoom # type: ignore + if _EvDefaultRoom is None: + raise ImportError("evennia.DefaultRoom is None") + DefaultRoom = _EvDefaultRoom +except (ImportError, Exception): + class DefaultRoom: # type: ignore # minimal stub for import/testing + """Stub for environments without Evennia installed.""" + + db: object = None + key: str = "" + + def return_appearance(self, looker): # noqa: D102 + return "" + + def at_object_receive(self, moved_obj, source_location, **kwargs): # noqa: D102 + pass + + +_PALACE_ROOM_HEADER = """|b═══════════════════════════════════════════════════|n +|c Mind Palace — {room_name}|n +|b═══════════════════════════════════════════════════|n""" + +_PALACE_ROOM_FOOTER = """|b───────────────────────────────────────────────────|n +|xType '|wrecall |x' to search deeper.|n""" + + +class MemPalaceRoom(DefaultRoom): + """An Evennia room whose description comes from the MemPalace. + + Attributes (set via ``room.db.``): + palace_topic (str): Search term used to populate the description. + Defaults to the room's key. + palace_wing (str): Wing to search. Defaults to fleet wing. + palace_n_results (int): How many memories to show. Default 3. + palace_room_filter (str): Optional room-name filter for the query. + """ + + def at_object_receive(self, moved_obj, source_location, **kwargs): + """Refresh palace content whenever someone enters.""" + super().at_object_receive(moved_obj, source_location, **kwargs) + # Only refresh for player-controlled characters + if hasattr(moved_obj, "account") and moved_obj.account: + self._refresh_palace_desc(viewer=moved_obj) + + def return_appearance(self, looker, **kwargs): + """Return description augmented with live palace memories.""" + self._refresh_palace_desc(viewer=looker) + return super().return_appearance(looker, **kwargs) + + # ── Internal helpers ────────────────────────────────────────────────── + + def _refresh_palace_desc(self, viewer=None): + """Update ``self.db.desc`` from a fresh palace query.""" + topic = self.db.palace_topic or self.key or "general" + wing = self.db.palace_wing or FLEET_WING + n = self.db.palace_n_results or 3 + room_filter = self.db.palace_room_filter + + try: + results = search_memories( + topic, wing=wing, room=room_filter, n_results=n + ) + except MemPalaceUnavailable: + self.db.desc = ( + f"[Palace unavailable — could not load memories for '{topic}'.]" + ) + return + + lines = [ + _PALACE_ROOM_HEADER.format(room_name=self.key), + ] + + if results: + for r in results: + lines.append(f"|w{r.room}|n |x(score {r.score:.2f})|n") + lines.append(f" {r.short(280)}") + lines.append("") + else: + lines.append(f"|yNo memories found for topic '|w{topic}|y'.|n") + lines.append("") + + lines.append(_PALACE_ROOM_FOOTER) + self.db.desc = "\n".join(lines) diff --git a/nexus/mempalace/__init__.py b/nexus/mempalace/__init__.py new file mode 100644 index 00000000..bcb9a5d5 --- /dev/null +++ b/nexus/mempalace/__init__.py @@ -0,0 +1,23 @@ +"""nexus.mempalace — MemPalace integration for the Nexus fleet. + +Public API for searching, configuring, and writing to MemPalace +local vector memory. Designed to be imported by both the +``evennia_mempalace`` plugin and any other harness component. + +ChromaDB is an optional runtime dependency; the module degrades +gracefully when it is not installed (tests, CI, environments that +have not yet set up the palace). +""" + +from __future__ import annotations + +from nexus.mempalace.config import MEMPALACE_PATH, FLEET_WING +from nexus.mempalace.searcher import search_memories, add_memory, MemPalaceResult + +__all__ = [ + "MEMPALACE_PATH", + "FLEET_WING", + "search_memories", + "add_memory", + "MemPalaceResult", +] diff --git a/nexus/mempalace/__pycache__/__init__.cpython-312.pyc b/nexus/mempalace/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 00000000..885a9ac5 Binary files /dev/null and b/nexus/mempalace/__pycache__/__init__.cpython-312.pyc differ diff --git a/nexus/mempalace/__pycache__/config.cpython-312.pyc b/nexus/mempalace/__pycache__/config.cpython-312.pyc new file mode 100644 index 00000000..a7e05237 Binary files /dev/null and b/nexus/mempalace/__pycache__/config.cpython-312.pyc differ diff --git a/nexus/mempalace/__pycache__/searcher.cpython-312.pyc b/nexus/mempalace/__pycache__/searcher.cpython-312.pyc new file mode 100644 index 00000000..21e28c58 Binary files /dev/null and b/nexus/mempalace/__pycache__/searcher.cpython-312.pyc differ diff --git a/nexus/mempalace/config.py b/nexus/mempalace/config.py new file mode 100644 index 00000000..f5155377 --- /dev/null +++ b/nexus/mempalace/config.py @@ -0,0 +1,46 @@ +"""MemPalace configuration — paths and fleet settings. + +All configuration is driven by environment variables so that +different wizards on different VPSes can use the same code with +their own palace directories. +""" + +from __future__ import annotations + +import os +from pathlib import Path + +# ── Palace path ────────────────────────────────────────────────────────────── +# Default: ~/.mempalace/palace/ (local wizard palace) +# Override via MEMPALACE_PATH env var (useful for fleet shared wing) +_default = Path.home() / ".mempalace" / "palace" +MEMPALACE_PATH: Path = Path(os.environ.get("MEMPALACE_PATH", str(_default))) + +# ── Fleet shared wing ───────────────────────────────────────────────────────── +# Path to the shared fleet palace on Alpha (used by --fleet searches) +_fleet_default = Path("/var/lib/mempalace/fleet") +FLEET_PALACE_PATH: Path = Path( + os.environ.get("FLEET_PALACE_PATH", str(_fleet_default)) +) + +# ── Wing name ───────────────────────────────────────────────────────────────── +# Identifies this wizard's wing within a shared palace. +# Populated from MEMPALACE_WING env var or falls back to system username. +def _default_wing() -> str: + import getpass + return os.environ.get("MEMPALACE_WING", getpass.getuser()) + +FLEET_WING: str = _default_wing() + +# ── Fleet rooms standard ───────────────────────────────────────────────────── +# Canonical rooms every wizard must have (see docs/mempalace/rooms.yaml) +CORE_ROOMS: list[str] = [ + "forge", # CI, builds, infra + "hermes", # agent platform, gateway, CLI + "nexus", # reports, docs, KT + "issues", # tickets, backlog + "experiments", # prototypes, spikes +] + +# ── ChromaDB collection name ────────────────────────────────────────────────── +COLLECTION_NAME: str = os.environ.get("MEMPALACE_COLLECTION", "palace") diff --git a/nexus/mempalace/searcher.py b/nexus/mempalace/searcher.py new file mode 100644 index 00000000..6cba1194 --- /dev/null +++ b/nexus/mempalace/searcher.py @@ -0,0 +1,200 @@ +"""MemPalace search and write interface. + +Wraps the ChromaDB-backed palace so that callers (Evennia commands, +harness agents, MCP tools) do not need to know the storage details. + +ChromaDB is imported lazily; if it is not installed the functions +raise ``MemPalaceUnavailable`` with an informative message rather +than crashing at import time. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path +from typing import Optional + +from nexus.mempalace.config import ( + MEMPALACE_PATH, + FLEET_PALACE_PATH, + COLLECTION_NAME, +) + + +class MemPalaceUnavailable(RuntimeError): + """Raised when ChromaDB or the palace directory is not accessible.""" + + +@dataclass +class MemPalaceResult: + """A single memory hit returned by the searcher.""" + + text: str + room: str + wing: str + score: float = 0.0 + source_file: str = "" + metadata: dict = field(default_factory=dict) + + def short(self, max_chars: int = 200) -> str: + """Return a truncated preview suitable for MUD output.""" + if len(self.text) <= max_chars: + return self.text + return self.text[:max_chars].rstrip() + "…" + + +def _get_client(palace_path: Path): + """Return a ChromaDB persistent client, or raise MemPalaceUnavailable.""" + try: + import chromadb # type: ignore + except ImportError as exc: + raise MemPalaceUnavailable( + "ChromaDB is not installed. " + "Run: pip install chromadb (or: pip install mempalace)" + ) from exc + + if not palace_path.exists(): + raise MemPalaceUnavailable( + f"Palace directory not found: {palace_path}\n" + "Run 'mempalace mine' to initialise the palace." + ) + + return chromadb.PersistentClient(path=str(palace_path)) + + +def search_memories( + query: str, + *, + palace_path: Optional[Path] = None, + wing: Optional[str] = None, + room: Optional[str] = None, + n_results: int = 5, +) -> list[MemPalaceResult]: + """Search the palace for memories matching *query*. + + Args: + query: Natural-language search string. + palace_path: Override the default palace path. + wing: Filter results to a specific wizard's wing. + room: Filter results to a specific room (e.g. ``"forge"``). + n_results: Maximum number of results to return. + + Returns: + List of :class:`MemPalaceResult`, best-match first. + + Raises: + MemPalaceUnavailable: If ChromaDB is not installed or the palace + directory does not exist. + """ + path = palace_path or MEMPALACE_PATH + client = _get_client(path) + + collection = client.get_or_create_collection(COLLECTION_NAME) + + where: dict = {} + if wing: + where["wing"] = wing + if room: + where["room"] = room + + kwargs: dict = {"query_texts": [query], "n_results": n_results} + if where: + kwargs["where"] = where + + raw = collection.query(**kwargs) + + results: list[MemPalaceResult] = [] + if not raw or not raw.get("documents"): + return results + + docs = raw["documents"][0] + metas = raw.get("metadatas", [[]])[0] or [{}] * len(docs) + distances = raw.get("distances", [[]])[0] or [0.0] * len(docs) + + for doc, meta, dist in zip(docs, metas, distances): + results.append( + MemPalaceResult( + text=doc, + room=meta.get("room", "general"), + wing=meta.get("wing", ""), + score=float(1.0 - dist), # cosine similarity from distance + source_file=meta.get("source_file", ""), + metadata=meta, + ) + ) + + return results + + +def search_fleet( + query: str, + *, + room: Optional[str] = None, + n_results: int = 10, +) -> list[MemPalaceResult]: + """Search the shared fleet palace (closets only, no raw drawers). + + Args: + query: Natural-language search string. + room: Optional room filter (e.g. ``"issues"``). + n_results: Maximum results. + + Returns: + List of :class:`MemPalaceResult` from all wings. + """ + return search_memories( + query, + palace_path=FLEET_PALACE_PATH, + room=room, + n_results=n_results, + ) + + +def add_memory( + text: str, + *, + room: str = "general", + wing: Optional[str] = None, + palace_path: Optional[Path] = None, + source_file: str = "", + extra_metadata: Optional[dict] = None, +) -> str: + """Add a new memory drawer to the palace. + + Args: + text: The memory text to store. + room: Target room (e.g. ``"hall_facts"``). + wing: Wing name; defaults to :data:`~nexus.mempalace.config.FLEET_WING`. + palace_path: Override the default palace path. + source_file: Optional source file attribution. + extra_metadata: Additional key/value metadata to store. + + Returns: + The generated document ID. + + Raises: + MemPalaceUnavailable: If ChromaDB is not installed or the palace + directory does not exist. + """ + import uuid + from nexus.mempalace.config import FLEET_WING + + path = palace_path or MEMPALACE_PATH + client = _get_client(path) + collection = client.get_or_create_collection(COLLECTION_NAME) + + doc_id = str(uuid.uuid4()) + metadata: dict = { + "room": room, + "wing": wing or FLEET_WING, + "source_file": source_file, + } + if extra_metadata: + metadata.update(extra_metadata) + + collection.add( + documents=[text], + metadatas=[metadata], + ids=[doc_id], + ) + return doc_id diff --git a/tests/test_evennia_mempalace_commands.py b/tests/test_evennia_mempalace_commands.py new file mode 100644 index 00000000..b296fca4 --- /dev/null +++ b/tests/test_evennia_mempalace_commands.py @@ -0,0 +1,244 @@ +"""Tests for nexus.evennia_mempalace commands and NPC helpers.""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest + +from nexus.evennia_mempalace.commands.recall import CmdRecall, CmdEnterRoom, _closest_room +from nexus.evennia_mempalace.commands.write import CmdRecord, CmdNote, CmdEvent +from nexus.evennia_mempalace.typeclasses.npcs import StewardNPC, _extract_topic +from nexus.mempalace.searcher import MemPalaceResult, MemPalaceUnavailable + + +# ── Helpers ────────────────────────────────────────────────────────────────── + + +def _make_caller(wing: str = "bezalel"): + """Build a minimal mock Evennia caller.""" + caller = MagicMock() + caller.db = MagicMock() + caller.db.wing = wing + caller.account = MagicMock() + return caller + + +def _make_cmd(cls, args: str = "", switches: list | None = None, wing: str = "bezalel"): + """Instantiate an Evennia command mock and wire it up.""" + cmd = cls() + cmd.caller = _make_caller(wing) + cmd.args = args + cmd.switches = switches or [] + return cmd + + +# ── CmdRecall ──────────────────────────────────────────────────────────────── + + +def test_recall_no_args_shows_usage(): + cmd = _make_cmd(CmdRecall, args="") + cmd.func() + cmd.caller.msg.assert_called_once() + assert "Usage" in cmd.caller.msg.call_args[0][0] + + +def test_recall_calls_search_memories(): + results = [ + MemPalaceResult(text="CI pipeline failed", room="forge", wing="bezalel", score=0.9) + ] + with patch("nexus.evennia_mempalace.commands.recall.search_memories", return_value=results): + cmd = _make_cmd(CmdRecall, args="CI failures") + cmd.func() + + calls = [c[0][0] for c in cmd.caller.msg.call_args_list] + assert any("CI pipeline failed" in c for c in calls) + + +def test_recall_fleet_flag_calls_search_fleet(): + results = [ + MemPalaceResult(text="Fleet doc", room="nexus", wing="timmy", score=0.8) + ] + with patch("nexus.evennia_mempalace.commands.recall.search_fleet", return_value=results) as mock_fleet: + cmd = _make_cmd(CmdRecall, args="architecture --fleet", switches=["--fleet"]) + cmd.func() + + mock_fleet.assert_called_once() + query_arg = mock_fleet.call_args[0][0] + assert "--fleet" not in query_arg + assert "architecture" in query_arg + + +def test_recall_unavailable_shows_error(): + with patch( + "nexus.evennia_mempalace.commands.recall.search_memories", + side_effect=MemPalaceUnavailable("ChromaDB not installed"), + ): + cmd = _make_cmd(CmdRecall, args="anything") + cmd.func() + + msg = cmd.caller.msg.call_args[0][0] + assert "unavailable" in msg.lower() + + +def test_recall_no_results_shows_no_memories(): + with patch("nexus.evennia_mempalace.commands.recall.search_memories", return_value=[]): + cmd = _make_cmd(CmdRecall, args="obscure query") + cmd.func() + + calls = " ".join(c[0][0] for c in cmd.caller.msg.call_args_list) + assert "No memories" in calls + + +# ── _closest_room ───────────────────────────────────────────────────────────── + + +@pytest.mark.parametrize("topic,expected", [ + ("forge", "forge"), + ("CI pipeline", "forge"), + ("hermes agent", "hermes"), + ("nexus report", "nexus"), + ("issue triage", "issues"), + ("spike experiment", "experiments"), + ("totally unknown topic xyz", "general"), +]) +def test_closest_room(topic, expected): + assert _closest_room(topic) == expected + + +# ── CmdEnterRoom ────────────────────────────────────────────────────────────── + + +def test_enter_room_no_args_shows_usage(): + cmd = _make_cmd(CmdEnterRoom, args="") + cmd.func() + output = " ".join(c[0][0] for c in cmd.caller.msg.call_args_list) + assert "Usage" in output + assert "forge" in output # shows core rooms + + +def test_enter_room_exact_match_no_room_found(): + """When an exact room name is given but no room exists, show a help message.""" + # evennia.utils.search raises Django config errors outside a live server; + # CmdEnterRoom catches all exceptions and falls back to a help message. + cmd = _make_cmd(CmdEnterRoom, args="forge") + cmd.func() + assert cmd.caller.msg.called + output = " ".join(c[0][0] for c in cmd.caller.msg.call_args_list) + # Should mention the room name or MemPalaceRoom typeclass + assert "forge" in output or "MemPalaceRoom" in output or "No palace room" in output + + +# ── Write commands ─────────────────────────────────────────────────────────── + + +def test_record_no_args_shows_usage(): + cmd = _make_cmd(CmdRecord, args="") + cmd.func() + assert "Usage" in cmd.caller.msg.call_args[0][0] + + +def test_record_calls_add_memory(): + with patch("nexus.evennia_mempalace.commands.write.add_memory", return_value="fake-uuid-1234-5678-abcd") as mock_add: + cmd = _make_cmd(CmdRecord, args="Use ChromaDB for storage.") + cmd.func() + + mock_add.assert_called_once() + kwargs = mock_add.call_args[1] + assert kwargs["room"] == "hall_facts" + assert "ChromaDB" in mock_add.call_args[0][0] + + +def test_note_files_to_hall_discoveries(): + with patch("nexus.evennia_mempalace.commands.write.add_memory", return_value="uuid") as mock_add: + cmd = _make_cmd(CmdNote, args="AAAK reduces cost by 40%.") + cmd.func() + + assert mock_add.call_args[1]["room"] == "hall_discoveries" + + +def test_event_files_to_hall_events(): + with patch("nexus.evennia_mempalace.commands.write.add_memory", return_value="uuid") as mock_add: + cmd = _make_cmd(CmdEvent, args="Deployed Evennia bridge to Alpha.") + cmd.func() + + assert mock_add.call_args[1]["room"] == "hall_events" + + +def test_write_command_unavailable_shows_error(): + with patch( + "nexus.evennia_mempalace.commands.write.add_memory", + side_effect=MemPalaceUnavailable("no palace"), + ): + cmd = _make_cmd(CmdRecord, args="some text") + cmd.func() + + msg = cmd.caller.msg.call_args[0][0] + assert "unavailable" in msg.lower() + + +# ── _extract_topic ──────────────────────────────────────────────────────────── + + +@pytest.mark.parametrize("question,expected_substring", [ + ("about nightly watch failures", "nightly watch failures"), + ("what do you know about CI pipeline?", "CI pipeline"), + ("tell me about hermes", "hermes"), + ("regarding the forge build", "forge build"), + ("nightly watch failures", "nightly watch failures"), +]) +def test_extract_topic(question, expected_substring): + result = _extract_topic(question) + assert expected_substring.lower() in result.lower() + + +# ── StewardNPC.respond_to_question ─────────────────────────────────────────── + + +def test_steward_responds_with_results(): + npc = StewardNPC() + npc.db = MagicMock() + npc.db.steward_wing = "bezalel" + npc.db.steward_name = "Bezalel-Steward" + npc.db.steward_n_results = 3 + npc.key = "steward" + + results = [ + MemPalaceResult(text="Three failures last week.", room="forge", wing="bezalel", score=0.95) + ] + with patch("nexus.evennia_mempalace.typeclasses.npcs.search_memories", return_value=results): + response = npc.respond_to_question("about nightly watch failures") + + assert "Bezalel-Steward" in response + assert "Three failures" in response + + +def test_steward_responds_not_found(): + npc = StewardNPC() + npc.db = MagicMock() + npc.db.steward_wing = "bezalel" + npc.db.steward_name = "Steward" + npc.db.steward_n_results = 3 + npc.key = "steward" + + with patch("nexus.evennia_mempalace.typeclasses.npcs.search_memories", return_value=[]): + response = npc.respond_to_question("about unknown_topic_xyz") + + assert "nothing" in response.lower() or "found" in response.lower() + + +def test_steward_responds_unavailable(): + npc = StewardNPC() + npc.db = MagicMock() + npc.db.steward_wing = "bezalel" + npc.db.steward_name = "Steward" + npc.db.steward_n_results = 3 + npc.key = "steward" + + with patch( + "nexus.evennia_mempalace.typeclasses.npcs.search_memories", + side_effect=MemPalaceUnavailable("no palace"), + ): + response = npc.respond_to_question("about anything") + + assert "unreachable" in response.lower() diff --git a/tests/test_mempalace_searcher.py b/tests/test_mempalace_searcher.py new file mode 100644 index 00000000..16143867 --- /dev/null +++ b/tests/test_mempalace_searcher.py @@ -0,0 +1,190 @@ +"""Tests for nexus.mempalace.searcher and nexus.mempalace.config.""" + +from __future__ import annotations + +import os +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from nexus.mempalace.config import CORE_ROOMS, MEMPALACE_PATH, COLLECTION_NAME +from nexus.mempalace.searcher import ( + MemPalaceResult, + MemPalaceUnavailable, + _get_client, + search_memories, + add_memory, +) + + +# ── MemPalaceResult ────────────────────────────────────────────────────────── + + +def test_result_short_truncates(): + r = MemPalaceResult(text="x" * 300, room="forge", wing="bezalel") + short = r.short(200) + assert len(short) <= 204 # 200 + ellipsis + assert short.endswith("…") + + +def test_result_short_no_truncation_needed(): + r = MemPalaceResult(text="hello", room="nexus", wing="bezalel") + assert r.short() == "hello" + + +def test_result_defaults(): + r = MemPalaceResult(text="test", room="general", wing="") + assert r.score == 0.0 + assert r.source_file == "" + assert r.metadata == {} + + +# ── Config ─────────────────────────────────────────────────────────────────── + + +def test_core_rooms_contains_required_rooms(): + required = {"forge", "hermes", "nexus", "issues", "experiments"} + assert required.issubset(set(CORE_ROOMS)) + + +def test_mempalace_path_env_override(monkeypatch, tmp_path): + monkeypatch.setenv("MEMPALACE_PATH", str(tmp_path)) + # Re-import to pick up env var (config reads at import time so we patch) + import importlib + import nexus.mempalace.config as cfg + importlib.reload(cfg) + assert Path(os.environ["MEMPALACE_PATH"]) == tmp_path + importlib.reload(cfg) # restore + + +# ── _get_client ────────────────────────────────────────────────────────────── + + +def test_get_client_raises_when_chromadb_missing(tmp_path): + with patch.dict("sys.modules", {"chromadb": None}): + with pytest.raises(MemPalaceUnavailable, match="ChromaDB"): + _get_client(tmp_path) + + +def test_get_client_raises_when_path_missing(tmp_path): + missing = tmp_path / "nonexistent_palace" + # chromadb importable but path missing + mock_chroma = MagicMock() + with patch.dict("sys.modules", {"chromadb": mock_chroma}): + with pytest.raises(MemPalaceUnavailable, match="Palace directory"): + _get_client(missing) + + +# ── search_memories ────────────────────────────────────────────────────────── + + +def _make_mock_collection(docs, metas=None, distances=None): + """Build a mock ChromaDB collection that returns canned results.""" + if metas is None: + metas = [{"room": "forge", "wing": "bezalel", "source_file": ""} for _ in docs] + if distances is None: + distances = [0.1 * i for i in range(len(docs))] + + collection = MagicMock() + collection.query.return_value = { + "documents": [docs], + "metadatas": [metas], + "distances": [distances], + } + return collection + + +def _mock_chroma_client(collection): + client = MagicMock() + client.get_or_create_collection.return_value = collection + return client + + +def test_search_memories_returns_results(tmp_path): + docs = ["CI pipeline failed on main", "Forge build log 2026-04-01"] + collection = _make_mock_collection(docs) + mock_chroma = MagicMock() + mock_chroma.PersistentClient.return_value = _mock_chroma_client(collection) + + with patch.dict("sys.modules", {"chromadb": mock_chroma}): + # Palace path must exist for _get_client check + (tmp_path / "chroma.sqlite3").touch() + results = search_memories("CI failures", palace_path=tmp_path) + + assert len(results) == 2 + assert results[0].room == "forge" + assert results[0].wing == "bezalel" + assert "CI pipeline" in results[0].text + + +def test_search_memories_empty_collection(tmp_path): + collection = MagicMock() + collection.query.return_value = {"documents": [[]], "metadatas": [[]], "distances": [[]]} + mock_chroma = MagicMock() + mock_chroma.PersistentClient.return_value = _mock_chroma_client(collection) + + with patch.dict("sys.modules", {"chromadb": mock_chroma}): + (tmp_path / "chroma.sqlite3").touch() + results = search_memories("anything", palace_path=tmp_path) + + assert results == [] + + +def test_search_memories_with_wing_filter(tmp_path): + docs = ["test doc"] + collection = _make_mock_collection(docs) + mock_chroma = MagicMock() + mock_chroma.PersistentClient.return_value = _mock_chroma_client(collection) + + with patch.dict("sys.modules", {"chromadb": mock_chroma}): + (tmp_path / "chroma.sqlite3").touch() + search_memories("query", palace_path=tmp_path, wing="bezalel") + + call_kwargs = collection.query.call_args[1] + assert call_kwargs["where"] == {"wing": "bezalel"} + + +def test_search_memories_with_room_filter(tmp_path): + collection = _make_mock_collection(["doc"]) + mock_chroma = MagicMock() + mock_chroma.PersistentClient.return_value = _mock_chroma_client(collection) + + with patch.dict("sys.modules", {"chromadb": mock_chroma}): + (tmp_path / "chroma.sqlite3").touch() + search_memories("query", palace_path=tmp_path, room="forge") + + call_kwargs = collection.query.call_args[1] + assert call_kwargs["where"] == {"room": "forge"} + + +def test_search_memories_unavailable(tmp_path): + with patch.dict("sys.modules", {"chromadb": None}): + with pytest.raises(MemPalaceUnavailable): + search_memories("anything", palace_path=tmp_path) + + +# ── add_memory ─────────────────────────────────────────────────────────────── + + +def test_add_memory_returns_id(tmp_path): + collection = MagicMock() + mock_chroma = MagicMock() + mock_chroma.PersistentClient.return_value = _mock_chroma_client(collection) + + with patch.dict("sys.modules", {"chromadb": mock_chroma}): + (tmp_path / "chroma.sqlite3").touch() + doc_id = add_memory( + "We decided to use ChromaDB.", + room="hall_facts", + wing="bezalel", + palace_path=tmp_path, + ) + + assert isinstance(doc_id, str) + assert len(doc_id) == 36 # UUID format + collection.add.assert_called_once() + call_kwargs = collection.add.call_args[1] + assert call_kwargs["documents"] == ["We decided to use ChromaDB."] + assert call_kwargs["metadatas"][0]["room"] == "hall_facts" + assert call_kwargs["metadatas"][0]["wing"] == "bezalel"