Commit a82d3ddc authored by Lisa (AI Assistant)'s avatar Lisa (AI Assistant)

fix: filter tier-1 memory scaffolding from compaction

parent 5b80c375
...@@ -12,6 +12,18 @@ from typing import Any ...@@ -12,6 +12,18 @@ from typing import Any
ENTRY_DELIMITER = "\n§\n" ENTRY_DELIMITER = "\n§\n"
DEFAULT_RETENTION_DAYS = 7 DEFAULT_RETENTION_DAYS = 7
MAX_DAILY_LINE_LEN = 220 MAX_DAILY_LINE_LEN = 220
SKIP_EXACT_ENTRIES = {
"# Active session only",
"## Memory system reference",
"## Today's active session",
}
SKIP_PREFIXES = (
"Today: active session scratchpad only.",
"- Tier 1:",
"- Tier 2:",
"- Tier 3:",
"- Durable facts belong in daily memory and/or MemPalace, not here.",
)
STOPWORDS = { STOPWORDS = {
"a", "an", "and", "are", "as", "at", "be", "by", "for", "from", "how", "i", "in", "is", "it", "a", "an", "and", "are", "as", "at", "be", "by", "for", "from", "how", "i", "in", "is", "it",
"of", "on", "or", "that", "the", "this", "to", "was", "we", "what", "when", "where", "which", "of", "on", "or", "that", "the", "this", "to", "was", "we", "what", "when", "where", "which",
...@@ -29,7 +41,9 @@ def load_entries(path: Path) -> list[str]: ...@@ -29,7 +41,9 @@ def load_entries(path: Path) -> list[str]:
text = path.read_text(encoding="utf-8").strip() text = path.read_text(encoding="utf-8").strip()
if not text: if not text:
return [] return []
return [chunk.strip() for chunk in text.split(ENTRY_DELIMITER) if chunk.strip()] if ENTRY_DELIMITER in text:
return [chunk.strip() for chunk in text.split(ENTRY_DELIMITER) if chunk.strip()]
return [line.strip() for line in text.splitlines() if line.strip()]
def write_entries(path: Path, entries: list[str]) -> None: def write_entries(path: Path, entries: list[str]) -> None:
...@@ -40,9 +54,32 @@ def write_entries(path: Path, entries: list[str]) -> None: ...@@ -40,9 +54,32 @@ def write_entries(path: Path, entries: list[str]) -> None:
path.write_text(body, encoding="utf-8") path.write_text(body, encoding="utf-8")
def _is_scaffolding_entry(entry: str) -> bool:
raw = (entry or "").strip()
if not raw:
return True
if "# Active session only" in raw or "## Memory system reference" in raw:
return True
raw_lines = [line.strip() for line in raw.splitlines() if line.strip()]
if raw_lines and all(
line in SKIP_EXACT_ENTRIES or any(line.startswith(prefix) for prefix in SKIP_PREFIXES)
for line in raw_lines
):
return True
normalized = re.sub(r"\s+", " ", raw)
if normalized in SKIP_EXACT_ENTRIES:
return True
return any(normalized.startswith(prefix) for prefix in SKIP_PREFIXES)
def compress_entry(entry: str) -> str: def compress_entry(entry: str) -> str:
line = " ".join(part.strip() for part in entry.splitlines() if part.strip()) line = " ".join(part.strip() for part in entry.splitlines() if part.strip())
line = re.sub(r"\s+", " ", line).strip() line = re.sub(r"\s+", " ", line).strip()
if _is_scaffolding_entry(line):
return ""
if len(line) > MAX_DAILY_LINE_LEN: if len(line) > MAX_DAILY_LINE_LEN:
line = line[: MAX_DAILY_LINE_LEN - 3].rstrip() + "..." line = line[: MAX_DAILY_LINE_LEN - 3].rstrip() + "..."
return line return line
...@@ -139,10 +176,12 @@ def score_entry(query: str, entry: str, day: str) -> dict[str, Any]: ...@@ -139,10 +176,12 @@ def score_entry(query: str, entry: str, day: str) -> dict[str, Any]:
def compact(memory_file: Path, daily_dir: Path, day: str | None, retention_days: int) -> dict[str, Any]: def compact(memory_file: Path, daily_dir: Path, day: str | None, retention_days: int) -> dict[str, Any]:
day_key = day or utc_today_key() day_key = day or utc_today_key()
active_entries = load_entries(memory_file) active_entries = load_entries(memory_file)
compressed = dedupe_keep_order([compress_entry(entry) for entry in active_entries if compress_entry(entry)]) compressed_entries = [compress_entry(entry) for entry in active_entries]
compressed = dedupe_keep_order([entry for entry in compressed_entries if entry])
daily_file = daily_dir / f"{day_key}.md" daily_file = daily_dir / f"{day_key}.md"
existing = load_entries(daily_file) existing = load_entries(daily_file)
merged = dedupe_keep_order(existing + compressed) existing_filtered = [entry for entry in existing if compress_entry(entry)]
merged = dedupe_keep_order(existing_filtered + compressed)
write_entries(daily_file, merged) write_entries(daily_file, merged)
removed = prune_daily_dir(daily_dir, retention_days) removed = prune_daily_dir(daily_dir, retention_days)
summary = summarize_entries(merged) summary = summarize_entries(merged)
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment