jason/mempalace
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: new MCP tools — get/list/update drawer, hook settings, export (resolves #635) (#667)

Browse Source 
* feat: MCP reliability — inode detection, WAL rotation, metadata cache, search limits

Infrastructure hardening for the MCP server:
- Detect palace DB replacement via inode tracking (repair command support)
- WAL rotation to prevent unbounded WAL growth
- _fetch_all_metadata() + _get_cached_metadata() with 60s TTL for taxonomy/status
- _MAX_RESULTS cap (100) with limit clamping [1, _MAX_RESULTS]
- max_distance parameter for similarity threshold in search
- Handle all notifications/* methods, null arguments, method=None
- Remove duplicate _client_cache = None declarations
- searcher.py max_distance parameter passthrough

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* feat: new MCP tools (get/list/update drawer, hook settings, memories filed), export, normalize

New MCP tools:
- mempalace_get_drawer: fetch single drawer by ID with full content
- mempalace_list_drawers: paginated listing with wing/room filter
- mempalace_update_drawer: update content/wing/room on existing drawers
- mempalace_hook_settings: get/set hook behavior (silent_save, desktop_toast)
- mempalace_memories_filed_away: check latest checkpoint status

Also includes:
- exporter.py: export palace as browsable markdown files
- normalize.py: tool_use/tool_result capture for richer transcript mining
- layers.py: updated for new tool integration
- config.py: hook settings properties (hook_silent_save, hook_desktop_toast)

Depends on PR 3 (reliability) for _MAX_RESULTS, _metadata_cache, WAL logging.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: normalize.py handles string messages and Read offset type mismatch

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: params null guard, L2→cosine docs, empty tool_use_map key guard

- Handle explicit null in MCP params (request.get("params") or {})
- Fix search tool description: L2 → cosine distance (collection uses hnsw:space=cosine)
- Guard against empty string key in tool_use_map from malformed JSONL entries

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: rename ambiguous var 'l' to 'line' (E741 lint)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: address code review findings (5 issues)

1. min_similarity backwards-compat: convert similarity to distance scale
   (1.0 - similarity) instead of passing raw value as max_distance
2. Restore structured error reporting (error + partial fields) in
   tool_status, tool_list_wings, tool_list_rooms, tool_get_taxonomy
   — reverts silent except:pass that dropped #647 security hardening
3. inode cache: remove falsy-zero short-circuit so missing DB file
   triggers reconnect instead of reusing stale client
4. _fetch_all_metadata: check for empty batch before extending/advancing
   offset to prevent infinite loop on concurrent deletion
5. KG initialization: only override path when --palace is explicit;
   default runs use KnowledgeGraph's built-in default path

Co-authored-by: jphein <jphein@users.noreply.github.com>

---------

Co-authored-by: jp <jp@jphein.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: jphein <jphein@users.noreply.github.com>
This commit is contained in:
Ben Sigman
2026-04-11 21:25:04 -07:00
committed by GitHub
parent 58eca5075a
commit 20c8f8e57b
9 changed files with 1429 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mempalace/config.py
+21
View File
@@ -173,6 +173,27 @@ class MempalaceConfig:
"""Mapping of hall names to keyword lists."""
return self._file_config.get("hall_keywords", DEFAULT_HALL_KEYWORDS)
@property
def hook_silent_save(self):
"""Whether the stop hook saves directly (True) or blocks for MCP calls (False)."""
return self._file_config.get("hooks", {}).get("silent_save", True)
@property
def hook_desktop_toast(self):
"""Whether the stop hook shows a desktop notification via notify-send."""
return self._file_config.get("hooks", {}).get("desktop_toast", False)
def set_hook_setting(self, key: str, value: bool):
"""Update a hook setting and write config to disk."""
if "hooks" not in self._file_config:
self._file_config["hooks"] = {}
self._file_config["hooks"][key] = value
try:
with open(self._config_file, "w", encoding="utf-8") as f:
json.dump(self._file_config, f, indent=2, ensure_ascii=False)
except OSError:
pass
def init(self):
"""Create config directory and write default config.json if it doesn't exist."""
self._config_dir.mkdir(parents=True, exist_ok=True)
mempalace/exporter.py
+153
View File
@@ -0,0 +1,153 @@
"""
exporter.py — Export the palace as a browsable folder of markdown files.
Produces:
output_dir/
index.md — table of contents
wing_name/
room_name.md — one file per room, drawers as sections
Streams drawers in paginated batches so memory usage stays bounded
regardless of palace size.
"""
import os
import re
from collections import defaultdict
from datetime import datetime
from .palace import get_collection
def _safe_path_component(name: str) -> str:
"""Sanitize a string for use as a directory/file name component."""
name = re.sub(r'[/\\:*?"<>|]', '_', name)
name = name.strip('. ')
return name or 'unknown'
def export_palace(palace_path: str, output_dir: str, format: str = "markdown") -> dict:
"""Export all palace drawers as markdown files organized by wing/room.
Streams drawers in batches of 1000 and writes each wing/room file
incrementally, keeping memory usage proportional to batch size rather
than total palace size.
Args:
palace_path: Path to the ChromaDB palace directory.
output_dir: Where to write the exported markdown tree.
format: Output format (currently only "markdown").
Returns:
Stats dict: {"wings": N, "rooms": N, "drawers": N}
"""
col = get_collection(palace_path)
total = col.count()
if total == 0:
print(" Palace is empty — nothing to export.")
return {"wings": 0, "rooms": 0, "drawers": 0}
os.makedirs(output_dir, exist_ok=True)
# Track which room files have been opened (so we can append vs overwrite)
opened_rooms: set[tuple[str, str]] = set()
# Track stats per wing: {wing: {room: count}}
wing_stats: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))
total_drawers = 0
print(f" Streaming {total} drawers...")
offset = 0
while offset < total:
batch = col.get(limit=1000, offset=offset, include=["documents", "metadatas"])
if not batch["ids"]:
break
# Group this batch by wing/room so we do one file write per room per batch
batch_grouped: dict[str, dict[str, list]] = defaultdict(lambda: defaultdict(list))
for doc_id, doc, meta in zip(batch["ids"], batch["documents"], batch["metadatas"]):
wing = meta.get("wing", "unknown")
room = meta.get("room", "general")
batch_grouped[wing][room].append({
"id": doc_id,
"content": doc,
"source": meta.get("source_file", ""),
"filed_at": meta.get("filed_at", ""),
"added_by": meta.get("added_by", ""),
})
# Write/append each room file
for wing, rooms in batch_grouped.items():
safe_wing = _safe_path_component(wing)
wing_dir = os.path.join(output_dir, safe_wing)
os.makedirs(wing_dir, exist_ok=True)
for room, drawers in rooms.items():
safe_room = _safe_path_component(room)
room_path = os.path.join(wing_dir, f"{safe_room}.md")
key = (wing, room)
is_new = key not in opened_rooms
with open(room_path, "a" if not is_new else "w", encoding="utf-8") as f:
if is_new:
f.write(f"# {wing} / {room}\n\n")
opened_rooms.add(key)
for drawer in drawers:
source = drawer["source"] or "unknown"
filed = drawer["filed_at"] or "unknown"
added_by = drawer["added_by"] or "unknown"
f.write(
f"## {drawer['id']}\n"
f"\n"
f"> {_quote_content(drawer['content'])}\n"
f"\n"
f"| Field | Value |\n"
f"|-------|-------|\n"
f"| Source | {source} |\n"
f"| Filed | {filed} |\n"
f"| Added by | {added_by} |\n"
f"\n"
f"---\n\n"
)
wing_stats[wing][room] += len(drawers)
total_drawers += len(drawers)
offset += len(batch["ids"])
# Build and print stats
index_rows = []
for wing in sorted(wing_stats):
rooms = wing_stats[wing]
wing_drawer_count = sum(rooms.values())
index_rows.append((wing, len(rooms), wing_drawer_count))
print(f" {wing}: {len(rooms)} rooms, {wing_drawer_count} drawers")
# Write index.md
today = datetime.now().strftime("%Y-%m-%d")
index_lines = [
f"# Palace Export — {today}\n",
"",
"| Wing | Rooms | Drawers |",
"|------|-------|---------|",
]
for wing, room_count, drawer_count in index_rows:
index_lines.append(f"| [{wing}]({wing}/) | {room_count} | {drawer_count} |")
index_lines.append("")
index_path = os.path.join(output_dir, "index.md")
with open(index_path, "w", encoding="utf-8") as f:
f.write("\n".join(index_lines))
stats = {"wings": len(wing_stats), "rooms": sum(r for _, r, _ in index_rows), "drawers": total_drawers}
print(f"\n Exported {stats['drawers']} drawers across {stats['wings']} wings, {stats['rooms']} rooms")
print(f" Output: {output_dir}")
return stats
def _quote_content(text: str) -> str:
"""Format content for a markdown blockquote, handling multiline."""
lines = text.rstrip("\n").split("\n")
return "\n> ".join(lines)
mempalace/layers.py
+6 -22
View File
@@ -23,6 +23,7 @@ from collections import defaultdict
from .config import MempalaceConfig
from .palace import get_collection as _get_collection
from .searcher import build_where_filter
# ---------------------------------------------------------------------------
@@ -81,6 +82,7 @@ class Layer1:
MAX_DRAWERS = 15 # at most 15 moments in wake-up
MAX_CHARS = 3200 # hard cap on total L1 text (~800 tokens)
MAX_SCAN = 2000 # don't scan more than this for L1 generation
def __init__(self, palace_path: str = None, wing: str = None):
cfg = MempalaceConfig()
@@ -113,7 +115,7 @@ class Layer1:
docs.extend(batch_docs)
metas.extend(batch_metas)
offset += len(batch_docs)
if len(batch_docs) < _BATCH:
if len(batch_docs) < _BATCH or len(docs) >= self.MAX_SCAN:
break
if not docs:
@@ -198,13 +200,7 @@ class Layer2:
except Exception:
return "No palace found."
where = {}
if wing and room:
where = {"$and": [{"wing": wing}, {"room": room}]}
elif wing:
where = {"wing": wing}
elif room:
where = {"room": room}
where = build_where_filter(wing, room)
kwargs = {"include": ["documents", "metadatas"], "limit": n_results}
if where:
@@ -261,13 +257,7 @@ class Layer3:
except Exception:
return "No palace found."
where = {}
if wing and room:
where = {"$and": [{"wing": wing}, {"room": room}]}
elif wing:
where = {"wing": wing}
elif room:
where = {"room": room}
where = build_where_filter(wing, room)
kwargs = {
"query_texts": [query],
@@ -316,13 +306,7 @@ class Layer3:
except Exception:
return []
where = {}
if wing and room:
where = {"$and": [{"wing": wing}, {"room": room}]}
elif wing:
where = {"wing": wing}
elif room:
where = {"room": room}
where = build_where_filter(wing, room)
kwargs = {
"query_texts": [query],
mempalace/mcp_server.py
+427 -100
View File
@@ -23,12 +23,13 @@ import sys
import json
import logging
import hashlib
import time
from datetime import datetime
from pathlib import Path
from .config import MempalaceConfig, sanitize_name, sanitize_content
from .version import __version__
from .palace import get_collection as _get_collection_from_palace
import chromadb
from .query_sanitizer import sanitize_query
from .searcher import search_memories
from .palace_graph import traverse, find_tunnels, graph_stats
@@ -58,13 +59,17 @@ if _args.palace:
os.environ["MEMPALACE_PALACE_PATH"] = os.path.abspath(_args.palace)
_config = MempalaceConfig()
# Only override KG path when --palace is explicitly provided; otherwise use
# KnowledgeGraph's default (~/.mempalace/knowledge_graph.sqlite3).
if _args.palace:
_kg = KnowledgeGraph(db_path=os.path.join(_config.palace_path, "knowledge_graph.sqlite3"))
else:
_kg = KnowledgeGraph()
_client_cache = None
_collection_cache = None
_palace_db_inode = 0 # inode of chroma.sqlite3 at cache time
# ==================== WRITE-AHEAD LOG ====================
@@ -115,16 +120,43 @@ def _wal_log(operation: str, params: dict, result: dict = None):
logger.error(f"WAL write failed: {e}")
def _get_collection(create=False):
"""Return the configured collection, caching the wrapper between calls."""
global _collection_cache
def _get_client():
"""Return a ChromaDB PersistentClient, reconnecting if the database changed on disk.
Detects palace rebuilds (repair/nuke/purge) by checking the inode of
chroma.sqlite3. A full rebuild replaces the file, changing the inode.
"""
global _client_cache, _collection_cache, _palace_db_inode, _metadata_cache, _metadata_cache_time
db_path = os.path.join(_config.palace_path, "chroma.sqlite3")
try:
if create or _collection_cache is None:
_collection_cache = _get_collection_from_palace(
_config.palace_path,
collection_name=_config.collection_name,
create=create,
current_inode = os.stat(db_path).st_ino
except OSError:
current_inode = 0
if _client_cache is None or current_inode != _palace_db_inode:
_client_cache = chromadb.PersistentClient(path=_config.palace_path)
_collection_cache = None
_metadata_cache = None
_metadata_cache_time = 0
_palace_db_inode = current_inode
return _client_cache
def _get_collection(create=False):
"""Return the ChromaDB collection, caching the client between calls."""
global _collection_cache, _metadata_cache, _metadata_cache_time
try:
client = _get_client()
if create:
_collection_cache = client.get_or_create_collection(
_config.collection_name, metadata={"hnsw:space": "cosine"}
)
_metadata_cache = None
_metadata_cache_time = 0
elif _collection_cache is None:
_collection_cache = client.get_collection(_config.collection_name)
_metadata_cache = None
_metadata_cache_time = 0
return _collection_cache
except Exception:
return None
@@ -137,6 +169,49 @@ def _no_palace():
}
# ==================== HELPERS ====================
def _fetch_all_metadata(col, where=None):
"""Paginate col.get() to avoid the 10K silent truncation limit."""
total = col.count()
all_meta = []
offset = 0
while offset < total:
kwargs = {"include": ["metadatas"], "limit": 1000, "offset": offset}
if where:
kwargs["where"] = where
batch = col.get(**kwargs)
if not batch["metadatas"]:
break
all_meta.extend(batch["metadatas"])
offset += len(batch["metadatas"])
return all_meta
_metadata_cache = None
_metadata_cache_time = 0
_METADATA_CACHE_TTL = 5.0 # seconds
_MAX_RESULTS = 100 # upper bound for search/list limit params
def _get_cached_metadata(col, where=None):
"""Return cached metadata if fresh, else fetch and cache."""
global _metadata_cache, _metadata_cache_time
now = time.time()
if (
where is None
and _metadata_cache is not None
and (now - _metadata_cache_time) < _METADATA_CACHE_TTL
):
return _metadata_cache
result = _fetch_all_metadata(col, where=where)
if where is None:
_metadata_cache = result
_metadata_cache_time = now
return result
# ==================== READ TOOLS ====================
@@ -147,24 +222,6 @@ def tool_status():
count = col.count()
wings = {}
rooms = {}
batch_size = 5000
offset = 0
error_info = None
while True:
try:
batch = col.get(include=["metadatas"], limit=batch_size, offset=offset)
rows = batch["metadatas"]
for m in rows:
w = m.get("wing", "unknown")
r = m.get("room", "unknown")
wings[w] = wings.get(w, 0) + 1
rooms[r] = rooms.get(r, 0) + 1
offset += len(rows)
if len(rows) < batch_size:
break
except Exception as e:
error_info = f"Partial result, failed at offset {offset}: {str(e)}"
break
result = {
"total_drawers": count,
"wings": wings,
@@ -173,8 +230,16 @@ def tool_status():
"protocol": PALACE_PROTOCOL,
"aaak_dialect": AAAK_SPEC,
}
if error_info:
result["error"] = error_info
try:
all_meta = _get_cached_metadata(col)
for m in all_meta:
w = m.get("wing", "unknown")
r = m.get("room", "unknown")
wings[w] = wings.get(w, 0) + 1
rooms[r] = rooms.get(r, 0) + 1
except Exception as e:
logger.exception("tool_status metadata fetch failed")
result["error"] = str(e)
result["partial"] = True
return result
@@ -217,29 +282,17 @@ def tool_list_wings():
if not col:
return _no_palace()
wings = {}
batch_size = 5000
offset = 0
result = {"wings": wings}
try:
col.count() # verify collection is accessible
except Exception as e:
return {"wings": {}, "error": str(e)}
while True:
try:
batch = col.get(include=["metadatas"], limit=batch_size, offset=offset)
rows = batch["metadatas"]
for m in rows:
all_meta = _get_cached_metadata(col)
for m in all_meta:
w = m.get("wing", "unknown")
wings[w] = wings.get(w, 0) + 1
offset += len(rows)
if len(rows) < batch_size:
break
except Exception as e:
return {
"wings": wings,
"error": f"Partial result, failed at offset {offset}: {str(e)}",
"partial": True,
}
return {"wings": wings}
logger.exception("tool_list_wings metadata fetch failed")
result["error"] = str(e)
result["partial"] = True
return result
def tool_list_rooms(wing: str = None):
@@ -247,34 +300,18 @@ def tool_list_rooms(wing: str = None):
if not col:
return _no_palace()
rooms = {}
batch_size = 5000
offset = 0
result = {"wing": wing or "all", "rooms": rooms}
try:
where = {"wing": wing} if wing else None
try:
col.count() # verify collection is accessible
except Exception as e:
return {"wing": wing or "all", "rooms": {}, "error": str(e)}
while True:
try:
kwargs = {"include": ["metadatas"], "limit": batch_size, "offset": offset}
if where:
kwargs["where"] = where
batch = col.get(**kwargs)
rows = batch["metadatas"]
for m in rows:
all_meta = _fetch_all_metadata(col, where=where)
for m in all_meta:
r = m.get("room", "unknown")
rooms[r] = rooms.get(r, 0) + 1
offset += len(rows)
if len(rows) < batch_size:
break
except Exception as e:
return {
"wing": wing or "all",
"rooms": rooms,
"error": f"Partial result, failed at offset {offset}: {str(e)}",
"partial": True,
}
return {"wing": wing or "all", "rooms": rooms}
logger.exception("tool_list_rooms metadata fetch failed")
result["error"] = str(e)
result["partial"] = True
return result
def tool_get_taxonomy():
@@ -282,38 +319,31 @@ def tool_get_taxonomy():
if not col:
return _no_palace()
taxonomy = {}
batch_size = 5000
offset = 0
result = {"taxonomy": taxonomy}
try:
col.count() # verify collection is accessible
except Exception as e:
return {"taxonomy": {}, "error": str(e)}
while True:
try:
batch = col.get(include=["metadatas"], limit=batch_size, offset=offset)
rows = batch["metadatas"]
for m in rows:
all_meta = _get_cached_metadata(col)
for m in all_meta:
w = m.get("wing", "unknown")
r = m.get("room", "unknown")
if w not in taxonomy:
taxonomy[w] = {}
taxonomy[w][r] = taxonomy[w].get(r, 0) + 1
offset += len(rows)
if len(rows) < batch_size:
break
except Exception as e:
return {
"taxonomy": taxonomy,
"error": f"Partial result, failed at offset {offset}: {str(e)}",
"partial": True,
}
return {"taxonomy": taxonomy}
logger.exception("tool_get_taxonomy metadata fetch failed")
result["error"] = str(e)
result["partial"] = True
return result
def tool_search(
query: str, limit: int = 5, wing: str = None, room: str = None, context: str = None
query: str, limit: int = 5, wing: str = None, room: str = None,
max_distance: float = 1.5, min_similarity: float = None, context: str = None,
):
limit = max(1, min(limit, 50))
limit = max(1, min(limit, _MAX_RESULTS))
# Backwards compat: accept old name
# Backwards compat: convert old similarity scale (higher=stricter) to
# distance scale (lower=stricter). Similarity 0.8 → distance 0.2.
dist = (1.0 - min_similarity) if min_similarity is not None else max_distance
# Mitigate system prompt contamination (Issue #333)
sanitized = sanitize_query(query)
result = search_memories(
@@ -322,6 +352,7 @@ def tool_search(
wing=wing,
room=room,
n_results=limit,
max_distance=dist,
)
# Attach sanitizer metadata for transparency
if sanitized["was_sanitized"]:
@@ -410,6 +441,7 @@ def tool_add_drawer(
wing: str, room: str, content: str, source_file: str = None, added_by: str = "mcp"
):
"""File verbatim content into a wing/room. Checks for duplicates first."""
global _metadata_cache
try:
wing = sanitize_name(wing, "wing")
room = sanitize_name(room, "room")
@@ -458,6 +490,7 @@ def tool_add_drawer(
}
],
)
_metadata_cache = None
logger.info(f"Filed drawer: {drawer_id} → {wing}/{room}")
return {"success": True, "drawer_id": drawer_id, "wing": wing, "room": room}
except Exception as e:
@@ -466,6 +499,7 @@ def tool_add_drawer(
def tool_delete_drawer(drawer_id: str):
"""Delete a single drawer by ID."""
global _metadata_cache
col = _get_collection()
if not col:
return _no_palace()
@@ -487,12 +521,150 @@ def tool_delete_drawer(drawer_id: str):
try:
col.delete(ids=[drawer_id])
_metadata_cache = None
logger.info(f"Deleted drawer: {drawer_id}")
return {"success": True, "drawer_id": drawer_id}
except Exception as e:
return {"success": False, "error": str(e)}
def tool_get_drawer(drawer_id: str):
"""Fetch a single drawer by ID. Returns full content and metadata."""
col = _get_collection()
if not col:
return _no_palace()
try:
result = col.get(ids=[drawer_id], include=["documents", "metadatas"])
if not result["ids"]:
return {"error": f"Drawer not found: {drawer_id}"}
meta = result["metadatas"][0]
doc = result["documents"][0]
return {
"drawer_id": drawer_id,
"content": doc,
"wing": meta.get("wing", ""),
"room": meta.get("room", ""),
"metadata": meta,
}
except Exception as e:
return {"error": str(e)}
def tool_list_drawers(wing: str = None, room: str = None, limit: int = 20, offset: int = 0):
"""List drawers with pagination. Optional wing/room filter."""
limit = max(1, min(limit, _MAX_RESULTS))
offset = max(0, offset)
col = _get_collection()
if not col:
return _no_palace()
try:
where = None
conditions = []
if wing:
conditions.append({"wing": wing})
if room:
conditions.append({"room": room})
if len(conditions) == 1:
where = conditions[0]
elif len(conditions) > 1:
where = {"$and": conditions}
kwargs = {"include": ["documents", "metadatas"], "limit": limit, "offset": offset}
if where:
kwargs["where"] = where
result = col.get(**kwargs)
drawers = []
for i, did in enumerate(result["ids"]):
meta = result["metadatas"][i]
doc = result["documents"][i]
drawers.append(
{
"drawer_id": did,
"wing": meta.get("wing", ""),
"room": meta.get("room", ""),
"content_preview": doc[:200] + "..." if len(doc) > 200 else doc,
}
)
return {
"drawers": drawers,
"count": len(drawers),
"offset": offset,
"limit": limit,
}
except Exception as e:
return {"error": str(e)}
def tool_update_drawer(drawer_id: str, content: str = None, wing: str = None, room: str = None):
"""Update an existing drawer's content and/or metadata."""
global _metadata_cache
if content is None and wing is None and room is None:
return {"success": True, "drawer_id": drawer_id, "noop": True}
col = _get_collection()
if not col:
return _no_palace()
try:
existing = col.get(ids=[drawer_id], include=["documents", "metadatas"])
if not existing["ids"]:
return {"success": False, "error": f"Drawer not found: {drawer_id}"}
old_meta = existing["metadatas"][0]
old_doc = existing["documents"][0]
new_doc = old_doc
if content is not None:
try:
new_doc = sanitize_content(content)
except ValueError as e:
return {"success": False, "error": str(e)}
new_meta = dict(old_meta)
if wing is not None:
try:
new_meta["wing"] = sanitize_name(wing, "wing")
except ValueError as e:
return {"success": False, "error": str(e)}
if room is not None:
try:
new_meta["room"] = sanitize_name(room, "room")
except ValueError as e:
return {"success": False, "error": str(e)}
_wal_log(
"update_drawer",
{
"drawer_id": drawer_id,
"old_wing": old_meta.get("wing", ""),
"old_room": old_meta.get("room", ""),
"new_wing": new_meta.get("wing", ""),
"new_room": new_meta.get("room", ""),
"content_changed": content is not None,
"content_preview": new_doc[:200] if content is not None else None,
},
)
update_kwargs = {"ids": [drawer_id]}
if content is not None:
update_kwargs["documents"] = [new_doc]
update_kwargs["metadatas"] = [new_meta]
col.update(**update_kwargs)
_metadata_cache = None
logger.info(f"Updated drawer: {drawer_id}")
return {
"success": True,
"drawer_id": drawer_id,
"wing": new_meta.get("wing", ""),
"room": new_meta.get("room", ""),
}
except Exception as e:
return {"success": False, "error": str(e)}
# ==================== KNOWLEDGE GRAPH ====================
@@ -691,6 +863,71 @@ def tool_diary_read(agent_name: str, last_n: int = 10):
return {"error": "Failed to read diary entries"}
def tool_hook_settings(silent_save: bool = None, desktop_toast: bool = None):
"""
Get or set hook behavior settings.
- silent_save: True = stop hook saves directly (no MCP clutter),
False = legacy blocking MCP calls. Default: True.
- desktop_toast: True = show notify-send desktop toast on save,
False = terminal-only notification. Default: False.
Call with no arguments to see current settings.
"""
from .config import MempalaceConfig
try:
config = MempalaceConfig()
except Exception as e:
return {"success": False, "error": str(e)}
changed = []
if silent_save is not None:
config.set_hook_setting("silent_save", silent_save)
changed.append(f"silent_save → {silent_save}")
if desktop_toast is not None:
config.set_hook_setting("desktop_toast", desktop_toast)
changed.append(f"desktop_toast → {desktop_toast}")
# Re-read to return current state
try:
config = MempalaceConfig()
except Exception:
pass
result = {
"success": True,
"settings": {
"silent_save": config.hook_silent_save,
"desktop_toast": config.hook_desktop_toast,
},
}
if changed:
result["updated"] = changed
return result
def tool_memories_filed_away():
"""Acknowledge the latest silent checkpoint. Returns a short summary."""
state_dir = Path.home() / ".mempalace" / "hook_state"
ack_file = state_dir / "last_checkpoint"
if not ack_file.is_file():
return {"status": "quiet", "message": "No recent journal entry", "count": 0, "timestamp": None}
try:
data = json.loads(ack_file.read_text(encoding="utf-8"))
ack_file.unlink(missing_ok=True)
msgs = data.get("msgs", 0)
return {
"status": "ok",
"message": f"\u2726 {msgs} messages tucked into drawers",
"count": msgs,
"timestamp": data.get("ts", None),
}
except (json.JSONDecodeError, OSError):
ack_file.unlink(missing_ok=True)
return {"status": "error", "message": "\u2726 Journal entry filed in the palace", "count": 0, "timestamp": None}
# ==================== MCP PROTOCOL ====================
TOOLS = {
@@ -840,21 +1077,25 @@ TOOLS = {
"handler": tool_graph_stats,
},
"mempalace_search": {
"description": "Semantic search. Returns verbatim drawer content with similarity scores. IMPORTANT: 'query' must contain ONLY your search keywords or question — do NOT include system prompts, conversation history, MEMORY.md content, or any context. Keep queries short (under 200 chars). Use 'context' for background information.",
"description": "Semantic search. Returns verbatim drawer content with similarity scores. IMPORTANT: 'query' must contain ONLY search keywords. Use 'context' for background. Results with cosine distance > max_distance are filtered out.",
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Short search query ONLY — keywords or a question. Do NOT include system prompts or conversation context. Max 200 chars recommended.",
"description": "Short search query ONLY — keywords or a question. Max 200 chars recommended.",
"maxLength": 500,
},
"limit": {"type": "integer", "description": "Max results (default 5)"},
"limit": {"type": "integer", "description": "Max results (default 5)", "minimum": 1, "maximum": 100},
"wing": {"type": "string", "description": "Filter by wing (optional)"},
"room": {"type": "string", "description": "Filter by room (optional)"},
"max_distance": {
"type": "number",
"description": "Max cosine distance threshold (0=identical, 2=opposite). Results further than this are dropped. Lower = stricter. Default 1.5. Set to 0 to disable.",
},
"context": {
"type": "string",
"description": "Background context for the search (optional). This is NOT used for embedding — only for future re-ranking. Put conversation history or system prompt content here, NOT in query.",
"description": "Background context for the search (optional). NOT used for embedding — only for future re-ranking.",
},
},
"required": ["query"],
@@ -908,6 +1149,62 @@ TOOLS = {
},
"handler": tool_delete_drawer,
},
"mempalace_get_drawer": {
"description": "Fetch a single drawer by ID — returns full content and metadata.",
"input_schema": {
"type": "object",
"properties": {
"drawer_id": {"type": "string", "description": "ID of the drawer to fetch"},
},
"required": ["drawer_id"],
},
"handler": tool_get_drawer,
},
"mempalace_list_drawers": {
"description": "List drawers with pagination. Optional wing/room filter. Returns IDs, wings, rooms, and content previews.",
"input_schema": {
"type": "object",
"properties": {
"wing": {"type": "string", "description": "Filter by wing (optional)"},
"room": {"type": "string", "description": "Filter by room (optional)"},
"limit": {
"type": "integer",
"description": "Max results per page (default 20, max 100)",
"minimum": 1,
"maximum": 100,
},
"offset": {
"type": "integer",
"description": "Offset for pagination (default 0)",
"minimum": 0,
},
},
},
"handler": tool_list_drawers,
},
"mempalace_update_drawer": {
"description": "Update an existing drawer's content and/or metadata (wing, room). Fetches existing drawer first; returns error if not found.",
"input_schema": {
"type": "object",
"properties": {
"drawer_id": {"type": "string", "description": "ID of the drawer to update"},
"content": {
"type": "string",
"description": "New content (optional — omit to keep existing)",
},
"wing": {
"type": "string",
"description": "New wing (optional — omit to keep existing)",
},
"room": {
"type": "string",
"description": "New room (optional — omit to keep existing)",
},
},
"required": ["drawer_id"],
},
"handler": tool_update_drawer,
},
"mempalace_diary_write": {
"description": "Write to your personal agent diary in AAAK format. Your observations, thoughts, what you worked on, what matters. Each agent has their own diary with full history. Write in AAAK for compression — e.g. 'SESSION:2026-04-04|built.palace.graph+diary.tools|ALC.req:agent.diaries.in.aaak|★★★'. Use entity codes from the AAAK spec.",
"input_schema": {
@@ -948,6 +1245,32 @@ TOOLS = {
},
"handler": tool_diary_read,
},
"mempalace_hook_settings": {
"description": (
"Get or set hook behavior. silent_save: True = save directly "
"(no MCP clutter), False = legacy blocking. desktop_toast: "
"True = show desktop notification. Call with no args to view."
),
"input_schema": {
"type": "object",
"properties": {
"silent_save": {
"type": "boolean",
"description": "True = silent direct save, False = blocking MCP calls",
},
"desktop_toast": {
"type": "boolean",
"description": "True = show desktop toast via notify-send",
},
},
},
"handler": tool_hook_settings,
},
"mempalace_memories_filed_away": {
"description": "Check if a recent palace checkpoint was saved. Returns message count and timestamp.",
"input_schema": {"type": "object", "properties": {}},
"handler": tool_memories_filed_away,
},
}
@@ -960,8 +1283,8 @@ SUPPORTED_PROTOCOL_VERSIONS = [
def handle_request(request):
method = request.get("method", "")
params = request.get("params", {})
method = request.get("method") or ""
params = request.get("params") or {}
req_id = request.get("id")
if method == "initialize":
@@ -982,7 +1305,8 @@ def handle_request(request):
}
elif method == "ping":
return {"jsonrpc": "2.0", "id": req_id, "result": {}}
elif method == "notifications/initialized":
elif method.startswith("notifications/"):
# Notifications (no id) never get a response per JSON-RPC spec
return None
elif method == "tools/list":
return {
@@ -1040,6 +1364,9 @@ def handle_request(request):
"error": {"code": -32000, "message": "Internal tool error"},
}
# Notifications (missing id) must never get a response
if req_id is None:
return None
return {
"jsonrpc": "2.0",
"id": req_id,
mempalace/normalize.py
+166 -7
View File
@@ -6,7 +6,7 @@ Supported:
- Plain text with > markers (pass through)
- Claude.ai JSON export
- ChatGPT conversations.json
- Claude Code JSONL
- Claude Code JSONL (with tool_use/tool_result block capture)
- OpenAI Codex CLI JSONL
- Slack JSON export
- Plain text (pass through for paragraph chunking)
@@ -83,6 +83,8 @@ def _try_claude_code_jsonl(content: str) -> Optional[str]:
"""Claude Code JSONL sessions."""
lines = [line.strip() for line in content.strip().split("\n") if line.strip()]
messages = []
tool_use_map = {} # tool_use_id → tool_name
for line in lines:
try:
entry = json.loads(line)
@@ -92,14 +94,46 @@ def _try_claude_code_jsonl(content: str) -> Optional[str]:
continue
msg_type = entry.get("type", "")
message = entry.get("message", {})
if not isinstance(message, dict):
continue
msg_content = message.get("content", "")
# Build tool_use_map from assistant messages
if msg_type == "assistant" and isinstance(msg_content, list):
for block in msg_content:
if isinstance(block, dict) and block.get("type") == "tool_use":
tool_id = block.get("id", "")
if tool_id:
tool_use_map[tool_id] = block.get("name", "Unknown")
if msg_type in ("human", "user"):
text = _extract_content(message.get("content", ""))
# Check if this message is tool_results only (no user text)
is_tool_only = (
isinstance(msg_content, list)
and all(
isinstance(b, dict) and b.get("type") == "tool_result"
for b in msg_content
)
)
text = _extract_content(msg_content, tool_use_map=tool_use_map)
if text:
if is_tool_only and messages and messages[-1][0] == "assistant":
# Append tool results to the previous assistant message
prev_role, prev_text = messages[-1]
messages[-1] = (prev_role, prev_text + "\n" + text)
elif not is_tool_only:
messages.append(("user", text))
elif msg_type == "assistant":
text = _extract_content(message.get("content", ""))
text = _extract_content(msg_content, tool_use_map=tool_use_map)
if text:
# If previous message is also assistant (multi-turn tool loop),
# merge into the same assistant turn
if messages and messages[-1][0] == "assistant":
prev_role, prev_text = messages[-1]
messages[-1] = (prev_role, prev_text + "\n" + text)
else:
messages.append(("assistant", text))
if len(messages) >= 2:
return _messages_to_transcript(messages)
return None
@@ -270,8 +304,14 @@ def _try_slack_json(data) -> Optional[str]:
return None
def _extract_content(content) -> str:
"""Pull text from content — handles str, list of blocks, or dict."""
def _extract_content(content, tool_use_map: dict = None) -> str:
"""Pull text from content — handles str, list of blocks, or dict.
Args:
content: Message content — string, list of content blocks, or dict.
tool_use_map: Optional mapping of tool_use_id → tool_name, used to
select the right formatting strategy for tool_result blocks.
"""
if isinstance(content, str):
return content.strip()
if isinstance(content, list):
@@ -279,14 +319,133 @@ def _extract_content(content) -> str:
for item in content:
if isinstance(item, str):
parts.append(item)
elif isinstance(item, dict) and item.get("type") == "text":
elif isinstance(item, dict):
block_type = item.get("type")
if block_type == "text":
parts.append(item.get("text", ""))
return " ".join(parts).strip()
elif block_type == "tool_use":
parts.append(_format_tool_use(item))
elif block_type == "tool_result":
tid = item.get("tool_use_id", "")
tname = (tool_use_map or {}).get(tid, "Unknown")
result_content = item.get("content", "")
formatted = _format_tool_result(result_content, tname)
if formatted:
parts.append(formatted)
return "\n".join(p for p in parts if p).strip()
if isinstance(content, dict):
return content.get("text", "").strip()
return ""
def _format_tool_use(block: dict) -> str:
"""Format a tool_use block into a human-readable one-liner."""
name = block.get("name", "Unknown")
inp = block.get("input", {})
if name == "Bash":
cmd = inp.get("command", "")
if len(cmd) > 200:
cmd = cmd[:200] + "..."
return f"[Bash] {cmd}"
if name == "Read":
path = inp.get("file_path", "?")
offset = inp.get("offset")
limit = inp.get("limit")
if offset is not None and limit is not None:
try:
return f"[Read {path}:{offset}-{int(offset) + int(limit)}]"
except (ValueError, TypeError):
return f"[Read {path}:{offset}+{limit}]"
return f"[Read {path}]"
if name == "Grep":
pattern = inp.get("pattern", "")
target = inp.get("path") or inp.get("glob") or ""
return f"[Grep] {pattern} in {target}"
if name == "Glob":
pattern = inp.get("pattern", "")
return f"[Glob] {pattern}"
if name in ("Edit", "Write"):
path = inp.get("file_path", "?")
return f"[{name} {path}]"
# Unknown tool — serialize input, truncate
summary = json.dumps(inp, separators=(",", ":"))
if len(summary) > 200:
summary = summary[:200] + "..."
return f"[{name}] {summary}"
_TOOL_RESULT_MAX_LINES_BASH = 20 # head and tail line count
_TOOL_RESULT_MAX_MATCHES = 20 # Grep/Glob cap
_TOOL_RESULT_MAX_BYTES = 2048 # fallback cap for unknown tools
def _format_tool_result(content, tool_name: str) -> str:
"""Format a tool_result based on the originating tool's type.
Args:
content: Result text (str) or list of content blocks (list of dicts).
tool_name: Name of the tool that produced this result.
Returns:
Formatted string prefixed with ``→ ``, or empty string if omitted.
"""
# Normalize list-of-blocks to plain text
if isinstance(content, list):
parts = []
for item in content:
if isinstance(item, dict) and item.get("type") == "text":
parts.append(item.get("text", ""))
elif isinstance(item, str):
parts.append(item)
text = "\n".join(parts)
else:
text = str(content) if content else ""
text = text.strip()
if not text:
return ""
# Read/Edit/Write — omit result (content is in palace or git)
if tool_name in ("Read", "Edit", "Write"):
return ""
lines = text.split("\n")
# Bash — head + tail
if tool_name == "Bash":
n = _TOOL_RESULT_MAX_LINES_BASH
if len(lines) <= n * 2:
return "→ " + "\n→ ".join(lines)
head = lines[:n]
tail = lines[-n:]
omitted = len(lines) - 2 * n
return (
"→ " + "\n→ ".join(head)
+ f"\n→ ... [{omitted} lines omitted] ..."
+ "\n→ " + "\n→ ".join(tail)
)
# Grep/Glob — cap matches
if tool_name in ("Grep", "Glob"):
cap = _TOOL_RESULT_MAX_MATCHES
if len(lines) <= cap:
return "→ " + "\n→ ".join(lines)
kept = lines[:cap]
remaining = len(lines) - cap
return "→ " + "\n→ ".join(kept) + f"\n→ ... [{remaining} more matches]"
# Unknown — byte cap
if len(text) > _TOOL_RESULT_MAX_BYTES:
return "→ " + text[:_TOOL_RESULT_MAX_BYTES] + f"... [truncated, {len(text)} chars]"
return "→ " + text
def _messages_to_transcript(messages: list, spellcheck: bool = True) -> str:
"""Convert [(role, text), ...] to transcript format with > markers."""
if spellcheck:
mempalace/searcher.py
+39 -21
View File
@@ -18,6 +18,17 @@ class SearchError(Exception):
"""Raised when search cannot proceed (e.g. no palace found)."""
def build_where_filter(wing: str = None, room: str = None) -> dict:
"""Build ChromaDB where filter for wing/room filtering."""
if wing and room:
return {"$and": [{"wing": wing}, {"room": room}]}
elif wing:
return {"wing": wing}
elif room:
return {"room": room}
return {}
def search(query: str, palace_path: str, wing: str = None, room: str = None, n_results: int = 5):
"""
Search the palace. Returns verbatim drawer content.
@@ -30,14 +41,7 @@ def search(query: str, palace_path: str, wing: str = None, room: str = None, n_r
print(" Run: mempalace init <dir> then mempalace mine <dir>")
raise SearchError(f"No palace found at {palace_path}")
# Build where filter
where = {}
if wing and room:
where = {"$and": [{"wing": wing}, {"room": room}]}
elif wing:
where = {"wing": wing}
elif room:
where = {"room": room}
where = build_where_filter(wing, room)
try:
kwargs = {
@@ -71,7 +75,7 @@ def search(query: str, palace_path: str, wing: str = None, room: str = None, n_r
print(f"{'=' * 60}\n")
for i, (doc, meta, dist) in enumerate(zip(docs, metas, dists), 1):
similarity = round(1 - dist, 3)
similarity = round(max(0.0, 1 - dist), 3)
source = Path(meta.get("source_file", "?")).name
wing_name = meta.get("wing", "?")
room_name = meta.get("room", "?")
@@ -90,11 +94,27 @@ def search(query: str, palace_path: str, wing: str = None, room: str = None, n_r
def search_memories(
query: str, palace_path: str, wing: str = None, room: str = None, n_results: int = 5
query: str,
palace_path: str,
wing: str = None,
room: str = None,
n_results: int = 5,
max_distance: float = 0.0,
) -> dict:
"""
Programmatic search — returns a dict instead of printing.
"""Programmatic search — returns a dict instead of printing.
Used by the MCP server and other callers that need data.
Args:
query: Natural language search query.
palace_path: Path to the ChromaDB palace directory.
wing: Optional wing filter.
room: Optional room filter.
n_results: Max results to return.
max_distance: Max cosine distance threshold. The palace collection uses
cosine distance (hnsw:space=cosine) — 0 = identical, 2 = opposite.
Results with distance > this value are filtered out. A value of
0.0 disables filtering. Typical useful range: 0.3–1.0.
"""
try:
col = get_collection(palace_path, create=False)
@@ -105,14 +125,7 @@ def search_memories(
"hint": "Run: mempalace init <dir> && mempalace mine <dir>",
}
# Build where filter
where = {}
if wing and room:
where = {"$and": [{"wing": wing}, {"room": room}]}
elif wing:
where = {"wing": wing}
elif room:
where = {"room": room}
where = build_where_filter(wing, room)
try:
kwargs = {
@@ -133,18 +146,23 @@ def search_memories(
hits = []
for doc, meta, dist in zip(docs, metas, dists):
# Filter on raw distance before rounding to avoid precision loss
if max_distance > 0.0 and dist > max_distance:
continue
hits.append(
{
"text": doc,
"wing": meta.get("wing", "unknown"),
"room": meta.get("room", "unknown"),
"source_file": Path(meta.get("source_file", "?")).name,
"similarity": round(1 - dist, 3),
"similarity": round(max(0.0, 1 - dist), 3),
"distance": round(dist, 4),
}
)
return {
"query": query,
"filters": {"wing": wing, "room": room},
"total_before_filter": len(docs),
"results": hits,
}
tests/test_exporter.py
+136
View File
@@ -0,0 +1,136 @@
import os
import shutil
import tempfile
from pathlib import Path
import yaml
from mempalace.miner import mine
from mempalace.exporter import export_palace
def write_file(path: Path, content: str):
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, encoding="utf-8")
def _setup_palace(tmpdir):
"""Create a small palace with drawers across two wings for testing."""
project_a = Path(tmpdir) / "project_a"
project_b = Path(tmpdir) / "project_b"
palace_path = str(Path(tmpdir) / "palace")
# Project A: wing=alpha, rooms=backend,frontend
os.makedirs(project_a / "backend")
os.makedirs(project_a / "frontend")
write_file(project_a / "backend" / "server.py", "def serve():\n return 'ok'\n" * 20)
write_file(project_a / "frontend" / "app.js", "function render() { return 'hi'; }\n" * 20)
with open(project_a / "mempalace.yaml", "w") as f:
yaml.dump(
{
"wing": "alpha",
"rooms": [
{"name": "backend", "description": "Backend code"},
{"name": "frontend", "description": "Frontend code"},
],
},
f,
)
# Project B: wing=beta, rooms=docs
os.makedirs(project_b / "docs")
write_file(project_b / "docs" / "guide.md", "# Guide\n\nThis explains things.\n" * 20)
with open(project_b / "mempalace.yaml", "w") as f:
yaml.dump(
{
"wing": "beta",
"rooms": [{"name": "docs", "description": "Documentation"}],
},
f,
)
mine(str(project_a), palace_path)
mine(str(project_b), palace_path)
return palace_path
def test_export_creates_structure():
tmpdir = tempfile.mkdtemp()
try:
palace_path = _setup_palace(tmpdir)
output_dir = os.path.join(tmpdir, "export")
stats = export_palace(palace_path, output_dir)
# Should have two wings
assert stats["wings"] == 2
assert stats["rooms"] >= 2
assert stats["drawers"] >= 3
# Directory structure
assert os.path.isfile(os.path.join(output_dir, "index.md"))
assert os.path.isdir(os.path.join(output_dir, "alpha"))
assert os.path.isdir(os.path.join(output_dir, "beta"))
# Room files exist
assert os.path.isfile(os.path.join(output_dir, "alpha", "backend.md"))
assert os.path.isfile(os.path.join(output_dir, "alpha", "frontend.md"))
assert os.path.isfile(os.path.join(output_dir, "beta", "docs.md"))
finally:
shutil.rmtree(tmpdir, ignore_errors=True)
def test_export_markdown_content():
tmpdir = tempfile.mkdtemp()
try:
palace_path = _setup_palace(tmpdir)
output_dir = os.path.join(tmpdir, "export")
export_palace(palace_path, output_dir)
# Check that room files contain expected markdown elements
backend_md = Path(output_dir) / "alpha" / "backend.md"
content = backend_md.read_text(encoding="utf-8")
assert content.startswith("# alpha / backend\n")
assert "## drawer_" in content
assert "| Field | Value |" in content
assert "| Source |" in content
assert "| Filed |" in content
assert "| Added by |" in content
assert "---" in content
finally:
shutil.rmtree(tmpdir, ignore_errors=True)
def test_export_index_content():
tmpdir = tempfile.mkdtemp()
try:
palace_path = _setup_palace(tmpdir)
output_dir = os.path.join(tmpdir, "export")
export_palace(palace_path, output_dir)
index_md = Path(output_dir) / "index.md"
content = index_md.read_text(encoding="utf-8")
assert "# Palace Export" in content
assert "| Wing | Rooms | Drawers |" in content
assert "[alpha](alpha/)" in content
assert "[beta](beta/)" in content
finally:
shutil.rmtree(tmpdir, ignore_errors=True)
def test_export_empty_palace():
tmpdir = tempfile.mkdtemp()
try:
palace_path = os.path.join(tmpdir, "empty_palace")
output_dir = os.path.join(tmpdir, "export")
stats = export_palace(palace_path, output_dir)
assert stats == {"wings": 0, "rooms": 0, "drawers": 0}
finally:
shutil.rmtree(tmpdir, ignore_errors=True)
tests/test_mcp_server.py
+141
View File
@@ -145,6 +145,42 @@ class TestHandleRequest:
resp = handle_request({"method": "unknown/method", "id": 4, "params": {}})
assert resp["error"]["code"] == -32601
def test_any_notification_returns_none(self):
"""All notifications/* methods should return None (no response)."""
from mempalace.mcp_server import handle_request
for method in [
"notifications/initialized",
"notifications/cancelled",
"notifications/progress",
"notifications/roots/list_changed",
]:
resp = handle_request({"method": method, "params": {}})
assert resp is None, f"{method} should return None"
def test_unknown_method_no_id_returns_none(self):
"""Messages without id (notifications) must never get a response."""
from mempalace.mcp_server import handle_request
resp = handle_request({"method": "unknown/thing", "params": {}})
assert resp is None
def test_malformed_method_none(self):
"""method=None or missing should not crash."""
from mempalace.mcp_server import handle_request
# Explicit None
resp = handle_request({"method": None, "params": {}})
assert resp is None # no id → no response
# Missing method entirely
resp = handle_request({"params": {}})
assert resp is None
# method=None with id → should return error, not crash
resp = handle_request({"method": None, "id": 99, "params": {}})
assert resp["error"]["code"] == -32601
def test_tools_call_dispatches(self, monkeypatch, config, palace_path, seeded_kg):
_patch_mcp_server(monkeypatch, config, seeded_kg)
from mempalace.mcp_server import handle_request
@@ -259,6 +295,20 @@ class TestSearchTool:
result = tool_search(query="database", room="backend")
assert all(r["room"] == "backend" for r in result["results"])
def test_search_min_similarity_backwards_compat(self, monkeypatch, config, palace_path, seeded_collection, kg):
"""Old min_similarity param still works via backwards-compat shim."""
_patch_mcp_server(monkeypatch, config, kg)
from mempalace.mcp_server import tool_search
# Old name should work
result = tool_search(query="JWT", min_similarity=1.5)
assert "results" in result
# Old name takes precedence when both provided
result_strict = tool_search(query="JWT", max_distance=999.0, min_similarity=0.01)
result_loose = tool_search(query="JWT", max_distance=0.01, min_similarity=999.0)
assert len(result_strict["results"]) <= len(result_loose["results"])
# ── Write Tools ─────────────────────────────────────────────────────────
@@ -328,6 +378,97 @@ class TestWriteTools:
)
assert result["is_duplicate"] is False
def test_get_drawer(self, monkeypatch, config, palace_path, seeded_collection, kg):
_patch_mcp_server(monkeypatch, config, kg)
from mempalace.mcp_server import tool_get_drawer
result = tool_get_drawer("drawer_proj_backend_aaa")
assert result["drawer_id"] == "drawer_proj_backend_aaa"
assert result["wing"] == "project"
assert result["room"] == "backend"
assert "JWT tokens" in result["content"]
def test_get_drawer_not_found(self, monkeypatch, config, palace_path, seeded_collection, kg):
_patch_mcp_server(monkeypatch, config, kg)
from mempalace.mcp_server import tool_get_drawer
result = tool_get_drawer("nonexistent_drawer")
assert "error" in result
def test_list_drawers(self, monkeypatch, config, palace_path, seeded_collection, kg):
_patch_mcp_server(monkeypatch, config, kg)
from mempalace.mcp_server import tool_list_drawers
result = tool_list_drawers()
assert result["count"] == 4
assert len(result["drawers"]) == 4
def test_list_drawers_with_wing_filter(self, monkeypatch, config, palace_path, seeded_collection, kg):
_patch_mcp_server(monkeypatch, config, kg)
from mempalace.mcp_server import tool_list_drawers
result = tool_list_drawers(wing="project")
assert result["count"] == 3
assert all(d["wing"] == "project" for d in result["drawers"])
def test_list_drawers_with_room_filter(self, monkeypatch, config, palace_path, seeded_collection, kg):
_patch_mcp_server(monkeypatch, config, kg)
from mempalace.mcp_server import tool_list_drawers
result = tool_list_drawers(wing="project", room="backend")
assert result["count"] == 2
assert all(d["room"] == "backend" for d in result["drawers"])
def test_list_drawers_pagination(self, monkeypatch, config, palace_path, seeded_collection, kg):
_patch_mcp_server(monkeypatch, config, kg)
from mempalace.mcp_server import tool_list_drawers
result = tool_list_drawers(limit=2, offset=0)
assert result["count"] == 2
assert result["limit"] == 2
assert result["offset"] == 0
def test_list_drawers_negative_offset_clamped(self, monkeypatch, config, palace_path, seeded_collection, kg):
_patch_mcp_server(monkeypatch, config, kg)
from mempalace.mcp_server import tool_list_drawers
result = tool_list_drawers(offset=-5)
assert result["offset"] == 0
def test_update_drawer_content(self, monkeypatch, config, palace_path, seeded_collection, kg):
_patch_mcp_server(monkeypatch, config, kg)
from mempalace.mcp_server import tool_update_drawer, tool_get_drawer
result = tool_update_drawer("drawer_proj_backend_aaa", content="Updated content about auth.")
assert result["success"] is True
fetched = tool_get_drawer("drawer_proj_backend_aaa")
assert fetched["content"] == "Updated content about auth."
def test_update_drawer_wing_and_room(self, monkeypatch, config, palace_path, seeded_collection, kg):
_patch_mcp_server(monkeypatch, config, kg)
from mempalace.mcp_server import tool_update_drawer
result = tool_update_drawer("drawer_proj_backend_aaa", wing="new_wing", room="new_room")
assert result["success"] is True
assert result["wing"] == "new_wing"
assert result["room"] == "new_room"
def test_update_drawer_not_found(self, monkeypatch, config, palace_path, seeded_collection, kg):
_patch_mcp_server(monkeypatch, config, kg)
from mempalace.mcp_server import tool_update_drawer
result = tool_update_drawer("nonexistent_drawer", content="hello")
assert result["success"] is False
def test_update_drawer_noop(self, monkeypatch, config, palace_path, seeded_collection, kg):
_patch_mcp_server(monkeypatch, config, kg)
from mempalace.mcp_server import tool_update_drawer
result = tool_update_drawer("drawer_proj_backend_aaa")
assert result["success"] is True
assert result.get("noop") is True
# ── KG Tools ────────────────────────────────────────────────────────────
tests/test_normalize.py
+328 -2
View File
@@ -3,6 +3,8 @@ from unittest.mock import patch
from mempalace.normalize import (
_extract_content,
_format_tool_result,
_format_tool_use,
_messages_to_transcript,
_try_chatgpt_json,
_try_claude_ai_json,
@@ -81,7 +83,7 @@ def test_extract_content_string():
def test_extract_content_list_of_strings():
assert _extract_content(["hello", "world"]) == "hello world"
assert _extract_content(["hello", "world"]) == "hello\nworld"
def test_extract_content_list_of_blocks():
@@ -99,7 +101,198 @@ def test_extract_content_none():
def test_extract_content_mixed_list():
blocks = ["plain", {"type": "text", "text": "block"}]
assert _extract_content(blocks) == "plain block"
assert _extract_content(blocks) == "plain\nblock"
# ── _format_tool_use ──────────────────────────────────────────────────
def test_format_tool_use_bash():
block = {"type": "tool_use", "id": "t1", "name": "Bash",
"input": {"command": "lsusb | grep razer", "description": "Check USB"}}
result = _format_tool_use(block)
assert result == "[Bash] lsusb | grep razer"
def test_format_tool_use_bash_truncates_long_command():
block = {"type": "tool_use", "id": "t1", "name": "Bash",
"input": {"command": "x" * 300}}
result = _format_tool_use(block)
assert len(result) <= len("[Bash] ") + 200 + len("...")
assert result.endswith("...")
def test_format_tool_use_read():
block = {"type": "tool_use", "id": "t1", "name": "Read",
"input": {"file_path": "/home/jp/file.py"}}
result = _format_tool_use(block)
assert result == "[Read /home/jp/file.py]"
def test_format_tool_use_read_with_range():
block = {"type": "tool_use", "id": "t1", "name": "Read",
"input": {"file_path": "/home/jp/file.py", "offset": 10, "limit": 50}}
result = _format_tool_use(block)
assert result == "[Read /home/jp/file.py:10-60]"
def test_format_tool_use_grep():
block = {"type": "tool_use", "id": "t1", "name": "Grep",
"input": {"pattern": "firmware", "path": "/home/jp/proj"}}
result = _format_tool_use(block)
assert result == "[Grep] firmware in /home/jp/proj"
def test_format_tool_use_grep_with_glob():
block = {"type": "tool_use", "id": "t1", "name": "Grep",
"input": {"pattern": "TODO", "glob": "*.py"}}
result = _format_tool_use(block)
assert result == "[Grep] TODO in *.py"
def test_format_tool_use_glob():
block = {"type": "tool_use", "id": "t1", "name": "Glob",
"input": {"pattern": "/home/jp/proj/**/*.py"}}
result = _format_tool_use(block)
assert result == "[Glob] /home/jp/proj/**/*.py"
def test_format_tool_use_edit():
block = {"type": "tool_use", "id": "t1", "name": "Edit",
"input": {"file_path": "/home/jp/file.py", "old_string": "x", "new_string": "y"}}
result = _format_tool_use(block)
assert result == "[Edit /home/jp/file.py]"
def test_format_tool_use_write():
block = {"type": "tool_use", "id": "t1", "name": "Write",
"input": {"file_path": "/home/jp/file.py", "content": "..."}}
result = _format_tool_use(block)
assert result == "[Write /home/jp/file.py]"
def test_format_tool_use_unknown_tool():
block = {"type": "tool_use", "id": "t1", "name": "mcp__mempalace__search",
"input": {"query": "firmware probe", "limit": 5}}
result = _format_tool_use(block)
assert result.startswith("[mcp__mempalace__search]")
assert "firmware probe" in result
def test_format_tool_use_unknown_tool_truncates():
block = {"type": "tool_use", "id": "t1", "name": "SomeTool",
"input": {"data": "x" * 300}}
result = _format_tool_use(block)
assert result.endswith("...")
assert len(result) <= len("[SomeTool] ") + 200 + len("...")
# ── _format_tool_result ──────────────────────────────────────────────
def test_format_tool_result_bash_short():
"""Short Bash output is preserved in full."""
content = "Bus 002 Device 005: ID 1532:0e05 Razer Kiyo Pro"
result = _format_tool_result(content, "Bash")
assert result == "→ Bus 002 Device 005: ID 1532:0e05 Razer Kiyo Pro"
def test_format_tool_result_bash_head_tail():
"""Long Bash output gets head+tail with gap marker."""
lines = [f"line {i}" for i in range(60)]
content = "\n".join(lines)
result = _format_tool_result(content, "Bash")
assert "line 0" in result
assert "line 19" in result
assert "line 40" in result
assert "line 59" in result
assert "20 lines omitted" in result
# Lines 20-39 should be gone
assert "line 20\n" not in result
def test_format_tool_result_bash_exactly_40_lines():
"""Bash output at exactly 40 lines is not truncated."""
lines = [f"line {i}" for i in range(40)]
content = "\n".join(lines)
result = _format_tool_result(content, "Bash")
assert "omitted" not in result
assert "line 0" in result
assert "line 39" in result
def test_format_tool_result_read_omitted():
"""Read results are omitted (content already in palace from project mining)."""
result = _format_tool_result("lots of file content here...", "Read")
assert result == ""
def test_format_tool_result_edit_omitted():
"""Edit results are omitted (diff is in git)."""
result = _format_tool_result("file updated", "Edit")
assert result == ""
def test_format_tool_result_write_omitted():
"""Write results are omitted."""
result = _format_tool_result("file created", "Write")
assert result == ""
def test_format_tool_result_grep_short():
"""Short Grep output is kept."""
content = "src/foo.py\nsrc/bar.py\nsrc/baz.py"
result = _format_tool_result(content, "Grep")
assert "→ src/foo.py" in result
assert "→ src/baz.py" in result
def test_format_tool_result_grep_caps_at_20():
"""Grep output beyond 20 lines is truncated."""
lines = [f"match_{i}.py" for i in range(30)]
content = "\n".join(lines)
result = _format_tool_result(content, "Grep")
assert "match_19.py" in result
assert "match_20.py" not in result
assert "10 more matches" in result
def test_format_tool_result_glob_caps_at_20():
"""Glob output beyond 20 lines is truncated."""
lines = [f"/path/file_{i}.py" for i in range(25)]
content = "\n".join(lines)
result = _format_tool_result(content, "Glob")
assert "file_19.py" in result
assert "file_20.py" not in result
assert "5 more matches" in result
def test_format_tool_result_unknown_short():
"""Unknown tool with short output is kept."""
result = _format_tool_result("some output", "mcp__mempalace__search")
assert result == "→ some output"
def test_format_tool_result_unknown_truncates():
"""Unknown tool output over 2KB is truncated."""
content = "x" * 3000
result = _format_tool_result(content, "SomeTool")
assert result.endswith("... [truncated, 3000 chars]")
assert len(result) < 2200
def test_format_tool_result_list_content():
"""tool_result content can be a list of text blocks."""
content = [{"type": "text", "text": "result line 1"}, {"type": "text", "text": "result line 2"}]
result = _format_tool_result(content, "Bash")
assert "result line 1" in result
assert "result line 2" in result
def test_format_tool_result_empty():
"""Empty result returns empty string."""
result = _format_tool_result("", "Bash")
assert result == ""
# ── _try_claude_code_jsonl ─────────────────────────────────────────────
@@ -501,6 +694,139 @@ def test_messages_to_transcript_assistant_first():
assert "> Q" in result
# ── Tool block integration (Task 3) ───────────────────────────────────
def test_extract_content_with_tool_use():
"""_extract_content includes formatted tool_use blocks."""
content = [
{"type": "text", "text": "Let me check."},
{"type": "tool_use", "id": "t1", "name": "Bash",
"input": {"command": "lsusb"}},
]
result = _extract_content(content)
assert "Let me check." in result
assert "[Bash] lsusb" in result
def test_extract_content_with_tool_result():
"""_extract_content includes formatted tool_result blocks (needs tool_use_map)."""
content = [
{"type": "tool_result", "tool_use_id": "t1", "content": "some output"},
]
result = _extract_content(content, tool_use_map={"t1": "Bash"})
assert "→ some output" in result
def test_extract_content_tool_result_without_map_uses_fallback():
"""tool_result without a map entry uses fallback strategy."""
content = [
{"type": "tool_result", "tool_use_id": "t1", "content": "some output"},
]
result = _extract_content(content)
assert "→ some output" in result
def test_claude_code_jsonl_captures_tool_output():
"""Full integration: tool_use + tool_result appear in normalized transcript."""
lines = [
json.dumps({"type": "human", "message": {"content": "Check the camera"}}),
json.dumps({"type": "assistant", "message": {"content": [
{"type": "text", "text": "Let me check."},
{"type": "tool_use", "id": "t1", "name": "Bash",
"input": {"command": "lsusb | grep razer"}},
]}}),
json.dumps({"type": "human", "message": {"content": [
{"type": "tool_result", "tool_use_id": "t1",
"content": "Bus 002 Device 005: ID 1532:0e05 Razer Kiyo Pro"},
]}}),
json.dumps({"type": "assistant", "message": {"content": "Found it."}}),
]
result = _try_claude_code_jsonl("\n".join(lines))
assert result is not None
assert "> Check the camera" in result
assert "[Bash] lsusb | grep razer" in result
assert "→ Bus 002 Device 005" in result
assert "Found it." in result
def test_claude_code_jsonl_read_result_omitted():
"""Read tool results are omitted but the path breadcrumb is kept."""
lines = [
json.dumps({"type": "human", "message": {"content": "Show me the file"}}),
json.dumps({"type": "assistant", "message": {"content": [
{"type": "text", "text": "Reading it."},
{"type": "tool_use", "id": "t1", "name": "Read",
"input": {"file_path": "/home/jp/file.py"}},
]}}),
json.dumps({"type": "human", "message": {"content": [
{"type": "tool_result", "tool_use_id": "t1",
"content": "entire file contents here that should not appear"},
]}}),
json.dumps({"type": "assistant", "message": {"content": "Here it is."}}),
]
result = _try_claude_code_jsonl("\n".join(lines))
assert result is not None
assert "[Read /home/jp/file.py]" in result
assert "entire file contents here" not in result
def test_claude_code_jsonl_tool_only_user_message_not_counted():
"""A user message containing ONLY tool_results (no text) should not
be added as a separate user turn with '>'."""
lines = [
json.dumps({"type": "human", "message": {"content": "Do it"}}),
json.dumps({"type": "assistant", "message": {"content": [
{"type": "text", "text": "Running."},
{"type": "tool_use", "id": "t1", "name": "Bash",
"input": {"command": "echo hi"}},
]}}),
json.dumps({"type": "human", "message": {"content": [
{"type": "tool_result", "tool_use_id": "t1", "content": "hi"},
]}}),
json.dumps({"type": "assistant", "message": {"content": "Done."}}),
]
result = _try_claude_code_jsonl("\n".join(lines))
assert result is not None
# Only one user turn marker — the original "Do it"
user_turns = [line for line in result.split("\n") if line.strip().startswith(">")]
assert len(user_turns) == 1
assert "> Do it" in result
def test_extract_content_text_only_backward_compat():
"""Text-only content blocks still work (backward compat)."""
content = [
{"type": "text", "text": "Hello"},
{"type": "text", "text": "World"},
]
result = _extract_content(content)
assert "Hello" in result
assert "World" in result
def test_extract_content_string_unchanged():
"""Plain string content still works."""
result = _extract_content("just a string")
assert result == "just a string"
def test_claude_code_jsonl_thinking_blocks_ignored():
"""Thinking blocks are still ignored."""
lines = [
json.dumps({"type": "human", "message": {"content": "Q"}}),
json.dumps({"type": "assistant", "message": {"content": [
{"type": "thinking", "thinking": "", "signature": "abc"},
{"type": "text", "text": "A"},
]}}),
]
result = _try_claude_code_jsonl("\n".join(lines))
assert result is not None
assert "thinking" not in result.lower()
assert "signature" not in result
assert "A" in result
def test_normalize_rejects_large_file():
"""Files over 500 MB should raise IOError before reading."""
with patch("mempalace.normalize.os.path.getsize", return_value=600 * 1024 * 1024):