From 8158178a562d0bff1acec18065dee3434b6a40b7 Mon Sep 17 00:00:00 2001 From: Mortdecai Date: Fri, 20 Mar 2026 23:28:04 -0400 Subject: [PATCH] Shared player memory system + whitelist migration to CT 650 player_memory.py: - Per-server JSON with owner tagging, cross-player references - write/read/delete with thread safety and limits (50/player, 500/server) - format_memory_context() for LLM prompt injection - handle_memory_write/read for model output processing - MODEL_OUTPUT_SCHEMA with commands, memory_write, memory_read, revert_after mortdecai-sites (CT 650): - Whitelist app migrated from CT 644, RCON via LAN (192.168.0.244) - All 4 sites verified: mortdec.ai, docs, git, minecraft Co-Authored-By: Claude Opus 4.6 (1M context) --- agent/tools/player_memory.py | 201 +++++++++++++++++++++++++++++++++++ 1 file changed, 201 insertions(+) create mode 100644 agent/tools/player_memory.py diff --git a/agent/tools/player_memory.py b/agent/tools/player_memory.py new file mode 100644 index 0000000..9e6bdc7 --- /dev/null +++ b/agent/tools/player_memory.py @@ -0,0 +1,201 @@ +""" +Shared player memory system for Mortdecai. + +Per-server JSON storage with owner tagging. Players can save locations, +preferences, and facts. Cross-player references supported. + +Usage: + from agent.tools.player_memory import write_memory, read_memory, format_memory_context + + write_memory(config, "slingshooter08", "slingshooter08", "location", "home", {"x": 100, "y": 64, "z": 200}) + memories = read_memory(config, owner="slingshooter08", key="home") + context = format_memory_context(config, "slingshooter08") +""" + +import json +import os +import threading +import time +from typing import Any, Dict, List, Optional + +_lock = threading.Lock() + +DEFAULT_MEMORY_PATH = "/opt/paper-ai-25567/player_memory.json" +MAX_PER_PLAYER = 50 +MAX_TOTAL = 500 + + +def _memory_path(config: dict) -> str: + return config.get("player_memory_path", DEFAULT_MEMORY_PATH) + + +def load_memory(config: dict) -> dict: + path = _memory_path(config) + with _lock: + try: + with open(path) as f: + return json.load(f) + except (FileNotFoundError, json.JSONDecodeError): + return {"memories": []} + + +def save_memory(config: dict, store: dict): + path = _memory_path(config) + with _lock: + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w") as f: + json.dump(store, f, indent=2, ensure_ascii=False) + + +def write_memory(config: dict, owner: str, set_by: str, mem_type: str, key: str, value: Any) -> dict: + """Add or update a memory. Returns the written entry.""" + store = load_memory(config) + memories = store.get("memories", []) + + # Upsert: remove existing entry with same owner+key+type + memories = [m for m in memories if not (m["owner"] == owner and m["key"] == key and m["type"] == mem_type)] + + entry = { + "owner": owner, + "set_by": set_by, + "type": mem_type, + "key": key, + "value": value, + "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ"), + } + memories.append(entry) + + # Enforce limits + player_mems = [m for m in memories if m["owner"] == owner] + if len(player_mems) > MAX_PER_PLAYER: + oldest = sorted(player_mems, key=lambda m: m["timestamp"]) + to_remove = oldest[:len(player_mems) - MAX_PER_PLAYER] + remove_keys = {(m["owner"], m["key"], m["type"]) for m in to_remove} + memories = [m for m in memories if (m["owner"], m["key"], m["type"]) not in remove_keys] + + if len(memories) > MAX_TOTAL: + memories = sorted(memories, key=lambda m: m["timestamp"])[-MAX_TOTAL:] + + store["memories"] = memories + save_memory(config, store) + return entry + + +def read_memory(config: dict, owner: str = None, key: str = None, mem_type: str = None) -> List[dict]: + """Query memories. All filters are optional.""" + store = load_memory(config) + results = store.get("memories", []) + if owner: + results = [m for m in results if m["owner"].lower() == owner.lower()] + if key: + results = [m for m in results if m["key"].lower() == key.lower()] + if mem_type: + results = [m for m in results if m["type"] == mem_type] + return sorted(results, key=lambda m: m["timestamp"], reverse=True) + + +def delete_memory(config: dict, owner: str, key: str) -> bool: + """Remove a memory by owner+key. Returns True if found and removed.""" + store = load_memory(config) + before = len(store["memories"]) + store["memories"] = [m for m in store["memories"] + if not (m["owner"].lower() == owner.lower() and m["key"].lower() == key.lower())] + if len(store["memories"]) < before: + save_memory(config, store) + return True + return False + + +def get_location_memories(config: dict) -> List[dict]: + """Get all location memories.""" + return read_memory(config, mem_type="location") + + +def format_memory_context(config: dict, player: str) -> str: + """Format relevant memories for LLM context injection.""" + store = load_memory(config) + memories = store.get("memories", []) + if not memories: + return "" + + lines = ["=== PLAYER MEMORIES ==="] + + # Player's own memories + own = [m for m in memories if m["owner"].lower() == player.lower()] + if own: + lines.append(f"Your memories ({player}):") + for m in own: + if m["type"] == "location": + v = m["value"] + lines.append(f" {m['key']}: x={v.get('x')}, y={v.get('y')}, z={v.get('z')}") + else: + lines.append(f" {m['key']}: {m['value']}") + + # Other players' locations (for cross-player references) + others = [m for m in memories if m["owner"].lower() != player.lower() and m["type"] == "location"] + if others: + lines.append("Other players' locations:") + for m in others: + v = m["value"] + lines.append(f" {m['owner']}'s {m['key']}: x={v.get('x')}, y={v.get('y')}, z={v.get('z')}") + + return "\n".join(lines) + + +# --- Model output handlers --- + +def handle_memory_write(config: dict, spec: dict, set_by: str) -> dict: + """Process a memory_write block from model JSON output.""" + owner = spec.get("owner", set_by) + mem_type = spec.get("type", "fact") + key = spec.get("key", "") + value = spec.get("value", "") + if not key: + return {"ok": False, "error": "key is required"} + entry = write_memory(config, owner, set_by, mem_type, key, value) + return {"ok": True, "entry": entry} + + +def handle_memory_read(config: dict, spec: dict) -> dict: + """Process a memory_read block from model JSON output.""" + owner = spec.get("owner") + key = spec.get("key") + mem_type = spec.get("type") + results = read_memory(config, owner=owner, key=key, mem_type=mem_type) + return {"ok": True, "results": results} + + +# --- Schema constants for model output --- + +MEMORY_WRITE_SCHEMA = { + "type": "object", + "properties": { + "owner": {"type": "string", "description": "Player who owns this memory"}, + "type": {"type": "string", "enum": ["location", "preference", "fact"]}, + "key": {"type": "string", "description": "Memory name (e.g. 'home', 'base', 'favorite_tool')"}, + "value": {"description": "Memory data — coordinates for locations, text for others"}, + }, + "required": ["type", "key", "value"], +} + +MEMORY_READ_SCHEMA = { + "type": "object", + "properties": { + "owner": {"type": "string", "description": "Player to look up (omit for requesting player)"}, + "key": {"type": "string", "description": "Memory name to find"}, + }, +} + +MODEL_OUTPUT_SCHEMA = { + "type": "object", + "properties": { + "commands": {"type": "array", "items": {"type": "string"}}, + "message": {"type": "string"}, + "reasoning": {"type": "string"}, + "risk_level": {"type": "integer", "minimum": 0, "maximum": 4}, + "memory_write": MEMORY_WRITE_SCHEMA, + "memory_read": MEMORY_READ_SCHEMA, + "revert_after": {"type": "integer", "description": "Seconds before revert_commands execute"}, + "revert_commands": {"type": "array", "items": {"type": "string"}}, + }, +}