feat: MemPalace fleet memory scaffold — taxonomy, Evennia plugin, privacy tools

Delivers milestone artifacts for #1075 (MemPalace × Evennia — Fleet Memory):

**#1082 — Palace taxonomy standard**
- `mempalace/rooms.yaml` — fleet-wide room vocabulary (5 core rooms: forge,
  hermes, nexus, issues, experiments) with optional domain rooms and tunnel
  routing table
- `mempalace/validate_rooms.py` — validates a wizard's mempalace.yaml against
  the standard; exits non-zero on missing core rooms (CI-ready)

**#1077 — Evennia plugin scaffold**
- `mempalace/evennia_mempalace/` — contrib module connecting Evennia to MemPalace
  - `CmdRecall` — `recall <query>` / `recall <query> --fleet` in-world commands
  - `CmdEnterRoom` — teleport to semantic room by topic
  - `MemPalaceRoom` — typeclass whose description auto-populates from palace search
  - `searcher.py` — thin subprocess wrapper around the mempalace binary
  - `settings.py` — MEMPALACE_PATH / MEMPALACE_WING configuration bridge

**#1083 — Privacy boundary tools**
- `mempalace/export_closets.sh` — closet-only export enforcing policy that raw
  drawers never leave the local VPS; aborts on violations
- `mempalace/audit_privacy.py` — weekly audit of fleet palace for raw drawers,
  full-text closets, and private source_file paths

**Tests:** 21 new tests covering validate_rooms and audit_privacy logic.

Refs #1075, #1077, #1082, #1083

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Alexander Whitestone
2026-04-07 10:09:10 -04:00
parent c1116b8bde
commit c05febf86f
18 changed files with 1281 additions and 0 deletions

5
mempalace/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
"""
mempalace — Fleet memory tools for the MemPalace × Evennia integration.
Refs: #1075 (MemPalace × Evennia — Fleet Memory milestone)
"""

Binary file not shown.

Binary file not shown.

Binary file not shown.

177
mempalace/audit_privacy.py Normal file
View File

