diff --git a/docs/bezalel/evennia/cmd_palace.py b/docs/bezalel/evennia/cmd_palace.py new file mode 100644 index 0000000..1afcaff --- /dev/null +++ b/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 + palace/search [--room ] + palace/recall + palace/file = + 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 | palace/recall | palace/file = | 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 = ") + 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 + recall --fleet + recall --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 [--fleet] [--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 + """ + + 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 ") + 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.") diff --git a/docs/bezalel/evennia/cmd_record.py b/docs/bezalel/evennia/cmd_record.py new file mode 100644 index 0000000..9a12b23 --- /dev/null +++ b/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 + 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 ") + 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 + 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 ") + 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 + 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 ") + 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 = + """ + + key = "palace/write" + locks = "cmd:all()" + help_category = "Mind Palace" + + def func(self): + if "=" not in self.args: + self.caller.msg("Usage: palace/write = ") + 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.") diff --git a/docs/bezalel/evennia/cmd_steward.py b/docs/bezalel/evennia/cmd_steward.py new file mode 100644 index 0000000..6886e7a --- /dev/null +++ b/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 about + ask about --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 about ' 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 about [--fleet]") + return + + self.parse() + + if not self.target_name: + self.caller.msg("Ask whom? Usage: ask about ") + 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, + ) diff --git a/docs/bezalel/evennia/hall_of_wings.py b/docs/bezalel/evennia/hall_of_wings.py new file mode 100644 index 0000000..2a45ee9 --- /dev/null +++ b/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 about |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 diff --git a/docs/bezalel/evennia/palace_room.py b/docs/bezalel/evennia/palace_room.py new file mode 100644 index 0000000..edfbb7c --- /dev/null +++ b/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 |n or |wrecall |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 |n to search deeper, or |wpalace/search |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 diff --git a/docs/bezalel/evennia/steward_npc.py b/docs/bezalel/evennia/steward_npc.py new file mode 100644 index 0000000..4b29b2c --- /dev/null +++ b/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, + ) diff --git a/scripts/audit_mempalace_privacy.py b/scripts/audit_mempalace_privacy.py new file mode 100644 index 0000000..5b1b61d --- /dev/null +++ b/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() diff --git a/scripts/backup_databases.sh b/scripts/backup_databases.sh new file mode 100755 index 0000000..e15cd30 --- /dev/null +++ b/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 diff --git a/scripts/ci_auto_revert.py b/scripts/ci_auto_revert.py new file mode 100644 index 0000000..bb74e49 --- /dev/null +++ b/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 / + 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]} ") + 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() diff --git a/scripts/mempalace_export.py b/scripts/mempalace_export.py new file mode 100644 index 0000000..3588ec6 --- /dev/null +++ b/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() diff --git a/scripts/mempalace_nightly.sh b/scripts/mempalace_nightly.sh new file mode 100755 index 0000000..6a509eb --- /dev/null +++ b/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 diff --git a/scripts/meta_heartbeat.sh b/scripts/meta_heartbeat.sh new file mode 100755 index 0000000..e61ec40 --- /dev/null +++ b/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 diff --git a/scripts/runner_health_probe.sh b/scripts/runner_health_probe.sh new file mode 100755 index 0000000..524d90c --- /dev/null +++ b/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 diff --git a/scripts/secret_guard.sh b/scripts/secret_guard.sh new file mode 100755 index 0000000..b9d56c8 --- /dev/null +++ b/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 diff --git a/scripts/sync_fleet_to_alpha.sh b/scripts/sync_fleet_to_alpha.sh new file mode 100644 index 0000000..d1472c6 --- /dev/null +++ b/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." diff --git a/scripts/validate_mempalace_taxonomy.py b/scripts/validate_mempalace_taxonomy.py new file mode 100644 index 0000000..33a7fad --- /dev/null +++ b/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()