feat(bezalel): MemPalace ecosystem — validation, audit, sync, auto-revert, Evennia integration

docs/bezalel/evennia/cmd_palace.py

@@ -0,0 +1,246 @@
+"""
+Palace commands — bridge Evennia to the local MemPalace memory system.
+"""
+
+import json
+import subprocess
+from evennia.commands.command import Command
+from evennia import create_object, search_object
+
+PALACE_SCRIPT = "/root/wizards/bezalel/evennia/palace_search.py"
+
+
+def _search_mempalace(query, wing=None, room=None, n=5, fleet=False):
+    """Call the helper script and return parsed results."""
+    cmd = ["/root/wizards/bezalel/hermes/venv/bin/python", PALACE_SCRIPT, query]
+    cmd.append(wing or "none")
+    cmd.append(room or "none")
+    cmd.append(str(n))
+    if fleet:
+        cmd.append("--fleet")
+    try:
+        result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
+        data = json.loads(result.stdout)
+        return data.get("results", [])
+    except Exception:
+        return []
+
+
+def _get_wing(caller):
+    """Return the caller's wing, defaulting to their key or 'general'."""
+    return caller.db.wing if caller.attributes.has("wing") else (caller.key.lower() if caller.key else "general")
+
+
+class CmdPalaceSearch(Command):
+    """
+    Search your memory palace.
+
+    Usage:
+      palace/search <query>
+      palace/search <query> [--room <room>]
+      palace/recall <topic>
+      palace/file <name> = <content>
+      palace/status
+    """
+
+    key = "palace"
+    aliases = ["pal"]
+    locks = "cmd:all()"
+    help_category = "Mind Palace"
+
+    def func(self):
+        if not self.args.strip():
+            self.caller.msg("Usage: palace/search <query> | palace/recall <topic> | palace/file <name> = <content> | palace/status")
+            return
+
+        parts = self.args.strip().split(" ", 1)
+        subcmd = parts[0].lower()
+        rest = parts[1] if len(parts) > 1 else ""
+
+        if subcmd == "search":
+            self._do_search(rest)
+        elif subcmd == "recall":
+            self._do_recall(rest)
+        elif subcmd == "file":
+            self._do_file(rest)
+        elif subcmd == "status":
+            self._do_status()
+        else:
+            self._do_search(self.args.strip())
+
+    def _do_search(self, query):
+        if not query:
+            self.caller.msg("Search for what?")
+            return
+        self.caller.msg(f"Searching the palace for: |c{query}|n...")
+        wing = _get_wing(self.caller)
+        results = _search_mempalace(query, wing=wing)
+        if not results:
+            self.caller.msg("The palace is silent on that matter.")
+            return
+
+        lines = []
+        for i, r in enumerate(results[:5], 1):
+            room = r.get("room", "unknown")
+            source = r.get("source", "unknown")
+            content = r.get("content", "")[:400]
+            lines.append(f"\n|g[{i}]|n |c{room}|n — |x{source}|n")
+            lines.append(f"{content}\n")
+        self.caller.msg("\n".join(lines))
+
+    def _do_recall(self, topic):
+        if not topic:
+            self.caller.msg("Recall what topic?")
+            return
+        results = _search_mempalace(topic, wing=_get_wing(self.caller), n=1)
+        if not results:
+            self.caller.msg("Nothing to recall.")
+            return
+
+        r = results[0]
+        content = r.get("content", "")
+        source = r.get("source", "unknown")
+
+        from typeclasses.memory_object import MemoryObject
+        obj = create_object(
+            MemoryObject,
+            key=f"memory:{topic}",
+            location=self.caller.location,
+        )
+        obj.db.memory_content = content
+        obj.db.source_file = source
+        obj.db.room_name = r.get("room", "general")
+        self.caller.location.msg_contents(
+            f"$You() conjure() a memory shard from the palace: |m{obj.key}|n.",
+            from_obj=self.caller,
+        )
+
+    def _do_file(self, rest):
+        if "=" not in rest:
+            self.caller.msg("Usage: palace/file <name> = <content>")
+            return
+        name, content = rest.split("=", 1)
+        name = name.strip()
+        content = content.strip()
+        if not name or not content:
+            self.caller.msg("Both name and content are required.")
+            return
+
+        from typeclasses.memory_object import MemoryObject
+        obj = create_object(
+            MemoryObject,
+            key=f"memory:{name}",
+            location=self.caller.location,
+        )
+        obj.db.memory_content = content
+        obj.db.source_file = f"filed by {self.caller.key}"
+        obj.db.room_name = self.caller.location.key if self.caller.location else "general"
+        self.caller.location.msg_contents(
+            f"$You() file() a new memory in the palace: |m{obj.key}|n.",
+            from_obj=self.caller,
+        )
+
+    def _do_status(self):
+        cmd = [
+            "/root/wizards/bezalel/hermes/venv/bin/mempalace",
+            "--palace", "/root/wizards/bezalel/.mempalace/palace",
+            "status"
+        ]
+        try:
+            result = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
+            self.caller.msg(result.stdout or result.stderr)
+        except Exception as e:
+            self.caller.msg(f"Could not reach the palace: {e}")
+
+
+class CmdRecall(Command):
+    """
+    Recall a memory from the palace.
+
+    Usage:
+      recall <query>
+      recall <query> --fleet
+      recall <query> --room <room>
+    """
+
+    key = "recall"
+    aliases = ["remember", "mem"]
+    locks = "cmd:all()"
+    help_category = "Mind Palace"
+
+    def func(self):
+        if not self.args.strip():
+            self.caller.msg("Recall what? Usage: recall <query> [--fleet] [--room <room>]")
+            return
+
+        args = self.args.strip()
+        fleet = "--fleet" in args
+        room = None
+
+        if "--room" in args:
+            parts = args.split("--room")
+            args = parts[0].strip()
+            room = parts[1].strip().split()[0] if len(parts) > 1 else None
+
+        if "--fleet" in args:
+            args = args.replace("--fleet", "").strip()
+
+        self.caller.msg(f"Recalling from the {'fleet' if fleet else 'personal'} palace: |c{args}|n...")
+
+        wing = None if fleet else _get_wing(self.caller)
+        results = _search_mempalace(args, wing=wing, room=room, n=5, fleet=fleet)
+        if not results:
+            self.caller.msg("The palace is silent on that matter.")
+            return
+
+        lines = []
+        for i, r in enumerate(results[:5], 1):
+            room_name = r.get("room", "unknown")
+            source = r.get("source", "unknown")
+            content = r.get("content", "")[:400]
+            wing_label = r.get("wing", "unknown")
+            wing_tag = f" |y[{wing_label}]|n" if fleet else ""
+            lines.append(f"\n|g[{i}]|n |c{room_name}|n{wing_tag} — |x{source}|n")
+            lines.append(f"{content}\n")
+        self.caller.msg("\n".join(lines))
+
+
+class CmdEnterRoom(Command):
+    """
+    Enter a room in the mind palace by topic.
+
+    Usage:
+      enter room <topic>
+    """
+
+    key = "enter room"
+    aliases = ["enter palace", "go room"]
+    locks = "cmd:all()"
+    help_category = "Mind Palace"
+
+    def func(self):
+        if not self.args.strip():
+            self.caller.msg("Enter which room? Usage: enter room <topic>")
+            return
+
+        topic = self.args.strip().lower().replace(" ", "-")
+        wing = _get_wing(self.caller)
+        room_key = f"palace:{wing}:{topic}"
+
+        # Search for existing room
+        rooms = search_object(room_key, typeclass="typeclasses.palace_room.PalaceRoom")
+        if rooms:
+            room = rooms[0]
+        else:
+            # Create the room dynamically
+            from typeclasses.palace_room import PalaceRoom
+            room = create_object(
+                PalaceRoom,
+                key=room_key,
+            )
+            room.db.memory_topic = topic
+            room.db.wing = wing
+            room.update_description()
+
+        self.caller.move_to(room, move_type="teleport")
+        self.caller.msg(f"You step into the |c{topic}|n room of your mind palace.")

