[claude] MemPalace × Evennia fleet memory scaffold (#1075) #1088

Merged
claude merged 4 commits from claude/issue-1075 into main 2026-04-07 14:12:40 +00:00
41 changed files with 2957 additions and 0 deletions

1
.gitignore vendored
View File

@@ -2,4 +2,5 @@ node_modules/
test-results/
nexus/__pycache__/
tests/__pycache__/
mempalace/__pycache__/
.aider*

183
docs/mempalace/rooms.yaml Normal file
View File

@@ -0,0 +1,183 @@
# MemPalace Fleet Room Taxonomy Standard
# =======================================
# Version: 1.0
# Milestone: MemPalace × Evennia — Fleet Memory (#1075)
# Issue: #1082 [Infra] Palace taxonomy standard
#
# Every wizard's palace MUST contain the five core rooms listed below.
# Domain rooms are optional and wizard-specific.
#
# Format:
# rooms:
# <room_key>:
# required: true|false
# description: one-liner purpose
# example_topics: [list of things that belong here]
# tunnel: true if a cross-wizard tunnel should exist for this room
rooms:
# ── Core rooms (required in every wing) ────────────────────────────────────
forge:
required: true
description: "CI, builds, deployment, infra operations"
example_topics:
- "github actions failures"
- "docker build logs"
- "server deployment steps"
- "cron job setup"
tunnel: true
hermes:
required: true
description: "Agent platform, gateway, CLI tooling, harness internals"
example_topics:
- "hermes session logs"
- "agent wake cycle"
- "MCP tool calls"
- "gateway configuration"
tunnel: true
nexus:
required: true
description: "Reports, docs, knowledge transfer, SITREPs"
example_topics:
- "nightly watch report"
- "architecture docs"
- "handoff notes"
- "decision records"
tunnel: true
issues:
required: true
description: "Gitea tickets, backlog items, bug reports, PR reviews"
example_topics:
- "issue triage"
- "PR feedback"
- "bug root cause"
- "milestone planning"
tunnel: true
experiments:
required: true
description: "Prototypes, spikes, research, benchmarks"
example_topics:
- "spike results"
- "benchmark numbers"
- "proof of concept"
- "chromadb evaluation"
tunnel: true
# ── Write rooms (created on demand by CmdRecord/CmdNote/CmdEvent) ──────────
hall_facts:
required: false
description: "Decisions and facts recorded via 'record' command"
example_topics:
- "architectural decisions"
- "policy choices"
- "approved approaches"
tunnel: false
hall_discoveries:
required: false
description: "Breakthroughs and key findings recorded via 'note' command"
example_topics:
- "performance breakthroughs"
- "algorithmic insights"
- "unexpected results"
tunnel: false
hall_events:
required: false
description: "Significant events logged via 'event' command"
example_topics:
- "production deployments"
- "milestones reached"
- "incidents resolved"
tunnel: false
# ── Optional domain rooms (wizard-specific) ────────────────────────────────
evennia:
required: false
description: "Evennia MUD world: rooms, commands, NPCs, world design"
example_topics:
- "command implementation"
- "typeclass design"
- "world building notes"
wizard: ["bezalel"]
tunnel: false
game_portals:
required: false
description: "Portal/gameplay work: satflow, economy, portal registry"
example_topics:
- "portal specs"
- "satflow visualization"
- "economy rules"
wizard: ["bezalel", "timmy"]
tunnel: false
workspace:
required: false
description: "General wizard workspace notes that don't fit elsewhere"
example_topics:
- "daily notes"
- "scratch work"
- "reference lookups"
tunnel: false
general:
required: false
description: "Fallback room for unclassified memories"
example_topics:
- "uncategorized notes"
tunnel: false
# ── Tunnel policy ─────────────────────────────────────────────────────────────
#
# A tunnel is a cross-wing link that lets any wizard recall memories
# from an equivalent room in another wing.
#
# Rules:
# 1. Only CLOSETS (summaries) are synced through tunnels — never raw drawers.
# 2. Required rooms marked tunnel:true MUST have tunnels on Alpha.
# 3. Optional rooms are never tunnelled unless explicitly opted in.
# 4. Raw drawers (source_file metadata) never leave the local VPS.
tunnels:
policy: closets_only
sync_schedule: "04:00 UTC nightly"
destination: "/var/lib/mempalace/fleet"
rooms_synced:
- forge
- hermes
- nexus
- issues
- experiments
# ── Privacy rules ─────────────────────────────────────────────────────────────
#
# See issue #1083 for the full privacy boundary design.
#
# Summary:
# - hall_facts, hall_discoveries, hall_events: LOCAL ONLY (never synced)
# - workspace, general: LOCAL ONLY
# - Domain rooms (evennia, game_portals): LOCAL ONLY unless tunnel:true
# - source_file paths MUST be stripped before sync
privacy:
local_only_rooms:
- hall_facts
- hall_discoveries
- hall_events
- workspace
- general
strip_on_sync:
- source_file
retention_days: 90
archive_flag: "archive: true"

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)
"""

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,49 @@
"""nexus.evennia_mempalace — Evennia plugin for MemPalace fleet memory.
This contrib module provides:
Commands (add to ``settings.CMDSETS_DEFAULT`` or a CmdSet):
CmdRecall — ``recall <query>`` / ``recall <query> --fleet``
CmdEnterRoom — ``enter room <topic>`` teleports to a palace room
CmdRecord — ``record decision <text>`` writes to hall_facts
CmdNote — ``note breakthrough <text>`` writes to hall_discoveries
CmdEvent — ``event <text>`` writes to hall_events
Typeclasses (use in place of Evennia's default Room/Character):
MemPalaceRoom — Room whose description auto-populates from palace search
StewardNPC — Wizard steward that answers questions via palace search
Usage example (in your Evennia game's ``mygame/server/conf/settings.py``)::
MEMPALACE_PATH = "/root/wizards/bezalel/.mempalace/palace"
MEMPALACE_WING = "bezalel"
FLEET_PALACE_PATH = "/var/lib/mempalace/fleet"
Then import commands into a CmdSet::
from nexus.evennia_mempalace.commands import (
CmdRecall, CmdEnterRoom, CmdRecord, CmdNote, CmdEvent
)
"""
from __future__ import annotations
from nexus.evennia_mempalace.commands import (
CmdEnterRoom,
CmdEvent,
CmdNote,
CmdRecord,
CmdRecall,
)
from nexus.evennia_mempalace.typeclasses.rooms import MemPalaceRoom
from nexus.evennia_mempalace.typeclasses.npcs import StewardNPC
__all__ = [
"CmdRecall",
"CmdEnterRoom",
"CmdRecord",
"CmdNote",
"CmdEvent",
"MemPalaceRoom",
"StewardNPC",
]

View File

@@ -0,0 +1,14 @@
"""MemPalace Evennia commands."""
from __future__ import annotations
from nexus.evennia_mempalace.commands.recall import CmdRecall, CmdEnterRoom
from nexus.evennia_mempalace.commands.write import CmdRecord, CmdNote, CmdEvent
__all__ = [
"CmdRecall",
"CmdEnterRoom",
"CmdRecord",
"CmdNote",
"CmdEvent",
]

View File

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

View File

@@ -0,0 +1,124 @@
"""Evennia commands for writing new memories to the palace.
CmdRecord — record decision <text> → files into hall_facts
CmdNote — note breakthrough <text> → files into hall_discoveries
CmdEvent — event <text> → files into hall_events
Phase 4 deliverable (see issue #1080).
"""
from __future__ import annotations
from nexus.mempalace.searcher import MemPalaceUnavailable, add_memory
from nexus.mempalace.config import FLEET_WING
try:
from evennia import Command as _EvCommand # type: ignore
if _EvCommand is None:
raise ImportError("evennia.Command is None (Django not configured)")
Command = _EvCommand
except (ImportError, Exception):
class Command: # type: ignore
key = ""
aliases: list = []
locks = "cmd:all()"
help_category = "MemPalace"
def __init__(self):
self.caller = None
self.args = ""
self.switches: list[str] = []
def func(self):
pass
class _MemWriteCommand(Command):
"""Base class for palace write commands."""
_room: str = "general"
_label: str = "memory"
def func(self):
text = self.args.strip()
if not text:
self.caller.msg(f"Usage: {self.key} <text>")
return
wing = getattr(self.caller.db, "wing", None) or FLEET_WING
try:
doc_id = add_memory(
text,
room=self._room,
wing=wing,
extra_metadata={"via": "evennia_cmd", "cmd": self.key},
)
except MemPalaceUnavailable as exc:
self.caller.msg(f"|rPalace unavailable:|n {exc}")
return
self.caller.msg(
f"|gFiled {self._label} into |c{self._room}|g.|n (id: {doc_id[:8]}…)"
)
class CmdRecord(_MemWriteCommand):
"""Record a decision into the palace (hall_facts).
Usage:
record <text>
record decision <text>
Example:
record We decided to use ChromaDB for local palace storage.
The text is filed into the ``hall_facts`` room of your wing and
becomes searchable via ``recall``.
"""
key = "record"
aliases = ["record decision"]
locks = "cmd:all()"
help_category = "MemPalace"
_room = "hall_facts"
_label = "decision"
class CmdNote(_MemWriteCommand):
"""File a breakthrough note into the palace (hall_discoveries).
Usage:
note <text>
note breakthrough <text>
Example:
note breakthrough AAAK compression reduces token cost by 40%.
The text is filed into the ``hall_discoveries`` room of your wing.
"""
key = "note"
aliases = ["note breakthrough"]
locks = "cmd:all()"
help_category = "MemPalace"
_room = "hall_discoveries"
_label = "breakthrough"
class CmdEvent(_MemWriteCommand):
"""Log a significant event into the palace (hall_events).
Usage:
event <text>
Example:
event Deployed Evennia bridge to production on Alpha.
The text is filed into the ``hall_events`` room of your wing.
"""
key = "event"
locks = "cmd:all()"
help_category = "MemPalace"
_room = "hall_events"
_label = "event"

View File

@@ -0,0 +1 @@
"""MemPalace Evennia typeclasses."""

View File

@@ -0,0 +1,138 @@
"""StewardNPC — wizard steward that answers questions via palace search.
Each wizard wing has a steward NPC that players can interrogate about
the wing's history. The NPC:
1. Detects the topic from the player's question.
2. Calls ``search_memories`` with wing + optional room filters.
3. Formats the top results as an in-character response.
Phase 3 deliverable (see issue #1079).
"""
from __future__ import annotations
from nexus.mempalace.searcher import MemPalaceUnavailable, search_memories
from nexus.mempalace.config import FLEET_WING
try:
from evennia import DefaultCharacter as _EvDefaultCharacter # type: ignore
if _EvDefaultCharacter is None:
raise ImportError("evennia.DefaultCharacter is None")
DefaultCharacter = _EvDefaultCharacter
except (ImportError, Exception):
class DefaultCharacter: # type: ignore # minimal stub
db: object = None
key: str = ""
def msg(self, text: str, **kwargs):
pass
def execute_cmd(self, raw_string: str, **kwargs):
pass
# Steward response templates
_FOUND_TEMPLATE = (
"|c{name}|n glances inward, consulting the palace...\n\n"
"I find {count} relevant {plural} about |w{topic}|n:\n\n"
"{memories}\n"
"|xType '|wrecall {topic}|x' to search further.|n"
)
_NOT_FOUND_TEMPLATE = (
"|c{name}|n ponders a moment, then shakes their head.\n"
"\"I found nothing about |w{topic}|n in this wing's memory.\""
)
_UNAVAILABLE_TEMPLATE = (
"|c{name}|n frowns. \"The palace is unreachable right now.\""
)
class StewardNPC(DefaultCharacter):
"""An NPC that serves as the custodian of a wizard's memory wing.
Attributes (set via ``npc.db.<attr>``):
steward_wing (str): The wizard wing this steward guards.
Defaults to ``FLEET_WING``.
steward_name (str): Display name used in responses.
Defaults to ``self.key``.
steward_n_results (int): How many memories to surface.
Default 3.
Usage (from game)::
> ask bezalel-steward about nightly watch failures
> ask steward about CI pipeline
"""
# Evennia will call at_say when players speak near the NPC
def at_say(self, message: str, msg_type: str = "say", **kwargs):
"""Intercept nearby speech that looks like a question."""
super().at_say(message, msg_type=msg_type, **kwargs)
def respond_to_question(self, question: str, asker=None) -> str:
"""Answer a question by searching the wing's palace.
Args:
question: The player's raw question text.
asker: The asking character object (used to personalise output).
Returns:
Formatted response string.
"""
topic = _extract_topic(question)
wing = self.db.steward_wing or FLEET_WING
name = self.db.steward_name or self.key
n = self.db.steward_n_results or 3
try:
results = search_memories(topic, wing=wing, n_results=n)
except MemPalaceUnavailable:
return _UNAVAILABLE_TEMPLATE.format(name=name)
if not results:
return _NOT_FOUND_TEMPLATE.format(name=name, topic=topic)
memory_lines = []
for i, r in enumerate(results, start=1):
memory_lines.append(
f"|w{i}. [{r.room}]|n {r.short(220)}"
)
return _FOUND_TEMPLATE.format(
name=name,
count=len(results),
plural="memory" if len(results) == 1 else "memories",
topic=topic,
memories="\n".join(memory_lines),
)
# ── Helpers ───────────────────────────────────────────────────────────────────
_QUESTION_PREFIXES = (
"about ", "regarding ", "on ", "concerning ",
"related to ", "for ", "with ", "involving ",
)
def _extract_topic(question: str) -> str:
"""Extract the key topic from a natural-language question.
Strips common question prefixes so that the palace search receives
a clean keyword rather than noise words.
Examples:
"about nightly watch failures""nightly watch failures"
"what do you know about the CI pipeline?""CI pipeline"
"""
q = question.strip().rstrip("?").strip()
# Remove leading question words
for prefix in ("what do you know ", "tell me ", "do you know "):
if q.lower().startswith(prefix):
q = q[len(prefix):]
for prep in _QUESTION_PREFIXES:
if q.lower().startswith(prep):
q = q[len(prep):]
break
return q or question.strip()

View File

@@ -0,0 +1,99 @@
"""MemPalaceRoom — Evennia room typeclass backed by palace search.
When a character enters a MemPalaceRoom, the room's description is
automatically refreshed from a live palace search for the room's
topic keyword. This makes the room "alive" — its contents reflect
what the fleet actually knows about that topic.
Phase 1 deliverable (see issue #1077).
"""
from __future__ import annotations
from nexus.mempalace.searcher import MemPalaceUnavailable, search_memories
from nexus.mempalace.config import FLEET_WING
try:
from evennia import DefaultRoom as _EvDefaultRoom # type: ignore
if _EvDefaultRoom is None:
raise ImportError("evennia.DefaultRoom is None")
DefaultRoom = _EvDefaultRoom
except (ImportError, Exception):
class DefaultRoom: # type: ignore # minimal stub for import/testing
"""Stub for environments without Evennia installed."""
db: object = None
key: str = ""
def return_appearance(self, looker): # noqa: D102
return ""
def at_object_receive(self, moved_obj, source_location, **kwargs): # noqa: D102
pass
_PALACE_ROOM_HEADER = """|b═══════════════════════════════════════════════════|n
|c Mind Palace — {room_name}|n
|b═══════════════════════════════════════════════════|n"""
_PALACE_ROOM_FOOTER = """|b───────────────────────────────────────────────────|n
|xType '|wrecall <query>|x' to search deeper.|n"""
class MemPalaceRoom(DefaultRoom):
"""An Evennia room whose description comes from the MemPalace.
Attributes (set via ``room.db.<attr>``):
palace_topic (str): Search term used to populate the description.
Defaults to the room's key.
palace_wing (str): Wing to search. Defaults to fleet wing.
palace_n_results (int): How many memories to show. Default 3.
palace_room_filter (str): Optional room-name filter for the query.
"""
def at_object_receive(self, moved_obj, source_location, **kwargs):
"""Refresh palace content whenever someone enters."""
super().at_object_receive(moved_obj, source_location, **kwargs)
# Only refresh for player-controlled characters
if hasattr(moved_obj, "account") and moved_obj.account:
self._refresh_palace_desc(viewer=moved_obj)
def return_appearance(self, looker, **kwargs):
"""Return description augmented with live palace memories."""
self._refresh_palace_desc(viewer=looker)
return super().return_appearance(looker, **kwargs)
# ── Internal helpers ──────────────────────────────────────────────────
def _refresh_palace_desc(self, viewer=None):
"""Update ``self.db.desc`` from a fresh palace query."""
topic = self.db.palace_topic or self.key or "general"
wing = self.db.palace_wing or FLEET_WING
n = self.db.palace_n_results or 3
room_filter = self.db.palace_room_filter
try:
results = search_memories(
topic, wing=wing, room=room_filter, n_results=n
)
except MemPalaceUnavailable:
self.db.desc = (
f"[Palace unavailable — could not load memories for '{topic}'.]"
)
return
lines = [
_PALACE_ROOM_HEADER.format(room_name=self.key),
]
if results:
for r in results:
lines.append(f"|w{r.room}|n |x(score {r.score:.2f})|n")
lines.append(f" {r.short(280)}")
lines.append("")
else:
lines.append(f"|yNo memories found for topic '|w{topic}|y'.|n")
lines.append("")
lines.append(_PALACE_ROOM_FOOTER)
self.db.desc = "\n".join(lines)

View File

@@ -0,0 +1,23 @@
"""nexus.mempalace — MemPalace integration for the Nexus fleet.
Public API for searching, configuring, and writing to MemPalace
local vector memory. Designed to be imported by both the
``evennia_mempalace`` plugin and any other harness component.
ChromaDB is an optional runtime dependency; the module degrades
gracefully when it is not installed (tests, CI, environments that
have not yet set up the palace).
"""
from __future__ import annotations
from nexus.mempalace.config import MEMPALACE_PATH, FLEET_WING
from nexus.mempalace.searcher import search_memories, add_memory, MemPalaceResult
__all__ = [
"MEMPALACE_PATH",
"FLEET_WING",
"search_memories",
"add_memory",
"MemPalaceResult",
]

Binary file not shown.

Binary file not shown.

Binary file not shown.

46
nexus/mempalace/config.py Normal file
View File

@@ -0,0 +1,46 @@
"""MemPalace configuration — paths and fleet settings.
All configuration is driven by environment variables so that
different wizards on different VPSes can use the same code with
their own palace directories.
"""
from __future__ import annotations
import os
from pathlib import Path
# ── Palace path ──────────────────────────────────────────────────────────────
# Default: ~/.mempalace/palace/ (local wizard palace)
# Override via MEMPALACE_PATH env var (useful for fleet shared wing)
_default = Path.home() / ".mempalace" / "palace"
MEMPALACE_PATH: Path = Path(os.environ.get("MEMPALACE_PATH", str(_default)))
# ── Fleet shared wing ─────────────────────────────────────────────────────────
# Path to the shared fleet palace on Alpha (used by --fleet searches)
_fleet_default = Path("/var/lib/mempalace/fleet")
FLEET_PALACE_PATH: Path = Path(
os.environ.get("FLEET_PALACE_PATH", str(_fleet_default))
)
# ── Wing name ─────────────────────────────────────────────────────────────────
# Identifies this wizard's wing within a shared palace.
# Populated from MEMPALACE_WING env var or falls back to system username.
def _default_wing() -> str:
import getpass
return os.environ.get("MEMPALACE_WING", getpass.getuser())
FLEET_WING: str = _default_wing()
# ── Fleet rooms standard ─────────────────────────────────────────────────────
# Canonical rooms every wizard must have (see docs/mempalace/rooms.yaml)
CORE_ROOMS: list[str] = [
"forge", # CI, builds, infra
"hermes", # agent platform, gateway, CLI
"nexus", # reports, docs, KT
"issues", # tickets, backlog
"experiments", # prototypes, spikes
]
# ── ChromaDB collection name ──────────────────────────────────────────────────
COLLECTION_NAME: str = os.environ.get("MEMPALACE_COLLECTION", "palace")

200
nexus/mempalace/searcher.py Normal file
View File

@@ -0,0 +1,200 @@
"""MemPalace search and write interface.
Wraps the ChromaDB-backed palace so that callers (Evennia commands,
harness agents, MCP tools) do not need to know the storage details.
ChromaDB is imported lazily; if it is not installed the functions
raise ``MemPalaceUnavailable`` with an informative message rather
than crashing at import time.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
from nexus.mempalace.config import (
MEMPALACE_PATH,
FLEET_PALACE_PATH,
COLLECTION_NAME,
)
class MemPalaceUnavailable(RuntimeError):
"""Raised when ChromaDB or the palace directory is not accessible."""
@dataclass
class MemPalaceResult:
"""A single memory hit returned by the searcher."""
text: str
room: str
wing: str
score: float = 0.0
source_file: str = ""
metadata: dict = field(default_factory=dict)
def short(self, max_chars: int = 200) -> str:
"""Return a truncated preview suitable for MUD output."""
if len(self.text) <= max_chars:
return self.text
return self.text[:max_chars].rstrip() + ""
def _get_client(palace_path: Path):
"""Return a ChromaDB persistent client, or raise MemPalaceUnavailable."""
try:
import chromadb # type: ignore
except ImportError as exc:
raise MemPalaceUnavailable(
"ChromaDB is not installed. "
"Run: pip install chromadb (or: pip install mempalace)"
) from exc
if not palace_path.exists():
raise MemPalaceUnavailable(
f"Palace directory not found: {palace_path}\n"
"Run 'mempalace mine' to initialise the palace."
)
return chromadb.PersistentClient(path=str(palace_path))
def search_memories(
query: str,
*,
palace_path: Optional[Path] = None,
wing: Optional[str] = None,
room: Optional[str] = None,
n_results: int = 5,
) -> list[MemPalaceResult]:
"""Search the palace for memories matching *query*.
Args:
query: Natural-language search string.
palace_path: Override the default palace path.
wing: Filter results to a specific wizard's wing.
room: Filter results to a specific room (e.g. ``"forge"``).
n_results: Maximum number of results to return.
Returns:
List of :class:`MemPalaceResult`, best-match first.
Raises:
MemPalaceUnavailable: If ChromaDB is not installed or the palace
directory does not exist.
"""
path = palace_path or MEMPALACE_PATH
client = _get_client(path)
collection = client.get_or_create_collection(COLLECTION_NAME)
where: dict = {}
if wing:
where["wing"] = wing
if room:
where["room"] = room
kwargs: dict = {"query_texts": [query], "n_results": n_results}
if where:
kwargs["where"] = where
raw = collection.query(**kwargs)
results: list[MemPalaceResult] = []
if not raw or not raw.get("documents"):
return results
docs = raw["documents"][0]
metas = raw.get("metadatas", [[]])[0] or [{}] * len(docs)
distances = raw.get("distances", [[]])[0] or [0.0] * len(docs)
for doc, meta, dist in zip(docs, metas, distances):
results.append(
MemPalaceResult(
text=doc,
room=meta.get("room", "general"),
wing=meta.get("wing", ""),
score=float(1.0 - dist), # cosine similarity from distance
source_file=meta.get("source_file", ""),
metadata=meta,
)
)
return results
def search_fleet(
query: str,
*,
room: Optional[str] = None,
n_results: int = 10,
) -> list[MemPalaceResult]:
"""Search the shared fleet palace (closets only, no raw drawers).
Args:
query: Natural-language search string.
room: Optional room filter (e.g. ``"issues"``).
n_results: Maximum results.
Returns:
List of :class:`MemPalaceResult` from all wings.
"""
return search_memories(
query,
palace_path=FLEET_PALACE_PATH,
room=room,
n_results=n_results,
)
def add_memory(
text: str,
*,
room: str = "general",
wing: Optional[str] = None,
palace_path: Optional[Path] = None,
source_file: str = "",
extra_metadata: Optional[dict] = None,
) -> str:
"""Add a new memory drawer to the palace.
Args:
text: The memory text to store.
room: Target room (e.g. ``"hall_facts"``).
wing: Wing name; defaults to :data:`~nexus.mempalace.config.FLEET_WING`.
palace_path: Override the default palace path.
source_file: Optional source file attribution.
extra_metadata: Additional key/value metadata to store.
Returns:
The generated document ID.
Raises:
MemPalaceUnavailable: If ChromaDB is not installed or the palace
directory does not exist.
"""
import uuid
from nexus.mempalace.config import FLEET_WING
path = palace_path or MEMPALACE_PATH
client = _get_client(path)
collection = client.get_or_create_collection(COLLECTION_NAME)
doc_id = str(uuid.uuid4())
metadata: dict = {
"room": room,
"wing": wing or FLEET_WING,
"source_file": source_file,
}
if extra_metadata:
metadata.update(extra_metadata)
collection.add(
documents=[text],
metadatas=[metadata],
ids=[doc_id],
)
return doc_id

View File

@@ -0,0 +1,62 @@
# MemPalace Field Report
**Wizard:** Bezalel
**Date:** 2026-04-07
**Scope:** Forge / testbed evaluation of https://github.com/milla-jovovich/mempalace
**Issue:** #1072
---
## What I Did
1. **Cloned and audited the upstream repo** — ~13.4k LOC dumped in a single commit (co-authored by `Claude Opus 4.6`). Zero development history, 5,769 GitHub stars in 48 hours, and no real test coverage (~125 lines of tests). Verdict: astroturfed hype, but the underlying code is not malicious.
2. **Ran the benchmark runners** — The "96.6% raw LongMemEval R@5" claim is technically runnable; it indexes user turns only, which is a questionable methodology, but the scripts are reproducible.
3. **Operationalized it for Bezalel** — I configured a local palace and mined my actual workspace into it.
---
## Method
- **Palace path:** `/root/wizards/bezalel/.mempalace/palace/` (ChromaDB persistent store)
- **Room taxonomy:** Defined in `/root/wizards/bezalel/mempalace.yaml`
- `forge` — CI, builds, health checks, nightly watch
- `hermes` — Hermes agent source, gateway, CLI
- `evennia` — MUD engine and world-building code
- `workspace` — Active prototypes, experiments, scratch code
- `home` — Personal scripts, configs, utilities
- `nexus` — Reports, docs, KT artifacts
- `issues` — Gitea issues, PRs, backlog tracking
- `general` — Catch-all
- **Ingestion:** Used the upstream `mempalace mine` pipeline against `/root/wizards/bezalel/`.
- **Outcome:** **823 artifacts filed** into the palace with wing (`bezalel`), room, source file path, and timestamp metadata.
---
## Effectiveness
| Before | After |
|--------|-------|
| Context evaporates between sessions; I rely on grep and shell history. | I can query the palace for "auth migration decisions" or "nightly watch failures" and get semantically ranked results from my own files. |
| No structured memory across wizard restarts. | 823 workspace artifacts are now vector-searchable with room-level filtering. |
The upstream repo is **vaporware marketing**, but the local-first pattern is **genuinely useful**. Verbatim storage + ChromaDB embeddings is fast, offline, and good enough for operational memory.
---
## Recommendations for the Fleet
@Timmy @Allegro @Ezra @bezalel
Each wizard should:
1. Clone the repo locally (do not trust it as a dependency; vendor it).
2. Write a `mempalace.yaml` that matches your domain rooms.
3. Run `mempalace mine <your_workspace>` to populate your palace.
4. Wire the searcher into your heartbeat scripts or MCP server for on-demand recall.
An incremental re-mine helper script has been added at `scripts/mempalace-incremental-mine.sh` to keep the palace current without full re-ingestion.
---
## Status
**Next action:** Awaiting sovereign or council signal on whether to standardize this across the fleet or keep it wizard-opt-in.

View File

@@ -0,0 +1,96 @@
#!/usr/bin/env bash
# mempalace-incremental-mine.sh
# Re-mines only files changed since the last run, keeping the palace current
# without full re-ingestion.
#
# Usage:
# ./scripts/mempalace-incremental-mine.sh [workspace_dir] [palace_dir]
#
# Defaults:
# workspace_dir — $WIZARD_HOME or /root/wizards/bezalel
# palace_dir — $MEMPALACE_DIR or <workspace_dir>/.mempalace
#
# Dependencies: mempalace (vendored), find
#
# Refs: #1072 (Bezalel MemPalace Field Report)
set -euo pipefail
WORKSPACE="${1:-${WIZARD_HOME:-/root/wizards/bezalel}}"
PALACE_DIR="${2:-${MEMPALACE_DIR:-$WORKSPACE/.mempalace}}"
STAMP_FILE="$PALACE_DIR/.last_mine_ts"
PALACE_PATH="$PALACE_DIR/palace"
if [[ ! -d "$WORKSPACE" ]]; then
echo "[mempalace-incremental-mine] ERROR: workspace not found: $WORKSPACE" >&2
exit 1
fi
# Resolve mempalace binary — check vendored location first
MEMPALACE_BIN=""
for candidate in \
"$WORKSPACE/.vendor/mempalace/mempalace" \
"$WORKSPACE/.vendor/mempalace/bin/mempalace" \
"$(command -v mempalace 2>/dev/null || true)"; do
if [[ -x "$candidate" ]]; then
MEMPALACE_BIN="$candidate"
break
fi
done
if [[ -z "$MEMPALACE_BIN" ]]; then
echo "[mempalace-incremental-mine] ERROR: mempalace binary not found." >&2
echo " Vendor it at $WORKSPACE/.vendor/mempalace/ or install it globally." >&2
exit 1
fi
mkdir -p "$PALACE_DIR"
# Determine changed files since last run
if [[ -f "$STAMP_FILE" ]]; then
SINCE=$(cat "$STAMP_FILE")
echo "[mempalace-incremental-mine] Mining files changed since $SINCE"
# Find files newer than the stamp file itself
CHANGED_FILES=$(find "$WORKSPACE" \
-newer "$STAMP_FILE" \
-type f \
! -path "*/.mempalace/*" \
! -path "*/.git/*" \
! -path "*/node_modules/*" \
! -path "*/__pycache__/*" \
! -name "*.pyc" \
2>/dev/null || true)
else
echo "[mempalace-incremental-mine] No prior stamp found — running full mine."
CHANGED_FILES=""
fi
if [[ -z "$CHANGED_FILES" && -f "$STAMP_FILE" ]]; then
echo "[mempalace-incremental-mine] No changed files detected. Palace is current."
exit 0
fi
YAML_CONFIG="$WORKSPACE/mempalace.yaml"
if [[ ! -f "$YAML_CONFIG" ]]; then
echo "[mempalace-incremental-mine] WARNING: $YAML_CONFIG not found." >&2
echo " Room taxonomy will not be applied. Create mempalace.yaml to enable routing." >&2
YAML_ARGS=()
else
YAML_ARGS=(--config "$YAML_CONFIG")
fi
if [[ -n "$CHANGED_FILES" ]]; then
# Mine only the changed files
FILE_COUNT=$(echo "$CHANGED_FILES" | wc -l | tr -d ' ')
echo "[mempalace-incremental-mine] Mining $FILE_COUNT changed file(s)..."
echo "$CHANGED_FILES" | xargs -I{} "$MEMPALACE_BIN" mine "${YAML_ARGS[@]}" \
--palace "$PALACE_PATH" {} 2>&1
else
# Full mine (first run)
echo "[mempalace-incremental-mine] Running full mine of $WORKSPACE ..."
"$MEMPALACE_BIN" mine "${YAML_ARGS[@]}" --palace "$PALACE_PATH" "$WORKSPACE" 2>&1
fi
# Update stamp
date -u +"%Y-%m-%dT%H:%M:%SZ" > "$STAMP_FILE"
echo "[mempalace-incremental-mine] Done. Stamp updated: $(cat "$STAMP_FILE")"

View File

@@ -0,0 +1,244 @@
"""Tests for nexus.evennia_mempalace commands and NPC helpers."""
from __future__ import annotations
from unittest.mock import MagicMock, patch
import pytest
from nexus.evennia_mempalace.commands.recall import CmdRecall, CmdEnterRoom, _closest_room
from nexus.evennia_mempalace.commands.write import CmdRecord, CmdNote, CmdEvent
from nexus.evennia_mempalace.typeclasses.npcs import StewardNPC, _extract_topic
from nexus.mempalace.searcher import MemPalaceResult, MemPalaceUnavailable
# ── Helpers ──────────────────────────────────────────────────────────────────
def _make_caller(wing: str = "bezalel"):
"""Build a minimal mock Evennia caller."""
caller = MagicMock()
caller.db = MagicMock()
caller.db.wing = wing
caller.account = MagicMock()
return caller
def _make_cmd(cls, args: str = "", switches: list | None = None, wing: str = "bezalel"):
"""Instantiate an Evennia command mock and wire it up."""
cmd = cls()
cmd.caller = _make_caller(wing)
cmd.args = args
cmd.switches = switches or []
return cmd
# ── CmdRecall ────────────────────────────────────────────────────────────────
def test_recall_no_args_shows_usage():
cmd = _make_cmd(CmdRecall, args="")
cmd.func()
cmd.caller.msg.assert_called_once()
assert "Usage" in cmd.caller.msg.call_args[0][0]
def test_recall_calls_search_memories():
results = [
MemPalaceResult(text="CI pipeline failed", room="forge", wing="bezalel", score=0.9)
]
with patch("nexus.evennia_mempalace.commands.recall.search_memories", return_value=results):
cmd = _make_cmd(CmdRecall, args="CI failures")
cmd.func()
calls = [c[0][0] for c in cmd.caller.msg.call_args_list]
assert any("CI pipeline failed" in c for c in calls)
def test_recall_fleet_flag_calls_search_fleet():
results = [
MemPalaceResult(text="Fleet doc", room="nexus", wing="timmy", score=0.8)
]
with patch("nexus.evennia_mempalace.commands.recall.search_fleet", return_value=results) as mock_fleet:
cmd = _make_cmd(CmdRecall, args="architecture --fleet", switches=["--fleet"])
cmd.func()
mock_fleet.assert_called_once()
query_arg = mock_fleet.call_args[0][0]
assert "--fleet" not in query_arg
assert "architecture" in query_arg
def test_recall_unavailable_shows_error():
with patch(
"nexus.evennia_mempalace.commands.recall.search_memories",
side_effect=MemPalaceUnavailable("ChromaDB not installed"),
):
cmd = _make_cmd(CmdRecall, args="anything")
cmd.func()
msg = cmd.caller.msg.call_args[0][0]
assert "unavailable" in msg.lower()
def test_recall_no_results_shows_no_memories():
with patch("nexus.evennia_mempalace.commands.recall.search_memories", return_value=[]):
cmd = _make_cmd(CmdRecall, args="obscure query")
cmd.func()
calls = " ".join(c[0][0] for c in cmd.caller.msg.call_args_list)
assert "No memories" in calls
# ── _closest_room ─────────────────────────────────────────────────────────────
@pytest.mark.parametrize("topic,expected", [
("forge", "forge"),
("CI pipeline", "forge"),
("hermes agent", "hermes"),
("nexus report", "nexus"),
("issue triage", "issues"),
("spike experiment", "experiments"),
("totally unknown topic xyz", "general"),
])
def test_closest_room(topic, expected):
assert _closest_room(topic) == expected
# ── CmdEnterRoom ──────────────────────────────────────────────────────────────
def test_enter_room_no_args_shows_usage():
cmd = _make_cmd(CmdEnterRoom, args="")
cmd.func()
output = " ".join(c[0][0] for c in cmd.caller.msg.call_args_list)
assert "Usage" in output
assert "forge" in output # shows core rooms
def test_enter_room_exact_match_no_room_found():
"""When an exact room name is given but no room exists, show a help message."""
# evennia.utils.search raises Django config errors outside a live server;
# CmdEnterRoom catches all exceptions and falls back to a help message.
cmd = _make_cmd(CmdEnterRoom, args="forge")
cmd.func()
assert cmd.caller.msg.called
output = " ".join(c[0][0] for c in cmd.caller.msg.call_args_list)
# Should mention the room name or MemPalaceRoom typeclass
assert "forge" in output or "MemPalaceRoom" in output or "No palace room" in output
# ── Write commands ───────────────────────────────────────────────────────────
def test_record_no_args_shows_usage():
cmd = _make_cmd(CmdRecord, args="")
cmd.func()
assert "Usage" in cmd.caller.msg.call_args[0][0]
def test_record_calls_add_memory():
with patch("nexus.evennia_mempalace.commands.write.add_memory", return_value="fake-uuid-1234-5678-abcd") as mock_add:
cmd = _make_cmd(CmdRecord, args="Use ChromaDB for storage.")
cmd.func()
mock_add.assert_called_once()
kwargs = mock_add.call_args[1]
assert kwargs["room"] == "hall_facts"
assert "ChromaDB" in mock_add.call_args[0][0]
def test_note_files_to_hall_discoveries():
with patch("nexus.evennia_mempalace.commands.write.add_memory", return_value="uuid") as mock_add:
cmd = _make_cmd(CmdNote, args="AAAK reduces cost by 40%.")
cmd.func()
assert mock_add.call_args[1]["room"] == "hall_discoveries"
def test_event_files_to_hall_events():
with patch("nexus.evennia_mempalace.commands.write.add_memory", return_value="uuid") as mock_add:
cmd = _make_cmd(CmdEvent, args="Deployed Evennia bridge to Alpha.")
cmd.func()
assert mock_add.call_args[1]["room"] == "hall_events"
def test_write_command_unavailable_shows_error():
with patch(
"nexus.evennia_mempalace.commands.write.add_memory",
side_effect=MemPalaceUnavailable("no palace"),
):
cmd = _make_cmd(CmdRecord, args="some text")
cmd.func()
msg = cmd.caller.msg.call_args[0][0]
assert "unavailable" in msg.lower()
# ── _extract_topic ────────────────────────────────────────────────────────────
@pytest.mark.parametrize("question,expected_substring", [
("about nightly watch failures", "nightly watch failures"),
("what do you know about CI pipeline?", "CI pipeline"),
("tell me about hermes", "hermes"),
("regarding the forge build", "forge build"),
("nightly watch failures", "nightly watch failures"),
])
def test_extract_topic(question, expected_substring):
result = _extract_topic(question)
assert expected_substring.lower() in result.lower()
# ── StewardNPC.respond_to_question ───────────────────────────────────────────
def test_steward_responds_with_results():
npc = StewardNPC()
npc.db = MagicMock()
npc.db.steward_wing = "bezalel"
npc.db.steward_name = "Bezalel-Steward"
npc.db.steward_n_results = 3
npc.key = "steward"
results = [
MemPalaceResult(text="Three failures last week.", room="forge", wing="bezalel", score=0.95)
]
with patch("nexus.evennia_mempalace.typeclasses.npcs.search_memories", return_value=results):
response = npc.respond_to_question("about nightly watch failures")
assert "Bezalel-Steward" in response
assert "Three failures" in response
def test_steward_responds_not_found():
npc = StewardNPC()
npc.db = MagicMock()
npc.db.steward_wing = "bezalel"
npc.db.steward_name = "Steward"
npc.db.steward_n_results = 3
npc.key = "steward"
with patch("nexus.evennia_mempalace.typeclasses.npcs.search_memories", return_value=[]):
response = npc.respond_to_question("about unknown_topic_xyz")
assert "nothing" in response.lower() or "found" in response.lower()
def test_steward_responds_unavailable():
npc = StewardNPC()
npc.db = MagicMock()
npc.db.steward_wing = "bezalel"
npc.db.steward_name = "Steward"
npc.db.steward_n_results = 3
npc.key = "steward"
with patch(
"nexus.evennia_mempalace.typeclasses.npcs.search_memories",
side_effect=MemPalaceUnavailable("no palace"),
):
response = npc.respond_to_question("about anything")
assert "unreachable" in response.lower()

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,190 @@
"""Tests for nexus.mempalace.searcher and nexus.mempalace.config."""
from __future__ import annotations
import os
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from nexus.mempalace.config import CORE_ROOMS, MEMPALACE_PATH, COLLECTION_NAME
from nexus.mempalace.searcher import (
MemPalaceResult,
MemPalaceUnavailable,
_get_client,
search_memories,
add_memory,
)
# ── MemPalaceResult ──────────────────────────────────────────────────────────
def test_result_short_truncates():
r = MemPalaceResult(text="x" * 300, room="forge", wing="bezalel")
short = r.short(200)
assert len(short) <= 204 # 200 + ellipsis
assert short.endswith("")
def test_result_short_no_truncation_needed():
r = MemPalaceResult(text="hello", room="nexus", wing="bezalel")
assert r.short() == "hello"
def test_result_defaults():
r = MemPalaceResult(text="test", room="general", wing="")
assert r.score == 0.0
assert r.source_file == ""
assert r.metadata == {}
# ── Config ───────────────────────────────────────────────────────────────────
def test_core_rooms_contains_required_rooms():
required = {"forge", "hermes", "nexus", "issues", "experiments"}
assert required.issubset(set(CORE_ROOMS))
def test_mempalace_path_env_override(monkeypatch, tmp_path):
monkeypatch.setenv("MEMPALACE_PATH", str(tmp_path))
# Re-import to pick up env var (config reads at import time so we patch)
import importlib
import nexus.mempalace.config as cfg
importlib.reload(cfg)
assert Path(os.environ["MEMPALACE_PATH"]) == tmp_path
importlib.reload(cfg) # restore
# ── _get_client ──────────────────────────────────────────────────────────────
def test_get_client_raises_when_chromadb_missing(tmp_path):
with patch.dict("sys.modules", {"chromadb": None}):
with pytest.raises(MemPalaceUnavailable, match="ChromaDB"):
_get_client(tmp_path)
def test_get_client_raises_when_path_missing(tmp_path):
missing = tmp_path / "nonexistent_palace"
# chromadb importable but path missing
mock_chroma = MagicMock()
with patch.dict("sys.modules", {"chromadb": mock_chroma}):
with pytest.raises(MemPalaceUnavailable, match="Palace directory"):
_get_client(missing)
# ── search_memories ──────────────────────────────────────────────────────────
def _make_mock_collection(docs, metas=None, distances=None):
"""Build a mock ChromaDB collection that returns canned results."""
if metas is None:
metas = [{"room": "forge", "wing": "bezalel", "source_file": ""} for _ in docs]
if distances is None:
distances = [0.1 * i for i in range(len(docs))]
collection = MagicMock()
collection.query.return_value = {
"documents": [docs],
"metadatas": [metas],
"distances": [distances],
}
return collection
def _mock_chroma_client(collection):
client = MagicMock()
client.get_or_create_collection.return_value = collection
return client
def test_search_memories_returns_results(tmp_path):
docs = ["CI pipeline failed on main", "Forge build log 2026-04-01"]
collection = _make_mock_collection(docs)
mock_chroma = MagicMock()
mock_chroma.PersistentClient.return_value = _mock_chroma_client(collection)
with patch.dict("sys.modules", {"chromadb": mock_chroma}):
# Palace path must exist for _get_client check
(tmp_path / "chroma.sqlite3").touch()
results = search_memories("CI failures", palace_path=tmp_path)
assert len(results) == 2
assert results[0].room == "forge"
assert results[0].wing == "bezalel"
assert "CI pipeline" in results[0].text
def test_search_memories_empty_collection(tmp_path):
collection = MagicMock()
collection.query.return_value = {"documents": [[]], "metadatas": [[]], "distances": [[]]}
mock_chroma = MagicMock()
mock_chroma.PersistentClient.return_value = _mock_chroma_client(collection)
with patch.dict("sys.modules", {"chromadb": mock_chroma}):
(tmp_path / "chroma.sqlite3").touch()
results = search_memories("anything", palace_path=tmp_path)
assert results == []
def test_search_memories_with_wing_filter(tmp_path):
docs = ["test doc"]
collection = _make_mock_collection(docs)
mock_chroma = MagicMock()
mock_chroma.PersistentClient.return_value = _mock_chroma_client(collection)
with patch.dict("sys.modules", {"chromadb": mock_chroma}):
(tmp_path / "chroma.sqlite3").touch()
search_memories("query", palace_path=tmp_path, wing="bezalel")
call_kwargs = collection.query.call_args[1]
assert call_kwargs["where"] == {"wing": "bezalel"}
def test_search_memories_with_room_filter(tmp_path):
collection = _make_mock_collection(["doc"])
mock_chroma = MagicMock()
mock_chroma.PersistentClient.return_value = _mock_chroma_client(collection)
with patch.dict("sys.modules", {"chromadb": mock_chroma}):
(tmp_path / "chroma.sqlite3").touch()
search_memories("query", palace_path=tmp_path, room="forge")
call_kwargs = collection.query.call_args[1]
assert call_kwargs["where"] == {"room": "forge"}
def test_search_memories_unavailable(tmp_path):
with patch.dict("sys.modules", {"chromadb": None}):
with pytest.raises(MemPalaceUnavailable):
search_memories("anything", palace_path=tmp_path)
# ── add_memory ───────────────────────────────────────────────────────────────
def test_add_memory_returns_id(tmp_path):
collection = MagicMock()
mock_chroma = MagicMock()
mock_chroma.PersistentClient.return_value = _mock_chroma_client(collection)
with patch.dict("sys.modules", {"chromadb": mock_chroma}):
(tmp_path / "chroma.sqlite3").touch()
doc_id = add_memory(
"We decided to use ChromaDB.",
room="hall_facts",
wing="bezalel",
palace_path=tmp_path,
)
assert isinstance(doc_id, str)
assert len(doc_id) == 36 # UUID format
collection.add.assert_called_once()
call_kwargs = collection.add.call_args[1]
assert call_kwargs["documents"] == ["We decided to use ChromaDB."]
assert call_kwargs["metadatas"][0]["room"] == "hall_facts"
assert call_kwargs["metadatas"][0]["wing"] == "bezalel"

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)