@@ -0,0 +1,177 @@
#!/usr/bin/env python3
"""
audit_privacy.py — Weekly privacy audit for the shared fleet palace.
Scans a palace directory (typically the shared Alpha fleet palace) and
reports any files that violate the closet-only sync policy:
1. Raw drawer files (.drawer.json) — must never exist in fleet palace.
2. Closet files containing full-text content (> threshold characters).
3. Closet files exposing private source_file paths.
Exits 0 if clean, 1 if violations found.
Usage:
python mempalace/audit_privacy.py [fleet_palace_dir]
Default: /var/lib/mempalace/fleet
Refs: #1083, #1075
"""
from __future__ import annotations
import argparse
import json
import sys
from dataclasses import dataclass, field
from pathlib import Path
# Closets should be compressed summaries, not full text.
# Flag any text field exceeding this character count as suspicious.
MAX_CLOSET_TEXT_CHARS = 2000
# Private path indicators — if a source_file contains any of these,
# it is considered a private VPS path that should not be in the fleet palace.
PRIVATE_PATH_PREFIXES = [
"/root/",
"/home/",
"/Users/",
"/var/home/",
]
@dataclass
class Violation:
path: Path
rule: str
detail: str
@dataclass
class AuditResult:
scanned: int = 0
violations: list[Violation] = field(default_factory=list)
@property
def clean(self) -> bool:
return len(self.violations) == 0
def _is_private_path(path_str: str) -> bool:
for prefix in PRIVATE_PATH_PREFIXES:
if path_str.startswith(prefix):
return True
return False
def audit_file(path: Path) -> list[Violation]:
violations: list[Violation] = []
# Rule 1: raw drawer files must not exist in fleet palace
if path.name.endswith(".drawer.json"):
violations.append(Violation(
path=path,
rule="RAW_DRAWER",
detail="Raw drawer file present — only closets allowed in fleet palace.",
))
return violations # no further checks needed
if not path.name.endswith(".closet.json"):
return violations # not a palace file, skip
try:
data = json.loads(path.read_text())
except (json.JSONDecodeError, OSError) as exc:
violations.append(Violation(
path=path,
rule="PARSE_ERROR",
detail=f"Could not parse file: {exc}",
))
return violations
drawers = data.get("drawers", []) if isinstance(data, dict) else []
if not isinstance(drawers, list):
drawers = []
for i, drawer in enumerate(drawers):
if not isinstance(drawer, dict):
continue
# Rule 2: closets must not contain full-text content
text = drawer.get("text", "")
if len(text) > MAX_CLOSET_TEXT_CHARS:
violations.append(Violation(
path=path,
rule="FULL_TEXT_IN_CLOSET",
detail=(
f"Drawer [{i}] text is {len(text)} chars "
f"(limit {MAX_CLOSET_TEXT_CHARS}). "
"Closets must be compressed summaries, not raw content."
),
))
# Rule 3: private source_file paths must not appear in fleet data
source_file = drawer.get("source_file", "")
if source_file and _is_private_path(source_file):
violations.append(Violation(
path=path,
rule="PRIVATE_SOURCE_PATH",
detail=f"Drawer [{i}] exposes private source_file: {source_file!r}",
))
return violations
def audit_palace(palace_dir: Path) -> AuditResult:
result = AuditResult()
for f in sorted(palace_dir.rglob("*.json")):
violations = audit_file(f)
result.scanned += 1
result.violations.extend(violations)
return result
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
description="Audit the fleet palace for privacy violations."
)
parser.add_argument(
"palace_dir",
nargs="?",
default="/var/lib/mempalace/fleet",
help="Path to the fleet palace directory (default: /var/lib/mempalace/fleet)",
)
parser.add_argument(
"--max-text",
type=int,
default=MAX_CLOSET_TEXT_CHARS,
metavar="N",
help=f"Maximum closet text length (default: {MAX_CLOSET_TEXT_CHARS})",
)
args = parser.parse_args(argv)
palace_dir = Path(args.palace_dir)
if not palace_dir.exists():
print(f"[audit_privacy] ERROR: palace directory not found: {palace_dir}", file=sys.stderr)
return 2
print(f"[audit_privacy] Scanning: {palace_dir}")
result = audit_palace(palace_dir)
if result.clean:
print(f"[audit_privacy] OK — {result.scanned} file(s) scanned, no violations.")
return 0
print(
f"[audit_privacy] FAIL — {len(result.violations)} violation(s) in {result.scanned} file(s):",
file=sys.stderr,
)
for v in result.violations:
print(f" [{v.rule}] {v.path}", file=sys.stderr)
print(f" {v.detail}", file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(main())

View File

@@ -0,0 +1,22 @@
"""
evennia_mempalace — Evennia contrib module for MemPalace fleet memory.
Connects Evennia's spatial world to the local MemPalace vector backend,
enabling in-world recall, room exploration, and steward NPC queries.
Phase 1 deliverables (Issue #1077):
- CmdRecall — `recall <query>` / `recall <query> --fleet`
- CmdEnterRoom — `enter room <topic>` — teleport to semantic room
- MemPalaceRoom — typeclass whose description auto-populates from palace
Installation (add to your Evennia game's settings.py):
INSTALLED_APPS += ["evennia_mempalace"]
MEMPALACE_PATH = "/root/wizards/bezalel/.mempalace/palace"
MEMPALACE_FLEET_PATH = "/var/lib/mempalace/fleet" # optional shared wing
MEMPALACE_WING = "bezalel" # default wing name
Refs: #1077, #1075
"""
VERSION = "0.1.0"

View File

@@ -0,0 +1,14 @@
"""
evennia_mempalace.commands — In-world commands for MemPalace fleet memory.
Commands:
CmdRecall — recall <query> [--fleet]
CmdEnterRoom — enter room <topic>
Refs: #1077, #1075
"""
from .recall import CmdRecall
from .enter_room import CmdEnterRoom
__all__ = ["CmdRecall", "CmdEnterRoom"]

View File

@@ -0,0 +1,86 @@
"""
CmdEnterRoom — teleport to a MemPalace room inside Evennia.
Usage:
enter room <topic>
Searches for a MemPalaceRoom whose key matches <topic> in the current
location's area, then teleports the caller there and auto-renders the
room description from top-3 palace memories for that topic.
Examples:
enter room forge
enter room hermes
enter room experiments
Refs: #1077, #1075
"""
from __future__ import annotations
try:
from evennia import Command, search_object
from evennia.utils.utils import inherits_from
except ImportError:
Command = object # type: ignore[assignment,misc]
search_object = None # type: ignore[assignment]
def inherits_from(obj, cls): # type: ignore[misc]
return False
from ..settings import get_palace_path, get_wing
MEMPALACE_ROOM_TYPECLASS = "evennia_mempalace.typeclasses.room.MemPalaceRoom"
class CmdEnterRoom(Command):
"""
Teleport into a MemPalace room and see its knowledge.
Usage:
enter room <topic>
Finds a MemPalaceRoom matching <topic> and moves you there.
The room description is populated from the top-3 memories in that room.
Examples:
enter room forge
enter room nexus
enter room experiments
"""
key = "enter room"
aliases = ["go to room", "visit room"]
locks = "cmd:all()"
help_category = "MemPalace"
def func(self) -> None:
topic = self.args.strip().lower()
if not topic:
self.caller.msg("Usage: enter room <topic> (e.g. 'enter room forge')")
return
# Look for a MemPalaceRoom with matching key in the current location
# or globally tagged with the topic.
candidates = []
if search_object is not None:
candidates = search_object(
topic,
typeclass=MEMPALACE_ROOM_TYPECLASS,
attribute_name="palace_room_key",
attribute_value=topic,
)
if not candidates:
self.caller.msg(
f"No palace room found for topic '|w{topic}|n'. "
f"Available rooms depend on your local palace configuration."
)
return
destination = candidates[0]
if destination == self.caller.location:
self.caller.msg(f"You are already in the {destination.key}.")
return
self.caller.move_to(destination, quiet=False)

View File

@@ -0,0 +1,93 @@
"""
CmdRecall — search the caller's MemPalace wing from inside Evennia.
Usage:
recall <query>
recall <query> --fleet
Examples:
recall nightly watch failures
recall GraphQL schema --fleet
The plain form searches only the caller's own wing.
--fleet searches the shared fleet wing (closets only, privacy-safe).
Refs: #1077, #1075
"""
from __future__ import annotations
try:
from evennia import Command
from evennia.utils import evtable
except ImportError: # allow import outside Evennia for testing
Command = object # type: ignore[assignment,misc]
evtable = None # type: ignore[assignment]
from ..searcher import search_memories
from ..settings import get_palace_path, get_fleet_palace_path, get_wing
class CmdRecall(Command):
"""
Search your MemPalace wing for relevant memories.
Usage:
recall <query>
recall <query> --fleet
The plain form searches your own wing. Add --fleet to search the
shared fleet wing (closets only).
Examples:
recall nightly watch
recall hermes gateway --fleet
"""
key = "recall"
aliases = ["remember"]
locks = "cmd:all()"
help_category = "MemPalace"
def func(self) -> None:
raw = self.args.strip()
if not raw:
self.caller.msg("Usage: recall <query> (add --fleet to search all wings)")
return
fleet_mode = "--fleet" in raw
query = raw.replace("--fleet", "").strip()
if not query:
self.caller.msg("Usage: recall <query> (add --fleet to search all wings)")
return
if fleet_mode:
palace_path = get_fleet_palace_path()
wing = "fleet"
scope_label = "|y[fleet]|n"
else:
palace_path = get_palace_path()
wing = get_wing(self.caller)
scope_label = f"|c[{wing}]|n"
if not palace_path:
self.caller.msg(
"|rMemPalace is not configured. Set MEMPALACE_PATH in settings.py.|n"
)
return
self.caller.msg(f"Searching {scope_label} for: |w{query}|n …")
results = search_memories(query, palace_path, wing=wing, top_k=5)
if not results:
self.caller.msg("No memories found.")
return
lines = []
for i, r in enumerate(results, 1):
snippet = r.text[:200].replace("\n", " ")
if len(r.text) > 200:
snippet += ""
lines.append(f"|w{i}.|n |c{r.room}|n {snippet}")
self.caller.msg("\n".join(lines))

View File

@@ -0,0 +1,106 @@
"""
searcher.py — Thin wrapper around MemPalace search for Evennia commands.
Provides a single `search_memories` function that returns ranked results
from a local palace. Evennia commands call this; the MemPalace binary/lib
handles ChromaDB and SQLite under the hood.
Refs: #1077, #1075
"""
from __future__ import annotations
import json
import subprocess
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
@dataclass
class MemoryResult:
room: str
text: str
score: float
source: str = ""
metadata: dict = field(default_factory=dict)
def _find_mempalace_bin(palace_path: Path) -> Optional[Path]:
"""Resolve the mempalace binary via vendored path or system PATH."""
# Check vendored binary alongside palace
candidates = [
palace_path.parent.parent / ".vendor" / "mempalace" / "mempalace",
palace_path.parent.parent / ".vendor" / "mempalace" / "bin" / "mempalace",
]
for c in candidates:
if c.is_file() and c.stat().st_mode & 0o111:
return c
# Fallback: system PATH
import shutil
found = shutil.which("mempalace")
return Path(found) if found else None
def search_memories(
query: str,
palace_path: str | Path,
wing: str = "general",
room: Optional[str] = None,
top_k: int = 5,
) -> list[MemoryResult]:
"""Search a MemPalace for memories matching *query*.
Args:
query: Natural-language search string.
palace_path: Absolute path to the palace directory (contains
chromadb/ and sqlite/).
wing: Wizard wing to scope the search to.
room: Optional room filter (e.g. "forge").
top_k: Maximum number of results to return.
Returns:
Ranked list of MemoryResult objects, best match first.
Returns an empty list (not an exception) if the palace is
unavailable or the binary is missing.
"""
palace_path = Path(palace_path)
bin_path = _find_mempalace_bin(palace_path)
if bin_path is None:
return []
cmd = [
str(bin_path),
"search",
"--palace", str(palace_path),
"--wing", wing,
"--top-k", str(top_k),
"--format", "json",
query,
]
if room:
cmd.extend(["--room", room])
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=10,
)
if result.returncode != 0:
return []
data = json.loads(result.stdout)
return [
MemoryResult(
room=item.get("room", "general"),
text=item.get("text", ""),
score=float(item.get("score", 0.0)),
source=item.get("source", ""),
metadata=item.get("metadata", {}),
)
for item in data.get("results", [])
]
except (subprocess.TimeoutExpired, json.JSONDecodeError, OSError):
return []

View File

@@ -0,0 +1,64 @@
"""
settings.py — Configuration bridge between Evennia settings and MemPalace.
Read MEMPALACE_* values from Django/Evennia settings with safe fallbacks.
All values can be overridden in your game's settings.py.
Settings:
MEMPALACE_PATH (str) — Absolute path to the local palace directory.
e.g. "/root/wizards/bezalel/.mempalace/palace"
MEMPALACE_FLEET_PATH (str) — Path to the shared fleet palace (optional).
e.g. "/var/lib/mempalace/fleet"
MEMPALACE_WING (str) — Default wing name for this server instance.
e.g. "bezalel"
Refs: #1077, #1075
"""
from __future__ import annotations
from pathlib import Path
from typing import Optional
try:
from django.conf import settings as _django_settings
_HAS_DJANGO = True
except ImportError:
_django_settings = None # type: ignore[assignment]
_HAS_DJANGO = False
def _get_setting(key: str, default=None):
if _HAS_DJANGO:
return getattr(_django_settings, key, default)
return default
def get_palace_path() -> Optional[Path]:
"""Return the local palace path, or None if not configured."""
val = _get_setting("MEMPALACE_PATH")
if not val:
return None
p = Path(val)
return p if p.exists() else p # return path even if not yet created
def get_fleet_palace_path() -> Optional[Path]:
"""Return the shared fleet palace path, or None if not configured."""
val = _get_setting("MEMPALACE_FLEET_PATH")
if not val:
return None
return Path(val)
def get_wing(caller=None) -> str:
"""Return the wing name for the given caller.
Falls back to MEMPALACE_WING setting, then "general".
Future: resolve per-character wing from caller.db.wing.
"""
if caller is not None:
wing = getattr(getattr(caller, "db", None), "wing", None)
if wing:
return wing
return _get_setting("MEMPALACE_WING", "general") or "general"