docs/bezalel/evennia/cmd_record.py

@@ -0,0 +1,166 @@
+"""
+Live memory commands — write new memories into the palace from Evennia.
+"""
+
+import json
+import subprocess
+from evennia.commands.command import Command
+from evennia import create_object
+
+PALACE_SCRIPT = "/root/wizards/bezalel/evennia/palace_search.py"
+PALACE_PATH = "/root/wizards/bezalel/.mempalace/palace"
+ADDER_SCRIPT = "/root/wizards/bezalel/evennia/palace_add.py"
+
+
+def _add_drawer(content, wing, room, source):
+    """Add a verbatim drawer to the palace via the helper script."""
+    cmd = [
+        "/root/wizards/bezalel/hermes/venv/bin/python",
+        ADDER_SCRIPT,
+        content,
+        wing,
+        room,
+        source,
+    ]
+    try:
+        result = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
+        return result.returncode == 0 and "OK" in result.stdout
+    except Exception:
+        return False
+
+
+class CmdRecord(Command):
+    """
+    Record a decision into the palace hall_facts.
+
+    Usage:
+      record <text>
+      record We decided to use PostgreSQL over MySQL.
+    """
+
+    key = "record"
+    aliases = ["decide"]
+    locks = "cmd:all()"
+    help_category = "Mind Palace"
+
+    def func(self):
+        if not self.args.strip():
+            self.caller.msg("Record what decision? Usage: record <text>")
+            return
+
+        wing = self.caller.db.wing if self.caller.attributes.has("wing") else (self.caller.key.lower() if self.caller.key else "general")
+        text = self.args.strip()
+        full_text = f"DECISION ({wing}): {text}\nRecorded by {self.caller.key} via Evennia."
+
+        ok = _add_drawer(full_text, wing, "general", f"evennia:{self.caller.key}")
+        if ok:
+            self.caller.location.msg_contents(
+                f"$You() record() a decision in the palace archives.",
+                from_obj=self.caller,
+            )
+        else:
+            self.caller.msg("The palace scribes could not write that down.")
+
+
+class CmdNote(Command):
+    """
+    Note a breakthrough into the palace hall_discoveries.
+
+    Usage:
+      note <text>
+      note The GraphQL schema can be auto-generated from our typeclasses.
+    """
+
+    key = "note"
+    aliases = ["jot"]
+    locks = "cmd:all()"
+    help_category = "Mind Palace"
+
+    def func(self):
+        if not self.args.strip():
+            self.caller.msg("Note what? Usage: note <text>")
+            return
+
+        wing = self.caller.db.wing if self.caller.attributes.has("wing") else (self.caller.key.lower() if self.caller.key else "general")
+        text = self.args.strip()
+        full_text = f"BREAKTHROUGH ({wing}): {text}\nNoted by {self.caller.key} via Evennia."
+
+        ok = _add_drawer(full_text, wing, "general", f"evennia:{self.caller.key}")
+        if ok:
+            self.caller.location.msg_contents(
+                f"$You() inscribe() a breakthrough into the palace scrolls.",
+                from_obj=self.caller,
+            )
+        else:
+            self.caller.msg("The palace scribes could not write that down.")
+
+
+class CmdEvent(Command):
+    """
+    Log an event into the palace hall_events.
+
+    Usage:
+      event <text>
+      event Gitea runner came back online after being offline for 6 hours.
+    """
+
+    key = "event"
+    aliases = ["log"]
+    locks = "cmd:all()"
+    help_category = "Mind Palace"
+
+    def func(self):
+        if not self.args.strip():
+            self.caller.msg("Log what event? Usage: event <text>")
+            return
+
+        wing = self.caller.db.wing if self.caller.attributes.has("wing") else (self.caller.key.lower() if self.caller.key else "general")
+        text = self.args.strip()
+        full_text = f"EVENT ({wing}): {text}\nLogged by {self.caller.key} via Evennia."
+
+        ok = _add_drawer(full_text, wing, "general", f"evennia:{self.caller.key}")
+        if ok:
+            self.caller.location.msg_contents(
+                f"$You() chronicle() an event in the palace records.",
+                from_obj=self.caller,
+            )
+        else:
+            self.caller.msg("The palace scribes could not write that down.")
+
+
+class CmdPalaceWrite(Command):
+    """
+    Directly write a memory into a specific palace room.
+
+    Usage:
+      palace/write <room> = <text>
+    """
+
+    key = "palace/write"
+    locks = "cmd:all()"
+    help_category = "Mind Palace"
+
+    def func(self):
+        if "=" not in self.args:
+            self.caller.msg("Usage: palace/write <room> = <text>")
+            return
+
+        room, text = self.args.split("=", 1)
+        room = room.strip()
+        text = text.strip()
+
+        if not room or not text:
+            self.caller.msg("Both room and text are required.")
+            return
+
+        wing = self.caller.db.wing if self.caller.attributes.has("wing") else (self.caller.key.lower() if self.caller.key else "general")
+        full_text = f"MEMORY ({wing}/{room}): {text}\nWritten by {self.caller.key} via Evennia."
+
+        ok = _add_drawer(full_text, wing, room, f"evennia:{self.caller.key}")
+        if ok:
+            self.caller.location.msg_contents(
+                f"$You() etch() a memory into the |c{room}|n room of the palace.",
+                from_obj=self.caller,
+            )
+        else:
+            self.caller.msg("The palace scribes could not write that down.")

