feat(normalize): auto-rebuild stale drawers via NORMALIZE_VERSION schema gate
Without this, the strip_noise improvement only helps new mines. Every user who had already mined Claude Code JSONL sessions would keep their noise-polluted drawers forever, because convo_miner's file_already_mined skip short-circuits before re-processing. Adds a versioned schema gate so upgrades propagate silently: - palace.NORMALIZE_VERSION=2 — bumped when the normalization pipeline changes shape (this PR's strip_noise is the v1→v2 bump). - file_already_mined now returns False if the stored normalize_version is missing or less than current, triggering a rebuild on next mine. - Both miners stamp drawers with the current normalize_version. - convo_miner now purges stale drawers before inserting fresh chunks (mirrors miner.py's existing delete+insert), extracted into _file_convo_chunks helper to keep mine_convos under ruff's C901 limit. User experience: upgrade mempalace, run `mempalace mine` as usual, old noisy drawers get silently replaced with clean ones. No erase needed, no "you need to rebuild" changelog footgun. Tests: - test_file_already_mined_returns_false_for_stale_normalize_version — pins the version gate contract for missing/v1/current. - test_add_drawer_stamps_normalize_version — fresh project-miner drawers carry the field. - test_mine_convos_rebuilds_stale_drawers_after_schema_bump — end-to-end proof that a pre-v2 palace gets silently cleaned on next mine, with orphan drawers purged and NOT skipped. Existing test_file_already_mined_check_mtime updated to include the new field; all other tests unaffected.
This commit is contained in:
+54
-29
@@ -16,7 +16,7 @@ from datetime import datetime
|
||||
from collections import defaultdict
|
||||
|
||||
from .normalize import normalize
|
||||
from .palace import SKIP_DIRS, get_collection, file_already_mined
|
||||
from .palace import NORMALIZE_VERSION, SKIP_DIRS, file_already_mined, get_collection
|
||||
|
||||
|
||||
# File types that might contain conversations
|
||||
@@ -51,6 +51,7 @@ def _register_file(collection, source_file: str, wing: str, agent: str):
|
||||
"added_by": agent,
|
||||
"filed_at": datetime.now().isoformat(),
|
||||
"ingest_mode": "registry",
|
||||
"normalize_version": NORMALIZE_VERSION,
|
||||
}
|
||||
],
|
||||
)
|
||||
@@ -272,6 +273,52 @@ def scan_convos(convo_dir: str) -> list:
|
||||
# =============================================================================
|
||||
|
||||
|
||||
def _file_convo_chunks(collection, source_file, chunks, wing, room, agent, extract_mode):
|
||||
"""Purge stale drawers for ``source_file`` then upsert fresh chunks.
|
||||
|
||||
Returns (drawers_added, room_counts_delta).
|
||||
"""
|
||||
# Purge stale drawers first. When the normalize schema bumps,
|
||||
# file_already_mined() returns False for pre-v2 drawers and we land
|
||||
# here — clean them out so the source doesn't end up with a mix of
|
||||
# old-noise and new-clean drawers.
|
||||
try:
|
||||
collection.delete(where={"source_file": source_file})
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
room_counts_delta: dict = defaultdict(int)
|
||||
drawers_added = 0
|
||||
for chunk in chunks:
|
||||
chunk_room = chunk.get("memory_type", room) if extract_mode == "general" else room
|
||||
if extract_mode == "general":
|
||||
room_counts_delta[chunk_room] += 1
|
||||
drawer_id = f"drawer_{wing}_{chunk_room}_{hashlib.sha256((source_file + str(chunk['chunk_index'])).encode()).hexdigest()[:24]}"
|
||||
try:
|
||||
collection.upsert(
|
||||
documents=[chunk["content"]],
|
||||
ids=[drawer_id],
|
||||
metadatas=[
|
||||
{
|
||||
"wing": wing,
|
||||
"room": chunk_room,
|
||||
"source_file": source_file,
|
||||
"chunk_index": chunk["chunk_index"],
|
||||
"added_by": agent,
|
||||
"filed_at": datetime.now().isoformat(),
|
||||
"ingest_mode": "convos",
|
||||
"extract_mode": extract_mode,
|
||||
"normalize_version": NORMALIZE_VERSION,
|
||||
}
|
||||
],
|
||||
)
|
||||
drawers_added += 1
|
||||
except Exception as e:
|
||||
if "already exists" not in str(e).lower():
|
||||
raise
|
||||
return drawers_added, room_counts_delta
|
||||
|
||||
|
||||
def mine_convos(
|
||||
convo_dir: str,
|
||||
palace_path: str,
|
||||
@@ -375,34 +422,12 @@ def mine_convos(
|
||||
if extract_mode != "general":
|
||||
room_counts[room] += 1
|
||||
|
||||
# File each chunk
|
||||
drawers_added = 0
|
||||
for chunk in chunks:
|
||||
chunk_room = chunk.get("memory_type", room) if extract_mode == "general" else room
|
||||
if extract_mode == "general":
|
||||
room_counts[chunk_room] += 1
|
||||
drawer_id = f"drawer_{wing}_{chunk_room}_{hashlib.sha256((source_file + str(chunk['chunk_index'])).encode()).hexdigest()[:24]}"
|
||||
try:
|
||||
collection.upsert(
|
||||
documents=[chunk["content"]],
|
||||
ids=[drawer_id],
|
||||
metadatas=[
|
||||
{
|
||||
"wing": wing,
|
||||
"room": chunk_room,
|
||||
"source_file": source_file,
|
||||
"chunk_index": chunk["chunk_index"],
|
||||
"added_by": agent,
|
||||
"filed_at": datetime.now().isoformat(),
|
||||
"ingest_mode": "convos",
|
||||
"extract_mode": extract_mode,
|
||||
}
|
||||
],
|
||||
)
|
||||
drawers_added += 1
|
||||
except Exception as e:
|
||||
if "already exists" not in str(e).lower():
|
||||
raise
|
||||
# Purge stale drawers + file fresh chunks.
|
||||
drawers_added, room_delta = _file_convo_chunks(
|
||||
collection, source_file, chunks, wing, room, agent, extract_mode
|
||||
)
|
||||
for r, n in room_delta.items():
|
||||
room_counts[r] += n
|
||||
|
||||
total_drawers += drawers_added
|
||||
print(f" ✓ [{i:4}/{len(files)}] {filepath.name[:50]:50} +{drawers_added}")
|
||||
|
||||
+2
-1
@@ -15,7 +15,7 @@ from pathlib import Path
|
||||
from datetime import datetime
|
||||
from collections import defaultdict
|
||||
|
||||
from .palace import SKIP_DIRS, get_collection, file_already_mined
|
||||
from .palace import NORMALIZE_VERSION, SKIP_DIRS, file_already_mined, get_collection
|
||||
|
||||
READABLE_EXTENSIONS = {
|
||||
".txt",
|
||||
@@ -381,6 +381,7 @@ def add_drawer(
|
||||
"chunk_index": chunk_index,
|
||||
"added_by": agent,
|
||||
"filed_at": datetime.now().isoformat(),
|
||||
"normalize_version": NORMALIZE_VERSION,
|
||||
}
|
||||
# Store file mtime so we can detect modifications later.
|
||||
try:
|
||||
|
||||
+24
-4
@@ -36,6 +36,16 @@ SKIP_DIRS = {
|
||||
|
||||
_DEFAULT_BACKEND = ChromaBackend()
|
||||
|
||||
# Schema version for drawer normalization. Bump when the normalization
|
||||
# pipeline changes in a way that existing drawers should be rebuilt to pick up
|
||||
# (e.g., new noise-stripping rules). `file_already_mined` treats drawers with
|
||||
# a missing or stale `normalize_version` as "not mined", so the next mine pass
|
||||
# silently rebuilds them — users don't need to manually erase + re-mine.
|
||||
#
|
||||
# v2 (2026-04): introduced strip_noise() for Claude Code JSONL; previous
|
||||
# drawers stored system tags / hook chrome verbatim.
|
||||
NORMALIZE_VERSION = 2
|
||||
|
||||
|
||||
def get_collection(
|
||||
palace_path: str,
|
||||
@@ -53,16 +63,26 @@ def get_collection(
|
||||
def file_already_mined(collection, source_file: str, check_mtime: bool = False) -> bool:
|
||||
"""Check if a file has already been filed in the palace.
|
||||
|
||||
When check_mtime=True (used by project miner), returns False if the file
|
||||
has been modified since it was last mined, so it gets re-mined.
|
||||
When check_mtime=False (used by convo miner), just checks existence.
|
||||
Returns False (so the file gets re-mined) when:
|
||||
- no drawers exist for this source_file
|
||||
- the stored `normalize_version` is missing or older than the current
|
||||
schema (triggers silent rebuild after a normalization upgrade)
|
||||
- `check_mtime=True` and the file's mtime differs from the stored one
|
||||
|
||||
When check_mtime=True (used by project miner), also re-mines on content
|
||||
change. When check_mtime=False (used by convo miner), transcripts are
|
||||
assumed immutable, so only the version gate triggers a rebuild.
|
||||
"""
|
||||
try:
|
||||
results = collection.get(where={"source_file": source_file}, limit=1)
|
||||
if not results.get("ids"):
|
||||
return False
|
||||
stored_meta = results.get("metadatas", [{}])[0] or {}
|
||||
# Pre-v2 drawers have no version field — treat them as stale.
|
||||
stored_version = stored_meta.get("normalize_version", 1)
|
||||
if stored_version < NORMALIZE_VERSION:
|
||||
return False
|
||||
if check_mtime:
|
||||
stored_meta = results.get("metadatas", [{}])[0]
|
||||
stored_mtime = stored_meta.get("source_mtime")
|
||||
if stored_mtime is None:
|
||||
return False
|
||||
|
||||
@@ -75,3 +75,86 @@ def test_mine_convos_does_not_reprocess_empty_chunk_files(capsys):
|
||||
assert "Files skipped (already filed): 1" in out2
|
||||
finally:
|
||||
shutil.rmtree(tmpdir, ignore_errors=True)
|
||||
|
||||
|
||||
def test_mine_convos_rebuilds_stale_drawers_after_schema_bump(capsys):
|
||||
"""When stored drawers have an older normalize_version, the next mine
|
||||
silently purges them and refiles — no manual erase required.
|
||||
|
||||
This is what makes the strip_noise upgrade apply to existing corpora:
|
||||
users just run `mempalace mine` again and old noise-filled drawers get
|
||||
replaced with clean ones."""
|
||||
from mempalace.palace import NORMALIZE_VERSION
|
||||
|
||||
tmpdir = tempfile.mkdtemp()
|
||||
try:
|
||||
convo_path = Path(tmpdir) / "chat.txt"
|
||||
convo_path.write_text(
|
||||
"> What is memory?\nMemory is persistence.\n\n"
|
||||
"> Why does it matter?\nIt enables continuity.\n\n"
|
||||
"> How do we build it?\nWith structured storage.\n"
|
||||
)
|
||||
palace_path = os.path.join(tmpdir, "palace")
|
||||
|
||||
# First mine — stamps drawers with NORMALIZE_VERSION
|
||||
mine_convos(tmpdir, palace_path, wing="test")
|
||||
capsys.readouterr()
|
||||
|
||||
client = chromadb.PersistentClient(path=palace_path)
|
||||
col = client.get_collection("mempalace_drawers")
|
||||
resolved = str(Path(tmpdir).resolve() / "chat.txt")
|
||||
first_pass = col.get(where={"source_file": resolved})
|
||||
first_ids = set(first_pass["ids"])
|
||||
assert first_ids, "first mine should produce drawers"
|
||||
for meta in first_pass["metadatas"]:
|
||||
assert meta.get("normalize_version") == NORMALIZE_VERSION
|
||||
|
||||
# Simulate pre-v2 drawers: rewrite metadata to an older version,
|
||||
# and replace content with "noise" so we can see it get cleaned up.
|
||||
stale_metas = []
|
||||
for meta in first_pass["metadatas"]:
|
||||
stale = dict(meta)
|
||||
stale["normalize_version"] = 1
|
||||
stale_metas.append(stale)
|
||||
col.update(
|
||||
ids=list(first_pass["ids"]),
|
||||
documents=["STALE NOISE"] * len(first_pass["ids"]),
|
||||
metadatas=stale_metas,
|
||||
)
|
||||
# Add an extra orphan drawer that should also be purged.
|
||||
col.add(
|
||||
ids=["orphan_drawer"],
|
||||
documents=["OLD ORPHAN"],
|
||||
metadatas=[
|
||||
{
|
||||
"wing": "test",
|
||||
"room": "default",
|
||||
"source_file": resolved,
|
||||
"chunk_index": 999,
|
||||
"normalize_version": 1,
|
||||
}
|
||||
],
|
||||
)
|
||||
del col, client
|
||||
|
||||
# Second mine — version gate should trigger rebuild
|
||||
mine_convos(tmpdir, palace_path, wing="test")
|
||||
out = capsys.readouterr().out
|
||||
assert (
|
||||
"Files skipped (already filed): 0" in out
|
||||
), "stale drawers should force a rebuild, not a skip"
|
||||
|
||||
client = chromadb.PersistentClient(path=palace_path)
|
||||
col = client.get_collection("mempalace_drawers")
|
||||
rebuilt = col.get(where={"source_file": resolved})
|
||||
# Orphan is gone
|
||||
assert "orphan_drawer" not in rebuilt["ids"]
|
||||
# No stale content survived
|
||||
assert all("STALE NOISE" not in d for d in rebuilt["documents"])
|
||||
assert all("OLD ORPHAN" not in d for d in rebuilt["documents"])
|
||||
# All rebuilt drawers carry the current version
|
||||
for meta in rebuilt["metadatas"]:
|
||||
assert meta.get("normalize_version") == NORMALIZE_VERSION
|
||||
del col, client
|
||||
finally:
|
||||
shutil.rmtree(tmpdir, ignore_errors=True)
|
||||
|
||||
+90
-4
@@ -7,7 +7,7 @@ import chromadb
|
||||
import yaml
|
||||
|
||||
from mempalace.miner import mine, scan_project, status
|
||||
from mempalace.palace import file_already_mined
|
||||
from mempalace.palace import NORMALIZE_VERSION, file_already_mined
|
||||
|
||||
|
||||
def write_file(path: Path, content: str):
|
||||
@@ -227,11 +227,17 @@ def test_file_already_mined_check_mtime():
|
||||
assert file_already_mined(col, test_file) is False
|
||||
assert file_already_mined(col, test_file, check_mtime=True) is False
|
||||
|
||||
# Add it with mtime
|
||||
# Add it with mtime + current normalize_version
|
||||
col.add(
|
||||
ids=["d1"],
|
||||
documents=["hello world"],
|
||||
metadatas=[{"source_file": test_file, "source_mtime": str(mtime)}],
|
||||
metadatas=[
|
||||
{
|
||||
"source_file": test_file,
|
||||
"source_mtime": str(mtime),
|
||||
"normalize_version": NORMALIZE_VERSION,
|
||||
}
|
||||
],
|
||||
)
|
||||
|
||||
# Already mined (no mtime check)
|
||||
@@ -253,7 +259,12 @@ def test_file_already_mined_check_mtime():
|
||||
col.add(
|
||||
ids=["d2"],
|
||||
documents=["other"],
|
||||
metadatas=[{"source_file": "/fake/no_mtime.txt"}],
|
||||
metadatas=[
|
||||
{
|
||||
"source_file": "/fake/no_mtime.txt",
|
||||
"normalize_version": NORMALIZE_VERSION,
|
||||
}
|
||||
],
|
||||
)
|
||||
assert file_already_mined(col, "/fake/no_mtime.txt", check_mtime=True) is False
|
||||
finally:
|
||||
@@ -296,3 +307,78 @@ def test_status_missing_palace_does_not_create_empty_collection(tmp_path, capsys
|
||||
out = capsys.readouterr().out
|
||||
assert "No palace found" in out
|
||||
assert not palace_path.exists()
|
||||
|
||||
|
||||
# ── normalize_version schema gate ───────────────────────────────────────
|
||||
#
|
||||
# When the normalization pipeline changes shape (e.g., strip_noise lands),
|
||||
# `NORMALIZE_VERSION` is bumped so pre-existing drawers can be silently
|
||||
# rebuilt on the next mine. These tests pin that contract.
|
||||
|
||||
|
||||
def test_file_already_mined_returns_false_for_stale_normalize_version():
|
||||
"""Pre-v2 drawers (no field, or older integer) must not short-circuit."""
|
||||
tmpdir = tempfile.mkdtemp()
|
||||
try:
|
||||
palace_path = os.path.join(tmpdir, "palace")
|
||||
os.makedirs(palace_path)
|
||||
client = chromadb.PersistentClient(path=palace_path)
|
||||
col = client.get_or_create_collection("mempalace_drawers")
|
||||
|
||||
# Pre-v2 drawer: no normalize_version field at all
|
||||
col.add(
|
||||
ids=["d_old"],
|
||||
documents=["old"],
|
||||
metadatas=[{"source_file": "/fake/old.jsonl"}],
|
||||
)
|
||||
assert file_already_mined(col, "/fake/old.jsonl") is False
|
||||
|
||||
# Explicitly older version
|
||||
col.add(
|
||||
ids=["d_v1"],
|
||||
documents=["v1"],
|
||||
metadatas=[{"source_file": "/fake/v1.jsonl", "normalize_version": 1}],
|
||||
)
|
||||
assert file_already_mined(col, "/fake/v1.jsonl") is False
|
||||
|
||||
# Current version — short-circuits
|
||||
col.add(
|
||||
ids=["d_current"],
|
||||
documents=["cur"],
|
||||
metadatas=[
|
||||
{
|
||||
"source_file": "/fake/current.jsonl",
|
||||
"normalize_version": NORMALIZE_VERSION,
|
||||
}
|
||||
],
|
||||
)
|
||||
assert file_already_mined(col, "/fake/current.jsonl") is True
|
||||
finally:
|
||||
del col, client
|
||||
shutil.rmtree(tmpdir, ignore_errors=True)
|
||||
|
||||
|
||||
def test_add_drawer_stamps_normalize_version(tmp_path):
|
||||
"""Fresh drawers carry the current schema version so future upgrades work."""
|
||||
from mempalace.miner import add_drawer
|
||||
|
||||
palace_path = tmp_path / "palace"
|
||||
palace_path.mkdir()
|
||||
client = chromadb.PersistentClient(path=str(palace_path))
|
||||
col = client.get_or_create_collection("mempalace_drawers")
|
||||
try:
|
||||
added = add_drawer(
|
||||
collection=col,
|
||||
wing="test",
|
||||
room="notes",
|
||||
content="hello",
|
||||
source_file=str(tmp_path / "src.md"),
|
||||
chunk_index=0,
|
||||
agent="unit",
|
||||
)
|
||||
assert added is True
|
||||
stored = col.get(limit=1)
|
||||
meta = stored["metadatas"][0]
|
||||
assert meta["normalize_version"] == NORMALIZE_VERSION
|
||||
finally:
|
||||
del col, client
|
||||
|
||||
Reference in New Issue
Block a user