View File

@@ -0,0 +1 @@
"""evennia_mempalace.typeclasses — Evennia typeclasses for palace rooms."""

View File

@@ -0,0 +1,87 @@
"""
MemPalaceRoom — Evennia room typeclass whose description auto-populates
from the top palace memories for its topic on each player entrance.
Usage (in your Evennia game's world-building script):
from evennia import create_object
from evennia_mempalace.typeclasses.room import MemPalaceRoom
forge = create_object(
MemPalaceRoom,
key="The Forge",
attributes=[
("palace_room_key", "forge"), # matches rooms.yaml key
("palace_top_k", 3), # memories shown on enter
],
)
Players who enter the room see a dynamically-generated description
synthesised from the top-3 matching palace memories.
Refs: #1077, #1075
"""
from __future__ import annotations
try:
from evennia.objects.objects import DefaultRoom
except ImportError:
DefaultRoom = object # type: ignore[assignment,misc]
from ..searcher import search_memories
from ..settings import get_palace_path, get_wing
# Shown when the palace has no memories for this room yet.
_EMPTY_DESC = (
"A quiet chamber, its shelves bare. No memories have been filed here yet. "
"Use |wrecall|n to search across all rooms, or mine new artefacts to populate this space."
)
class MemPalaceRoom(DefaultRoom):
"""
A room whose description is drawn from MemPalace search results.
Attributes (set via room.db or create_object attributes):
palace_room_key (str): The room key from rooms.yaml (e.g. "forge").
palace_top_k (int): Number of memories to show. Default: 3.
"""
def at_object_receive(self, moved_obj, source_location, **kwargs):
"""Re-generate description when a character enters."""
super().at_object_receive(moved_obj, source_location, **kwargs)
# Only refresh for player characters
if not hasattr(moved_obj, "account") or moved_obj.account is None:
return
self._refresh_description(moved_obj)
def _refresh_description(self, viewer) -> None:
"""Fetch top palace memories and update db.desc."""
room_key = self.db.palace_room_key or ""
top_k = int(self.db.palace_top_k or 3)
palace_path = get_palace_path()
wing = get_wing(viewer)
if not palace_path or not room_key:
self.db.desc = _EMPTY_DESC
return
results = search_memories("", palace_path, wing=wing, room=room_key, top_k=top_k)
if not results:
self.db.desc = _EMPTY_DESC
return
lines = [f"|c[{self.db.palace_room_key}]|n — Top memories:\n"]
for i, r in enumerate(results, 1):
snippet = r.text[:300].replace("\n", " ")
if len(r.text) > 300:
snippet += ""
lines.append(f"|w{i}.|n {snippet}\n")
self.db.desc = "\n".join(lines)
def return_appearance(self, looker, **kwargs):
"""Refresh description before rendering to the viewer."""
self._refresh_description(looker)
return super().return_appearance(looker, **kwargs)