docs/bezalel/evennia/cmd_steward.py

@@ -0,0 +1,105 @@
+"""
+Steward commands — ask a palace steward about memories.
+"""
+
+from evennia.commands.command import Command
+from evennia import search_object
+
+
+class CmdAskSteward(Command):
+    """
+    Ask a steward NPC about a topic from the palace memory.
+
+    Usage:
+      ask <steward> about <topic>
+      ask <steward> about <topic> --fleet
+
+    Example:
+      ask bezalel-steward about nightly watch
+      ask bezalel-steward about runner outage --fleet
+    """
+
+    key = "ask"
+    aliases = ["question"]
+    locks = "cmd:all()"
+    help_category = "Mind Palace"
+
+    def parse(self):
+        """Parse 'ask <target> about <topic>' syntax."""
+        raw = self.args.strip()
+        fleet = "--fleet" in raw
+        if fleet:
+            raw = raw.replace("--fleet", "").strip()
+
+        if " about " in raw.lower():
+            parts = raw.split(" about ", 1)
+            self.target_name = parts[0].strip()
+            self.topic = parts[1].strip()
+        else:
+            self.target_name = ""
+            self.topic = raw
+        self.fleet = fleet
+
+    def func(self):
+        if not self.args.strip():
+            self.caller.msg("Usage: ask <steward> about <topic> [--fleet]")
+            return
+
+        self.parse()
+
+        if not self.target_name:
+            self.caller.msg("Ask whom? Usage: ask <steward> about <topic>")
+            return
+
+        # Find steward NPC in current room
+        stewards = [
+            obj for obj in self.caller.location.contents
+            if hasattr(obj, "respond_to_question")
+            and self.target_name.lower() in obj.key.lower()
+        ]
+
+        if not stewards:
+            self.caller.msg(f"There is no steward here matching '{self.target_name}'.")
+            return
+
+        steward = stewards[0]
+        self.caller.msg(f"You ask |c{steward.key}|n about '{self.topic}'...")
+        steward.respond_to_question(self.topic, self.caller, fleet=self.fleet)
+
+
+class CmdSummonSteward(Command):
+    """
+    Summon your wing's steward NPC to your current location.
+
+    Usage:
+      summon steward
+    """
+
+    key = "summon steward"
+    locks = "cmd:all()"
+    help_category = "Mind Palace"
+
+    def func(self):
+        wing = self.caller.db.wing if self.caller.attributes.has("wing") else (self.caller.key.lower() if self.caller.key else "general")
+        steward_key = f"{wing}-steward"
+
+        # Search for existing steward
+        from typeclasses.steward_npc import StewardNPC
+        stewards = search_object(steward_key, typeclass="typeclasses.steward_npc.StewardNPC")
+
+        if stewards:
+            steward = stewards[0]
+            steward.move_to(self.caller.location, move_type="teleport")
+            self.caller.location.msg_contents(
+                f"A shimmer of light coalesces into |c{steward.key}|n.",
+                from_obj=self.caller,
+            )
+        else:
+            steward = StewardNPC.create(steward_key)[0]
+            steward.db.wing = wing
+            steward.db.steward_name = self.caller.key
+            steward.move_to(self.caller.location, move_type="teleport")
+            self.caller.location.msg_contents(
+                f"You call forth |c{steward.key}|n from the palace archives.",
+                from_obj=self.caller,
+            )

docs/bezalel/evennia/hall_of_wings.py

@@ -0,0 +1,83 @@
+"""
+Hall of Wings — Builds the central MemPalace zone in Evennia.
+
+Usage (from Evennia shell or script):
+    from world.hall_of_wings import build_hall_of_wings
+    build_hall_of_wings()
+"""
+
+from evennia import create_object
+from typeclasses.palace_room import PalaceRoom
+from typeclasses.steward_npc import StewardNPC
+from typeclasses.rooms import Room
+from typeclasses.exits import Exit
+
+HALL_KEY = "hall_of_wings"
+HALL_NAME = "Hall of Wings"
+
+DEFAULT_WINGS = [
+    "bezalel",
+    "timmy",
+    "allegro",
+    "ezra",
+]
+
+
+def build_hall_of_wings():
+    """Create or update the central Hall of Wings and attach steward chambers."""
+    # Find or create the hall
+    from evennia import search_object
+    halls = search_object(HALL_KEY, typeclass="typeclasses.rooms.Room")
+    if halls:
+        hall = halls[0]
+    else:
+        hall = create_object(Room, key=HALL_KEY)
+        hall.db.desc = (
+            "|cThe Hall of Wings|n\n"
+            "A vast circular chamber of pale stone and shifting starlight.\n"
+            "Arched doorways line the perimeter, each leading to a steward's chamber.\n"
+            "Here, the memories of the fleet converge.\n\n"
+            "Use |wsummon steward|n to call your wing's steward, or\n"
+            "|wask <steward> about <topic>|n to query the palace archives."
+        )
+
+    for wing in DEFAULT_WINGS:
+        chamber_key = f"chamber:{wing}"
+        chambers = search_object(chamber_key, typeclass="typeclasses.palace_room.PalaceRoom")
+        if chambers:
+            chamber = chambers[0]
+        else:
+            chamber = create_object(PalaceRoom, key=chamber_key)
+            chamber.db.memory_topic = wing
+            chamber.db.wing = wing
+            chamber.db.desc = (
+                f"|cThe Chamber of {wing.title()}|n\n"
+                f"This room holds the accumulated memories of the {wing} wing.\n"
+                f"A steward stands ready to answer questions."
+            )
+            chamber.update_description()
+
+        # Link hall <-> chamber with exits
+        exit_name = f"{wing}-chamber"
+        existing_exits = [ex for ex in hall.exits if ex.key == exit_name]
+        if not existing_exits:
+            create_object(Exit, key=exit_name, location=hall, destination=chamber)
+
+        return_exits = [ex for ex in chamber.exits if ex.key == "hall"]
+        if not return_exits:
+            create_object(Exit, key="hall", location=chamber, destination=hall)
+
+        # Place or summon steward
+        steward_key = f"{wing}-steward"
+        stewards = search_object(steward_key, typeclass="typeclasses.steward_npc.StewardNPC")
+        if stewards:
+            steward = stewards[0]
+            if steward.location != chamber:
+                steward.move_to(chamber, move_type="teleport")
+        else:
+            steward = create_object(StewardNPC, key=steward_key)
+            steward.db.wing = wing
+            steward.db.steward_name = wing.title()
+            steward.move_to(chamber, move_type="teleport")
+
+    return hall

docs/bezalel/evennia/palace_room.py

