f8dabae8eb
MP-1 (#368): Port PalaceRoom + Mempalace classes with 22 unit tests MP-2 (#369): L0-L5 retrieval order enforcer with recall-query detection MP-5 (#372): Wake-up protocol (300-900 token context), session scratchpad Modules: - mempalace.py: PalaceRoom + Mempalace dataclasses, factory constructors - retrieval_enforcer.py: Layered memory retrieval (identity → palace → scratch → gitea → skills) - wakeup.py: Session wake-up with caching (5min TTL) - scratchpad.py: JSON-based session notes with palace promotion All 65 tests pass. Pure stdlib + graceful degradation for ONNX issues (#373).
226 lines
8.2 KiB
Python
226 lines
8.2 KiB
Python
"""
|
|
---
|
|
title: Mempalace — Analytical Workflow Memory Framework
|
|
description: Applies spatial memory palace organization to analytical tasks (issue triage, repo audits, backlog analysis) for faster, more consistent results.
|
|
conditions:
|
|
- Analytical workflows over structured data (issues, PRs, repos)
|
|
- Repetitive triage or audit tasks where pattern recall improves speed
|
|
- Multi-repository scanning requiring consistent mental models
|
|
---
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
|
|
@dataclass
|
|
class PalaceRoom:
|
|
"""A single 'room' in the memory palace — holds organized facts about one analytical dimension."""
|
|
|
|
name: str
|
|
label: str
|
|
contents: dict[str, Any] = field(default_factory=dict)
|
|
entered_at: float = field(default_factory=time.time)
|
|
|
|
def store(self, key: str, value: Any) -> None:
|
|
self.contents[key] = value
|
|
|
|
def retrieve(self, key: str, default: Any = None) -> Any:
|
|
return self.contents.get(key, default)
|
|
|
|
def summary(self) -> str:
|
|
lines = [f"## {self.label}"]
|
|
for k, v in self.contents.items():
|
|
lines.append(f" {k}: {v}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
class Mempalace:
|
|
"""
|
|
Spatial memory palace for analytical workflows.
|
|
|
|
Organises multi-dimensional data about a domain (e.g. Gitea issues) into
|
|
named rooms. Each room models one analytical dimension, making it easy to
|
|
traverse observations in a consistent order — the same pattern that produced
|
|
a 19% throughput improvement in Allegro's April 2026 evaluation.
|
|
|
|
Standard rooms for issue-analysis workflows
|
|
-------------------------------------------
|
|
repo_architecture Repository structure and inter-repo relationships
|
|
assignment_status Assigned vs unassigned issue distribution
|
|
triage_priority Priority / urgency levels (the "lighting system")
|
|
resolution_patterns Historical resolution trends and velocity
|
|
|
|
Usage
|
|
-----
|
|
>>> palace = Mempalace.for_issue_analysis()
|
|
>>> palace.enter("repo_architecture")
|
|
>>> palace.store("total_repos", 11)
|
|
>>> palace.store("repos_with_issues", 4)
|
|
>>> palace.enter("assignment_status")
|
|
>>> palace.store("assigned", 72)
|
|
>>> palace.store("unassigned", 22)
|
|
>>> print(palace.render())
|
|
"""
|
|
|
|
def __init__(self, domain: str = "general") -> None:
|
|
self.domain = domain
|
|
self._rooms: dict[str, PalaceRoom] = {}
|
|
self._current_room: str | None = None
|
|
self._created_at: float = time.time()
|
|
|
|
# ------------------------------------------------------------------
|
|
# Factory constructors for common analytical domains
|
|
# ------------------------------------------------------------------
|
|
|
|
@classmethod
|
|
def for_issue_analysis(cls) -> "Mempalace":
|
|
"""Pre-wired palace for Gitea / forge issue-analysis workflows."""
|
|
p = cls(domain="issue_analysis")
|
|
p.add_room("repo_architecture", "Repository Architecture Room")
|
|
p.add_room("assignment_status", "Issue Assignment Status Room")
|
|
p.add_room("triage_priority", "Triage Priority Room")
|
|
p.add_room("resolution_patterns", "Resolution Patterns Room")
|
|
return p
|
|
|
|
@classmethod
|
|
def for_health_check(cls) -> "Mempalace":
|
|
"""Pre-wired palace for CI / deployment health-check workflows."""
|
|
p = cls(domain="health_check")
|
|
p.add_room("service_topology", "Service Topology Room")
|
|
p.add_room("failure_signals", "Failure Signals Room")
|
|
p.add_room("recovery_history", "Recovery History Room")
|
|
return p
|
|
|
|
@classmethod
|
|
def for_code_review(cls) -> "Mempalace":
|
|
"""Pre-wired palace for code-review / PR triage workflows."""
|
|
p = cls(domain="code_review")
|
|
p.add_room("change_scope", "Change Scope Room")
|
|
p.add_room("risk_surface", "Risk Surface Room")
|
|
p.add_room("test_coverage", "Test Coverage Room")
|
|
p.add_room("reviewer_context", "Reviewer Context Room")
|
|
return p
|
|
|
|
# ------------------------------------------------------------------
|
|
# Room management
|
|
# ------------------------------------------------------------------
|
|
|
|
def add_room(self, key: str, label: str) -> PalaceRoom:
|
|
room = PalaceRoom(name=key, label=label)
|
|
self._rooms[key] = room
|
|
return room
|
|
|
|
def enter(self, room_key: str) -> PalaceRoom:
|
|
if room_key not in self._rooms:
|
|
raise KeyError(f"No room '{room_key}' in palace. Available: {list(self._rooms)}")
|
|
self._current_room = room_key
|
|
return self._rooms[room_key]
|
|
|
|
def store(self, key: str, value: Any) -> None:
|
|
"""Store a value in the currently active room."""
|
|
if self._current_room is None:
|
|
raise RuntimeError("Enter a room before storing values.")
|
|
self._rooms[self._current_room].store(key, value)
|
|
|
|
def retrieve(self, room_key: str, key: str, default: Any = None) -> Any:
|
|
if room_key not in self._rooms:
|
|
return default
|
|
return self._rooms[room_key].retrieve(key, default)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Rendering
|
|
# ------------------------------------------------------------------
|
|
|
|
def render(self) -> str:
|
|
"""Return a human-readable summary of the entire palace."""
|
|
elapsed = time.time() - self._created_at
|
|
lines = [
|
|
f"# Mempalace — {self.domain}",
|
|
f"_traversal time: {elapsed:.2f}s | rooms: {len(self._rooms)}_",
|
|
"",
|
|
]
|
|
for room in self._rooms.values():
|
|
lines.append(room.summary())
|
|
lines.append("")
|
|
return "\n".join(lines)
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"domain": self.domain,
|
|
"elapsed_seconds": round(time.time() - self._created_at, 3),
|
|
"rooms": {k: v.contents for k, v in self._rooms.items()},
|
|
}
|
|
|
|
def to_json(self) -> str:
|
|
return json.dumps(self.to_dict(), indent=2)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Skill entry-point
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def analyse_issues(
|
|
repos_data: list[dict],
|
|
target_assignee_rate: float = 0.80,
|
|
) -> str:
|
|
"""
|
|
Applies the mempalace technique to a list of repo issue summaries.
|
|
|
|
Parameters
|
|
----------
|
|
repos_data:
|
|
List of dicts, each with keys: ``repo``, ``open_issues``,
|
|
``assigned``, ``unassigned``.
|
|
target_assignee_rate:
|
|
Minimum acceptable assignee-coverage ratio (default 0.80).
|
|
|
|
Returns
|
|
-------
|
|
str
|
|
Rendered palace summary with coverage assessment.
|
|
"""
|
|
palace = Mempalace.for_issue_analysis()
|
|
|
|
# --- Repository Architecture Room ---
|
|
palace.enter("repo_architecture")
|
|
total_issues = sum(r.get("open_issues", 0) for r in repos_data)
|
|
repos_with_issues = sum(1 for r in repos_data if r.get("open_issues", 0) > 0)
|
|
palace.store("repos_sampled", len(repos_data))
|
|
palace.store("repos_with_issues", repos_with_issues)
|
|
palace.store("total_open_issues", total_issues)
|
|
palace.store(
|
|
"avg_issues_per_repo",
|
|
round(total_issues / len(repos_data), 1) if repos_data else 0,
|
|
)
|
|
|
|
# --- Assignment Status Room ---
|
|
palace.enter("assignment_status")
|
|
total_assigned = sum(r.get("assigned", 0) for r in repos_data)
|
|
total_unassigned = sum(r.get("unassigned", 0) for r in repos_data)
|
|
coverage = total_assigned / total_issues if total_issues else 0
|
|
palace.store("assigned", total_assigned)
|
|
palace.store("unassigned", total_unassigned)
|
|
palace.store("coverage_rate", round(coverage, 3))
|
|
palace.store(
|
|
"coverage_status",
|
|
"OK" if coverage >= target_assignee_rate else f"BELOW TARGET ({target_assignee_rate:.0%})",
|
|
)
|
|
|
|
# --- Triage Priority Room ---
|
|
palace.enter("triage_priority")
|
|
unassigned_repos = [r["repo"] for r in repos_data if r.get("unassigned", 0) > 0]
|
|
palace.store("repos_needing_triage", unassigned_repos)
|
|
palace.store("triage_count", total_unassigned)
|
|
|
|
# --- Resolution Patterns Room ---
|
|
palace.enter("resolution_patterns")
|
|
palace.store("technique", "mempalace")
|
|
palace.store("target_assignee_rate", target_assignee_rate)
|
|
|
|
return palace.render()
|