104
mempalace/export_closets.sh Executable file
View File

@@ -0,0 +1,104 @@
#!/usr/bin/env bash
# export_closets.sh — Privacy-safe export of wizard closets for fleet sync.
#
# Exports ONLY closet (summary) files from a wizard's local MemPalace to
# a bundle directory suitable for rsync to the shared Alpha fleet palace.
#
# POLICY: Raw drawers (full-text source content) NEVER leave the local VPS.
# Only closets (compressed summaries) are exported.
#
# Usage:
# ./mempalace/export_closets.sh [palace_dir] [export_dir]
#
# Defaults:
# palace_dir — $MEMPALACE_DIR or /root/wizards/bezalel/.mempalace/palace
# export_dir — /tmp/mempalace_export_closets
#
# After export, sync with:
# rsync -avz --delete /tmp/mempalace_export_closets/ alpha:/var/lib/mempalace/fleet/bezalel/
#
# Refs: #1083, #1075
set -euo pipefail
PALACE_DIR="${1:-${MEMPALACE_DIR:-/root/wizards/bezalel/.mempalace/palace}}"
EXPORT_DIR="${2:-/tmp/mempalace_export_closets}"
WIZARD="${MEMPALACE_WING:-bezalel}"
echo "[export_closets] Wizard: $WIZARD"
echo "[export_closets] Palace: $PALACE_DIR"
echo "[export_closets] Export: $EXPORT_DIR"
if [[ ! -d "$PALACE_DIR" ]]; then
echo "[export_closets] ERROR: palace not found: $PALACE_DIR" >&2
exit 1
fi
# Validate closets-only policy: abort if any raw drawer files are present in export scope.
# Closets are files named *.closet.json or stored under a closets/ subdirectory.
# Raw drawers are everything else (*.drawer.json, *.md source files, etc.).
DRAWER_COUNT=0
while IFS= read -r -d '' f; do
# Raw drawer check: any .json file that is NOT a closet
basename_f="$(basename "$f")"
if [[ "$basename_f" == *.drawer.json ]]; then
echo "[export_closets] POLICY VIOLATION: raw drawer found in export scope: $f" >&2
DRAWER_COUNT=$((DRAWER_COUNT + 1))
fi
done < <(find "$PALACE_DIR" -type f -name "*.json" -print0 2>/dev/null)
if [[ "$DRAWER_COUNT" -gt 0 ]]; then
echo "[export_closets] ABORT: $DRAWER_COUNT raw drawer(s) detected. Only closets may be exported." >&2
echo "[export_closets] Run mempalace compress to generate closets before exporting." >&2
exit 1
fi
# Also check for source_file metadata in closet JSON that would expose private paths.
SOURCE_FILE_LEAKS=0
while IFS= read -r -d '' f; do
if python3 -c "
import json, sys
try:
data = json.load(open('$f'))
drawers = data.get('drawers', []) if isinstance(data, dict) else []
for d in drawers:
if 'source_file' in d and not d.get('closet', False):
sys.exit(1)
except Exception:
pass
sys.exit(0)
" 2>/dev/null; then
:
else
echo "[export_closets] POLICY VIOLATION: source_file metadata in non-closet: $f" >&2
SOURCE_FILE_LEAKS=$((SOURCE_FILE_LEAKS + 1))
fi
done < <(find "$PALACE_DIR" -type f -name "*.closet.json" -print0 2>/dev/null)
if [[ "$SOURCE_FILE_LEAKS" -gt 0 ]]; then
echo "[export_closets] ABORT: $SOURCE_FILE_LEAKS file(s) contain private source_file paths." >&2
exit 1
fi
# Collect closet files
mkdir -p "$EXPORT_DIR/$WIZARD"
CLOSET_COUNT=0
while IFS= read -r -d '' f; do
rel_path="${f#$PALACE_DIR/}"
dest="$EXPORT_DIR/$WIZARD/$rel_path"
mkdir -p "$(dirname "$dest")"
cp "$f" "$dest"
CLOSET_COUNT=$((CLOSET_COUNT + 1))
done < <(find "$PALACE_DIR" -type f -name "*.closet.json" -print0 2>/dev/null)
if [[ "$CLOSET_COUNT" -eq 0 ]]; then
echo "[export_closets] WARNING: no closet files found in $PALACE_DIR" >&2
echo "[export_closets] Run 'mempalace compress' to generate closets from drawers." >&2
exit 0
fi
echo "[export_closets] Exported $CLOSET_COUNT closet(s) to $EXPORT_DIR/$WIZARD/"
echo "[export_closets] OK — ready for fleet sync."
echo ""
echo " rsync -avz --delete $EXPORT_DIR/$WIZARD/ alpha:/var/lib/mempalace/fleet/$WIZARD/"