@@ -0,0 +1,87 @@
+"""
+PalaceRoom
+
+A Room that represents a topic in the memory palace.
+Memory objects spawned here embody concepts retrieved from mempalace.
+Its description auto-populates from a palace search on the memory topic.
+"""
+
+import json
+import subprocess
+from evennia.objects.objects import DefaultRoom
+from .objects import ObjectParent
+
+PALACE_SCRIPT = "/root/wizards/bezalel/evennia/palace_search.py"
+
+
+class PalaceRoom(ObjectParent, DefaultRoom):
+    """
+    A room in the mind palace. Its db.memory_topic describes what
+    kind of memories are stored here. The description is populated
+    from a live MemPalace search.
+    """
+
+    def at_object_creation(self):
+        super().at_object_creation()
+        self.db.memory_topic = ""
+        self.db.wing = "bezalel"
+        self.db.desc = (
+            f"This is the |c{self.key}|n room of your mind palace.\n"
+            "Memories and concepts drift here like motes of light.\n"
+            "Use |wpalace/search <query>|n or |wrecall <topic>|n to summon memories."
+        )
+
+    def _search_palace(self, query, wing=None, room=None, n=3):
+        """Call the helper script and return parsed results."""
+        cmd = ["/root/wizards/bezalel/hermes/venv/bin/python", PALACE_SCRIPT, query]
+        cmd.append(wing or "none")
+        cmd.append(room or "none")
+        cmd.append(str(n))
+        try:
+            result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
+            data = json.loads(result.stdout)
+            return data.get("results", [])
+        except Exception:
+            return []
+
+    def update_description(self):
+        """Refresh the room description from a palace search on its topic."""
+        topic = self.db.memory_topic or self.key.split(":")[-1] if ":" in self.key else self.key
+        wing = self.db.wing or "bezalel"
+        results = self._search_palace(topic, wing=wing, n=3)
+
+        header = (
+            f"=|c {topic.upper()} |n="
+        )
+        desc_lines = [
+            header,
+            f"You stand in the |c{topic}|n room of the |y{wing}|n wing.",
+            "Memories drift here like motes of light.",
+            "",
+        ]
+
+        if results:
+            desc_lines.append("|gNearby memories:|n")
+            for i, r in enumerate(results, 1):
+                content = r.get("content", "")[:200]
+                source = r.get("source", "unknown")
+                room_name = r.get("room", "unknown")
+                desc_lines.append(f"  |m[{i}]|n |c{room_name}|n — {content}... |x({source})|n")
+        else:
+            desc_lines.append("|xThe palace is quiet here. No memories resonate with this topic yet.|n")
+
+        desc_lines.append("")
+        desc_lines.append("Use |wrecall <query>|n to search deeper, or |wpalace/search <query>|n.")
+        self.db.desc = "\n".join(desc_lines)
+
+    def at_object_receive(self, moved_obj, source_location, **kwargs):
+        """Refresh description when someone enters."""
+        if moved_obj.has_account:
+            self.update_description()
+        super().at_object_receive(moved_obj, source_location, **kwargs)
+
+    def return_appearance(self, looker):
+        text = super().return_appearance(looker)
+        if self.db.memory_topic:
+            text += f"\n|xTopic: {self.db.memory_topic}|n"
+        return text

docs/bezalel/evennia/steward_npc.py

@@ -0,0 +1,70 @@
+"""
+StewardNPC
+
+A palace steward NPC that answers questions by querying the local
+or fleet MemPalace backend. One steward per wizard wing.
+"""
+
+import json
+import subprocess
+from evennia.objects.objects import DefaultCharacter
+from typeclasses.objects import ObjectParent
+
+PALACE_SCRIPT = "/root/wizards/bezalel/evennia/palace_search.py"
+
+
+class StewardNPC(ObjectParent, DefaultCharacter):
+    """
+    A steward of the mind palace. Ask it about memories,
+    decisions, or events from its wing.
+    """
+
+    def at_object_creation(self):
+        super().at_object_creation()
+        self.db.wing = "bezalel"
+        self.db.steward_name = "Bezalel"
+        self.db.desc = (
+            f"|c{self.key}|n stands here quietly, eyes like polished steel, "
+            "waiting to recall anything from the palace archives."
+        )
+        self.locks.add("get:false();delete:perm(Admin)")
+
+    def _search_palace(self, query, fleet=False, n=3):
+        cmd = [
+            "/root/wizards/bezalel/hermes/venv/bin/python",
+            PALACE_SCRIPT,
+            query,
+            "none" if fleet else self.db.wing,
+            "none",
+            str(n),
+        ]
+        if fleet:
+            cmd.append("--fleet")
+        try:
+            result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
+            data = json.loads(result.stdout)
+            return data.get("results", [])
+        except Exception:
+            return []
+
+    def _summarize_for_speech(self, results, query):
+        """Convert search results into in-character dialogue."""
+        if not results:
+            return "I find no memory of that in the palace."
+
+        lines = [f"Regarding '{query}':"]
+        for r in results:
+            room = r.get("room", "unknown")
+            content = r.get("content", "")[:300]
+            source = r.get("source", "unknown")
+            lines.append(f"  From the |c{room}|n room: {content}... |x[{source}]|n")
+        return "\n".join(lines)
+
+    def respond_to_question(self, question, asker, fleet=False):
+        results = self._search_palace(question, fleet=fleet, n=3)
+        speech = self._summarize_for_speech(results, question)
+        self.location.msg_contents(
+            f"|c{self.key}|n says to $you(asker): \"{speech}\"",
+            mapping={"asker": asker},
+            from_obj=self,
+        )

scripts/audit_mempalace_privacy.py

