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

style: ruff format all Python files (#675)

Browse Source
This commit is contained in:
Ben Sigman
2026-04-11 22:59:34 -07:00
committed by GitHub
parent 53229f725a
commit 4621f85d7c
6 changed files with 226 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mempalace/exporter.py
+20 -12
View File
@@ -21,9 +21,9 @@ 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'
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:
@@ -68,13 +68,15 @@ def export_palace(palace_path: str, output_dir: str, format: str = "markdown") -
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", ""),
})
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():
@@ -141,8 +143,14 @@ def export_palace(palace_path: str, output_dir: str, format: str = "markdown") -
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")
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
mempalace/layers.py
+1 -1
View File
@@ -82,7 +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
MAX_SCAN = 2000 # don't scan more than this for L1 generation
def __init__(self, palace_path: str = None, wing: str = None):
cfg = MempalaceConfig()
mempalace/mcp_server.py
+25 -5
View File
@@ -336,8 +336,13 @@ def tool_get_taxonomy():
def tool_search(
query: str, limit: int = 5, wing: str = None, room: str = None,
max_distance: float = 1.5, min_similarity: float = 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, _MAX_RESULTS))
# Backwards compat: accept old name
@@ -912,7 +917,12 @@ def tool_memories_filed_away():
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}
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)
@@ -925,7 +935,12 @@ def tool_memories_filed_away():
}
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}
return {
"status": "error",
"message": "\u2726 Journal entry filed in the palace",
"count": 0,
"timestamp": None,
}
# ==================== MCP PROTOCOL ====================
@@ -1086,7 +1101,12 @@ TOOLS = {
"description": "Short search query ONLY — keywords or a question. Max 200 chars recommended.",
"maxLength": 500,
},
"limit": {"type": "integer", "description": "Max results (default 5)", "minimum": 1, "maximum": 100},
"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": {
mempalace/normalize.py
+8 -10
View File
@@ -108,12 +108,8 @@ def _try_claude_code_jsonl(content: str) -> Optional[str]:
if msg_type in ("human", "user"):
# 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
)
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:
@@ -381,8 +377,8 @@ def _format_tool_use(block: dict) -> str:
_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
_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:
@@ -426,9 +422,11 @@ def _format_tool_result(content, tool_name: str) -> str:
tail = lines[-n:]
omitted = len(lines) - 2 * n
return (
"→ " + "\n→ ".join(head)
"→ "
+ "\n→ ".join(head)
+ f"\n→ ... [{omitted} lines omitted] ..."
+ "\n→ " + "\n→ ".join(tail)
+ "\n→ "
+ "\n→ ".join(tail)
)
# Grep/Glob — cap matches
tests/test_mcp_server.py
+18 -6
View File
@@ -295,7 +295,9 @@ 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):
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
@@ -403,7 +405,9 @@ class TestWriteTools:
assert result["count"] == 4
assert len(result["drawers"]) == 4
def test_list_drawers_with_wing_filter(self, monkeypatch, config, palace_path, seeded_collection, kg):
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
@@ -411,7 +415,9 @@ class TestWriteTools:
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):
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
@@ -428,7 +434,9 @@ class TestWriteTools:
assert result["limit"] == 2
assert result["offset"] == 0
def test_list_drawers_negative_offset_clamped(self, monkeypatch, config, palace_path, seeded_collection, kg):
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
@@ -439,13 +447,17 @@ class TestWriteTools:
_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.")
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):
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
tests/test_normalize.py
+154 -54
View File
@@ -108,80 +108,114 @@ def test_extract_content_mixed_list():
def test_format_tool_use_bash():
block = {"type": "tool_use", "id": "t1", "name": "Bash",
"input": {"command": "lsusb | grep razer", "description": "Check USB"}}
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}}
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"}}
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}}
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"}}
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"}}
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"}}
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"}}
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": "..."}}
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}}
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}}
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("...")
@@ -701,8 +735,7 @@ 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"}},
{"type": "tool_use", "id": "t1", "name": "Bash", "input": {"command": "lsusb"}},
]
result = _extract_content(content)
assert "Let me check." in result
@@ -731,15 +764,36 @@ 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": [
{"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))
@@ -754,15 +808,36 @@ 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": [
{"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))
@@ -776,14 +851,32 @@ def test_claude_code_jsonl_tool_only_user_message_not_counted():
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": [
{"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))
@@ -815,10 +908,17 @@ 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"},
]}}),
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