114
mempalace/rooms.yaml Normal file
View File

@@ -0,0 +1,114 @@
# MemPalace Fleet Taxonomy Standard
# Refs: #1082, #1075 (MemPalace × Evennia — Fleet Memory milestone)
#
# Every wizard palace MUST contain the 5 core rooms listed under `core_rooms`.
# Optional domain-specific rooms are listed under `optional_rooms` for reference.
# Wizards may add additional rooms beyond this taxonomy.
#
# Room schema fields:
# key — machine-readable slug (used for tunnel routing and fleet search)
# label — human-readable display name
# purpose — one-line description of what belongs here
# examples — sample artifact types filed in this room
version: "1"
core_rooms:
- key: forge
label: Forge
purpose: CI pipelines, builds, infra configuration, deployment artefacts
examples:
- build logs
- CI run summaries
- Dockerfile changes
- cron job definitions
- server provisioning notes
- key: hermes
label: Hermes
purpose: Agent platform, Hermes gateway, harness CLI, inter-agent messaging
examples:
- harness config snapshots
- agent boot reports
- MCP tool definitions
- Hermes gateway events
- worker health logs
- key: nexus
label: Nexus
purpose: Project reports, documentation, knowledge transfer, field reports
examples:
- SITREP documents
- architecture decision records
- field reports
- onboarding docs
- milestone summaries
- key: issues
label: Issues
purpose: Tickets, backlog items, PR summaries, bug reports
examples:
- Gitea issue summaries
- PR merge notes
- bug reproduction steps
- acceptance criteria
- key: experiments
label: Experiments
purpose: Prototypes, spikes, sandbox work, exploratory research
examples:
- spike results
- A/B test notes
- proof-of-concept code snippets
- benchmark data
optional_rooms:
- key: evennia
label: Evennia
purpose: MUD world state, room descriptions, NPC dialogue, game events
wizards: [bezalel, timmy]
- key: game-portals
label: Game Portals
purpose: Portal registry, zone configs, dungeon layouts, loot tables
wizards: [timmy]
- key: lazarus-pit
label: Lazarus Pit
purpose: Dead/parked work, archived experiments, deprecated configs
wizards: [timmy, allegro, bezalel]
- key: satflow
label: SatFlow
purpose: Economy visualizations, satoshi flow tracking, L402 audit trails
wizards: [timmy, allegro]
- key: workspace
label: Workspace
purpose: General scratch notes, daily logs, personal coordination
wizards: ["*"]
- key: home
label: Home
purpose: Personal identity, agent persona, preferences, capability docs
wizards: ["*"]
- key: general
label: General
purpose: Catch-all for artefacts not yet assigned to a named room
wizards: ["*"]
# Tunnel routing table
# Defines which room pairs are connected across wizard wings.
# A tunnel lets `recall <query> --fleet` search both wings at once.
tunnels:
- rooms: [forge, forge]
description: Build and infra knowledge shared across all wizards
- rooms: [hermes, hermes]
description: Harness platform knowledge shared across all wizards
- rooms: [nexus, nexus]
description: Cross-wizard documentation and field reports
- rooms: [issues, issues]
description: Fleet-wide issue and PR knowledge
- rooms: [experiments, experiments]
description: Cross-wizard spike and prototype results

119
mempalace/validate_rooms.py Normal file
View File