@@ -0,0 +1,95 @@
+#!/usr/bin/env python3
+"""
+Audit the fleet shared palace for privacy violations.
+Ensures no raw drawers, full source paths, or private workspace leaks exist.
+
+Usage:
+    python audit_mempalace_privacy.py /path/to/fleet/palace
+
+Exit codes:
+    0 = clean
+    1 = violations found
+"""
+
+import sys
+from pathlib import Path
+
+try:
+    import chromadb
+except ImportError:
+    print("ERROR: chromadb not installed")
+    sys.exit(1)
+
+VIOLATION_KEYWORDS = [
+    "/root/wizards/",
+    "/home/",
+    "/Users/",
+    "private_key",
+    "-----BEGIN",
+    "GITEA_TOKEN=",
+    "OPENAI_API_KEY",
+    "ANTHROPIC_API_KEY",
+]
+
+
+def audit(palace_path: Path):
+    violations = []
+    client = chromadb.PersistentClient(path=str(palace_path))
+    try:
+        col = client.get_collection("mempalace_drawers")
+    except Exception as e:
+        print(f"ERROR: Could not open collection: {e}")
+        sys.exit(1)
+
+    all_data = col.get(include=["documents", "metadatas"])
+    docs = all_data["documents"]
+    metas = all_data["metadatas"]
+
+    for doc, meta in zip(docs, metas):
+        source = meta.get("source_file", "")
+        doc_type = meta.get("type", "")
+
+        # Rule 1: Fleet palace should only contain closets or explicitly typed entries
+        if doc_type not in ("closet", "summary", "fleet"):
+            violations.append(
+                f"VIOLATION: Document type is '{doc_type}' (expected closet/summary/fleet). "
+                f"Source: {source}"
+            )
+
+        # Rule 2: No full absolute paths from private workspaces
+        if any(abs_path in source for abs_path in ["/root/wizards/", "/home/", "/Users/"]):
+            violations.append(
+                f"VIOLATION: Source contains absolute path: {source}"
+            )
+
+        # Rule 3: No raw secrets in document text
+        for kw in VIOLATION_KEYWORDS:
+            if kw in doc:
+                violations.append(
+                    f"VIOLATION: Document contains sensitive keyword '{kw}'. Source: {source}"
+                )
+                break  # one violation per doc is enough
+
+    return violations
+
+
+def main():
+    import argparse
+    parser = argparse.ArgumentParser(description="Audit fleet palace privacy")
+    parser.add_argument("palace", default="/var/lib/mempalace/fleet", nargs="?", help="Path to fleet palace")
+    args = parser.parse_args()
+
+    violations = audit(Path(args.palace))
+
+    if violations:
+        print(f"FAIL: {len(violations)} privacy violation(s) found")
+        for v in violations:
+            print(f"  {v}")
+        sys.exit(1)
+    else:
+        print("PASS: No privacy violations detected")
+        sys.exit(0)
+
+
+if __name__ == "__main__":
+    main()

scripts/backup_databases.sh

@@ -0,0 +1,50 @@
+#!/usr/bin/env bash
+# Bezalel Database Backup — MemPalace + Evennia + Fleet
+# Runs nightly after re-mine completes. Keeps 7 days of rolling backups.
+set -euo pipefail
+
+BACKUP_BASE="/root/wizards/bezalel/home/backups"
+DATE=$(date +%Y%m%d_%H%M%S)
+LOG="/var/log/bezalel_db_backup.log"
+
+# Sources
+LOCAL_PALACE="/root/wizards/bezalel/.mempalace/palace"
+FLEET_PALACE="/var/lib/mempalace/fleet"
+EVENNIA_DB="/root/wizards/bezalel/evennia/bezalel_world/server/evennia.db3"
+
+# Destinations
+LOCAL_BACKUP="${BACKUP_BASE}/mempalace/mempalace_${DATE}.tar.gz"
+FLEET_BACKUP="${BACKUP_BASE}/fleet/fleet_${DATE}.tar.gz"
+EVENNIA_BACKUP="${BACKUP_BASE}/evennia/evennia_${DATE}.db3.gz"
+
+log() {
+    echo "[$(date -Iseconds)] $1" | tee -a "$LOG"
+}
+
+log "Starting database backup cycle..."
+
+# 1. Backup local MemPalace
+tar -czf "$LOCAL_BACKUP" -C "$(dirname "$LOCAL_PALACE")" "$(basename "$LOCAL_PALACE")"
+log "Local palace backed up: ${LOCAL_BACKUP} ($(du -h "$LOCAL_BACKUP" | cut -f1))"
+
+# 2. Backup fleet MemPalace
+tar -czf "$FLEET_BACKUP" -C "$(dirname "$FLEET_PALACE")" "$(basename "$FLEET_PALACE")"
+log "Fleet palace backed up: ${FLEET_BACKUP} ($(du -h "$FLEET_BACKUP" | cut -f1))"
+
+# 3. Backup Evennia DB (gzip for space)
+gzip -c "$EVENNIA_DB" > "$EVENNIA_BACKUP"
+log "Evennia DB backed up: ${EVENNIA_BACKUP} ($(du -h "$EVENNIA_BACKUP" | cut -f1))"
+
+# 4. Prune backups older than 7 days
+find "${BACKUP_BASE}/mempalace" -name 'mempalace_*.tar.gz' -mtime +7 -delete
+find "${BACKUP_BASE}/fleet" -name 'fleet_*.tar.gz' -mtime +7 -delete
+find "${BACKUP_BASE}/evennia" -name 'evennia_*.db3.gz' -mtime +7 -delete
+log "Pruned backups older than 7 days"
+
+# 5. Report counts
+MP_COUNT=$(find "${BACKUP_BASE}/mempalace" -name 'mempalace_*.tar.gz' | wc -l)
+FL_COUNT=$(find "${BACKUP_BASE}/fleet" -name 'fleet_*.tar.gz' | wc -l)
+EV_COUNT=$(find "${BACKUP_BASE}/evennia" -name 'evennia_*.db3.gz' | wc -l)
+log "Backup cycle complete. Retained: mempalace=${MP_COUNT}, fleet=${FL_COUNT}, evennia=${EV_COUNT}"
+
+touch /var/lib/bezalel/heartbeats/db_backup.last

scripts/ci_auto_revert.py

