Merge pull request #826 from MemPalace/pr/multi-agent-lock

chore: forward closet layer (#788) into develop

docs/CLOSETS.md

@@ -0,0 +1,88 @@
+# Closets — The Searchable Index Layer
+
+## What closets are
+
+Drawers hold your verbatim content. Closets are the index — compact pointers that tell the searcher which drawers to open.
+
+```
+CLOSET: "built auth system|Ben;Igor|→drawer_api_auth_a1b2c3"
+         ↑ topic           ↑ entities  ↑ points to this drawer
+```
+
+An agent searching "who built the auth?" hits the closet first (fast scan of short text), then opens the referenced drawer to get the full verbatim content.
+
+## Lifecycle
+
+### When are closets created?
+
+Closets are created during `mempalace mine`. For each file mined:
+1. Content is chunked into drawers (verbatim, ~800 chars each)
+2. Topics, entities, and quotes are extracted from the content
+3. A closet is created with pointer lines to those drawers
+
+### What's inside a closet?
+
+Each line is one atomic topic pointer:
+```
+topic description|entity1;entity2|→drawer_id_1,drawer_id_2
+"verbatim quote from the content"|entity1|→drawer_id_3
+```
+
+Topics are never split across closets. If adding a topic would exceed 1,500 characters, a new closet is created.
+
+### When do closets update?
+
+When a file is re-mined (content changed, or `NORMALIZE_VERSION` was bumped), the miner first deletes every closet for that source file (`purge_file_closets`) and then writes a fresh set. Stale topics from the prior mine are gone — closets are always a snapshot of the current content, never an accumulation across runs.
+
+### What about stale topics?
+
+There are no stale topics: each re-mine is a clean rebuild for that source file. If a file gets larger and produces fewer or more closets than last time, the leftover numbered closets from the larger run are still purged because the delete is done by `source_file`, not by ID.
+
+### Do closets survive palace rebuilds?
+
+Closets are stored in the `mempalace_closets` ChromaDB collection alongside `mempalace_drawers`. If you delete and rebuild the palace, closets are recreated during the next `mempalace mine`.
+
+## How search uses closets
+
+```
+Query → search mempalace_closets (fast, small documents)
+         ↓
+    top closet hits → parse `→drawer_id_a,drawer_id_b` pointers
+         ↓
+    fetch exactly those drawers from mempalace_drawers (verbatim content)
+         ↓
+    apply max_distance filter
+         ↓
+    return chunk-level results (same shape as direct search)
+```
+
+Hits carry `matched_via: "closet"` (or `"drawer"` for the fallback path) plus a `closet_preview` field showing the line that surfaced them.
+
+If no closets exist (palace created before this feature) — or all closet hits get filtered out by `max_distance` — search falls back to direct drawer search. Closets are created on next mine.
+
+> **BM25 hybrid re-rank** is on the roadmap (deferred to a follow-up PR alongside generic `LLM_*` env-var support); the current closet search ranks purely by ChromaDB cosine distance against the closet text.
+
+## Limits
+
+| Setting | Value | Reason |
+|---------|-------|--------|
+| Max closet size | 1,500 chars (`CLOSET_CHAR_LIMIT`) | Leaves buffer under ChromaDB's working limit |
+| Source content scanned | 5,000 chars (`CLOSET_EXTRACT_WINDOW`) | Caps regex extraction cost on long files; back-of-file content is currently invisible to closet extraction (tracked for follow-up) |
+| Max topics per file | 12 | Keeps closets focused |
+| Max quotes per file | 3 | Most relevant only |
+| Max entities per pointer | 5 | Top names by frequency, after stoplist filtering |
+
+## For developers
+
+Closet functions live in `mempalace/palace.py`:
+- `get_closets_collection()` — get the closets ChromaDB collection
+- `build_closet_lines()` — extract topics/entities/quotes into pointer lines
+- `upsert_closet_lines()` — write lines to closets respecting the char limit (overwrites existing IDs; does not append — call `purge_file_closets` first when re-mining)
+- `purge_file_closets()` — delete every closet for a given source file before rebuild
+- `CLOSET_CHAR_LIMIT` / `CLOSET_EXTRACT_WINDOW` — size constants
+
+The closet-first search path lives in `mempalace/searcher.py`:
+- `_extract_drawer_ids_from_closet()` — parse `→drawer_a,drawer_b` pointers out of a closet document
+- `_closet_first_hits()` — query closets, parse pointers, hydrate matching drawers, return chunk-level hits or `None` to fall back
+
+Note: only the project miner (`miner.py::process_file`) builds closets today. Conversation-mined wings (Claude Code JSONL, ChatGPT export, etc.) will keep using direct drawer search via the searcher fallback until the convo-closet PR lands.

mempalace/miner.py

@@ -18,9 +18,13 @@ from collections import defaultdict
 from .palace import (
     NORMALIZE_VERSION,
     SKIP_DIRS,
+    build_closet_lines,
     file_already_mined,
+    get_closets_collection,
     get_collection,
     mine_lock,
+    purge_file_closets,
+    upsert_closet_lines,
 )
 
 READABLE_EXTENSIONS = {
@@ -417,6 +421,7 @@ def process_file(
     rooms: list,
     agent: str,
     dry_run: bool,
+    closets_col=None,
 ) -> tuple:
     """Read, chunk, route, and file one file. Returns (drawer_count, room_name)."""
 
@@ -473,6 +478,33 @@ def process_file(
             if added:
                 drawers_added += 1
 
+        # Build closet — the searchable index pointing to these drawers.
+        # Purge first: a re-mine (mtime change or normalize_version bump) must
+        # fully replace the prior closets, not append to them.
+        if closets_col and drawers_added > 0:
+            drawer_ids = [
+                f"drawer_{wing}_{room}_{hashlib.sha256((source_file + str(c['chunk_index'])).encode()).hexdigest()[:24]}"
+                for c in chunks
+            ]
+            closet_lines = build_closet_lines(source_file, drawer_ids, content, wing, room)
+            closet_id_base = (
+                f"closet_{wing}_{room}_{hashlib.sha256(source_file.encode()).hexdigest()[:24]}"
+            )
+            purge_file_closets(closets_col, source_file)
+            upsert_closet_lines(
+                closets_col,
+                closet_id_base,
+                closet_lines,
+                {
+                    "wing": wing,
+                    "room": room,
+                    "source_file": source_file,
+                    "drawer_count": drawers_added,
+                    "filed_at": datetime.now().isoformat(),
+                    "normalize_version": NORMALIZE_VERSION,
+                },
+            )
+
     return drawers_added, room
 
 
@@ -593,8 +625,10 @@ def mine(
 
     if not dry_run:
         collection = get_collection(palace_path)
+        closets_col = get_closets_collection(palace_path)
     else:
         collection = None
+        closets_col = None
 
     total_drawers = 0
     files_skipped = 0
@@ -609,6 +643,7 @@ def mine(
             rooms=rooms,
             agent=agent,
             dry_run=dry_run,
+            closets_col=closets_col,
         )
         if drawers == 0 and not dry_run:
             files_skipped += 1

mempalace/palace.py

@@ -62,6 +62,185 @@ def get_collection(
     )
 
 
+def get_closets_collection(palace_path: str, create: bool = True):
+    """Get the closets collection — the searchable index layer."""
+    return get_collection(palace_path, collection_name="mempalace_closets", create=create)
+
+
+CLOSET_CHAR_LIMIT = 1500  # fill closet until ~1500 chars, then start a new one
+CLOSET_EXTRACT_WINDOW = 5000  # how many chars of source content to scan for entities/topics
+
+# Common capitalized words that look like proper nouns but are usually
+# sentence-starters or filler. Filtered out of entity extraction.
+_ENTITY_STOPLIST = frozenset(
+    {
+        "The",
+        "This",
+        "That",
+        "These",
+        "Those",
+        "When",
+        "Where",
+        "What",
+        "Why",
+        "Who",
+        "Which",
+        "How",
+        "After",
+        "Before",
+        "Then",
+        "Now",
+        "Here",
+        "There",
+        "And",
+        "But",
+        "Or",
+        "Yet",
+        "So",
+        "If",
+        "Else",
+        "Yes",
+        "No",
+        "Maybe",
+        "Okay",
+        "User",
+        "Assistant",
+        "System",
+        "Tool",
+        "Monday",
+        "Tuesday",
+        "Wednesday",
+        "Thursday",
+        "Friday",
+        "Saturday",
+        "Sunday",
+        "January",
+        "February",
+        "March",
+        "April",
+        "May",
+        "June",
+        "July",
+        "August",
+        "September",
+        "October",
+        "November",
+        "December",
+    }
+)
+
+
+def build_closet_lines(source_file, drawer_ids, content, wing, room):
+    """Build compact closet pointer lines from drawer content.
+
+    Returns a LIST of lines (not joined). Each line is one complete topic
+    pointer — never split across closets.
+
+    Format: topic|entities|→drawer_ids
+    """
+    import re
+    from pathlib import Path
+
+    drawer_ref = ",".join(drawer_ids[:3])
+    window = content[:CLOSET_EXTRACT_WINDOW]
+
+    # Extract proper nouns (capitalized words, 2+ occurrences). Filter out
+    # common sentence-starters that aren't real entities.
+    words = re.findall(r"\b[A-Z][a-z]{2,}\b", window)
+    word_freq = {}
+    for w in words:
+        if w in _ENTITY_STOPLIST:
+            continue
+        word_freq[w] = word_freq.get(w, 0) + 1
+    entities = sorted(
+        [w for w, c in word_freq.items() if c >= 2],
+        key=lambda w: -word_freq[w],
+    )[:5]
+    entity_str = ";".join(entities) if entities else ""
+
+    # Extract key phrases — action verbs + context
+    topics = []
+    for pattern in [
+        r"(?:built|fixed|wrote|added|pushed|tested|created|decided|migrated|reviewed|deployed|configured|removed|updated)\s+[\w\s]{3,40}",
+    ]:
+        topics.extend(re.findall(pattern, window, re.IGNORECASE))
+    # Also grab section headers if present
+    for header in re.findall(r"^#{1,3}\s+(.{5,60})$", window, re.MULTILINE):
+        topics.append(header.strip())
+    # Dedupe preserving order
+    topics = list(dict.fromkeys(t.strip().lower() for t in topics))[:12]
+
+    # Extract quotes
+    quotes = re.findall(r'"([^"]{15,150})"', window)
+
+    # Build pointer lines — each one is atomic, never split
+    lines = []
+    for topic in topics:
+        lines.append(f"{topic}|{entity_str}|→{drawer_ref}")
+    for quote in quotes[:3]:
+        lines.append(f'"{quote}"|{entity_str}|→{drawer_ref}')
+
+    # Always have at least one line
+    if not lines:
+        name = Path(source_file).stem[:40]
+        lines.append(f"{wing}/{room}/{name}|{entity_str}|→{drawer_ref}")
+
+    return lines
+
+
+def purge_file_closets(closets_col, source_file: str) -> None:
+    """Delete every closet associated with ``source_file``.
+
+    Call this before ``upsert_closet_lines`` on a re-mine so stale topics
+    from a prior schema/version don't survive in the closet collection.
+    Mirrors the drawer-purge step in process_file().
+    """
+    try:
+        closets_col.delete(where={"source_file": source_file})
+    except Exception:
+        pass
+
+
+def upsert_closet_lines(closets_col, closet_id_base, lines, metadata):
+    """Write topic lines to closets, packed greedily without splitting a line.
+
+    Closets are deterministically numbered (``..._01``, ``..._02``, …) and
+    each ``upsert`` fully overwrites the prior content at that ID. Callers
+    are expected to ``purge_file_closets`` first when re-mining a source
+    file so stale-numbered closets from larger prior runs don't leak.
+
+    Returns the number of closets written.
+    """
+    closet_num = 1
+    current_lines: list = []
+    current_chars = 0
+    closets_written = 0
+
+    def _flush():
+        nonlocal closets_written
+        if not current_lines:
+            return
+        closet_id = f"{closet_id_base}_{closet_num:02d}"
+        text = "\n".join(current_lines)
+        closets_col.upsert(documents=[text], ids=[closet_id], metadatas=[metadata])
+        closets_written += 1
+
+    for line in lines:
+        line_len = len(line)
+        # Would this line fit whole in the current closet?
+        if current_chars > 0 and current_chars + line_len + 1 > CLOSET_CHAR_LIMIT:
+            _flush()
+            closet_num += 1
+            current_lines = []
+            current_chars = 0
+
+        current_lines.append(line)
+        current_chars += line_len + 1  # +1 for newline
+
+    _flush()
+    return closets_written
+
+
 @contextlib.contextmanager
 def mine_lock(source_file: str):
     """Cross-platform file lock for mine operations.

mempalace/searcher.py

@@ -7,9 +7,14 @@ Returns verbatim text — the actual words, never summaries.
 """
 
 import logging
+import re
 from pathlib import Path
 
-from .palace import get_collection
+from .palace import get_closets_collection, get_collection
+
+# Closet pointer line format: "topic|entities|→drawer_id_a,drawer_id_b"
+# Multiple lines may join with newlines inside one closet document.
+_CLOSET_DRAWER_REF_RE = re.compile(r"→([\w,]+)")
 
 logger = logging.getLogger("mempalace_mcp")
 
@@ -29,6 +34,116 @@ def build_where_filter(wing: str = None, room: str = None) -> dict:
     return {}
 
 
+def _extract_drawer_ids_from_closet(closet_doc: str) -> list:
+    """Parse all `→drawer_id_a,drawer_id_b` pointers out of a closet document.
+
+    Preserves order and dedupes.
+    """
+    seen: dict = {}
+    for match in _CLOSET_DRAWER_REF_RE.findall(closet_doc):
+        for did in match.split(","):
+            did = did.strip()
+            if did and did not in seen:
+                seen[did] = None
+    return list(seen.keys())
+
+
+def _closet_first_hits(
+    palace_path: str,
+    query: str,
+    where: dict,
+    drawers_col,
+    n_results: int,
+    max_distance: float,
+):
+    """Run a closet-first search and return chunk-level drawer hits.
+
+    Returns:
+        non-empty list of hits when the closet path produced usable matches.
+        ``None`` when the closet collection is empty/missing OR when every
+        candidate drawer was filtered out (e.g. by max_distance); the
+        caller should fall back to direct drawer search.
+    """
+    try:
+        closets_col = get_closets_collection(palace_path, create=False)
+    except Exception:
+        return None
+
+    try:
+        ckwargs = {
+            "query_texts": [query],
+            "n_results": max(n_results * 2, 5),
+            "include": ["documents", "metadatas", "distances"],
+        }
+        if where:
+            ckwargs["where"] = where
+        closet_results = closets_col.query(**ckwargs)
+    except Exception:
+        return None
+
+    closet_docs = closet_results["documents"][0] if closet_results["documents"] else []
+    if not closet_docs:
+        return None
+
+    closet_metas = closet_results["metadatas"][0]
+    closet_dists = closet_results["distances"][0]
+
+    # Collect candidate drawer IDs in closet-rank order, dedupe, remember
+    # which closet (and its distance/preview) introduced each one.
+    drawer_id_order: list = []
+    drawer_provenance: dict = {}
+    for cdoc, cmeta, cdist in zip(closet_docs, closet_metas, closet_dists):
+        for did in _extract_drawer_ids_from_closet(cdoc):
+            if did in drawer_provenance:
+                continue
+            drawer_provenance[did] = (cdist, cdoc, cmeta)
+            drawer_id_order.append(did)
+
+    if not drawer_id_order:
+        return None
+
+    # Hydrate exactly those drawers — chunk-level, not whole-file.
+    try:
+        fetched = drawers_col.get(
+            ids=drawer_id_order,
+            include=["documents", "metadatas"],
+        )
+    except Exception:
+        return None
+
+    fetched_ids = fetched.get("ids") or []
+    fetched_docs = fetched.get("documents") or []
+    fetched_metas = fetched.get("metadatas") or []
+    fetched_map = {
+        did: (doc, meta) for did, doc, meta in zip(fetched_ids, fetched_docs, fetched_metas)
+    }
+
+    hits: list = []
+    for did in drawer_id_order:
+        if did not in fetched_map:
+            continue  # closet pointed to a drawer that no longer exists
+        doc, meta = fetched_map[did]
+        cdist, cdoc, _ = drawer_provenance[did]
+        if max_distance > 0.0 and cdist > 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(max(0.0, 1 - cdist), 3),
+                "distance": round(cdist, 4),
+                "matched_via": "closet",
+                "closet_preview": cdoc[:200],
+            }
+        )
+        if len(hits) >= n_results:
+            break
+
+    return hits if hits else None
+
+
 def search(query: str, palace_path: str, wing: str = None, room: str = None, n_results: int = 5):
     """
     Search the palace. Returns verbatim drawer content.
@@ -117,7 +232,7 @@ def search_memories(
             0.0 disables filtering. Typical useful range: 0.3–1.0.
     """
     try:
-        col = get_collection(palace_path, create=False)
+        drawers_col = get_collection(palace_path, create=False)
     except Exception as e:
         logger.error("No palace found at %s: %s", palace_path, e)
         return {
@@ -127,6 +242,27 @@ def search_memories(
 
     where = build_where_filter(wing, room)
 
+    # Closet-first search: scan the compact index, parse drawer pointers
+    # from each matching line, then hydrate exactly those drawers. This
+    # keeps the result shape chunk-level (consistent with direct search)
+    # and applies the same max_distance filter.
+    closet_hits = _closet_first_hits(
+        palace_path=palace_path,
+        query=query,
+        where=where,
+        drawers_col=drawers_col,
+        n_results=n_results,
+        max_distance=max_distance,
+    )
+    if closet_hits is not None:
+        return {
+            "query": query,
+            "filters": {"wing": wing, "room": room},
+            "total_before_filter": len(closet_hits),
+            "results": closet_hits,
+        }
+
+    # Fallback: direct drawer search (no closets yet, or closets empty)
     try:
         kwargs = {
             "query_texts": [query],
@@ -136,7 +272,7 @@ def search_memories(
         if where:
             kwargs["where"] = where
 
-        results = col.query(**kwargs)
+        results = drawers_col.query(**kwargs)
     except Exception as e:
         return {"error": f"Search error: {e}"}
 
@@ -157,6 +293,7 @@ def search_memories(
                 "source_file": Path(meta.get("source_file", "?")).name,
                 "similarity": round(max(0.0, 1 - dist), 3),
                 "distance": round(dist, 4),
+                "matched_via": "drawer",
             }
         )
 

tests/test_closets.py

@@ -0,0 +1,316 @@
+"""
+test_closets.py — Tests for the closet (searchable index) layer.
+
+Covers:
+  * build_closet_lines — pointer-line shape, entity extraction, stoplist,
+    quote/header pickup, and the "always emit one line" guarantee.
+  * upsert_closet_lines — pure overwrite (no append), char-limit packing,
+    atomic-line guarantee.
+  * purge_file_closets — wipes prior closets so a re-mine starts clean.
+  * The end-to-end rebuild: re-mining a file fully replaces its closets,
+    including when the prior run produced more numbered closets.
+  * search_memories closet-first path — returns chunk-level hits parsed
+    from `→drawer_ids` pointers, falls back when closets are empty,
+    respects max_distance.
+"""
+
+from mempalace.miner import mine
+from mempalace.palace import (
+    CLOSET_CHAR_LIMIT,
+    build_closet_lines,
+    get_closets_collection,
+    purge_file_closets,
+    upsert_closet_lines,
+)
+from mempalace.searcher import _extract_drawer_ids_from_closet, search_memories
+
+
+# ── build_closet_lines ─────────────────────────────────────────────────
+
+
+class TestBuildClosetLines:
+    def test_emits_pointer_line_shape(self, tmp_path):
+        content = (
+            "# Auth rewrite\n\n"
+            "Decided we need to migrate to passkeys. "
+            "Built the prototype with WebAuthn. "
+            "Reviewed the API surface."
+        )
+        lines = build_closet_lines(
+            "/proj/auth.md",
+            ["drawer_proj_backend_aaa", "drawer_proj_backend_bbb"],
+            content,
+            wing="proj",
+            room="backend",
+        )
+        assert lines, "should always emit at least one line"
+        for line in lines:
+            assert "→" in line, f"line missing pointer arrow: {line!r}"
+            parts = line.split("|")
+            assert len(parts) == 3, f"expected topic|entities|→refs, got {line!r}"
+            assert parts[2].startswith("→")
+
+    def test_extracts_section_headers_as_topics(self):
+        content = "# First Header\nbody\n## Second Header\nmore body"
+        lines = build_closet_lines("/x.md", ["d1"], content, "w", "r")
+        joined = "\n".join(lines).lower()
+        assert "first header" in joined
+        assert "second header" in joined
+
+    def test_entity_stoplist_filters_sentence_starters(self):
+        # "When", "After", "The" repeat 3+ times — old code would index them
+        # as entities. New code's stoplist drops them.
+        content = (
+            "When the pipeline ran, the result was good. "
+            "When the user logged in, the token was issued. "
+            "After the migration, the latency dropped. "
+            "After the rollback, the latency rose. "
+            "The new flow is stable. The audit cleared."
+        )
+        lines = build_closet_lines("/x.md", ["d1"], content, "w", "r")
+        # Entities sit between the two pipes
+        entity_segments = [line.split("|")[1] for line in lines]
+        for seg in entity_segments:
+            tokens = set(seg.split(";")) if seg else set()
+            assert "When" not in tokens
+            assert "After" not in tokens
+            assert "The" not in tokens
+
+    def test_real_proper_nouns_survive_stoplist(self):
+        content = (
+            "Igor reviewed the diff. Milla wrote the spec. "
+            "Igor pushed the fix. Milla approved the PR. "
+            "Igor and Milla shipped together."
+        )
+        lines = build_closet_lines("/x.md", ["d1"], content, "w", "r")
+        entity_segments = [line.split("|")[1] for line in lines]
+        joined_entities = ";".join(entity_segments)
+        assert "Igor" in joined_entities
+        assert "Milla" in joined_entities
+
+    def test_emits_fallback_line_when_nothing_extractable(self):
+        # No headers, no action verbs, no quotes, no repeated capitalized words
+        content = "lorem ipsum dolor sit amet consectetur adipiscing elit"
+        lines = build_closet_lines("/x/notes.txt", ["d1"], content, "wing", "room")
+        assert len(lines) == 1
+        assert "wing/room/notes" in lines[0]
+        assert "→d1" in lines[0]
+
+    def test_pointer_references_first_three_drawers(self):
+        ids = [f"drawer_{i}" for i in range(10)]
+        lines = build_closet_lines("/x.md", ids, "# A\n# B", "w", "r")
+        assert all("→drawer_0,drawer_1,drawer_2" in line for line in lines)
+
+
+# ── upsert_closet_lines ───────────────────────────────────────────────
+
+
+class TestUpsertClosetLines:
+    def test_overwrites_existing_closet_does_not_append(self, palace_path):
+        col = get_closets_collection(palace_path)
+        base = "closet_test_room_abc"
+        meta = {"wing": "test", "room": "room", "source_file": "/x.md"}
+
+        # First mine — three short lines.
+        upsert_closet_lines(col, base, ["alpha|;|→d1", "beta|;|→d2", "gamma|;|→d3"], meta)
+        first = col.get(ids=[f"{base}_01"])
+        assert "alpha" in first["documents"][0]
+        assert "beta" in first["documents"][0]
+
+        # Second mine — entirely different lines. Must replace, not append.
+        upsert_closet_lines(col, base, ["delta|;|→d4", "epsilon|;|→d5"], meta)
+        second = col.get(ids=[f"{base}_01"])
+        doc = second["documents"][0]
+        assert "delta" in doc
+        assert "epsilon" in doc
+        assert "alpha" not in doc, "old closet line leaked into rebuild"
+        assert "beta" not in doc
+
+    def test_packs_into_multiple_closets_without_splitting_lines(self, palace_path):
+        col = get_closets_collection(palace_path)
+        base = "closet_pack_room_def"
+        meta = {"wing": "test", "room": "room", "source_file": "/y.md"}
+
+        # Build lines that approach but never exceed the limit.
+        line = "x" * 600  # well under CLOSET_CHAR_LIMIT
+        n_written = upsert_closet_lines(col, base, [line, line, line, line], meta)
+        # 4 lines @ 600+1 chars = 2404 — should pack into 2 closets (≤1500 each)
+        assert n_written == 2
+
+        for i in range(1, n_written + 1):
+            doc = col.get(ids=[f"{base}_{i:02d}"])["documents"][0]
+            # Every line is intact (never split mid-line)
+            for chunk in doc.split("\n"):
+                assert len(chunk) == 600, f"line was truncated in closet {i}"
+            # Closet stays under the cap
+            assert len(doc) <= CLOSET_CHAR_LIMIT
+
+
+# ── purge_file_closets ────────────────────────────────────────────────
+
+
+class TestPurgeFileClosets:
+    def test_deletes_only_the_targeted_source(self, palace_path):
+        col = get_closets_collection(palace_path)
+        col.upsert(
+            ids=["closet_a_01", "closet_b_01"],
+            documents=["a|;|→d1", "b|;|→d2"],
+            metadatas=[
+                {"source_file": "/keep.md", "wing": "w", "room": "r"},
+                {"source_file": "/drop.md", "wing": "w", "room": "r"},
+            ],
+        )
+        purge_file_closets(col, "/drop.md")
+
+        remaining_ids = set(col.get()["ids"])
+        assert "closet_a_01" in remaining_ids
+        assert "closet_b_01" not in remaining_ids
+
+
+# ── End-to-end rebuild via the project miner ──────────────────────────
+
+
+class TestMinerClosetRebuild:
+    def test_remine_replaces_closets_completely(self, tmp_path):
+        import yaml
+
+        project = tmp_path / "proj"
+        project.mkdir()
+        (project / "mempalace.yaml").write_text(
+            yaml.dump({"wing": "proj", "rooms": [{"name": "general", "description": "x"}]})
+        )
+        target = project / "doc.md"
+
+        # First mine — long content produces multiple numbered closets.
+        first_topics = "\n\n".join(f"# Topic {i}\n" + ("filler text " * 30) for i in range(15))
+        target.write_text(first_topics)
+        palace = tmp_path / "palace"
+        mine(str(project), str(palace), wing_override="proj", agent="test")
+
+        col = get_closets_collection(str(palace))
+        first_pass = col.get(where={"source_file": str(target)})
+        assert first_pass["ids"], "first mine should have written closets"
+        first_ids = set(first_pass["ids"])
+        assert any("topic 0" in (d or "").lower() for d in first_pass["documents"])
+
+        # Touch mtime so file_already_mined doesn't short-circuit, and
+        # rewrite with fewer topics (so the rebuild produces fewer closets
+        # than the first run).
+        import os
+        import time
+
+        target.write_text("# Only Topic Now\n" + ("short body " * 5))
+        new_mtime = os.path.getmtime(target) + 60
+        os.utime(target, (new_mtime, new_mtime))
+        time.sleep(0.01)  # ensure mtime delta is visible
+
+        mine(str(project), str(palace), wing_override="proj", agent="test")
+
+        col = get_closets_collection(str(palace))
+        second_pass = col.get(where={"source_file": str(target)})
+        second_docs = "\n".join(second_pass["documents"]).lower()
+        assert "only topic now" in second_docs
+        for i in range(15):
+            assert (
+                f"topic {i}\n" not in second_docs
+            ), f"stale 'Topic {i}' from first mine survived the rebuild"
+        # Numbered closets that existed only in the larger first run must be gone.
+        leftover = first_ids - set(second_pass["ids"])
+        for stale_id in leftover:
+            assert not col.get(ids=[stale_id])[
+                "ids"
+            ], f"orphan closet {stale_id} from larger first run survived purge"
+
+
+# ── _extract_drawer_ids_from_closet ───────────────────────────────────
+
+
+class TestExtractDrawerIds:
+    def test_parses_single_pointer(self):
+        assert _extract_drawer_ids_from_closet("topic|;|→drawer_x") == ["drawer_x"]
+
+    def test_parses_multiple_pointers_per_line(self):
+        line = "topic|ent|→drawer_a,drawer_b,drawer_c"
+        assert _extract_drawer_ids_from_closet(line) == [
+            "drawer_a",
+            "drawer_b",
+            "drawer_c",
+        ]
+
+    def test_dedupes_across_lines(self):
+        doc = "one|;|→drawer_a,drawer_b\ntwo|;|→drawer_b,drawer_c"
+        assert _extract_drawer_ids_from_closet(doc) == [
+            "drawer_a",
+            "drawer_b",
+            "drawer_c",
+        ]
+
+    def test_empty_doc_returns_empty(self):
+        assert _extract_drawer_ids_from_closet("") == []
+        assert _extract_drawer_ids_from_closet("no arrows here") == []
+
+
+# ── search_memories closet-first path ────────────────────────────────
+
+
+class TestSearchMemoriesClosetFirst:
+    def test_falls_back_to_direct_when_no_closets(self, palace_path, seeded_collection):
+        # seeded_collection populates only mempalace_drawers, not closets.
+        result = search_memories("JWT authentication", palace_path)
+        assert result["results"], "should still find drawer hits via fallback"
+        for hit in result["results"]:
+            assert hit.get("matched_via") == "drawer"
+
+    def test_closet_first_returns_chunk_level_hits(self, palace_path, seeded_collection):
+        # Build a closet that points at the JWT drawer specifically.
+        closets = get_closets_collection(palace_path)
+        closets.upsert(
+            ids=["closet_proj_backend_aaa_01"],
+            documents=["JWT auth tokens|;|→drawer_proj_backend_aaa"],
+            metadatas=[
+                {
+                    "wing": "project",
+                    "room": "backend",
+                    "source_file": "auth.py",
+                }
+            ],
+        )
+
+        result = search_memories("JWT authentication", palace_path)
+        assert result["results"], "closet-first search should hydrate the drawer"
+        top = result["results"][0]
+        assert top["matched_via"] == "closet"
+        # Must be the chunk-level drawer text, not a concatenation of every
+        # drawer in the file.
+        assert "JWT" in top["text"]
+        assert (
+            "Database migrations" not in top["text"]
+        ), "closet path should not glue unrelated drawers together"
+        assert "closet_preview" in top
+        assert "→drawer_proj_backend_aaa" in top["closet_preview"]
+
+    def test_max_distance_filters_closet_hits(self, palace_path, seeded_collection):
+        closets = get_closets_collection(palace_path)
+        closets.upsert(
+            ids=["closet_proj_backend_aaa_01"],
+            documents=["JWT auth tokens|;|→drawer_proj_backend_aaa"],
+            metadatas=[
+                {
+                    "wing": "project",
+                    "room": "backend",
+                    "source_file": "auth.py",
+                }
+            ],
+        )
+
+        # max_distance=0.001 is essentially "must match exactly". The closet
+        # path should reject everything and the caller falls back to direct
+        # search (which also filters with the same threshold).
+        result = search_memories(
+            "completely unrelated query about quantum gardening",
+            palace_path,
+            max_distance=0.001,
+        )
+        # Either no results, or every result respected the threshold.
+        for hit in result["results"]:
+            assert hit["distance"] <= 0.001