@@ -0,0 +1,119 @@
#!/usr/bin/env python3
"""
validate_rooms.py — Fleet palace taxonomy validator.
Checks a wizard's mempalace.yaml against the fleet standard in rooms.yaml.
Exits 0 if valid, 1 if core rooms are missing or the config is malformed.
Usage:
python mempalace/validate_rooms.py <wizard_mempalace.yaml>
python mempalace/validate_rooms.py /root/wizards/bezalel/mempalace.yaml
Refs: #1082, #1075
"""
from __future__ import annotations
import argparse
import sys
from pathlib import Path
from typing import Any
try:
import yaml
except ImportError:
print("ERROR: PyYAML is required. Install with: pip install pyyaml", file=sys.stderr)
sys.exit(2)
FLEET_STANDARD = Path(__file__).parent / "rooms.yaml"
def load_yaml(path: Path) -> dict[str, Any]:
with path.open() as fh:
return yaml.safe_load(fh) or {}
def get_core_room_keys(standard: dict[str, Any]) -> list[str]:
return [r["key"] for r in standard.get("core_rooms", [])]
def get_wizard_room_keys(config: dict[str, Any]) -> list[str]:
"""Extract room keys from a wizard's mempalace.yaml.
Supports two common shapes:
rooms:
- key: forge
- key: hermes
or:
rooms:
forge: ...
hermes: ...
"""
rooms_field = config.get("rooms", {})
if isinstance(rooms_field, list):
return [r["key"] for r in rooms_field if isinstance(r, dict) and "key" in r]
if isinstance(rooms_field, dict):
return list(rooms_field.keys())
return []
def validate(wizard_config_path: Path, standard_path: Path = FLEET_STANDARD) -> list[str]:
"""Return a list of validation errors. Empty list means valid."""
errors: list[str] = []
if not standard_path.exists():
errors.append(f"Fleet standard not found: {standard_path}")
return errors
if not wizard_config_path.exists():
errors.append(f"Wizard config not found: {wizard_config_path}")
return errors
standard = load_yaml(standard_path)
config = load_yaml(wizard_config_path)
core_keys = get_core_room_keys(standard)
wizard_keys = get_wizard_room_keys(config)
missing = [k for k in core_keys if k not in wizard_keys]
for key in missing:
errors.append(f"Missing required core room: '{key}'")
return errors
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
description="Validate a wizard's mempalace.yaml against the fleet room standard."
)
parser.add_argument(
"config",
metavar="mempalace.yaml",
help="Path to the wizard's mempalace.yaml",
)
parser.add_argument(
"--standard",
default=str(FLEET_STANDARD),
metavar="rooms.yaml",
help="Path to the fleet rooms.yaml standard (default: mempalace/rooms.yaml)",
)
args = parser.parse_args(argv)
wizard_path = Path(args.config)
standard_path = Path(args.standard)
errors = validate(wizard_path, standard_path)
if errors:
print(f"[validate_rooms] FAIL: {wizard_path}", file=sys.stderr)
for err in errors:
print(f"{err}", file=sys.stderr)
return 1
core_count = len(get_core_room_keys(load_yaml(standard_path)))
print(f"[validate_rooms] OK: {wizard_path} — all {core_count} core rooms present.")
return 0
if __name__ == "__main__":
sys.exit(main())

View File

@@ -0,0 +1,129 @@
"""
Tests for mempalace/audit_privacy.py — fleet palace privacy auditor.
Refs: #1083, #1075
"""
import json
from pathlib import Path
import pytest
from mempalace.audit_privacy import (
Violation,
audit_file,
audit_palace,
_is_private_path,
)
# ---------------------------------------------------------------------------
# _is_private_path
# ---------------------------------------------------------------------------
def test_private_path_root():
assert _is_private_path("/root/wizards/bezalel/workspace.md") is True
def test_private_path_home():
assert _is_private_path("/home/apayne/projects/nexus") is True
def test_private_path_users():
assert _is_private_path("/Users/apayne/worktrees/nexus/foo.py") is True
def test_non_private_path():
assert _is_private_path("/var/lib/mempalace/fleet/bezalel/forge.closet.json") is False
assert _is_private_path("relative/path.md") is False
# ---------------------------------------------------------------------------
# audit_file — clean closet
# ---------------------------------------------------------------------------
def _write_closet(tmp_path: Path, name: str, drawers: list) -> Path:
p = tmp_path / name
p.write_text(json.dumps({"drawers": drawers}))
return p
def test_clean_closet_has_no_violations(tmp_path):
f = _write_closet(tmp_path, "forge.closet.json", [
{"text": "Build succeeded on commit abc123.", "closet": True},
])
assert audit_file(f) == []
# ---------------------------------------------------------------------------
# audit_file — raw drawer violation
# ---------------------------------------------------------------------------
def test_raw_drawer_file_is_violation(tmp_path):
f = tmp_path / "workspace.drawer.json"
f.write_text(json.dumps({"text": "some private content"}))
violations = audit_file(f)
assert len(violations) == 1
assert violations[0].rule == "RAW_DRAWER"
# ---------------------------------------------------------------------------
# audit_file — full text in closet
# ---------------------------------------------------------------------------
def test_full_text_closet_is_violation(tmp_path):
long_text = "x" * 3000 # exceeds 2000 char limit
f = _write_closet(tmp_path, "nexus.closet.json", [
{"text": long_text, "closet": True},
])
violations = audit_file(f)
assert any(v.rule == "FULL_TEXT_IN_CLOSET" for v in violations)
# ---------------------------------------------------------------------------
# audit_file — private source_file path
# ---------------------------------------------------------------------------
def test_private_source_file_is_violation(tmp_path):
f = _write_closet(tmp_path, "hermes.closet.json", [
{
"text": "Short summary.",
"source_file": "/root/wizards/bezalel/secret.md",
"closet": True,
}
])
violations = audit_file(f)
assert any(v.rule == "PRIVATE_SOURCE_PATH" for v in violations)
def test_fleet_source_file_is_ok(tmp_path):
f = _write_closet(tmp_path, "hermes.closet.json", [
{
"text": "Short summary.",
"source_file": "/var/lib/mempalace/fleet/bezalel/hermes.closet.json",
"closet": True,
}
])
violations = audit_file(f)
assert violations == []
# ---------------------------------------------------------------------------
# audit_palace
# ---------------------------------------------------------------------------
def test_audit_palace_clean(tmp_path):
_write_closet(tmp_path, "forge.closet.json", [{"text": "ok", "closet": True}])
_write_closet(tmp_path, "nexus.closet.json", [{"text": "ok", "closet": True}])
result = audit_palace(tmp_path)
assert result.clean
assert result.scanned == 2
def test_audit_palace_finds_violations(tmp_path):
_write_closet(tmp_path, "forge.closet.json", [{"text": "ok", "closet": True}])
bad = tmp_path / "secret.drawer.json"
bad.write_text(json.dumps({"text": "raw private data"}))
result = audit_palace(tmp_path)
assert not result.clean
assert any(v.rule == "RAW_DRAWER" for v in result.violations)