@@ -0,0 +1,135 @@
+#!/usr/bin/env python3
+"""
+CI Auto-Revert — Poka-yoke for broken merges.
+Monitors the main branch post-merge and auto-reverts via local git if CI fails.
+
+Usage:
+    python ci_auto_revert.py <repo_owner>/<repo_name>
+    python ci_auto_revert.py Timmy_Foundation/hermes-agent
+
+Recommended cron: */10 * * * *
+"""
+
+import os
+import sys
+import json
+import subprocess
+import tempfile
+from datetime import datetime, timedelta, timezone
+from urllib import request, error
+
+GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "")
+GITEA_URL = os.environ.get("GITEA_URL", "https://forge.alexanderwhitestone.com")
+REVERT_WINDOW_MINUTES = 10
+
+
+def api_call(method, path):
+    url = f"{GITEA_URL}/api/v1{path}"
+    headers = {"Authorization": f"token {GITEA_TOKEN}"}
+    req = request.Request(url, method=method, headers=headers)
+    try:
+        with request.urlopen(req, timeout=30) as resp:
+            return json.loads(resp.read().decode())
+    except error.HTTPError as e:
+        return {"error": e.read().decode(), "status": e.code}
+
+
+def get_recent_commits(owner, repo, since):
+    since_iso = since.strftime("%Y-%m-%dT%H:%M:%SZ")
+    return api_call("GET", f"/repos/{owner}/{repo}/commits?sha=main&since={since_iso}&limit=20")
+
+
+def get_commit_status(owner, repo, sha):
+    return api_call("GET", f"/repos/{owner}/{repo}/commits/{sha}/status")
+
+
+def revert_via_git(clone_url, sha, msg, owner, repo):
+    with tempfile.TemporaryDirectory() as tmpdir:
+        # Clone with token
+        auth_url = clone_url.replace("https://", f"https://bezalel:{GITEA_TOKEN}@")
+        subprocess.run(["git", "clone", "--depth", "10", auth_url, tmpdir], check=True, capture_output=True)
+
+        # Configure git
+        subprocess.run(["git", "-C", tmpdir, "config", "user.email", "bezalel@timmy.foundation"], check=True, capture_output=True)
+        subprocess.run(["git", "-C", tmpdir, "config", "user.name", "Bezalel"], check=True, capture_output=True)
+
+        # Revert the commit
+        revert_msg = f"[auto-revert] {msg}\n\nOriginal commit {sha} failed CI."
+        result = subprocess.run(
+            ["git", "-C", tmpdir, "revert", "--no-edit", "-m", revert_msg, sha],
+            capture_output=True,
+            text=True,
+        )
+        if result.returncode != 0:
+            return {"error": f"git revert failed: {result.stderr}"}
+
+        # Push
+        push_result = subprocess.run(
+            ["git", "-C", tmpdir, "push", "origin", "main"],
+            capture_output=True,
+            text=True,
+        )
+        if push_result.returncode != 0:
+            return {"error": f"git push failed: {push_result.stderr}"}
+
+        return {"ok": True, "reverted_sha": sha}
+
+
+def main():
+    if len(sys.argv) < 2:
+        print(f"Usage: {sys.argv[0]} <owner/repo>")
+        sys.exit(1)
+
+    repo_full = sys.argv[1]
+    owner, repo = repo_full.split("/", 1)
+
+    since = datetime.now(timezone.utc) - timedelta(minutes=REVERT_WINDOW_MINUTES + 5)
+    commits = get_recent_commits(owner, repo, since)
+
+    if not isinstance(commits, list):
+        print(f"ERROR fetching commits: {commits}")
+        sys.exit(1)
+
+    reverted = 0
+    for commit in commits:
+        sha = commit.get("sha", "")
+        msg = commit.get("commit", {}).get("message", "").split("\n")[0]
+        commit_time = commit.get("commit", {}).get("committer", {}).get("date", "")
+        if not commit_time:
+            continue
+
+        commit_dt = datetime.fromisoformat(commit_time.replace("Z", "+00:00"))
+        age_min = (datetime.now(timezone.utc) - commit_dt).total_seconds() / 60
+
+        if age_min > REVERT_WINDOW_MINUTES:
+            continue
+
+        status = get_commit_status(owner, repo, sha)
+        state = status.get("state", "")
+
+        if state == "failure":
+            print(f"ALERT: Commit {sha[:8]} '{msg}' failed CI ({age_min:.1f}m old). Initiating revert...")
+            repo_info = api_call("GET", f"/repos/{owner}/{repo}")
+            clone_url = repo_info.get("clone_url", "")
+            if not clone_url:
+                print(f"  Cannot find clone URL")
+                continue
+            result = revert_via_git(clone_url, sha, msg, owner, repo)
+            if "error" in result:
+                print(f"  Revert failed: {result['error']}")
+            else:
+                print(f"  Reverted successfully.")
+                reverted += 1
+        elif state == "success":
+            print(f"OK: Commit {sha[:8]} '{msg}' passed CI.")
+        elif state == "pending":
+            print(f"PENDING: Commit {sha[:8]} '{msg}' still running CI.")
+        else:
+            print(f"UNKNOWN: Commit {sha[:8]} '{msg}' has CI state '{state}'.")
+
+    if reverted == 0:
+        print("No broken merges found in the last 10 minutes.")
+
+
+if __name__ == "__main__":
+    main()

scripts/mempalace_export.py

@@ -0,0 +1,75 @@
+#!/usr/bin/env python3
+"""
+Export closets from a local MemPalace wing for fleet-wide sharing.
+
+Privacy rule: only summaries/closets are exported. No raw source_file paths.
+Source filenames are anonymized to just the basename.
+"""
+
+import json
+import sys
+from pathlib import Path
+import chromadb
+
+PALACE_PATH = "/root/wizards/bezalel/.mempalace/palace"
+FLEET_INCOMING = "/var/lib/mempalace/fleet/incoming"
+WING = "bezalel"
+DOCS_PER_ROOM = 5
+
+
+def main():
+    client = chromadb.PersistentClient(path=PALACE_PATH)
+    col = client.get_collection("mempalace_drawers")
+
+    # Discover rooms in this wing
+    all_meta = col.get(include=["metadatas"])["metadatas"]
+    rooms = set()
+    for m in all_meta:
+        if m.get("wing") == WING:
+            rooms.add(m.get("room", "general"))
+
+    Path(FLEET_INCOMING).mkdir(parents=True, exist_ok=True)
+
+    closets = []
+    for room in sorted(rooms):
+        results = col.query(
+            query_texts=[room],
+            n_results=DOCS_PER_ROOM,
+            where={"$and": [{"wing": WING}, {"room": room}]},
+            include=["documents", "metadatas"],
+        )
+        docs = results["documents"][0]
+        metas = results["metadatas"][0]
+
+        entries = []
+        for doc, meta in zip(docs, metas):
+            # Sanitize content: strip absolute workspace paths
+            sanitized = doc[:800]
+            sanitized = sanitized.replace("/root/wizards/bezalel/", "~/")
+            sanitized = sanitized.replace("/root/wizards/", "~/")
+            sanitized = sanitized.replace("/home/bezalel/", "~/")
+            sanitized = sanitized.replace("/home/", "~/")
+            entries.append({
+                "content": sanitized,
+                "source_basename": Path(meta.get("source_file", "?")).name,
+            })
+
+        closet = {
+            "wing": WING,
+            "room": room,
+            "type": "closet",
+            "entries": entries,
+        }
+        closets.append(closet)
+
+    out_file = Path(FLEET_INCOMING) / f"{WING}_closets.json"
+    with open(out_file, "w") as f:
+        json.dump(closets, f, indent=2)
+
+    print(f"Exported {len(closets)} closets to {out_file}")
+    for c in closets:
+        print(f"  {c['wing']} / {c['room']} : {len(c['entries'])} entries")
+
+
+if __name__ == "__main__":
+    main()

scripts/mempalace_nightly.sh

