[claude] MemPalace × Evennia fleet memory scaffold (#1075) #1088
1
.gitignore
vendored
1
.gitignore
vendored
@@ -2,4 +2,5 @@ node_modules/
|
||||
test-results/
|
||||
nexus/__pycache__/
|
||||
tests/__pycache__/
|
||||
mempalace/__pycache__/
|
||||
.aider*
|
||||
|
||||
183
docs/mempalace/rooms.yaml
Normal file
183
docs/mempalace/rooms.yaml
Normal 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
5
mempalace/__init__.py
Normal 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
177
mempalace/audit_privacy.py
Normal 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())
|
||||
22
mempalace/evennia_mempalace/__init__.py
Normal file
22
mempalace/evennia_mempalace/__init__.py
Normal 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"
|
||||
14
mempalace/evennia_mempalace/commands/__init__.py
Normal file
14
mempalace/evennia_mempalace/commands/__init__.py
Normal 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"]
|
||||
86
mempalace/evennia_mempalace/commands/enter_room.py
Normal file
86
mempalace/evennia_mempalace/commands/enter_room.py
Normal 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)
|
||||
93
mempalace/evennia_mempalace/commands/recall.py
Normal file
93
mempalace/evennia_mempalace/commands/recall.py
Normal 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))
|
||||
106
mempalace/evennia_mempalace/searcher.py
Normal file
106
mempalace/evennia_mempalace/searcher.py
Normal 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 []
|
||||
64
mempalace/evennia_mempalace/settings.py
Normal file
64
mempalace/evennia_mempalace/settings.py
Normal 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"
|
||||
1
mempalace/evennia_mempalace/typeclasses/__init__.py
Normal file
1
mempalace/evennia_mempalace/typeclasses/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""evennia_mempalace.typeclasses — Evennia typeclasses for palace rooms."""
|
||||
87
mempalace/evennia_mempalace/typeclasses/room.py
Normal file
87
mempalace/evennia_mempalace/typeclasses/room.py
Normal 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
104
mempalace/export_closets.sh
Executable 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
114
mempalace/rooms.yaml
Normal 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
119
mempalace/validate_rooms.py
Normal 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())
|
||||
49
nexus/evennia_mempalace/__init__.py
Normal file
49
nexus/evennia_mempalace/__init__.py
Normal 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",
|
||||
]
|
||||
BIN
nexus/evennia_mempalace/__pycache__/__init__.cpython-312.pyc
Normal file
BIN
nexus/evennia_mempalace/__pycache__/__init__.cpython-312.pyc
Normal file
Binary file not shown.
14
nexus/evennia_mempalace/commands/__init__.py
Normal file
14
nexus/evennia_mempalace/commands/__init__.py
Normal 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",
|
||||
]
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
206
nexus/evennia_mempalace/commands/recall.py
Normal file
206
nexus/evennia_mempalace/commands/recall.py
Normal 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"
|
||||
124
nexus/evennia_mempalace/commands/write.py
Normal file
124
nexus/evennia_mempalace/commands/write.py
Normal 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"
|
||||
1
nexus/evennia_mempalace/typeclasses/__init__.py
Normal file
1
nexus/evennia_mempalace/typeclasses/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""MemPalace Evennia typeclasses."""
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
138
nexus/evennia_mempalace/typeclasses/npcs.py
Normal file
138
nexus/evennia_mempalace/typeclasses/npcs.py
Normal 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()
|
||||
99
nexus/evennia_mempalace/typeclasses/rooms.py
Normal file
99
nexus/evennia_mempalace/typeclasses/rooms.py
Normal 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)
|
||||
23
nexus/mempalace/__init__.py
Normal file
23
nexus/mempalace/__init__.py
Normal 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",
|
||||
]
|
||||
BIN
nexus/mempalace/__pycache__/__init__.cpython-312.pyc
Normal file
BIN
nexus/mempalace/__pycache__/__init__.cpython-312.pyc
Normal file
Binary file not shown.
BIN
nexus/mempalace/__pycache__/config.cpython-312.pyc
Normal file
BIN
nexus/mempalace/__pycache__/config.cpython-312.pyc
Normal file
Binary file not shown.
BIN
nexus/mempalace/__pycache__/searcher.cpython-312.pyc
Normal file
BIN
nexus/mempalace/__pycache__/searcher.cpython-312.pyc
Normal file
Binary file not shown.
46
nexus/mempalace/config.py
Normal file
46
nexus/mempalace/config.py
Normal 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
200
nexus/mempalace/searcher.py
Normal 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
|
||||
62
reports/bezalel/2026-04-07-mempalace-field-report.md
Normal file
62
reports/bezalel/2026-04-07-mempalace-field-report.md
Normal 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.
|
||||
96
scripts/mempalace-incremental-mine.sh
Executable file
96
scripts/mempalace-incremental-mine.sh
Executable 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")"
|
||||
244
tests/test_evennia_mempalace_commands.py
Normal file
244
tests/test_evennia_mempalace_commands.py
Normal 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()
|
||||
129
tests/test_mempalace_audit_privacy.py
Normal file
129
tests/test_mempalace_audit_privacy.py
Normal 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)
|
||||
190
tests/test_mempalace_searcher.py
Normal file
190
tests/test_mempalace_searcher.py
Normal 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"
|
||||
160
tests/test_mempalace_validate_rooms.py
Normal file
160
tests/test_mempalace_validate_rooms.py
Normal 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)
|
||||
Reference in New Issue
Block a user