View File

@@ -0,0 +1,160 @@
"""
Tests for mempalace/validate_rooms.py — fleet room taxonomy validator.
Refs: #1082, #1075
"""
import json
import textwrap
from pathlib import Path
import pytest
import yaml
from mempalace.validate_rooms import (
get_core_room_keys,
get_wizard_room_keys,
validate,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
STANDARD_YAML = textwrap.dedent("""\
version: "1"
core_rooms:
- key: forge
label: Forge
purpose: CI and builds
- key: hermes
label: Hermes
purpose: Agent platform
- key: nexus
label: Nexus
purpose: Reports and docs
- key: issues
label: Issues
purpose: Tickets and backlog
- key: experiments
label: Experiments
purpose: Prototypes and spikes
""")
def write_standard(tmp_path: Path) -> Path:
p = tmp_path / "rooms.yaml"
p.write_text(STANDARD_YAML)
return p
# ---------------------------------------------------------------------------
# get_core_room_keys
# ---------------------------------------------------------------------------
def test_get_core_room_keys_returns_all_keys(tmp_path):
standard_path = write_standard(tmp_path)
standard = yaml.safe_load(standard_path.read_text())
keys = get_core_room_keys(standard)
assert keys == ["forge", "hermes", "nexus", "issues", "experiments"]
def test_get_core_room_keys_empty_if_no_core_rooms():
keys = get_core_room_keys({})
assert keys == []
# ---------------------------------------------------------------------------
# get_wizard_room_keys
# ---------------------------------------------------------------------------
def test_get_wizard_room_keys_list_style():
config = {
"rooms": [
{"key": "forge"},
{"key": "hermes"},
]
}
assert get_wizard_room_keys(config) == ["forge", "hermes"]
def test_get_wizard_room_keys_dict_style():
config = {
"rooms": {
"forge": {"purpose": "builds"},
"nexus": {"purpose": "docs"},
}
}
keys = get_wizard_room_keys(config)
assert set(keys) == {"forge", "nexus"}
def test_get_wizard_room_keys_empty_config():
assert get_wizard_room_keys({}) == []
# ---------------------------------------------------------------------------
# validate — happy path
# ---------------------------------------------------------------------------
def test_validate_passes_with_all_core_rooms(tmp_path):
standard_path = write_standard(tmp_path)
wizard_config = tmp_path / "mempalace.yaml"
wizard_config.write_text(textwrap.dedent("""\
rooms:
- key: forge
- key: hermes
- key: nexus
- key: issues
- key: experiments
"""))
errors = validate(wizard_config, standard_path)
assert errors == []
def test_validate_passes_with_extra_rooms(tmp_path):
standard_path = write_standard(tmp_path)
wizard_config = tmp_path / "mempalace.yaml"
wizard_config.write_text(textwrap.dedent("""\
rooms:
- key: forge
- key: hermes
- key: nexus
- key: issues
- key: experiments
- key: evennia
- key: workspace
"""))
errors = validate(wizard_config, standard_path)
assert errors == []
# ---------------------------------------------------------------------------
# validate — failure cases
# ---------------------------------------------------------------------------
def test_validate_reports_missing_core_rooms(tmp_path):
standard_path = write_standard(tmp_path)
wizard_config = tmp_path / "mempalace.yaml"
wizard_config.write_text(textwrap.dedent("""\
rooms:
- key: forge
"""))
errors = validate(wizard_config, standard_path)
missing_keys = [e for e in errors if "hermes" in e or "nexus" in e or "issues" in e or "experiments" in e]
assert len(missing_keys) == 4
def test_validate_missing_wizard_config(tmp_path):
standard_path = write_standard(tmp_path)
missing = tmp_path / "nonexistent.yaml"
errors = validate(missing, standard_path)
assert any("not found" in e for e in errors)
def test_validate_missing_standard(tmp_path):
wizard_config = tmp_path / "mempalace.yaml"
wizard_config.write_text("rooms:\n - key: forge\n")
missing_standard = tmp_path / "no_such_rooms.yaml"
errors = validate(wizard_config, missing_standard)
assert any("not found" in e for e in errors)