@@ -0,0 +1,24 @@
+#!/usr/bin/env bash
+# Bezalel MemPalace Nightly Re-mine + Fleet Sync
+set -euo pipefail
+
+PALACE="/root/wizards/bezalel/.mempalace/palace"
+MINER="/root/wizards/bezalel/hermes/venv/bin/mempalace"
+WING_DIR="/root/wizards/bezalel"
+LOG="/var/log/bezalel_mempalace.log"
+EXPORTER="/root/wizards/bezalel/hermes/venv/bin/python /root/wizards/bezalel/mempalace_export.py"
+IMPORTER="/root/wizards/bezalel/hermes/venv/bin/python /var/lib/mempalace/fleet_import.py"
+
+echo "[$(date -Iseconds)] Starting mempalace re-mine" >> "$LOG"
+cd "$WING_DIR"
+"$MINER" --palace "$PALACE" mine "$WING_DIR" --agent bezalel >> "$LOG" 2>&1 || true
+echo "[$(date -Iseconds)] Finished mempalace re-mine" >> "$LOG"
+"$MINER" --palace "$PALACE" status >> "$LOG" 2>&1 || true
+
+echo "[$(date -Iseconds)] Starting fleet closet export" >> "$LOG"
+$EXPORTER >> "$LOG" 2>&1 || true
+echo "[$(date -Iseconds)] Starting fleet closet import" >> "$LOG"
+$IMPORTER >> "$LOG" 2>&1 || true
+echo "[$(date -Iseconds)] Fleet sync complete" >> "$LOG"
+
+touch /var/lib/bezalel/heartbeats/mempalace_nightly.last

scripts/meta_heartbeat.sh

@@ -0,0 +1,53 @@
+#!/usr/bin/env bash
+# Meta-heartbeat — checks all Bezalel cron jobs for stale timestamps
+set -euo pipefail
+
+HEARTBEAT_DIR="/var/lib/bezalel/heartbeats"
+ALERT_LOG="/var/log/bezalel_meta_heartbeat.log"
+STALE_MINUTES=30
+
+log() {
+    echo "[$(date -Iseconds)] $1" | tee -a "$ALERT_LOG"
+}
+
+mkdir -p "$HEARTBEAT_DIR"
+
+# Define expected heartbeats: name => max_stale_minutes
+HEARTBEATS=(
+    "nightly_watch:150"      # 2.5h — runs at 02:00
+    "mempalace_nightly:150"  # 2.5h — runs at 03:00
+    "db_backup:150"          # 2.5h — runs at 03:30
+    "runner_health:15"       # 15m — every 5 min
+)
+
+NOW_EPOCH=$(date +%s)
+FAILURES=0
+
+for entry in "${HEARTBEATS[@]}"; do
+    name="${entry%%:*}"
+    max_minutes="${entry##*:}"
+    file="${HEARTBEAT_DIR}/${name}.last"
+
+    if [[ ! -f "$file" ]]; then
+        log "MISSING: $name heartbeat file not found ($file)"
+        FAILURES=$((FAILURES + 1))
+        continue
+    fi
+
+    LAST_EPOCH=$(stat -c %Y "$file")
+    AGE_MIN=$(( (NOW_EPOCH - LAST_EPOCH) / 60 ))
+
+    if [[ $AGE_MIN -gt $max_minutes ]]; then
+        log "STALE: $name is ${AGE_MIN}m old (max ${max_minutes}m)"
+        FAILURES=$((FAILURES + 1))
+    else
+        log "OK: $name is ${AGE_MIN}m old"
+    fi
+done
+
+if [[ $FAILURES -gt 0 ]]; then
+    log "ALERT: $FAILURES stale/missing heartbeat(s) detected."
+    exit 1
+else
+    log "ALL_OK: All heartbeats healthy."
+fi

scripts/runner_health_probe.sh

@@ -0,0 +1,46 @@
+#!/usr/bin/env bash
+# Gitea Runner Health Probe — Poka-yoke for unregistered runners
+set -euo pipefail
+
+GITEA_TOKEN="${GITEA_TOKEN:-}"
+GITEA_URL="https://forge.alexanderwhitestone.com"
+ALERT_LOG="/var/log/bezalel_runner_health.log"
+
+log() {
+    echo "[$(date -Iseconds)] $1" | tee -a "$ALERT_LOG"
+}
+
+if [[ -z "$GITEA_TOKEN" ]]; then
+    log "ERROR: GITEA_TOKEN not set"
+    exit 1
+fi
+
+ACTIVE_RUNNERS=$(curl -s -H "Authorization: token ${GITEA_TOKEN}" \
+    "${GITEA_URL}/api/v1/repos/Timmy_Foundation/hermes-agent/actions/runners" | \
+    python3 -c "import sys,json; d=json.load(sys.stdin); print(len([r for r in d.get('runners',[]) if r.get('status')=='online']))")
+
+log "Active runners: ${ACTIVE_RUNNERS}"
+
+if [[ "$ACTIVE_RUNNERS" -eq 0 ]]; then
+    log "CRITICAL: Zero active runners detected. Attempting self-healing restart."
+    pkill -f "act_runner daemon" 2>/dev/null || true
+    sleep 2
+    cd /opt/gitea-runner && nohup ./act_runner daemon > /var/log/gitea-runner.log 2>&1 &
+    sleep 3
+    # Re-check
+    ACTIVE_RUNNERS_AFTER=$(curl -s -H "Authorization: token ${GITEA_TOKEN}" \
+        "${GITEA_URL}/api/v1/repos/Timmy_Foundation/hermes-agent/actions/runners" | \
+        python3 -c "import sys,json; d=json.load(sys.stdin); print(len([r for r in d.get('runners',[]) if r.get('status')=='online']))")
+    log "Active runners after restart: ${ACTIVE_RUNNERS_AFTER}"
+    if [[ "$ACTIVE_RUNNERS_AFTER" -eq 0 ]]; then
+        log "CRITICAL: Self-healing failed. Runner still offline."
+        touch /var/lib/bezalel/heartbeats/runner_health.last
+        exit 1
+    else
+        log "RECOVERED: Runner back online."
+    fi
+else
+    log "OK: ${ACTIVE_RUNNERS} runner(s) online."
+fi
+
+touch /var/lib/bezalel/heartbeats/runner_health.last

scripts/secret_guard.sh

