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) <noreply@anthropic.com>
This commit is contained in:
Mortdecai
@@ -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"}},
|
||||
},
|
||||
}
|
||||
Reference in New Issue
Block a user