@@ -0,0 +1,50 @@
+#!/usr/bin/env bash
+# Secret Guard — Poka-yoke for world-readable credentials
+set -euo pipefail
+
+ALERT_LOG="/var/log/bezalel_secret_guard.log"
+QUARANTINE_DIR="/root/wizards/bezalel/home/quarantine"
+
+mkdir -p "$QUARANTINE_DIR"
+
+log() {
+    echo "[$(date -Iseconds)] $1" | tee -a "$ALERT_LOG"
+}
+
+# Scan for world-readable files with sensitive keywords in /root, /home, /etc, /tmp, /var/log
+# Exclude binary files, large files (>1MB), and known safe paths
+BAD_FILES=$(find /root /home /etc /tmp /var/log -maxdepth 4 -type f -perm /o+r 2>/dev/null \
+    ! -path "*/.git/*" \
+    ! -path "*/node_modules/*" \
+    ! -path "*/venv/*" \
+    ! -path "*/.venv/*" \
+    ! -path "*/__pycache__/*" \
+    ! -path "*/.pyc" \
+    ! -size +1M \
+    -exec grep -l -i -E 'password|token|secret|nsec|api_key|private_key|aws_access_key_id|aws_secret_access_key' {} + 2>/dev/null | head -50)
+
+VIOLATIONS=0
+for file in $BAD_FILES; do
+    # Skip if already quarantined
+    if [[ "$file" == "$QUARANTINE_DIR"* ]]; then
+        continue
+    fi
+    # Skip log files that are expected to be world-readable
+    if [[ "$file" == /var/log/* ]]; then
+        continue
+    fi
+
+    VIOLATIONS=$((VIOLATIONS + 1))
+    basename=$(basename "$file")
+    quarantine_path="${QUARANTINE_DIR}/${basename}.$(date +%s)"
+    cp "$file" "$quarantine_path"
+    chmod 600 "$quarantine_path"
+    chmod 600 "$file"
+    log "QUARANTINED: $file -> $quarantine_path (permissions fixed to 600)"
+done
+
+if [[ $VIOLATIONS -gt 0 ]]; then
+    log "ALERT: $VIOLATIONS world-readable secret file(s) detected and quarantined."
+else
+    log "OK: No world-readable secret files found."
+fi

scripts/sync_fleet_to_alpha.sh

@@ -0,0 +1,30 @@
+#!/usr/bin/env bash
+# Sync Fleet MemPalace from Beta to Alpha
+# Usage: ./sync_fleet_to_alpha.sh
+set -euo pipefail
+
+FLEET_DIR="/var/lib/mempalace/fleet"
+ALPHA_HOST="167.99.126.228"
+ALPHA_USER="root"
+ALPHA_DEST="/var/lib/mempalace/fleet"
+LOG="/var/log/bezalel_alpha_sync.log"
+
+log() {
+    echo "[$(date -Iseconds)] $1" | tee -a "$LOG"
+}
+
+log "Starting fleet palace sync to Alpha (${ALPHA_HOST})..."
+
+# Ensure Alpha destination exists (SSH must be configured key-based or agent-forwarded)
+ssh -o ConnectTimeout=10 "${ALPHA_USER}@${ALPHA_HOST}" "mkdir -p ${ALPHA_DEST}" || {
+    log "ERROR: Cannot reach Alpha host. Aborting."
+    exit 1
+}
+
+# rsync the fleet palace directory (ChromaDB files + incoming closets)
+rsync -avz --delete \
+    -e "ssh -o ConnectTimeout=10" \
+    "${FLEET_DIR}/" \
+    "${ALPHA_USER}@${ALPHA_HOST}:${ALPHA_DEST}/" >> "$LOG" 2>&1
+
+log "Fleet palace sync complete."

scripts/validate_mempalace_taxonomy.py

@@ -0,0 +1,123 @@
+#!/usr/bin/env python3
+"""
+Validate a wizard's mempalace.yaml against the fleet taxonomy standard.
+
+Usage:
+    python validate_mempalace_taxonomy.py /path/to/mempalace.yaml
+    python validate_mempalace_taxonomy.py --ci /path/to/mempalace.yaml
+
+Exit codes:
+    0 = valid
+    1 = missing required rooms or other violations
+"""
+
+import sys
+from pathlib import Path
+
+try:
+    import yaml
+except ImportError:
+    print("ERROR: PyYAML not installed. Run: pip install pyyaml")
+    sys.exit(1)
+
+
+REQUIRED_ROOMS = {
+    "forge",
+    "hermes",
+    "nexus",
+    "issues",
+    "experiments",
+}
+
+
+def load_standard():
+    # Try to find the fleet standard in the-nexus clone or local path
+    candidates = [
+        Path(__file__).parent.parent / "mempalace_taxonomy.yaml",
+        Path("/tmp/nexus_clone/docs/mempalace_taxonomy.yaml"),
+        Path(__file__).parent.parent.parent / "the-nexus" / "docs" / "mempalace_taxonomy.yaml",
+    ]
+    for c in candidates:
+        if c.exists():
+            with open(c) as f:
+                return yaml.safe_load(f)
+    return None
+
+
+def validate(path: Path):
+    errors = []
+    warnings = []
+
+    if not path.exists():
+        errors.append(f"File not found: {path}")
+        return errors, warnings
+
+    with open(path) as f:
+        data = yaml.safe_load(f)
+
+    if not data:
+        errors.append("Empty or invalid YAML")
+        return errors, warnings
+
+    rooms = data.get("rooms", data.get("wings", {}).get("bezalel", {}).get("rooms", []))
+    if isinstance(rooms, list) and rooms and isinstance(rooms[0], dict):
+        room_names = {r.get("name") for r in rooms if isinstance(r, dict)}
+    elif isinstance(rooms, dict):
+        room_names = set(rooms.keys())
+    else:
+        room_names = set()
+
+    missing = REQUIRED_ROOMS - room_names
+    if missing:
+        errors.append(f"Missing required rooms: {', '.join(sorted(missing))}")
+
+    # Check for duplicate room names
+    if len(room_names) < len(list(rooms) if isinstance(rooms, list) else rooms):
+        errors.append("Duplicate room names detected")
+
+    # Check for empty keywords
+    if isinstance(rooms, list):
+        for r in rooms:
+            if isinstance(r, dict):
+                kw = r.get("keywords", [])
+                if not kw:
+                    warnings.append(f"Room '{r.get('name')}' has no keywords")
+
+    standard = load_standard()
+    if standard:
+        std_optional = set(standard.get("optional_rooms", {}).keys())
+        unknown = room_names - REQUIRED_ROOMS - std_optional
+        if unknown:
+            warnings.append(f"Non-standard rooms (OK but not in fleet spec): {', '.join(sorted(unknown))}")
+
+    return errors, warnings
+
+
+def main():
+    import argparse
+    parser = argparse.ArgumentParser(description="Validate MemPalace taxonomy")
+    parser.add_argument("config", help="Path to mempalace.yaml")
+    parser.add_argument("--ci", action="store_true", help="CI mode: fail on warnings too")
+    args = parser.parse_args()
+
+    errors, warnings = validate(Path(args.config))
+
+    if warnings:
+        for w in warnings:
+            print(f"WARNING: {w}")
+
+    if errors:
+        for e in errors:
+            print(f"ERROR: {e}")
+        sys.exit(1)
+
+    if args.ci and warnings:
+        print("Validation failed in CI mode (warnings treated as errors)")
+        sys.exit(1)
+
+    print("OK: Taxonomy validation passed")
+    sys.exit(0)
+
+
+if __name__ == "__main__":
+    main()