From 181e00a9281b2db5e681d1f1d68df8c333466ec9 Mon Sep 17 00:00:00 2001
From: Cristian Deheleanu <160292664+colorpanda82@users.noreply.github.com>
Date: Wed, 15 Jul 2026 15:30:58 +0300
Subject: [PATCH 01/10] fix(mcp_server): reset chromadb System cache on
 staleness reconnect (#2002)

_get_client() detects a peer writer's inode/mtime change and rebuilds the
client via ChromaBackend.make_client(), but chromadb caches its System (and
the live in-memory HNSW segment) keyed by path. The rebuilt client is handed
back the same stale segment, which on its next _persist() overwrites the
on-disk index, destroying records other writers had already indexed. Observed
in a live multi-writer palace: the persisted index count went backwards (4 to 3).

Call the existing _force_chroma_cache_reset() on the staleness path, before
make_client(), so chromadb rebuilds the segment from the on-disk state. The
call is guarded by the existing inode_changed/mtime_changed check, so it has
no effect on first-open.

Adds test_get_client_resets_chroma_system_cache_on_reconnect, which asserts the
reset runs before make_client on an mtime reconnect (fails without the fix).

Refs #1963.
---
 mempalace/mcp_server.py  |  7 +++++++
 tests/test_mcp_server.py | 45 ++++++++++++++++++++++++++++++++++++++++
 2 files changed, 52 insertions(+)

diff --git a/mempalace/mcp_server.py b/mempalace/mcp_server.py
index 801a21e710..995d8ccb90 100644
--- a/mempalace/mcp_server.py
+++ b/mempalace/mcp_server.py
@@ -998,6 +998,13 @@ def _get_client():
         _refresh_vector_disabled_flag()
         if inode_changed or mtime_changed:
             ChromaBackend._quarantined_paths.discard(_config.palace_path)
+            # #2002: a peer process changed chroma.sqlite3 on disk. chromadb
+            # caches its System (and the live HNSW segment) keyed by path, so
+            # make_client() below would hand back the STALE segment, which then
+            # persists its outdated index over the peer's writes, driving the
+            # persisted count backwards. Drop chromadb's shared cache first so
+            # make_client() rebuilds the segment from the on-disk state.
+            _force_chroma_cache_reset()
         _client_cache = ChromaBackend.make_client(_config.palace_path)
         _collection_cache = None
         _collection_cache_backend = None
diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py
index 41c1c537ff..8f9da8999d 100644
--- a/tests/test_mcp_server.py
+++ b/tests/test_mcp_server.py
@@ -4176,6 +4176,51 @@ def spy_prepare(path):
             "_get_client should call _prepare_palace_for_open on reconnect"
         )
 
+    def test_get_client_resets_chroma_system_cache_on_reconnect(
+        self, monkeypatch, config, palace_path, kg
+    ):
+        """``_get_client`` must clear chromadb's path-keyed System/HNSW cache
+        (via ``_force_chroma_cache_reset``) *before* calling ``make_client`` on an
+        inode/mtime reconnect. Otherwise chromadb hands back the stale in-memory
+        HNSW segment, which persists its outdated index over a peer writer's
+        on-disk changes, driving the persisted count backwards (#2002)."""
+        _patch_mcp_server(monkeypatch, config, kg)
+        from mempalace import mcp_server
+        from mempalace.backends.chroma import ChromaBackend
+
+        _client, _col = _get_collection(palace_path, create=True)
+        del _client
+
+        # Prime the cache.
+        mcp_server._get_collection()
+
+        # Simulate a peer writer touching chroma.sqlite3 on disk.
+        old_mtime = mcp_server._palace_db_mtime
+        monkeypatch.setattr(mcp_server, "_palace_db_mtime", old_mtime - 10.0)
+
+        order: list[str] = []
+        real_reset = mcp_server._force_chroma_cache_reset
+        real_make = ChromaBackend.make_client
+
+        def spy_reset():
+            order.append("reset")
+            real_reset()
+
+        @staticmethod
+        def spy_make(path):
+            order.append("make_client")
+            return real_make(path)
+
+        monkeypatch.setattr(mcp_server, "_force_chroma_cache_reset", spy_reset)
+        monkeypatch.setattr(ChromaBackend, "make_client", spy_make)
+
+        mcp_server._get_client()
+
+        assert order == ["reset", "make_client"], (
+            "_get_client must reset chromadb's system cache BEFORE reopening the "
+            "client on a staleness reconnect (#2002)"
+        )
+
     def test_call_kg_retries_after_concurrent_close(self, monkeypatch):
         """A KG closed mid-handler must trigger a one-shot retry with a fresh
         instance — not surface a -32000 to the MCP client."""

From e5db07918277da5cfe9df325af18c46232abca28 Mon Sep 17 00:00:00 2001
From: colorpanda82 <160292664+colorpanda82@users.noreply.github.com>
Date: Wed, 15 Jul 2026 19:21:21 +0300
Subject: [PATCH 02/10] fix(chroma): reset chromadb System cache in
 ChromaBackend._client() on inode/mtime reopen

_client() reconstructs PersistentClient on an inode/mtime change but did not drop chromadb's process-global SharedSystemClient cache first, so the rebuilt client reused the stale path-keyed System (and its in-memory HNSW segment) and could persist an outdated index over on-disk changes -- the same class as #2002, reached via _client() instead of _get_client.

Add SharedSystemClient.clear_system_cache() to the external-change branch of _client(), mirroring mcp_server._force_chroma_cache_reset (#2026) and repair._close_chroma_handles. Backend-level regression test asserts the reset fires on the change reopen, strictly before the reconstruct, and not on first open (chroma-core/chroma#2536, #5843).

Fixes #2028.
---
 mempalace/backends/chroma.py | 35 +++++++++++++++++++
 tests/test_backends.py       | 65 ++++++++++++++++++++++++++++++++++++
 2 files changed, 100 insertions(+)

diff --git a/mempalace/backends/chroma.py b/mempalace/backends/chroma.py
index ee8f10b384..24d2ad6a48 100644
--- a/mempalace/backends/chroma.py
+++ b/mempalace/backends/chroma.py
@@ -1311,6 +1311,33 @@ def _close_client(client) -> None:
         logger.debug("client.close() unavailable or failed", exc_info=True)
 
 
+def _clear_chroma_system_cache() -> None:
+    """Drop chromadb's process-global ``SharedSystemClient`` cache.
+
+    chromadb caches its ``System`` (and the live HNSW segment) keyed by path.
+    A bare ``chromadb.PersistentClient(path=...)`` reopen reuses that cached
+    System, so after a peer/rebuild has changed ``chroma.sqlite3`` on disk we
+    would rebuild against the stale in-memory segment and persist an outdated
+    index over the on-disk changes -- the same data-loss class as #2002,
+    reached via :meth:`ChromaBackend._client` instead of
+    ``mcp_server._get_client``. This mirrors the reset already performed by
+    ``mcp_server._force_chroma_cache_reset`` and ``repair._close_chroma_handles``.
+
+    The clear is process-global (it evicts every palace's cached System, not
+    just this path); chromadb exposes no per-path eviction. It only fires on the
+    inode/mtime-change branch of ``_client``, never the steady-state hot path,
+    so the redundant rebuild cost is bounded to genuine external-change reopens.
+    """
+    try:
+        from chromadb.api.client import SharedSystemClient
+
+        clear = getattr(SharedSystemClient, "clear_system_cache", None)
+        if callable(clear):
+            clear()
+    except Exception:
+        logger.debug("Failed to clear chromadb SharedSystemClient cache", exc_info=True)
+
+
 class ChromaCollection(BaseCollection):
     """Thin adapter translating ChromaDB dict returns into typed results.
 
@@ -2054,6 +2081,14 @@ def _client(self, palace_path: str):
                 or (mtime_appeared and palace_path in self._freshness)
             ):
                 ChromaBackend._quarantined_paths.discard(palace_path)
+                # #2028: the same external change means chromadb's path-keyed
+                # System cache is now stale. Reconstructing PersistentClient
+                # below would reuse the cached System (and its in-memory HNSW
+                # segment), so drop the shared cache first -- otherwise the
+                # rebuilt client persists an outdated index over the on-disk
+                # change. Gated on genuine external change (not first open) so
+                # cold opens never pay the global-evict cost.
+                _clear_chroma_system_cache()
             ChromaBackend._prepare_palace_for_open(palace_path)
             cached = chromadb.PersistentClient(path=palace_path)
             self._clients[palace_path] = cached
diff --git a/tests/test_backends.py b/tests/test_backends.py
index 3168ec73dc..92a1b3610a 100644
--- a/tests/test_backends.py
+++ b/tests/test_backends.py
@@ -1827,6 +1827,71 @@ class DummyClient:
     ]
 
 
+def test_chroma_backend_resets_system_cache_on_inode_change(tmp_path, monkeypatch):
+    """#2028: ``_client`` must drop chromadb's path-keyed ``SharedSystemClient``
+    cache *before* reconstructing ``PersistentClient`` on an inode/mtime change.
+
+    chromadb caches its ``System`` (and live HNSW segment) keyed by path, so a
+    bare reopen reuses the stale segment and persists an outdated index over a
+    peer/rebuild's on-disk changes -- the #2002 data-loss class reached via
+    ``_client`` instead of ``mcp_server._get_client``. The reset must fire only
+    on a genuine external change (not first open) and must precede the reopen.
+    """
+    palace = tmp_path / "palace"
+    palace.mkdir()
+    (palace / "chroma.sqlite3").write_text("")
+
+    events = []
+
+    # Neutralize the on-disk HNSW pre-checks so the test exercises only the
+    # cache-reset / client-rebuild ordering.
+    for _name in (
+        "_fix_missing_collection_type",
+        "_fix_blob_seq_ids",
+        "quarantine_invalid_hnsw_metadata",
+        "quarantine_stale_hnsw",
+    ):
+        monkeypatch.setattr(
+            f"mempalace.backends.chroma.{_name}", lambda path, *a, **k: []
+        )
+
+    monkeypatch.setattr(ChromaBackend, "_quarantined_paths", set())
+
+    class DummyClient:
+        pass
+
+    def _record_open(path):
+        events.append(("open", path))
+        return DummyClient()
+
+    monkeypatch.setattr(
+        "mempalace.backends.chroma.chromadb.PersistentClient", _record_open
+    )
+
+    from chromadb.api.client import SharedSystemClient
+
+    def _record_clear(*args, **kwargs):
+        events.append(("clear", None))
+
+    monkeypatch.setattr(SharedSystemClient, "clear_system_cache", _record_clear)
+
+    backend = ChromaBackend()
+    # ``_db_stat`` is called twice per ``_client`` call (freshness check, then
+    # re-stat after reopen). Same inode on the first call (first open, no prior
+    # freshness -> no reset), changed inode on the second (external change).
+    stats = iter([(1, 1.0), (1, 1.0), (2, 2.0), (2, 2.0)])
+    monkeypatch.setattr(backend, "_db_stat", lambda path: next(stats))
+
+    backend._client(str(palace))  # first open: no external change -> no clear
+    backend._client(str(palace))  # inode 1 -> 2: clear, then reopen
+
+    assert events == [
+        ("open", str(palace)),  # first open, no cache reset
+        ("clear", None),  # #2028: reset fires on the inode change...
+        ("open", str(palace)),  # ...strictly before the PersistentClient reopen
+    ], events
+
+
 def test_explain_ef_mismatch_recognizes_chromadb_conflict():
     """When ChromaDB rejects a collection read due to an EF-name mismatch
     (user changed MEMPALACE_EMBEDDING_MODEL on an existing palace), the

From e672a3f7f098ea617d30ff288930ca286f364c7f Mon Sep 17 00:00:00 2001
From: Cristian Deheleanu <160292664+colorpanda82@users.noreply.github.com>
Date: Fri, 17 Jul 2026 21:37:31 +0300
Subject: [PATCH 03/10] test(chroma): reformat test_backends.py to satisfy ruff
 format

Two monkeypatch.setattr calls were wrapped across lines that fit within
the line length; ruff format --check flagged them. Formatter-only, no
behavior change.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Gg6g5efZ1rNbBTHqGz2Tjw
---
 tests/test_backends.py | 8 ++------
 1 file changed, 2 insertions(+), 6 deletions(-)

diff --git a/tests/test_backends.py b/tests/test_backends.py
index 92a1b3610a..262d43d286 100644
--- a/tests/test_backends.py
+++ b/tests/test_backends.py
@@ -1851,9 +1851,7 @@ def test_chroma_backend_resets_system_cache_on_inode_change(tmp_path, monkeypatc
         "quarantine_invalid_hnsw_metadata",
         "quarantine_stale_hnsw",
     ):
-        monkeypatch.setattr(
-            f"mempalace.backends.chroma.{_name}", lambda path, *a, **k: []
-        )
+        monkeypatch.setattr(f"mempalace.backends.chroma.{_name}", lambda path, *a, **k: [])
 
     monkeypatch.setattr(ChromaBackend, "_quarantined_paths", set())
 
@@ -1864,9 +1862,7 @@ def _record_open(path):
         events.append(("open", path))
         return DummyClient()
 
-    monkeypatch.setattr(
-        "mempalace.backends.chroma.chromadb.PersistentClient", _record_open
-    )
+    monkeypatch.setattr("mempalace.backends.chroma.chromadb.PersistentClient", _record_open)
 
     from chromadb.api.client import SharedSystemClient
 

From 77770946138a573ac39b14a5232cccf25b59b9be Mon Sep 17 00:00:00 2001
From: Cristian Deheleanu <160292664+colorpanda82@users.noreply.github.com>
Date: Fri, 17 Jul 2026 21:38:00 +0300
Subject: [PATCH 04/10] test(mcp_server): re-acquire closets handle after
 delete_by_source

The #2002 staleness reconnect makes _get_client() call
_force_chroma_cache_reset(), which clears chromadb's path-keyed
SharedSystemClient cache. Two TestDeleteBySource tests grabbed a
closets collection handle *before* calling tool_delete_by_source and
then asserted on it afterwards, by which point the reset had dropped
the Rust binding underneath the handle (AttributeError:
'RustBindingsAPI' object has no attribute 'bindings').

Re-acquire the closets collection after the tool call in both tests.
Production callers already re-acquire fresh handles per call, so this
is a test-lifetime issue, not a regression in the fix.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Gg6g5efZ1rNbBTHqGz2Tjw
---
 tests/test_mcp_server.py | 12 ++++++++++--
 1 file changed, 10 insertions(+), 2 deletions(-)

diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py
index 8f9da8999d..37c7ab88b8 100644
--- a/tests/test_mcp_server.py
+++ b/tests/test_mcp_server.py
@@ -2707,12 +2707,16 @@ def test_dry_run_reports_count_without_deleting(self, monkeypatch, config, palac
     def test_dry_run_reports_closet_match_count(self, monkeypatch, config, palace_path, kg):
         """Dry run surfaces the closet blast radius (#1722) without deleting."""
         self._seed(monkeypatch, config, palace_path, kg)
-        closets_col = self._seed_closets(palace_path)
+        self._seed_closets(palace_path)
         from mempalace.mcp_server import tool_delete_by_source
+        from mempalace.palace import get_closets_collection
 
         result = tool_delete_by_source("results_mempal_hybrid_v4_session_1.jsonl")
         assert result["dry_run"] is True
         assert result["closet_match_count"] == 2
+        # Re-acquire: the staleness reconnect drops chromadb's path-keyed System
+        # cache (#2002), so a handle taken before the call is dead by now.
+        closets_col = get_closets_collection(palace_path, create=False)
         # Nothing removed — all three closets still present.
         assert len(closets_col.get(include=[])["ids"]) == 3
 
@@ -2731,13 +2735,17 @@ def test_commit_purges_matching_closets(self, monkeypatch, config, palace_path,
         """Deleting by source purges the matching closets too, so the AAAK
         index keeps no stale pointers at the now-deleted drawers (#1722)."""
         self._seed(monkeypatch, config, palace_path, kg)
-        closets_col = self._seed_closets(palace_path)
+        self._seed_closets(palace_path)
         from mempalace.mcp_server import tool_delete_by_source
+        from mempalace.palace import get_closets_collection
 
         result = tool_delete_by_source("results_mempal_hybrid_v4_session_1.jsonl", dry_run=False)
         assert result["success"] is True
         assert result["deleted"] == 2
         assert result["closets_deleted"] == 2
+        # Re-acquire: the staleness reconnect drops chromadb's path-keyed System
+        # cache (#2002), so a handle taken before the call is dead by now.
+        closets_col = get_closets_collection(palace_path, create=False)
         # The two benchmark closets are gone; the real-client closet survives.
         remaining = closets_col.get(include=["metadatas"])
         sources = {m["source_file"] for m in remaining["metadatas"]}

From 3da1d79e15c31166e64778b65c12707f4c70f4e4 Mon Sep 17 00:00:00 2001
From: KeilerHirsch <KeilerHirsch@users.noreply.github.com>
Date: Tue, 28 Jul 2026 15:47:24 +0200
Subject: [PATCH 05/10] fix(miner): close four re-mine safety gaps in
 process_file

#21 (CRITICAL, data-loss): multi-batch re-mine had no completion marker.
A mid-file crash after batch 1 committed but before a later batch left
permanently silent partial data -- the surviving drawers shared the
file's unchanged on-disk mtime, so file_already_mined() treated the
file as fully mined forever. Every chunk now carries chunk_total, and
file_already_mined() verifies a matching-mtime group's drawer count
reaches chunk_total before reporting True. Drawers with no chunk_total
(legacy rows, single-shot add_drawer()) are trusted as before.

#22 (HIGH, correctness/TOCTOU): source_mtime was captured via a fresh
os.path.getmtime() well after content was read, chunked, and
room-detected. A file appended to in that window got its new drawers
stamped with an mtime that already matched the (now newer) on-disk
state, so the appended tail was silently, permanently skipped on every
future mine. _read_text_no_follow now returns (content, mtime) from
the same fstat() that validates the file; process_file threads that
single value through instead of re-stating.

#23 (HIGH, silent-failure): a failed stale-drawer purge was swallowed
to a debug log and mining proceeded anyway, silently producing
duplicate or orphaned drawers. A purge failure now aborts this file's
mine attempt (old drawers' stored mtime is untouched, so the next mine
still sees a mismatch and retries) and prints a visible warning,
matching every other degraded path in this module.

#24 (LOW, data-loss): the old-drawer delete ran unconditionally, but
the closet purge+rebuild only ran when drawers_added > 0 -- a file
whose chunks all landed below min_chunk_size after boundary-splitting
lost its drawers but kept stale closets pointing at now-deleted IDs.
purge_file_closets now runs whenever the delete-and-rebuild cycle
does, regardless of the new chunk count; only the rebuild itself stays
conditional.

169 tests pass across test_miner.py/test_convo_miner*.py/test_palace.py/
test_hallways.py/test_format_miner.py/test_miner_fts5_validation.py, no
regressions. Full suite: 1 unrelated pre-existing flake in
test_mcp_server.py (module-global peer-writer-lock state leaking across
test files in full-suite ordering -- passes standalone and as a full
file; the diff here never touches mcp_server.py).
---
 mempalace/miner.py  |  71 +++++++++---
 mempalace/palace.py |  23 +++-
 tests/test_miner.py | 272 ++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 349 insertions(+), 17 deletions(-)

diff --git a/mempalace/miner.py b/mempalace/miner.py
index 48d916460c..438a698a7b 100644
--- a/mempalace/miner.py
+++ b/mempalace/miner.py
@@ -57,7 +57,12 @@ def _path_within_root(path: Path, root: Path) -> bool:
         return False
 
 
-def _read_text_no_follow(filepath: Path, root: Path) -> Optional[str]:
+def _read_text_no_follow(filepath: Path, root: Path) -> Optional[tuple[str, float]]:
+    """Read ``filepath`` and return ``(content, mtime)`` from the SAME
+    ``fstat()`` call that validated the file, so callers never need a
+    separate, later ``os.path.getmtime()`` that could observe a file
+    modified in between (see #22: a stale re-stat lets appended content
+    be silently and permanently skipped)."""
     if not _path_within_root(filepath, root):
         return None
     flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
@@ -67,9 +72,10 @@ def _read_text_no_follow(filepath: Path, root: Path) -> Optional[str]:
         st = os.fstat(fd)
         if not stat.S_ISREG(st.st_mode) or st.st_size > MAX_FILE_SIZE:
             return None
+        mtime = st.st_mtime
         with os.fdopen(fd, "r", encoding="utf-8", errors="replace") as f:
             fd = -1
-            return f.read()
+            return f.read(), mtime
     except OSError:
         return None
     finally:
@@ -1368,6 +1374,7 @@ def _build_drawer_metadata(
     line_start: Optional[int] = None,
     line_end: Optional[int] = None,
     content_date: Optional[str] = None,
+    chunk_total: Optional[int] = None,
 ) -> dict:
     """Build the metadata dict for one drawer without upserting.
 
@@ -1383,6 +1390,14 @@ def _build_drawer_metadata(
     (legacy callers, pre-Tier-6a drawers), the keys are absent from the
     returned dict and downstream code falls back to ``filed_at`` for the
     date and the 3-segment closet pointer format.
+
+    ``chunk_total`` — the total number of chunks this mining pass expects
+    to write for ``source_file`` (see #21). Every chunk of the same pass
+    carries the same value so ``file_already_mined`` can tell "N of N
+    batches committed" from "crashed after batch 1 of N", instead of
+    treating any surviving drawer with a matching mtime as proof the file
+    is fully mined. ``None`` for legacy callers (e.g. ``add_drawer``,
+    which is inherently a single atomic write with no partial-batch risk).
     """
     metadata = {
         "wing": wing,
@@ -1402,6 +1417,8 @@ def _build_drawer_metadata(
         metadata["line_end"] = line_end
     if content_date:
         metadata["content_date"] = content_date
+    if chunk_total is not None:
+        metadata["chunk_total"] = chunk_total
     metadata["hall"] = detect_hall(content)
     entities = _extract_entities_for_metadata(content)
     if entities:
@@ -1469,9 +1486,10 @@ def process_file(
     if not dry_run and file_already_mined(collection, source_file, check_mtime=True):
         return 0, "general", None
 
-    content = _read_text_no_follow(filepath, project_path)
-    if content is None:
+    read_result = _read_text_no_follow(filepath, project_path)
+    if read_result is None:
         return 0, "general", None
+    content, read_mtime = read_result
 
     content = content.strip()
     if len(content) < effective_min:
@@ -1518,19 +1536,34 @@ def process_file(
         # hnswlib's thread-unsafe updatePoint path and can segfault on macOS ARM
         # with chromadb 0.6.3) into a clean delete+insert, bypassing the update
         # path entirely.
+        #
+        # A failed purge must abort this file's mine attempt rather than fall
+        # through to upsert: proceeding would either leave stale tail entries
+        # as permanent orphans (old chunk count > new) or silently overwrite
+        # only the overlapping chunk_index positions (not a real re-mine) --
+        # see #23. Returning here (without touching source_mtime/chunk_total)
+        # leaves the old drawers' stored mtime untouched, so the next mine
+        # still sees a mismatch against the current on-disk mtime and retries.
         try:
             collection.delete(where={"source_file": source_file})
-        except Exception:
+        except Exception as exc:
+            print(
+                f"  ! [skip] {filepath.name[:50]:50} stale-drawer purge failed "
+                f"({exc!r}); leaving existing drawers untouched, will retry "
+                f"on the next mine",
+                file=sys.stderr,
+            )
             logger.debug("Stale-drawer purge failed for %s", source_file, exc_info=True)
+            return 0, room, None
 
-        # Batch chunks into bounded upserts so the embedding model sees many
-        # chunks per forward pass without building one huge Chroma/SQLite
-        # request for pathological files. A bad chunk can fail its sub-batch;
-        # that is the deliberate trade-off for amortizing embedding overhead.
-        try:
-            source_mtime = os.path.getmtime(source_file)
-        except OSError:
-            source_mtime = None
+        # source_mtime is the mtime paired with the content actually read
+        # above (from _read_text_no_follow's own fstat), not a fresh re-stat
+        # here -- see #22. Re-statting separately can observe a file that was
+        # appended to between the read and this point, stamping drawers with
+        # an mtime that doesn't match what was actually chunked; the next
+        # mine's freshness check then sees stored-mtime == current-disk-mtime
+        # and silently, permanently skips the appended tail.
+        source_mtime = read_mtime
 
         # Tier 6a content-date: extract once per file (not per chunk) and
         # share across all chunks. Reads filename / frontmatter / content /
@@ -1565,6 +1598,7 @@ def process_file(
                         line_start=chunk.get("line_start"),
                         line_end=chunk.get("line_end"),
                         content_date=file_content_date,
+                        chunk_total=len(chunks),
                     )
                 )
             assert_no_collisions(list(zip(batch_ids, batch_metas)), collection)
@@ -1577,8 +1611,14 @@ def process_file(
             all_metas.extend(batch_metas)
 
         # 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.
+        # Purge unconditionally: the old drawers this closet pointed at were
+        # already deleted above regardless of how many chunks survived this
+        # pass's own length filter, so a re-mine that ends up with zero filed
+        # drawers must still end up with zero closets, not stale ones
+        # dangling on deleted drawer IDs (see #24). Only the closet
+        # rebuild itself is conditional on there being new drawers to point at.
+        if closets_col:
+            purge_file_closets(closets_col, source_file)
         if closets_col and drawers_added > 0:
             drawer_ids = [
                 make_drawer_id_from_chunk(wing, room, source_file, c["chunk_index"]) for c in chunks
@@ -1609,7 +1649,6 @@ def process_file(
             }
             if entities:
                 closet_meta["entities"] = entities
-            purge_file_closets(closets_col, source_file)
             upsert_closet_lines(closets_col, closet_id_base, closet_lines, closet_meta)
 
     return drawers_added, room, None
diff --git a/mempalace/palace.py b/mempalace/palace.py
index 49ad06c551..01d4fdd544 100644
--- a/mempalace/palace.py
+++ b/mempalace/palace.py
@@ -1248,6 +1248,15 @@ def file_already_mined(
     that extraction mode so exchange-mode and general-mode drawers can coexist
     for the same source transcript. Legacy drawers without extract_mode are
     treated as exchange-mode drawers.
+
+    A drawer whose metadata carries ``chunk_total`` (see #21) is only
+    counted toward a match once its stored_mtime group has accumulated at
+    least that many drawers -- guarding against a mid-file crash between
+    upsert batches, where the surviving drawers share the current mtime
+    (the file itself was never touched) but are short of the full set. A
+    drawer with no ``chunk_total`` (legacy rows, or a single-shot
+    ``add_drawer()`` call with no partial-batch risk) is trusted on its own,
+    exactly as before.
     """
     try:
         # Under the additive-mining model, a single ``source_file`` can have
@@ -1264,6 +1273,9 @@ def file_already_mined(
         # first matching group regardless of ordering.
         current_mtime = os.path.getmtime(source_file) if check_mtime else None
         offset = 0
+        # Tracks, per matching stored_mtime group, how many drawers have
+        # been seen so far toward that group's own chunk_total (#21).
+        group_counts: dict = {}
         while True:
             results = collection.get(
                 where={"source_file": source_file},
@@ -1289,7 +1301,16 @@ def file_already_mined(
                 stored_mtime = meta.get("source_mtime")
                 if stored_mtime is None:
                     continue
-                if abs(float(stored_mtime) - current_mtime) < 0.001:
+                if abs(float(stored_mtime) - current_mtime) >= 0.001:
+                    continue
+                chunk_total = meta.get("chunk_total")
+                if chunk_total is None:
+                    # No completion marker on this drawer — can't verify
+                    # completeness for its group, trust the match as before.
+                    return True
+                seen = group_counts.get(stored_mtime, 0) + 1
+                group_counts[stored_mtime] = seen
+                if seen >= chunk_total:
                     return True
             if not ids:
                 break
diff --git a/tests/test_miner.py b/tests/test_miner.py
index 4871e39f99..4319cb1dce 100644
--- a/tests/test_miner.py
+++ b/tests/test_miner.py
@@ -1218,6 +1218,278 @@ def upsert(self, documents, ids, metadatas):
     assert col.batch_sizes == [2, 2, 1]
 
 
+def test_process_file_stamps_chunk_total_for_completion_check(tmp_path, monkeypatch):
+    """Every chunk across every batch of one mining pass must carry the
+    same ``chunk_total`` so ``file_already_mined`` can tell a complete
+    multi-batch mine from one that crashed partway through (#21)."""
+    from mempalace import miner
+
+    class FakeCol:
+        def __init__(self):
+            self.metadatas: list = []
+
+        def get(self, *args, **kwargs):
+            return {"ids": []}
+
+        def delete(self, *args, **kwargs):
+            pass
+
+        def upsert(self, documents, ids, metadatas):
+            self.metadatas.extend(metadatas)
+
+    source = tmp_path / "src.py"
+    source.write_text("print('hello')\n" * 20, encoding="utf-8")
+    chunks = [{"content": f"chunk {i} " * 20, "chunk_index": i} for i in range(5)]
+    col = FakeCol()
+    monkeypatch.setattr(miner, "DRAWER_UPSERT_BATCH_SIZE", 2)
+    monkeypatch.setattr(miner, "chunk_text", lambda content, source_file, **kwargs: chunks)
+    monkeypatch.setattr(miner, "detect_hall", lambda content: "code")
+    monkeypatch.setattr(miner, "_extract_entities_for_metadata", lambda content: "")
+
+    miner.process_file(
+        source,
+        tmp_path,
+        col,
+        "wing",
+        [{"name": "general", "description": "General"}],
+        "agent",
+        False,
+    )
+
+    assert len(col.metadatas) == 5
+    assert all(m["chunk_total"] == 5 for m in col.metadatas), (
+        "not every chunk carries the pass's chunk_total — "
+        "file_already_mined can't verify completeness without it on every row"
+    )
+
+
+def test_process_file_stamps_metadata_with_read_time_mtime_not_a_later_restat(
+    tmp_path, monkeypatch
+):
+    """The stored source_mtime must be the one paired with the content
+    that was actually read and chunked, not a fresh os.path.getmtime()
+    call later in the function (#22). Otherwise a file appended to
+    between the read and the old re-stat point gets stamped with an
+    mtime that matches its (now newer) on-disk state, so the next
+    mine's freshness check thinks nothing changed and the appended tail
+    is silently, permanently skipped."""
+    from mempalace import miner
+
+    class FakeCol:
+        def __init__(self):
+            self.metadatas: list = []
+
+        def get(self, *args, **kwargs):
+            return {"ids": []}
+
+        def delete(self, *args, **kwargs):
+            pass
+
+        def upsert(self, documents, ids, metadatas):
+            self.metadatas.extend(metadatas)
+
+    source = tmp_path / "src.py"
+    source.write_text("print('hello')\n" * 20, encoding="utf-8")
+
+    read_time_mtime = 1_700_000_000.0
+    later_disk_mtime = 1_700_000_999.0  # simulates an append landing after the read
+
+    monkeypatch.setattr(
+        miner,
+        "_read_text_no_follow",
+        lambda filepath, root: ("print('hello')\n" * 20, read_time_mtime),
+    )
+    monkeypatch.setattr(os.path, "getmtime", lambda path: later_disk_mtime)
+    chunks = [{"content": "chunk 0 " * 20, "chunk_index": 0}]
+    col = FakeCol()
+    monkeypatch.setattr(miner, "chunk_text", lambda content, source_file, **kwargs: chunks)
+    monkeypatch.setattr(miner, "detect_hall", lambda content: "code")
+    monkeypatch.setattr(miner, "_extract_entities_for_metadata", lambda content: "")
+
+    miner.process_file(
+        source,
+        tmp_path,
+        col,
+        "wing",
+        [{"name": "general", "description": "General"}],
+        "agent",
+        False,
+    )
+
+    assert len(col.metadatas) == 1
+    assert col.metadatas[0]["source_mtime"] == read_time_mtime, (
+        "drawer was stamped with a re-stat'd mtime instead of the one "
+        "paired with the content actually read/chunked"
+    )
+
+
+def test_process_file_aborts_when_stale_drawer_purge_fails(tmp_path, monkeypatch):
+    """A failed stale-drawer purge must abort this file's mine attempt,
+    not silently proceed to upsert on top of it (#23). Proceeding either
+    orphans old tail entries beyond the new chunk count, or overwrites
+    only the overlapping chunk_index positions — not a real re-mine —
+    with zero operator-visible signal unless DEBUG logging happens to be
+    enabled."""
+    from mempalace import miner
+
+    class FailingPurgeCol:
+        def __init__(self):
+            self.upsert_called = False
+
+        def get(self, *args, **kwargs):
+            return {"ids": []}
+
+        def delete(self, *args, **kwargs):
+            raise RuntimeError("simulated transient backend error")
+
+        def upsert(self, documents, ids, metadatas):
+            self.upsert_called = True
+
+    source = tmp_path / "src.py"
+    source.write_text("print('hello')\n" * 20, encoding="utf-8")
+    chunks = [{"content": f"chunk {i} " * 20, "chunk_index": i} for i in range(3)]
+    col = FailingPurgeCol()
+    monkeypatch.setattr(miner, "chunk_text", lambda content, source_file, **kwargs: chunks)
+    monkeypatch.setattr(miner, "detect_hall", lambda content: "code")
+    monkeypatch.setattr(miner, "_extract_entities_for_metadata", lambda content: "")
+
+    drawers, room, skip_reason = miner.process_file(
+        source,
+        tmp_path,
+        col,
+        "wing",
+        [{"name": "general", "description": "General"}],
+        "agent",
+        False,
+    )
+
+    assert col.upsert_called is False, (
+        "process_file inserted new chunks even though the stale-drawer "
+        "purge raised — old and new rows can now coexist as duplicates/orphans"
+    )
+    assert drawers == 0
+
+
+def test_process_file_purges_closets_even_when_all_chunks_filtered_out(tmp_path, monkeypatch):
+    """Old closets must be purged whenever the old drawers were deleted,
+    even if the new content ends up producing zero filed drawers (#24).
+    Otherwise the stale closet entries point at drawer IDs that were
+    just deleted, permanently misdirecting search until the file
+    changes again in a way that produces at least one filed chunk."""
+    from mempalace import miner
+
+    class FakeCol:
+        def get(self, *args, **kwargs):
+            return {"ids": []}
+
+        def delete(self, *args, **kwargs):
+            pass
+
+        def upsert(self, documents, ids, metadatas):
+            raise AssertionError("no chunks should be upserted in this scenario")
+
+    purged: list = []
+
+    source = tmp_path / "src.py"
+    source.write_text("x" * 200, encoding="utf-8")
+    col = FakeCol()
+    # Content passes the file-level min-length gate, but every individual
+    # chunk gets filtered out downstream (e.g. each fragment falls below
+    # min_chunk_size after boundary-splitting) -- modeled directly here.
+    monkeypatch.setattr(miner, "chunk_text", lambda content, source_file, **kwargs: [])
+    monkeypatch.setattr(
+        miner, "purge_file_closets", lambda closets_col, source_file: purged.append(source_file)
+    )
+    monkeypatch.setattr(
+        miner,
+        "upsert_closet_lines",
+        lambda *a, **kw: pytest.fail("should not rebuild closets with zero drawers"),
+    )
+
+    drawers, room, skip_reason = miner.process_file(
+        source,
+        tmp_path,
+        col,
+        "wing",
+        [{"name": "general", "description": "General"}],
+        "agent",
+        False,
+        closets_col=object(),
+    )
+
+    assert drawers == 0
+    assert purged == [str(source)], (
+        "old closets for this source_file were left dangling — they point "
+        "at drawer IDs that collection.delete() already removed"
+    )
+
+
+def test_file_already_mined_detects_incomplete_multi_batch_remine():
+    """A crash between upsert batches must not be mistaken for 'fully
+    mined' (#21). process_file stamps every chunk's metadata with
+    chunk_total (the total chunks expected for this pass). If killed
+    after batch 1 commits but before a later batch, the surviving
+    drawers share the current on-disk mtime (the file itself was never
+    touched) but their count is short of chunk_total — file_already_mined
+    must detect that and report False so the file gets fully re-mined,
+    not silently skipped forever."""
+    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")
+
+        test_file = os.path.join(tmpdir, "big.md")
+        with open(test_file, "w") as f:
+            f.write("content")
+        mtime = os.path.getmtime(test_file)
+
+        # Simulate a crash after only 2 of 3 expected chunks committed.
+        col.add(
+            ids=["d0", "d1"],
+            documents=["chunk 0", "chunk 1"],
+            metadatas=[
+                {
+                    "source_file": test_file,
+                    "source_mtime": mtime,
+                    "normalize_version": NORMALIZE_VERSION,
+                    "chunk_total": 3,
+                },
+                {
+                    "source_file": test_file,
+                    "source_mtime": mtime,
+                    "normalize_version": NORMALIZE_VERSION,
+                    "chunk_total": 3,
+                },
+            ],
+        )
+
+        assert file_already_mined(col, test_file, check_mtime=True) is False, (
+            "2 of 3 expected chunks were treated as a complete mine — the "
+            "missing chunk is now permanently unreachable since the file's "
+            "on-disk mtime never changes again"
+        )
+
+        # The 3rd batch lands (mine resumes/retries and completes the set).
+        col.add(
+            ids=["d2"],
+            documents=["chunk 2"],
+            metadatas=[
+                {
+                    "source_file": test_file,
+                    "source_mtime": mtime,
+                    "normalize_version": NORMALIZE_VERSION,
+                    "chunk_total": 3,
+                }
+            ],
+        )
+        assert file_already_mined(col, test_file, check_mtime=True) is True
+    finally:
+        del col, client
+        shutil.rmtree(tmpdir, ignore_errors=True)
+
+
 # ── normalize_version schema gate ───────────────────────────────────────
 #
 # When the normalization pipeline changes shape (e.g., strip_noise lands),

From db299595baa4b4f825c5961e8d3f9634a8933074 Mon Sep 17 00:00:00 2001
From: Michael Valentsev <michael@valentsev.ru>
Date: Tue, 11 Aug 2026 14:07:53 +0500
Subject: [PATCH 06/10] fix(ingest): never block on a non-regular file (#2221)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`os.walk` and `glob` list a FIFO, a socket and a device node as ordinary
filenames, and MemPalace decides what to read from the suffix. Opening a
FIFO for reading parks in the kernel until a writer appears, so a named
pipe called `notes.md` in a mined directory wedged `mempalace mine`
forever — no output, no error, no progress. `mine --mode convos`,
`sweep`, `init`, `compress` and `split` blocked the same way.

Two shapes are at fault.

Four helpers already refused non-regular files with `fstat` + `S_ISREG`,
but the check sat *after* a blocking `os.open`, so it could never run.
Adding `O_NONBLOCK` to those opens makes the existing type check
reachable: the open returns immediately and the file mode decides, with
no errno guesswork. A FIFO that does have a live writer is refused just
the same. Linux open(2) states the flag has no effect on regular files;
the one exception is a write lease, where a non-blocking open fails
EAGAIN instead of waiting out lease-break-time. Leases are granted on
regular files only, so that branch re-checks the type and then opens
without the flag rather than silently dropping a file that used to be
mined.

The rest guard with `exists()`, which is true for a pipe, and then open
anyway. Those become type checks: a discovery walk drops non-regular
entries before any reader sees them, and a fixed-name read decides with
`is_file()` instead. `scan_project` and `scan_convos` already stat every
candidate for the size limit, so the type check costs no extra syscall.
Where the gate replaced an `open` that sat inside a `try`, it goes in
the same `try`: `is_file()` raises `PermissionError` on a directory
without `x`, which that handler already absorbed.

O_NONBLOCK: miner._read_text_no_follow, convo_miner._is_regular_source_file,
normalize._read_transcript_file, repair._open_regular_file_no_follow.
Type gate: miner.scan_project, miner.load_config, convo_miner.scan_convos,
sweeper.parse_claude_jsonl, sweeper.sweep_directory,
entity_detector.detect_entities, cli._gather_origin_samples,
cli._ensure_mempalace_files_gitignored, cli.cmd_compress, cli.cmd_init,
project_scanner._collect_manifest_names,
project_scanner._parse_gradle_root_project_name,
room_detector_local.detect_rooms_local, llm_refine.collect_corpus_text,
split_mega_files.main, hook_shell.count_human_messages.
---
 mempalace/cli.py                      |  49 +-
 mempalace/convo_miner.py              |  27 +-
 mempalace/entity_detector.py          |   7 +
 mempalace/hook_shell.py               |  10 +
 mempalace/llm_refine.py               |   7 +
 mempalace/miner.py                    |  47 +-
 mempalace/normalize.py                |  21 +-
 mempalace/project_scanner.py          |  14 +-
 mempalace/repair.py                   |  22 +-
 mempalace/room_detector_local.py      |   5 +
 mempalace/split_mega_files.py         |  10 +-
 mempalace/sweeper.py                  |  24 +
 tests/test_non_regular_file_guards.py | 716 ++++++++++++++++++++++++++
 13 files changed, 937 insertions(+), 22 deletions(-)
 create mode 100644 tests/test_non_regular_file_guards.py

diff --git a/mempalace/cli.py b/mempalace/cli.py
index 1fc9997d4b..b28e3b56af 100644
--- a/mempalace/cli.py
+++ b/mempalace/cli.py
@@ -131,6 +131,13 @@ def _gather_origin_samples(project_dir) -> list:
         if total_chars >= _PASS_ZERO_TOTAL_CAP:
             break
         try:
+            # ``scan_for_detection`` picks candidates by extension, so a FIFO
+            # named ``notes.md`` reaches this loop; opening one for reading
+            # blocks until a writer appears. ``is_file()`` stats instead.
+            # It belongs inside the try: it raises PermissionError on an
+            # unreadable directory, which the open below used to absorb.
+            if not filepath.is_file():
+                continue
             with open(filepath, encoding="utf-8", errors="replace") as f:
                 content = f.read(_PASS_ZERO_PER_FILE_CAP)
         except OSError:
@@ -263,9 +270,17 @@ def _ensure_mempalace_files_gitignored(project_dir) -> bool:
     if not (project_path / ".git").exists():
         return False
     gitignore = project_path / ".gitignore"
+    # ``exists()`` is true for a FIFO, and both the read below and the append
+    # at the end of this function would block in the kernel on one. Decide by
+    # type instead: an absent file still yields "" as before, a regular one
+    # is read, and anything else is left untouched.
+    if gitignore.exists() and not gitignore.is_file():
+        return False
     # Force UTF-8: Windows defaults to GBK and chokes on non-ASCII .gitignore
     # comments, killing auto-init even though the file is valid UTF-8.
-    existing = gitignore.read_text(encoding="utf-8", errors="replace") if gitignore.exists() else ""
+    existing = (
+        gitignore.read_text(encoding="utf-8", errors="replace") if gitignore.is_file() else ""
+    )
     existing_lines = {line.strip() for line in existing.splitlines()}
     missing = [p for p in _MEMPALACE_PROJECT_FILES if p not in existing_lines]
     if not missing:
@@ -420,9 +435,19 @@ def cmd_init(args):
         if confirmed["people"] or confirmed["projects"] or confirmed.get("topics"):
             project_path = Path(args.dir).expanduser().resolve()
             entities_path = project_path / "entities.json"
-            with open(entities_path, "w", encoding="utf-8") as f:
-                json.dump(confirmed, f, indent=2, ensure_ascii=False)
-            print(f"  Entities saved: {entities_path}")
+            # Opening a pre-existing FIFO for writing blocks in the kernel
+            # until a reader appears. Only a regular file is a valid target
+            # for the per-project audit trail; the global registry merge
+            # below is unaffected either way.
+            if entities_path.exists() and not entities_path.is_file():
+                print(
+                    f"  ! Not writing entities: {entities_path} is not a regular file",
+                    file=sys.stderr,
+                )
+            else:
+                with open(entities_path, "w", encoding="utf-8") as f:
+                    json.dump(confirmed, f, indent=2, ensure_ascii=False)
+                print(f"  Entities saved: {entities_path}")
 
             from .config import normalize_wing_name
             from .miner import add_to_known_entities
@@ -438,7 +463,14 @@ def cmd_init(args):
         print("  No entities detected -- proceeding with directory-based rooms.")
 
     # Pass 2: detect rooms from folder structure
-    detect_rooms_local(project_dir=args.dir, yes=getattr(args, "yes", False))
+    try:
+        detect_rooms_local(project_dir=args.dir, yes=getattr(args, "yes", False))
+    except OSError as exc:
+        # Writing mempalace.yaml is the point of init; a target it cannot
+        # write (a pre-existing pipe, a full disk) is a hard failure, and a
+        # message beats the traceback this used to produce.
+        print(f"\n  ERROR: {exc}", file=sys.stderr)
+        sys.exit(1)
     cfg.init()
     backend = _backend_arg(args)
     if backend:
@@ -2219,12 +2251,15 @@ def cmd_compress(args):
     # Load dialect (with optional entity config)
     config_path = args.config
     if not config_path:
+        # ``isfile`` rather than ``exists``: the latter is true for a FIFO,
+        # and ``Dialect.from_config`` opens whatever it is handed, which
+        # blocks in the kernel on a pipe named entities.json in the cwd.
         for candidate in ["entities.json", os.path.join(palace_path, "entities.json")]:
-            if os.path.exists(candidate):
+            if os.path.isfile(candidate):
                 config_path = candidate
                 break
 
-    if config_path and os.path.exists(config_path):
+    if config_path and os.path.isfile(config_path):
         dialect = Dialect.from_config(config_path)
         print(f"  Loaded entity config: {config_path}")
     else:
diff --git a/mempalace/convo_miner.py b/mempalace/convo_miner.py
index a3798842ce..c9e4995f49 100644
--- a/mempalace/convo_miner.py
+++ b/mempalace/convo_miner.py
@@ -8,6 +8,7 @@
 Same palace as project mining. Different ingest strategy.
 """
 
+import errno
 import os
 import sys
 import json
@@ -185,10 +186,20 @@ def _path_within_root(path: Path, root: Path) -> bool:
 def _is_regular_source_file(filepath: Path, root: Path) -> bool:
     if not _path_within_root(filepath, root):
         return False
-    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
+    # O_NONBLOCK keeps the S_ISREG verdict below reachable: a blocking open
+    # of a FIFO waits in the kernel for a writer, so a named pipe called
+    # ``session.jsonl`` would hang this check instead of failing it. See the
+    # matching comment in ``miner._read_text_no_follow``, including why the
+    # EAGAIN branch re-checks the type and then opens without the flag.
+    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0)
     fd = -1
     try:
-        fd = os.open(filepath, flags)
+        try:
+            fd = os.open(filepath, flags)
+        except OSError as exc:
+            if exc.errno != errno.EAGAIN or not stat.S_ISREG(os.lstat(filepath).st_mode):
+                raise
+            fd = os.open(filepath, flags & ~getattr(os, "O_NONBLOCK", 0))
         st = os.fstat(fd)
         return stat.S_ISREG(st.st_mode) and st.st_size <= MAX_FILE_SIZE
     except OSError:
@@ -543,7 +554,17 @@ def scan_convos(convo_dir: str, include_subagents: bool = False) -> list:
                 # to stderr to match the SKIP: (symlink) line above; silent
                 # drops at this gate were the original #923 complaint.
                 try:
-                    file_size = filepath.stat().st_size
+                    file_stat = filepath.stat()
+                    # Drop non-regular entries (FIFO, socket, device node)
+                    # before any reader touches them — see the matching
+                    # gate in ``miner.scan_project``.
+                    if not stat.S_ISREG(file_stat.st_mode):
+                        print(
+                            f"  SKIP: {filepath.name} (not a regular file)",
+                            file=sys.stderr,
+                        )
+                        continue
+                    file_size = file_stat.st_size
                     if file_size > MAX_FILE_SIZE:
                         print(
                             f"  SKIP: {filepath.name} ({file_size / (1024 * 1024):.1f} MB)"
diff --git a/mempalace/entity_detector.py b/mempalace/entity_detector.py
index 3854f6d5ec..53216e1ba6 100644
--- a/mempalace/entity_detector.py
+++ b/mempalace/entity_detector.py
@@ -628,6 +628,13 @@ def detect_entities(
         if files_read >= max_files:
             break
         try:
+            # Decide by file type before opening: ``scan_for_detection``
+            # picks candidates by extension, so a FIFO named ``notes.md``
+            # reaches this loop and a blocking open of one waits in the
+            # kernel for a writer that may never come. ``is_file()`` stats
+            # instead of opening and never blocks.
+            if not Path(filepath).is_file():
+                continue
             with open(filepath, encoding="utf-8", errors="replace") as f:
                 content = f.read(MAX_BYTES_PER_FILE)
             all_text.append(content)
diff --git a/mempalace/hook_shell.py b/mempalace/hook_shell.py
index 2f2f32ebcb..bfeea8acee 100644
--- a/mempalace/hook_shell.py
+++ b/mempalace/hook_shell.py
@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import json
+import os
 import re
 import sys
 
@@ -71,9 +72,18 @@ def count_human_messages(path: str) -> int:
     Claude transcripts are UTF-8. Windows Python defaults to cp1252 in many
     environments, so the encoding must be explicit. Invalid bytes are ignored
     to match the hooks' fail-soft behavior.
+
+    A path that exists but is not a regular file counts zero rather than
+    being opened: opening a FIFO for reading blocks in the kernel until a
+    writer appears, and this function has no timeout. A path that does not
+    exist still raises from the ``open`` below, as before.
+    ``mempal_save_hook.sh`` screens with ``[ -f ]``, which is false for a
+    pipe, so the guard here covers callers that do not.
     """
 
     count = 0
+    if os.path.exists(path) and not os.path.isfile(path):
+        return count
     with open(path, encoding="utf-8", errors="ignore") as fh:
         for line in fh:
             try:
diff --git a/mempalace/llm_refine.py b/mempalace/llm_refine.py
index e3afe6b8ec..9dadcb6526 100644
--- a/mempalace/llm_refine.py
+++ b/mempalace/llm_refine.py
@@ -479,6 +479,13 @@ def collect_corpus_text(
     chunks: list[str] = []
     for p in selected:
         try:
+            # ``_walk_prose`` selects by suffix and the stat above reads only
+            # st_mtime, so a FIFO named ``notes.md`` reaches this loop and a
+            # blocking open of one waits for a writer that may never come.
+            # Inside the try: is_file() raises on an unreadable directory,
+            # which the open below already absorbed.
+            if not p.is_file():
+                continue
             with open(p, encoding="utf-8", errors="replace") as f:
                 chunks.append(f.read(max_bytes_per_file))
         except OSError:
diff --git a/mempalace/miner.py b/mempalace/miner.py
index 6915fe6601..54002a10fb 100644
--- a/mempalace/miner.py
+++ b/mempalace/miner.py
@@ -7,6 +7,7 @@
 Stores verbatim chunks as drawers. No summaries. Ever.
 """
 
+import errno
 import os
 import re
 import sys
@@ -60,10 +61,30 @@ def _path_within_root(path: Path, root: Path) -> bool:
 def _read_text_no_follow(filepath: Path, root: Path) -> Optional[str]:
     if not _path_within_root(filepath, root):
         return None
-    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
+    # O_NONBLOCK is what makes the S_ISREG check below reachable. Opening a
+    # FIFO for reading parks in the kernel until a writer shows up, so
+    # without it the fstat never runs and a named pipe carrying a
+    # READABLE_EXTENSIONS suffix wedges the mine forever. With it the open
+    # returns immediately and the *file type* decides — no errno guesswork,
+    # and a FIFO that does have a live writer is rejected just the same.
+    # Linux open(2): "this flag has no effect for regular files and block
+    # devices". POSIX leaves it unspecified outside FIFOs and special files,
+    # and one Linux case is not a no-op — see the EAGAIN branch below.
+    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0)
     fd = -1
     try:
-        fd = os.open(filepath, flags)
+        try:
+            fd = os.open(filepath, flags)
+        except OSError as exc:
+            # A reader that breaks a write lease gets EAGAIN when it passes
+            # O_NONBLOCK, where a blocking open waits out lease-break-time
+            # and succeeds. The kernel grants leases on regular files only
+            # (F_SETLEASE on a pipe gives ENXIO), so re-check the type and
+            # then read it the way this code did before the flag existed;
+            # dropping it would silently lose a file that used to be mined.
+            if exc.errno != errno.EAGAIN or not stat.S_ISREG(os.lstat(filepath).st_mode):
+                raise
+            fd = os.open(filepath, flags & ~getattr(os, "O_NONBLOCK", 0))
         st = os.fstat(fd)
         if not stat.S_ISREG(st.st_mode) or st.st_size > MAX_FILE_SIZE:
             return None
@@ -486,10 +507,14 @@ def load_config(project_dir: str) -> dict:
 
     resolved_project_dir = Path(project_dir).expanduser().resolve()
     config_path = resolved_project_dir / "mempalace.yaml"
-    if not config_path.exists():
+    # ``is_file()`` rather than ``exists()``: the latter is true for a FIFO,
+    # and the ``open`` at the end of this function would then block in the
+    # kernel until a writer appears. A config that is not a regular file is
+    # treated as absent, which lands on the auto-detected defaults below.
+    if not config_path.is_file():
         # Fallback to legacy name
         legacy_path = resolved_project_dir / "mempal.yaml"
-        if legacy_path.exists():
+        if legacy_path.is_file():
             config_path = legacy_path
         else:
             from .config import normalize_wing_name
@@ -1708,7 +1733,19 @@ def scan_project(
             # match the SKIP: (symlink) line above; silent drops at this
             # gate were the original #923 complaint.
             try:
-                file_size = filepath.stat().st_size
+                file_stat = filepath.stat()
+                # Reject anything that is not a regular file before it can
+                # reach a reader. os.walk lists FIFOs, sockets and device
+                # nodes as plain filenames and the extension filter above
+                # decides by name, so ``notes.md`` can be a named pipe.
+                # stat() itself never blocks on one; opening it can.
+                if not stat.S_ISREG(file_stat.st_mode):
+                    print(
+                        f"  SKIP: {filepath.name} (not a regular file)",
+                        file=sys.stderr,
+                    )
+                    continue
+                file_size = file_stat.st_size
                 if file_size > MAX_FILE_SIZE:
                     print(
                         f"  SKIP: {filepath.name} ({file_size / (1024 * 1024):.1f} MB)"
diff --git a/mempalace/normalize.py b/mempalace/normalize.py
index 9503aeb0a5..943feec823 100644
--- a/mempalace/normalize.py
+++ b/mempalace/normalize.py
@@ -19,6 +19,7 @@
 No API key. No internet. Everything local.
 """
 
+import errno
 import json
 import os
 import re
@@ -120,17 +121,29 @@ def _read_transcript_file(filepath: str) -> str:
     and normalize_conversations() both need: no symlinks, regular files only,
     size-capped, BOM-tolerant.
     """
-    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
+    # O_NONBLOCK keeps the "not a regular file" check below reachable: a
+    # blocking open of a FIFO waits in the kernel for a writer, so the
+    # S_ISREG test never runs. See ``miner._read_text_no_follow``, including
+    # why the EAGAIN branch re-checks the type and retries without the flag.
+    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0)
     if os.path.islink(filepath):
         raise IOError(f"Could not read {filepath}: symlinked files are skipped")
     fd = -1
     try:
-        fd = os.open(filepath, flags)
+        try:
+            fd = os.open(filepath, flags)
+        except OSError as exc:
+            if exc.errno != errno.EAGAIN or not stat.S_ISREG(os.lstat(filepath).st_mode):
+                raise
+            fd = os.open(filepath, flags & ~getattr(os, "O_NONBLOCK", 0))
         file_stat = os.fstat(fd)
         if not stat.S_ISREG(file_stat.st_mode):
-            raise IOError(f"Could not read {filepath}: not a regular file")
+            # Text stays prefix-free: this raise is inside the ``try``, so the
+            # ``except OSError`` below composes "Could not read <path>: ...".
+            raise IOError("not a regular file")
         if file_stat.st_size > 500 * 1024 * 1024:  # 500 MB safety limit
-            raise IOError(f"File too large ({file_stat.st_size // (1024 * 1024)} MB): {filepath}")
+            # Prefix-free for the same reason as the branch above.
+            raise IOError(f"file too large ({file_stat.st_size // (1024 * 1024)} MB)")
         with os.fdopen(fd, "r", encoding="utf-8-sig", errors="replace") as f:
             fd = -1
             return f.read()
diff --git a/mempalace/project_scanner.py b/mempalace/project_scanner.py
index d92b8167b1..dd77c45240 100644
--- a/mempalace/project_scanner.py
+++ b/mempalace/project_scanner.py
@@ -188,6 +188,12 @@ def _parse_pom(path: Path) -> Optional[str]:
 
 
 def _parse_gradle_root_project_name(path: Path) -> Optional[str]:
+    # ``_parse_gradle`` reaches this with a SIBLING path it constructs itself
+    # (``build.gradle`` next to ``settings.gradle``), which the manifest walk
+    # never vetted. Opening a FIFO for reading blocks in the kernel until a
+    # writer appears; ``is_file()`` stats instead and never blocks.
+    if not path.is_file():
+        return None
     try:
         text = path.read_text(encoding="utf-8", errors="replace")
     except OSError:
@@ -421,7 +427,13 @@ def _collect_manifest_names(repo_root: Path) -> list[tuple[str, str, Path]]:
             parser = MANIFEST_PARSERS.get(fname)
             if not parser:
                 continue
-            name = parser(dirpath / fname)
+            manifest_path = dirpath / fname
+            # Every parser below opens the path. A FIFO named
+            # ``package.json`` would park that open in the kernel until a
+            # writer appears; ``is_file()`` stats instead and never blocks.
+            if not manifest_path.is_file():
+                continue
+            name = parser(manifest_path)
             if name:
                 found.append((fname, name, dirpath))
     return sorted(found, key=lambda entry: _manifest_sort_key(entry, repo_root))
diff --git a/mempalace/repair.py b/mempalace/repair.py
index 902c4f7b0d..d8048ee972 100644
--- a/mempalace/repair.py
+++ b/mempalace/repair.py
@@ -30,6 +30,7 @@
 """
 
 import argparse
+import errno
 import os
 import shutil
 import sqlite3
@@ -62,10 +63,29 @@ def _no_follow_flag() -> int:
     return getattr(os, "O_NOFOLLOW", 0)
 
 
+def _non_blocking_flag() -> int:
+    """Return O_NONBLOCK, or 0 where the platform has no such flag (Windows).
+
+    Without it the ``S_ISREG`` refusal in ``_open_regular_file_no_follow``
+    is unreachable for a FIFO: opening one for reading blocks in the kernel
+    until a writer appears, so ``repair`` would wedge instead of refusing.
+    """
+    return getattr(os, "O_NONBLOCK", 0)
+
+
 def _open_regular_file_no_follow(path: str) -> int:
     if os.path.islink(path):
         raise RuntimeError(f"Refusing symlinked file: {path}")
-    fd = os.open(path, os.O_RDONLY | _no_follow_flag())
+    flags = os.O_RDONLY | _no_follow_flag() | _non_blocking_flag()
+    try:
+        fd = os.open(path, flags)
+    except OSError as exc:
+        # EAGAIN here is a write-lease break, which the kernel grants on
+        # regular files only, so re-check the type and open the way this
+        # helper did before the flag existed. Anything else propagates.
+        if exc.errno != errno.EAGAIN or not stat.S_ISREG(os.lstat(path).st_mode):
+            raise
+        fd = os.open(path, flags & ~_non_blocking_flag())
     try:
         st = os.fstat(fd)
         if not stat.S_ISREG(st.st_mode):
diff --git a/mempalace/room_detector_local.py b/mempalace/room_detector_local.py
index f754f463a2..1c88f26557 100644
--- a/mempalace/room_detector_local.py
+++ b/mempalace/room_detector_local.py
@@ -292,6 +292,11 @@ def save_config(project_dir: str, project_name: str, rooms: list):
         ],
     }
     config_path = Path(project_dir).expanduser().resolve() / "mempalace.yaml"
+    # Opening a pre-existing FIFO for writing blocks in the kernel until a
+    # reader appears. Only a regular file is a valid config target; refuse
+    # loudly rather than park ``init`` with no output.
+    if config_path.exists() and not config_path.is_file():
+        raise OSError(f"Refusing to write config: {config_path} is not a regular file")
     with open(config_path, "w") as f:
         yaml.dump(config, f, default_flow_style=False, sort_keys=False)
 
diff --git a/mempalace/split_mega_files.py b/mempalace/split_mega_files.py
index 8ce0c859b3..d876c7722a 100644
--- a/mempalace/split_mega_files.py
+++ b/mempalace/split_mega_files.py
@@ -26,6 +26,7 @@
 import json
 import os
 import re
+import stat
 from pathlib import Path
 
 HOME = Path.home()
@@ -272,7 +273,14 @@ def main():
     mega_files = []
     max_scan_size = 500 * 1024 * 1024  # 500 MB
     for f in files:
-        if f.stat().st_size > max_scan_size:
+        file_stat = f.stat()
+        # ``glob`` lists a FIFO named ``x.txt`` like any other match, and
+        # read_text() on one blocks in the kernel until a writer appears.
+        # stat() never blocks, so the type decides before the open does.
+        if not stat.S_ISREG(file_stat.st_mode):
+            print(f"  SKIP: {f.name} (not a regular file)")
+            continue
+        if file_stat.st_size > max_scan_size:
             print(f"  SKIP: {f.name} exceeds {max_scan_size // (1024 * 1024)} MB limit")
             continue
         lines = f.read_text(errors="replace").splitlines(keepends=True)
diff --git a/mempalace/sweeper.py b/mempalace/sweeper.py
index d036b6e670..715e41d2ab 100644
--- a/mempalace/sweeper.py
+++ b/mempalace/sweeper.py
@@ -40,6 +40,7 @@
 
 import json
 import logging
+import stat
 import sys
 from datetime import datetime
 from pathlib import Path
@@ -101,7 +102,16 @@ def parse_claude_jsonl(path: str) -> Iterator[dict]:
     queue-operation, last-prompt) are filtered out. Malformed lines are
     skipped silently — data quality is the transcript writer's problem,
     not ours.
+
+    Raises ``OSError`` when ``path`` is not a regular file. ``rglob`` in
+    ``sweep_directory`` lists a FIFO named ``session.jsonl`` like any
+    other match, and opening one for reading blocks in the kernel until a
+    writer appears — an unbounded hang for the whole sweep. ``stat`` never
+    blocks on one, and it raises for a missing path exactly as ``open``
+    did before, so callers see the same error for the same mistake.
     """
+    if not stat.S_ISREG(Path(path).stat().st_mode):
+        raise OSError(f"Refusing non-regular file: {path}")
     with open(path, "r", encoding="utf-8", errors="replace") as f:
         for line in f:
             line = line.strip()
@@ -337,6 +347,20 @@ def sweep_directory(dir_path: str, palace_path: str) -> dict:
 
     failures: list[dict] = []
     for f in files:
+        # A non-regular match is not a sweep failure, it is nothing to sweep.
+        # ``rglob`` lists a FIFO or a symlink to /dev/null like any other
+        # ``*.jsonl``; report it the way ``miner.scan_project`` reports one
+        # and leave ``failures`` (and the exit status) for real errors.
+        # ``parse_claude_jsonl`` still refuses one, for callers arriving by
+        # another route.
+        try:
+            regular = stat.S_ISREG(f.stat().st_mode)
+        except OSError as exc:
+            print(f"  SKIP: {f.name} (stat error: {exc.strerror or exc})", file=sys.stderr)
+            continue
+        if not regular:
+            print(f"  SKIP: {f.name} (not a regular file)", file=sys.stderr)
+            continue
         try:
             result = sweep(str(f), palace_path, source_label=str(f))
         except Exception as exc:
diff --git a/tests/test_non_regular_file_guards.py b/tests/test_non_regular_file_guards.py
new file mode 100644
index 0000000000..9ea2703663
--- /dev/null
+++ b/tests/test_non_regular_file_guards.py
@@ -0,0 +1,716 @@
+"""Non-regular files must never wedge an ingest command.
+
+``os.walk``/``rglob`` list a FIFO, a socket and a device node as ordinary
+filenames, and MemPalace decides what to read by extension. Opening a FIFO
+for reading parks in the kernel until a writer appears, so a named pipe
+called ``notes.md`` sitting in a mined directory used to hang ``mine``,
+``sweep`` and ``init`` forever — no output, no error, no progress.
+
+Every check here is wrapped in :func:`hard_timeout`. A regression must turn
+this file red; it must not hang the suite (an unbounded blocking open would
+otherwise stall pytest itself, which reports as "still running", not as a
+failure).
+"""
+
+import argparse
+import errno
+import hashlib
+import os
+import signal
+import socket
+import stat as stat_module
+import threading
+from contextlib import contextmanager
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+import yaml
+
+from mempalace.cli import (
+    _ensure_mempalace_files_gitignored,
+    _gather_origin_samples,
+    cmd_compress,
+    cmd_init,
+)
+from mempalace.convo_miner import _is_regular_source_file, scan_convos
+from mempalace.entity_detector import detect_entities
+from mempalace.format_miner import ExtractionStatus, extract_text
+from mempalace.hook_shell import count_human_messages
+from mempalace.llm_refine import collect_corpus_text
+from mempalace.miner import _read_text_no_follow, load_config, mine, scan_project
+from mempalace.normalize import _read_transcript_file
+from mempalace.project_scanner import _collect_manifest_names
+from mempalace.repair import _copy_file_no_follow, _open_regular_file_no_follow
+from mempalace.room_detector_local import detect_rooms_local
+from mempalace.split_mega_files import main as split_main
+from mempalace.sweeper import parse_claude_jsonl, sweep_directory
+
+# ``os.mkfifo`` and ``SIGALRM`` are both POSIX-only. Windows has no FIFO in
+# the filesystem namespace at all (its named pipes live under \\.\pipe\ and
+# no directory walk can reach them), so there is nothing to reproduce there.
+posix_only = pytest.mark.skipif(
+    not hasattr(os, "mkfifo") or not hasattr(signal, "SIGALRM"),
+    reason="requires POSIX FIFOs and SIGALRM",
+)
+
+TIMEOUT_SECONDS = 10.0
+
+
+class Blocked(BaseException):
+    """Raised when a call under test blocks past the deadline.
+
+    Deliberately derived from ``BaseException`` rather than ``Exception``.
+    Two separate handlers would otherwise eat the deadline and leave a
+    reverted fix looking green while it blocked for the full timeout:
+
+    * ``except OSError`` guards every read site here, and ``TimeoutError``
+      *is* an ``OSError`` — that one alone cost four vacuous passes.
+    * ``except Exception`` guards the paths the end-to-end tests cross,
+      including ``sweeper.sweep_directory`` and four sites inside
+      ``miner._mine_impl``, so a plain ``Exception`` subclass would be
+      swallowed there just as thoroughly.
+    """
+
+
+@contextmanager
+def hard_timeout(seconds: float, what: str):
+    """Fail the test instead of blocking forever.
+
+    ``signal.setitimer`` fires SIGALRM even while the interpreter sits in a
+    blocking ``open(2)``; the handler raises, and PEP 475 propagates that
+    exception rather than restarting the syscall. Without this, reverting
+    the fix would hang pytest rather than fail it.
+    """
+
+    def _fire(signum, frame):
+        raise Blocked(f"{what} blocked for more than {seconds}s")
+
+    previous = signal.signal(signal.SIGALRM, _fire)
+    signal.setitimer(signal.ITIMER_REAL, seconds)
+    try:
+        yield
+    finally:
+        signal.setitimer(signal.ITIMER_REAL, 0)
+        signal.signal(signal.SIGALRM, previous)
+
+
+def make_fifo(directory: Path, name: str) -> Path:
+    path = directory / name
+    os.mkfifo(path)
+    return path
+
+
+def write_regular(directory: Path, name: str, content: str) -> Path:
+    path = directory / name
+    path.write_text(content, encoding="utf-8")
+    return path
+
+
+# ─────────────────────────────────────────────────────────────────────────
+# The four os.open read sites: the type check must be reachable
+# ─────────────────────────────────────────────────────────────────────────
+
+
+@posix_only
+def test_read_text_no_follow_rejects_fifo(tmp_path):
+    fifo = make_fifo(tmp_path, "notes.md")
+    with hard_timeout(TIMEOUT_SECONDS, "_read_text_no_follow on a FIFO"):
+        assert _read_text_no_follow(fifo, tmp_path) is None
+
+
+@posix_only
+def test_is_regular_source_file_rejects_fifo(tmp_path):
+    fifo = make_fifo(tmp_path, "session.jsonl")
+    with hard_timeout(TIMEOUT_SECONDS, "_is_regular_source_file on a FIFO"):
+        assert _is_regular_source_file(fifo, tmp_path) is False
+
+
+@posix_only
+def test_read_transcript_file_rejects_fifo(tmp_path):
+    fifo = make_fifo(tmp_path, "session.jsonl")
+    with hard_timeout(TIMEOUT_SECONDS, "_read_transcript_file on a FIFO"):
+        with pytest.raises(IOError) as excinfo:
+            _read_transcript_file(str(fifo))
+    message = str(excinfo.value)
+    assert "not a regular file" in message
+    # The path belongs in the message exactly once — the raise inside the
+    # try block is prefix-free so the wrapper composes it.
+    assert message.count(str(fifo)) == 1
+
+
+@posix_only
+def test_open_regular_file_no_follow_rejects_fifo(tmp_path):
+    fifo = make_fifo(tmp_path, "chroma.sqlite3")
+    with hard_timeout(TIMEOUT_SECONDS, "_open_regular_file_no_follow on a FIFO"):
+        with pytest.raises(RuntimeError, match="Refusing non-regular file"):
+            _open_regular_file_no_follow(str(fifo))
+
+
+@posix_only
+def test_copy_file_no_follow_rejects_fifo_source(tmp_path):
+    fifo = make_fifo(tmp_path, "chroma.sqlite3")
+    with hard_timeout(TIMEOUT_SECONDS, "_copy_file_no_follow from a FIFO"):
+        with pytest.raises(RuntimeError, match="Refusing non-regular file"):
+            _copy_file_no_follow(str(fifo), str(tmp_path / "backup.sqlite3"))
+    assert not (tmp_path / "backup.sqlite3").exists()
+
+
+@posix_only
+def test_read_text_no_follow_rejects_fifo_that_has_a_live_writer(tmp_path):
+    """The verdict comes from the file type, not from "nobody is writing".
+
+    With a writer attached the open succeeds even without ``O_NONBLOCK``, so
+    this is not the hang regression — it pins the gate itself. Anyone who
+    "fixes" the hang by swallowing an errno instead of checking ``S_ISREG``
+    turns this red, and a pipe whose content would otherwise be mined as a
+    verbatim drawer stays out of the palace.
+    """
+    fifo = make_fifo(tmp_path, "notes.md")
+    writer_attached = threading.Event()
+    release_writer = threading.Event()
+    failures = []
+
+    def _hold_write_end():
+        try:
+            fd = os.open(fifo, os.O_WRONLY)  # blocks until a reader opens
+        except OSError as exc:  # pragma: no cover - only on a broken setup
+            failures.append(exc)
+            writer_attached.set()
+            return
+        writer_attached.set()
+        release_writer.wait(TIMEOUT_SECONDS)
+        os.close(fd)
+
+    thread = threading.Thread(target=_hold_write_end, daemon=True)
+    thread.start()
+    # Opening our own read end is what lets the writer's open(2) return.
+    reader_fd = os.open(fifo, os.O_RDONLY | os.O_NONBLOCK)
+    try:
+        assert writer_attached.wait(TIMEOUT_SECONDS), "writer never attached"
+        assert not failures, failures
+        with hard_timeout(TIMEOUT_SECONDS, "_read_text_no_follow on a written FIFO"):
+            assert _read_text_no_follow(fifo, tmp_path) is None
+    finally:
+        release_writer.set()
+        os.close(reader_fd)
+        thread.join(TIMEOUT_SECONDS)
+
+
+# ─────────────────────────────────────────────────────────────────────────
+# O_NONBLOCK must not change what a regular file reads back
+# ─────────────────────────────────────────────────────────────────────────
+
+
+@posix_only
+def test_read_text_no_follow_still_reads_a_large_regular_file_whole(tmp_path):
+    """POSIX and Linux open(2) both say O_NONBLOCK has no effect on regular
+    files. This pins that: 2 MB is many buffered reads, and a short read or
+    an ``EAGAIN`` would truncate a drawer silently.
+    """
+    payload = "the quick brown fox jumps over the lazy dog\n" * 50_000
+    regular = write_regular(tmp_path, "big.md", payload)
+    with hard_timeout(TIMEOUT_SECONDS, "_read_text_no_follow on a 2 MB file"):
+        content = _read_text_no_follow(regular, tmp_path)
+    assert content == payload
+
+
+@posix_only
+def test_copy_file_no_follow_still_copies_a_large_regular_file_byte_for_byte(tmp_path):
+    """The palace backup path reads through the same non-blocking fd."""
+    payload = os.urandom(2 * 1024 * 1024)
+    src = tmp_path / "chroma.sqlite3"
+    src.write_bytes(payload)
+    dst = tmp_path / "chroma.sqlite3.backup"
+    with hard_timeout(TIMEOUT_SECONDS, "_copy_file_no_follow of a 2 MB file"):
+        _copy_file_no_follow(str(src), str(dst))
+    assert hashlib.sha256(dst.read_bytes()).hexdigest() == hashlib.sha256(payload).hexdigest()
+
+
+# ─────────────────────────────────────────────────────────────────────────
+# Discovery walks must not hand a non-regular file to a reader
+# ─────────────────────────────────────────────────────────────────────────
+
+
+@posix_only
+def test_scan_project_skips_fifo_and_keeps_regular_files(tmp_path, capsys):
+    make_fifo(tmp_path, "notes.md")
+    write_regular(tmp_path, "real.md", "# Real\n")
+    with hard_timeout(TIMEOUT_SECONDS, "scan_project over a FIFO"):
+        found = scan_project(str(tmp_path))
+    assert [path.name for path in found] == ["real.md"]
+    assert "SKIP: notes.md (not a regular file)" in capsys.readouterr().err
+
+
+@posix_only
+def test_scan_project_skips_unix_socket(tmp_path, capsys):
+    """Sockets fail the open with ENXIO rather than blocking, but they are
+    not readable either — the walk drops them at the same gate.
+    """
+    # Bind through a short-lived cwd rather than the absolute path: AF_UNIX
+    # caps sun_path at 104 bytes on macOS (108 on Linux), and pytest's
+    # tmp_path under the macOS runner's /var/folders/... TMPDIR overruns it.
+    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    previous_cwd = os.getcwd()
+    try:
+        os.chdir(tmp_path)
+        sock.bind("mcp.md")
+        write_regular(tmp_path, "real.md", "# Real\n")
+        with hard_timeout(TIMEOUT_SECONDS, "scan_project over a socket"):
+            found = scan_project(str(tmp_path))
+    finally:
+        os.chdir(previous_cwd)
+        sock.close()
+    assert (tmp_path / "mcp.md").is_socket()
+    assert [path.name for path in found] == ["real.md"]
+    assert "SKIP: mcp.md (not a regular file)" in capsys.readouterr().err
+
+
+@posix_only
+def test_scan_convos_skips_fifo_and_keeps_regular_files(tmp_path, capsys):
+    make_fifo(tmp_path, "piped.jsonl")
+    write_regular(tmp_path, "real.jsonl", '{"type": "user"}\n')
+    with hard_timeout(TIMEOUT_SECONDS, "scan_convos over a FIFO"):
+        found = scan_convos(str(tmp_path))
+    assert [path.name for path in found] == ["real.jsonl"]
+    assert "SKIP: piped.jsonl (not a regular file)" in capsys.readouterr().err
+
+
+@posix_only
+def test_parse_claude_jsonl_refuses_fifo(tmp_path):
+    fifo = make_fifo(tmp_path, "session.jsonl")
+    with hard_timeout(TIMEOUT_SECONDS, "parse_claude_jsonl on a FIFO"):
+        with pytest.raises(OSError, match="Refusing non-regular file"):
+            list(parse_claude_jsonl(str(fifo)))
+
+
+def test_parse_claude_jsonl_still_raises_for_a_missing_path(tmp_path):
+    """The type gate stats first, so a missing path must keep failing the
+    way the plain ``open`` used to.
+    """
+    with pytest.raises(FileNotFoundError):
+        list(parse_claude_jsonl(str(tmp_path / "nope.jsonl")))
+
+
+def test_parse_claude_jsonl_still_reads_a_regular_transcript(tmp_path):
+    path = write_regular(
+        tmp_path,
+        "session.jsonl",
+        '{"type": "user", "sessionId": "s1", "uuid": "u1", '
+        '"timestamp": "2026-01-01T00:00:00Z", '
+        '"message": {"role": "user", "content": "hello"}}\n',
+    )
+    records = list(parse_claude_jsonl(str(path)))
+    assert [record["content"] for record in records] == ["hello"]
+
+
+@posix_only
+def test_detect_entities_ignores_a_fifo_candidate(tmp_path):
+    """A FIFO must be invisible: same result as if it were never there, and
+    it must not consume the ``max_files`` budget.
+    """
+    regular = write_regular(
+        tmp_path,
+        "people.md",
+        "Met Sarah Connor and John Connor at the office today.\n" * 5,
+    )
+    fifo = make_fifo(tmp_path, "notes.md")
+    baseline = detect_entities([regular])
+    with hard_timeout(TIMEOUT_SECONDS, "detect_entities over a FIFO"):
+        with_fifo = detect_entities([fifo, regular])
+    assert with_fifo == baseline
+
+
+@posix_only
+def test_gather_origin_samples_ignores_a_fifo_candidate(tmp_path):
+    """``init``'s corpus-origin pass reads the same candidate list as
+    ``detect_entities`` and must drop a FIFO at the same gate.
+    """
+    write_regular(tmp_path, "real.md", "# Real\n\nSome prose about the project.\n")
+    make_fifo(tmp_path, "notes.md")
+    with hard_timeout(TIMEOUT_SECONDS, "_gather_origin_samples over a FIFO"):
+        samples = _gather_origin_samples(str(tmp_path))
+    assert len(samples) == 1
+    assert "Some prose about the project." in samples[0]
+
+
+@posix_only
+def test_split_mega_files_skips_fifo(tmp_path, capsys, monkeypatch):
+    make_fifo(tmp_path, "piped.txt")
+    # Two detectable sessions, so the regular file is reported rather than
+    # dropped for having nothing to split. Shape copied from
+    # tests/test_split_mega_files.py::test_find_session_boundaries_two_sessions.
+    session = "Claude Code v1.0\ncontent\n" + "\n" * 5
+    write_regular(tmp_path, "real.txt", session * 2)
+    monkeypatch.setattr("sys.argv", ["mempalace split", "--source", str(tmp_path), "--dry-run"])
+    with hard_timeout(TIMEOUT_SECONDS, "split_mega_files.main over a FIFO"):
+        split_main()
+    out = capsys.readouterr().out
+    assert "SKIP: piped.txt (not a regular file)" in out
+    # The gate must drop the pipe and nothing else: a version that skipped
+    # every entry would satisfy the SKIP assertion above on its own.
+    assert "SKIP: real.txt" not in out
+    assert "real.txt" in out
+
+
+@posix_only
+def test_format_miner_extract_text_does_not_block_on_fifo(tmp_path):
+    """``mine --mode extract`` was already immune — its zero-size gate fires
+    first, because a FIFO stats as 0 bytes. Pinned so a future reshuffle of
+    those checks cannot reintroduce the hang here.
+    """
+    fifo = make_fifo(tmp_path, "doc.pdf")
+    with hard_timeout(TIMEOUT_SECONDS, "extract_text on a FIFO"):
+        text, status = extract_text(fifo)
+    assert text is None
+    assert status is ExtractionStatus.SKIP_EMPTY
+
+
+@posix_only
+def test_collect_manifest_names_ignores_a_fifo_manifest(tmp_path):
+    real_repo = tmp_path / "real"
+    real_repo.mkdir()
+    (real_repo / "package.json").write_text('{"name": "real-project"}', encoding="utf-8")
+    piped = tmp_path / "piped"
+    piped.mkdir()
+    os.mkfifo(piped / "package.json")
+
+    with hard_timeout(TIMEOUT_SECONDS, "_collect_manifest_names over a FIFO manifest"):
+        found = _collect_manifest_names(tmp_path)
+    assert [entry[1] for entry in found] == ["real-project"]
+
+
+# ─────────────────────────────────────────────────────────────────────────
+# Fixed-name reads: `exists()` is not a type check
+#
+# The sites above are fed by a directory walk. These are read by a name the
+# code already knows, behind an `exists()` guard — which is true for a FIFO,
+# so the open right after it blocks anyway.
+# ─────────────────────────────────────────────────────────────────────────
+
+
+@posix_only
+def test_load_config_treats_a_fifo_yaml_as_absent(tmp_path):
+    """`mempalace.yaml` is the first thing `mine` reads."""
+    make_fifo(tmp_path, "mempalace.yaml")
+    with hard_timeout(TIMEOUT_SECONDS, "load_config on a FIFO mempalace.yaml"):
+        config = load_config(str(tmp_path))
+    assert [room["name"] for room in config["rooms"]] == ["general"]
+
+
+@posix_only
+def test_load_config_still_reads_a_regular_yaml(tmp_path):
+    (tmp_path / "mempalace.yaml").write_text(
+        yaml.dump({"wing": "realwing", "rooms": [{"name": "docs", "description": "d"}]}),
+        encoding="utf-8",
+    )
+    config = load_config(str(tmp_path))
+    assert config["wing"] == "realwing"
+    assert [room["name"] for room in config["rooms"]] == ["docs"]
+
+
+@posix_only
+def test_ensure_gitignore_leaves_a_fifo_gitignore_alone(tmp_path):
+    (tmp_path / ".git").mkdir()
+    make_fifo(tmp_path, ".gitignore")
+    with hard_timeout(TIMEOUT_SECONDS, "_ensure_mempalace_files_gitignored on a FIFO"):
+        assert _ensure_mempalace_files_gitignored(str(tmp_path)) is False
+
+
+def test_ensure_gitignore_still_appends_to_a_regular_gitignore(tmp_path):
+    (tmp_path / ".git").mkdir()
+    (tmp_path / ".gitignore").write_text("*.pyc\n", encoding="utf-8")
+    assert _ensure_mempalace_files_gitignored(str(tmp_path)) is True
+    written = (tmp_path / ".gitignore").read_text(encoding="utf-8")
+    assert "mempalace.yaml" in written and "entities.json" in written
+
+
+@posix_only
+def test_cmd_compress_ignores_a_fifo_entities_json(tmp_path, monkeypatch, capsys):
+    """`compress` picks up `./entities.json` when no --config is given."""
+    monkeypatch.chdir(tmp_path)
+    make_fifo(tmp_path, "entities.json")
+    args = argparse.Namespace(palace=None, wing=None, dry_run=False, config=None)
+    with patch("mempalace.cli.MempalaceConfig") as mock_config_cls:
+        mock_config_cls.return_value.palace_path = str(tmp_path / "nonexistent")
+        with hard_timeout(TIMEOUT_SECONDS, "cmd_compress with a FIFO entities.json"):
+            with pytest.raises(SystemExit):
+                cmd_compress(args)
+    out = capsys.readouterr().out
+    assert "No palace found" in out
+    assert "Loaded entity config" not in out
+
+
+@posix_only
+def test_cmd_compress_fifo_entities_json_does_not_shadow_the_palace_copy(
+    tmp_path, monkeypatch, capsys
+):
+    """The candidate loop must not stop at a pipe.
+
+    An `entities.json` FIFO in the cwd would otherwise be picked as the
+    config and then rejected by the load guard, hiding a perfectly good
+    `<palace>/entities.json` behind it.
+    """
+    monkeypatch.chdir(tmp_path)
+    make_fifo(tmp_path, "entities.json")
+    palace = tmp_path / "palace"
+    palace.mkdir()
+    (palace / "entities.json").write_text('{"entities": {"Alice": "ALC"}}', encoding="utf-8")
+    args = argparse.Namespace(palace=None, wing=None, dry_run=False, config=None)
+    with patch("mempalace.cli.MempalaceConfig") as mock_config_cls:
+        mock_config_cls.return_value.palace_path = str(palace)
+        with hard_timeout(TIMEOUT_SECONDS, "cmd_compress candidate loop"):
+            with pytest.raises(SystemExit):
+                cmd_compress(args)
+    assert f"Loaded entity config: {palace / 'entities.json'}" in capsys.readouterr().out
+
+
+@posix_only
+def test_parse_gradle_ignores_a_fifo_sibling_settings_file(tmp_path):
+    """`build.gradle` is a regular file and clears the manifest gate; the
+    parser then reads the SIBLING `settings.gradle`, which no walk vetted.
+    """
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    (repo / "build.gradle").write_text("plugins { id 'java' }\n", encoding="utf-8")
+    make_fifo(repo, "settings.gradle")
+    with hard_timeout(TIMEOUT_SECONDS, "_collect_manifest_names with a FIFO settings.gradle"):
+        found = _collect_manifest_names(tmp_path)
+    assert [entry[1] for entry in found] == ["repo"]
+
+
+@posix_only
+def test_count_human_messages_ignores_a_fifo_transcript(tmp_path):
+    fifo = make_fifo(tmp_path, "transcript.jsonl")
+    with hard_timeout(TIMEOUT_SECONDS, "count_human_messages on a FIFO"):
+        assert count_human_messages(str(fifo)) == 0
+
+
+def test_count_human_messages_still_raises_for_a_missing_path(tmp_path):
+    """The type gate is narrowed to paths that exist, so a missing transcript
+    keeps failing the way the plain ``open`` did.
+    """
+    with pytest.raises(FileNotFoundError):
+        count_human_messages(str(tmp_path / "nope.jsonl"))
+
+
+def test_count_human_messages_still_counts_a_regular_transcript(tmp_path):
+    path = write_regular(
+        tmp_path,
+        "transcript.jsonl",
+        '{"message": {"role": "user", "content": "one"}}\n'
+        '{"message": {"role": "assistant", "content": "two"}}\n'
+        '{"message": {"role": "user", "content": "three"}}\n',
+    )
+    assert count_human_messages(str(path)) == 2
+
+
+@posix_only
+def test_cmd_init_refuses_to_write_entities_over_a_fifo(tmp_path, capsys):
+    """`init` writes `<project>/entities.json`; opening a pre-existing FIFO
+    for writing blocks until a reader appears.
+    """
+    make_fifo(tmp_path, "entities.json")
+    args = argparse.Namespace(dir=str(tmp_path), yes=True, no_llm=True)
+    detected = {"people": [{"name": "Alice"}], "projects": [], "topics": [], "uncertain": []}
+    confirmed = {"people": ["Alice"], "projects": [], "topics": []}
+    with (
+        patch("mempalace.cli.MempalaceConfig"),
+        patch("mempalace.project_scanner.discover_entities", return_value=detected),
+        patch("mempalace.entity_detector.confirm_entities", return_value=confirmed),
+        patch("mempalace.room_detector_local.detect_rooms_local"),
+        patch("mempalace.cli._run_pass_zero", return_value=None),
+        patch("mempalace.cli._maybe_run_mine_after_init"),
+    ):
+        with hard_timeout(TIMEOUT_SECONDS, "cmd_init with a FIFO entities.json"):
+            cmd_init(args)
+    captured = capsys.readouterr()
+    assert "is not a regular file" in captured.err
+    assert "Entities saved" not in captured.out
+
+
+@posix_only
+def test_detect_rooms_local_refuses_a_fifo_config(tmp_path):
+    """`init` writes `mempalace.yaml`; a write open on a pipe blocks until a
+    reader appears, which parked `init` with no output at all.
+    """
+    write_regular(tmp_path, "README.md", "note\n")
+    make_fifo(tmp_path, "mempalace.yaml")
+    with hard_timeout(TIMEOUT_SECONDS, "detect_rooms_local over a FIFO config"):
+        with pytest.raises(OSError, match="not a regular file"):
+            detect_rooms_local(project_dir=str(tmp_path), yes=True)
+
+
+@posix_only
+def test_collect_corpus_text_ignores_a_fifo(tmp_path):
+    """LLM refinement walks prose by suffix and stats only for mtime."""
+    write_regular(tmp_path, "a.md", "real prose\n")
+    make_fifo(tmp_path, "notes.md")
+    with hard_timeout(TIMEOUT_SECONDS, "collect_corpus_text over a FIFO"):
+        text = collect_corpus_text(str(tmp_path))
+    assert text == "real prose\n"
+
+
+@posix_only
+def test_sweep_directory_skips_a_fifo_without_booking_a_failure(tmp_path, capsys):
+    """A pipe is nothing to sweep, not a sweep failure.
+
+    Booking it as a failure flips the command's exit status to 2 through
+    ``cli.cmd_sweep``, which breaks any script gating on it — and the same
+    input on a directory walk is a benign ``SKIP`` in ``mine``.
+    """
+    convos = tmp_path / "convos"
+    convos.mkdir()
+    write_regular(
+        convos,
+        "real.jsonl",
+        '{"type": "user", "sessionId": "s1", "uuid": "u1", '
+        '"timestamp": "2026-01-01T00:00:00Z", '
+        '"message": {"role": "user", "content": "hello"}}\n',
+    )
+    make_fifo(convos, "piped.jsonl")
+    with hard_timeout(TIMEOUT_SECONDS, "sweep_directory over a FIFO"):
+        result = sweep_directory(str(convos), str(tmp_path / "palace"))
+    # ``failures`` is what ``cli.cmd_sweep`` turns into ``sys.exit(2)``.
+    assert result["failures"] == []
+    # ``files_attempted`` counts discovery, per its docstring, so the pipe
+    # stays in it; ``files_succeeded`` counts what was actually swept.
+    assert result["files_attempted"] == 2
+    assert result["files_succeeded"] == 1
+    assert result["drawers_added"] == 1
+    assert "SKIP: piped.jsonl (not a regular file)" in capsys.readouterr().err
+
+
+# ─────────────────────────────────────────────────────────────────────────
+# O_NONBLOCK must not drop a regular file the blocking open would have read
+# ─────────────────────────────────────────────────────────────────────────
+
+
+@posix_only
+def test_read_text_no_follow_retries_when_a_lease_break_returns_eagain(tmp_path, monkeypatch):
+    """A write lease is the one case where the flag changes `open` itself.
+
+    Breaking a lease with ``O_NONBLOCK`` fails ``EAGAIN`` immediately, where
+    a blocking open waits out ``lease-break-time`` and succeeds. Left alone
+    that turns into a silently dropped file. The kernel grants leases on
+    regular files only, so the retry is authorised by the file *type*; the
+    errno only decides whether to look again.
+
+    ``EAGAIN`` is injected rather than staged with a real lease so the test
+    costs milliseconds instead of the 45 s default lease-break-time.
+    """
+    payload = "PAYLOAD THAT MUST STILL BE MINED\n" * 20
+    target = write_regular(tmp_path, "notes.md", payload)
+    real_open = os.open
+    calls = {"n": 0}
+
+    def _fake_open(path, flags, *args, **kwargs):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            assert flags & os.O_NONBLOCK, "first attempt should carry the flag"
+            raise OSError(errno.EAGAIN, os.strerror(errno.EAGAIN), str(path))
+        assert not flags & os.O_NONBLOCK, "retry should drop the flag"
+        return real_open(path, flags, *args, **kwargs)
+
+    monkeypatch.setattr("mempalace.miner.os.open", _fake_open)
+    with hard_timeout(TIMEOUT_SECONDS, "_read_text_no_follow under an injected EAGAIN"):
+        content = _read_text_no_follow(target, tmp_path)
+    assert content == payload
+    assert calls["n"] == 2
+
+
+@posix_only
+def test_read_text_no_follow_does_not_retry_eagain_on_a_fifo(tmp_path, monkeypatch):
+    """The retry is gated on the type, so a pipe never gets a blocking open.
+
+    The stand-in raises if it is ever called without the flag, so a retry
+    that trusted the errno alone would fail this test rather than hang it.
+    """
+    fifo = make_fifo(tmp_path, "notes.md")
+
+    def _fake_open(path, flags, *args, **kwargs):
+        if flags & os.O_NONBLOCK:
+            raise OSError(errno.EAGAIN, os.strerror(errno.EAGAIN), str(path))
+        raise AssertionError("must not retry a blocking open on a FIFO")
+
+    monkeypatch.setattr("mempalace.miner.os.open", _fake_open)
+    with hard_timeout(TIMEOUT_SECONDS, "_read_text_no_follow EAGAIN on a FIFO"):
+        assert _read_text_no_follow(fifo, tmp_path) is None
+
+
+@posix_only
+def test_gather_origin_samples_survives_an_unreadable_directory(tmp_path):
+    """The type gate must not turn a skipped file into a crash.
+
+    ``Path.is_file()`` raises ``PermissionError`` on a directory without
+    ``x``; the ``open`` it replaced was already inside the ``try`` that
+    absorbs exactly that, so the gate has to sit there too.
+    """
+    write_regular(tmp_path, "real.md", "# Real\n\nsome prose here\n")
+    walled = tmp_path / "walled"
+    walled.mkdir()
+    write_regular(walled, "notes.md", "y" * 200)
+    os.chmod(walled, 0o444)
+    try:
+        with hard_timeout(TIMEOUT_SECONDS, "_gather_origin_samples over an unreadable dir"):
+            samples = _gather_origin_samples(str(tmp_path))
+    finally:
+        os.chmod(walled, 0o755)
+    assert len(samples) == 1
+    assert "some prose here" in samples[0]
+
+
+def test_read_transcript_file_size_message_names_the_path_once(tmp_path):
+    """Both refusal branches compose the path through the same wrapper."""
+    big = write_regular(tmp_path, "huge.jsonl", "not actually huge")
+
+    class _HugeStat:
+        st_mode = stat_module.S_IFREG | 0o644
+        st_size = 600 * 1024 * 1024
+
+    with patch("mempalace.normalize.os.fstat", return_value=_HugeStat()):
+        with pytest.raises(IOError) as excinfo:
+            _read_transcript_file(str(big))
+    message = str(excinfo.value)
+    assert "too large" in message.lower()
+    assert message.count(str(big)) == 1
+
+
+# ─────────────────────────────────────────────────────────────────────────
+# End to end through the real miner
+# ─────────────────────────────────────────────────────────────────────────
+
+
+@posix_only
+def test_mine_completes_with_a_fifo_in_the_corpus(tmp_path):
+    """The original report: ``mempalace mine <dir>`` never returned."""
+    project_root = tmp_path / "corpus"
+    project_root.mkdir()
+    (project_root / "mempalace.yaml").write_text(
+        yaml.dump(
+            {
+                "wing": "fifo_repro",
+                "rooms": [{"name": "general", "description": "General"}],
+            }
+        ),
+        encoding="utf-8",
+    )
+    write_regular(
+        project_root,
+        "real.md",
+        "# Real note\n\n" + "The quick brown fox jumps over the lazy dog. " * 40,
+    )
+    make_fifo(project_root, "notes.md")
+
+    palace_path = tmp_path / "palace"
+    with hard_timeout(TIMEOUT_SECONDS, "mine() over a corpus holding a FIFO"):
+        mine(str(project_root), str(palace_path))
+
+    import chromadb
+
+    collection = chromadb.PersistentClient(path=str(palace_path)).get_collection(
+        "mempalace_drawers"
+    )
+    stored = collection.get(include=["metadatas"])
+    sources = {Path(meta["source_file"]).name for meta in stored["metadatas"]}
+    assert sources == {"real.md"}

From 0f3f0c6fbbad54ca91452eca41f17ef88850edae Mon Sep 17 00:00:00 2001
From: Michael Valentsev <michael@valentsev.ru>
Date: Tue, 11 Aug 2026 21:33:24 +0500
Subject: [PATCH 07/10] docs(changelog): note the non-regular-file ingest hang
 (#2221)

---
 CHANGELOG.md | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index dd11273ef3..82dc52b5a5 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,6 +8,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 
 ## [Unreleased]
 
+### Bug Fixes
+
+- **Ingest commands no longer hang on a named pipe.** `os.walk` and `glob` list a FIFO as an ordinary filename and MemPalace decides what to read from the suffix, so a pipe called `notes.md` in a mined directory wedged `mine` in the kernel forever: opening a FIFO for reading waits for a writer that never arrives, and the `S_ISREG` refusal written on the next line could never run. `mine --mode convos`, `sweep`, `init`, `compress` and `split` blocked the same way through their own readers. The four affected opens now pass `O_NONBLOCK`, which makes the existing type check reachable — a pipe is refused on its mode, with or without a live writer — and the discovery walks drop non-regular entries with a `SKIP: <name> (not a regular file)` line, so the readers that use a plain `open()` never see one. Regular files read back byte-identical; the one case where the flag is not inert, a reader breaking a write lease, re-checks the file type and retries without it rather than dropping the file. `mine --mode extract` was already immune through its zero-size gate. (#2221)
+
 ---
 
 ## [3.7.0] — 2026-08-11

From 1654cd233c6af579f595a69695fe500a7f8cf608 Mon Sep 17 00:00:00 2001
From: Igor Lins e Silva <4753812+igorls@users.noreply.github.com>
Date: Tue, 11 Aug 2026 14:11:01 -0300
Subject: [PATCH 08/10] =?UTF-8?q?fix:=203.7.1=20critical=20patch=20?=
 =?UTF-8?q?=E2=80=94=20re-mine=20honesty=20+=20SIGTERM=20lock=20release?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Stack the post-3.7.0 hang and silent-skip fixes for a fast patch release:

- Keep #2223 (non-regular file hang) and #2088 (chunk_total / same-fstat
  mtime / purge abort / closet purge) as the base.
- On multi-batch upsert failure, delete partial drawers and closets for
  that source before re-raising so the next mine retries (#2122, #2151).
- Install SIGTERM/SIGHUP handlers in mcp_server.main so atexit can release
  the palace writer lease (#2205).
- Adapt non-regular-file tests to the (content, mtime) read return type.
---
 CHANGELOG.md                          |  2 +
 mempalace/mcp_server.py               | 31 +++++++++++
 mempalace/miner.py                    | 79 +++++++++++++++++----------
 tests/test_mcp_server.py              | 34 ++++++++++++
 tests/test_miner.py                   | 76 ++++++++++++++++++++++++++
 tests/test_non_regular_file_guards.py | 10 +++-
 6 files changed, 201 insertions(+), 31 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 82dc52b5a5..ae8a90549d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,6 +11,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 ### Bug Fixes
 
 - **Ingest commands no longer hang on a named pipe.** `os.walk` and `glob` list a FIFO as an ordinary filename and MemPalace decides what to read from the suffix, so a pipe called `notes.md` in a mined directory wedged `mine` in the kernel forever: opening a FIFO for reading waits for a writer that never arrives, and the `S_ISREG` refusal written on the next line could never run. `mine --mode convos`, `sweep`, `init`, `compress` and `split` blocked the same way through their own readers. The four affected opens now pass `O_NONBLOCK`, which makes the existing type check reachable — a pipe is refused on its mode, with or without a live writer — and the discovery walks drop non-regular entries with a `SKIP: <name> (not a regular file)` line, so the readers that use a plain `open()` never see one. Regular files read back byte-identical; the one case where the flag is not inert, a reader breaking a write lease, re-checks the file type and retries without it rather than dropping the file. `mine --mode extract` was already immune through its zero-size gate. (#2221)
+- **Project re-mine no longer silently skips a partial or interrupted file.** Four related gaps in `process_file`: (1) multi-batch upserts now stamp every drawer with `chunk_total` so `file_already_mined` can tell "N of N committed" from "crashed after batch 1"; (2) `source_mtime` comes from the same `fstat` as the content read, so an append between read and a later re-stat cannot permanently hide the new tail; (3) a failed stale-drawer purge aborts the file instead of half-overwriting; (4) closets are purged even when the re-mine ends with zero drawers. A mid-file upsert failure also deletes the partial drawers and closets for that source before re-raising, so the next mine retries instead of treating the incomplete set as complete. (#2088, #2122, #2151)
+- **MCP releases the palace writer lease on SIGTERM/SIGHUP.** The lease was only released via `atexit`, which CPython skips on those signals' default disposition. SSH disconnect (SIGHUP) and container/systemd stop (SIGTERM) therefore left `mine_palace_*.lock` naming a dead PID until a contender's liveness check reclaimed it. `main()` now installs handlers that exit through `sys.exit`, so the existing `atexit` release path runs. (#2205)
 
 ---
 
diff --git a/mempalace/mcp_server.py b/mempalace/mcp_server.py
index d2a9262178..1953c7691c 100644
--- a/mempalace/mcp_server.py
+++ b/mempalace/mcp_server.py
@@ -8021,6 +8021,35 @@ def _warmup_with_lock():
                 _release_mcp_writer_lock()
 
 
+def _install_shutdown_signal_handlers() -> None:
+    """Route terminal signals through ``sys.exit`` so ``atexit`` runs.
+
+    The palace writer lease is released by an ``atexit`` callback registered
+    when the lock is acquired. CPython's default disposition for SIGTERM and
+    SIGHUP is immediate termination, which skips ``atexit`` and leaves
+    ``mine_palace_*.lock`` naming a dead PID until a contender's liveness
+    check reclaims it (#2205). Calling ``sys.exit(0)`` from the handler
+    unwinds the synchronous stdio/http loop and runs the existing release
+    path. SIGHUP is Unix-only (SSH session disconnect); Windows only gets
+    SIGTERM. Handlers are best-effort — signal registration only works from
+    the main thread and is a no-op when the platform omits the signal.
+    """
+    import signal
+
+    def _shutdown_handler(signum, frame):  # noqa: ARG001
+        raise SystemExit(0)
+
+    for name in ("SIGTERM", "SIGHUP"):
+        sig = getattr(signal, name, None)
+        if sig is None:
+            continue
+        try:
+            signal.signal(sig, _shutdown_handler)
+        except (ValueError, OSError):
+            # Not in the main thread, or the platform rejects the install.
+            pass
+
+
 def main():
     """MCP server entry point for the ``mempalace-mcp`` console script.
 
@@ -8043,6 +8072,8 @@ def main():
     # extend the protection to children.
     os.environ.pop("PYTHONPATH", None)
 
+    _install_shutdown_signal_handlers()
+
     if _args.transport == "http":
         _run_http_loop()
     else:
diff --git a/mempalace/miner.py b/mempalace/miner.py
index 6846b3c96f..707efde562 100644
--- a/mempalace/miner.py
+++ b/mempalace/miner.py
@@ -1603,37 +1603,58 @@ def process_file(
         # in production and the 4-segment pointer form lives only in tests.
         # Per PR #1584 review (Igor, 2026-05-22).
         all_metas: list = []
-        for batch_start in range(0, len(chunks), DRAWER_UPSERT_BATCH_SIZE):
-            batch_docs: list = []
-            batch_ids: list = []
-            batch_metas: list = []
-            for chunk in chunks[batch_start : batch_start + DRAWER_UPSERT_BATCH_SIZE]:
-                drawer_id = make_drawer_id_from_chunk(wing, room, source_file, chunk["chunk_index"])
-                batch_docs.append(chunk["content"])
-                batch_ids.append(drawer_id)
-                batch_metas.append(
-                    _build_drawer_metadata(
-                        wing,
-                        room,
-                        source_file,
-                        chunk["chunk_index"],
-                        agent,
-                        chunk["content"],
-                        source_mtime,
-                        line_start=chunk.get("line_start"),
-                        line_end=chunk.get("line_end"),
-                        content_date=file_content_date,
-                        chunk_total=len(chunks),
+        try:
+            for batch_start in range(0, len(chunks), DRAWER_UPSERT_BATCH_SIZE):
+                batch_docs: list = []
+                batch_ids: list = []
+                batch_metas: list = []
+                for chunk in chunks[batch_start : batch_start + DRAWER_UPSERT_BATCH_SIZE]:
+                    drawer_id = make_drawer_id_from_chunk(
+                        wing, room, source_file, chunk["chunk_index"]
+                    )
+                    batch_docs.append(chunk["content"])
+                    batch_ids.append(drawer_id)
+                    batch_metas.append(
+                        _build_drawer_metadata(
+                            wing,
+                            room,
+                            source_file,
+                            chunk["chunk_index"],
+                            agent,
+                            chunk["content"],
+                            source_mtime,
+                            line_start=chunk.get("line_start"),
+                            line_end=chunk.get("line_end"),
+                            content_date=file_content_date,
+                            chunk_total=len(chunks),
+                        )
                     )
+                assert_no_collisions(list(zip(batch_ids, batch_metas)), collection)
+                collection.upsert(
+                    documents=batch_docs,
+                    ids=batch_ids,
+                    metadatas=batch_metas,
                 )
-            assert_no_collisions(list(zip(batch_ids, batch_metas)), collection)
-            collection.upsert(
-                documents=batch_docs,
-                ids=batch_ids,
-                metadatas=batch_metas,
-            )
-            drawers_added += len(batch_docs)
-            all_metas.extend(batch_metas)
+                drawers_added += len(batch_docs)
+                all_metas.extend(batch_metas)
+        except Exception:
+            # A successful earlier batch has the source's current mtime (and
+            # often chunk_total). Leaving those drawers behind would make the
+            # next run skip this incomplete rebuild when chunk_total is absent
+            # on legacy rows, and would leave partial content searchable until
+            # the next mine. The source lock prevents this cleanup from
+            # deleting another miner's work for the same file. (#2122)
+            try:
+                collection.delete(where={"source_file": source_file})
+            except Exception:
+                logger.warning(
+                    "Failed to clean partial drawers after upsert error for %s",
+                    source_file,
+                    exc_info=True,
+                )
+            if closets_col:
+                purge_file_closets(closets_col, source_file)
+            raise
 
         # Build closet — the searchable index pointing to these drawers.
         # Purge unconditionally: the old drawers this closet pointed at were
diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py
index 6f131df065..e5154a9c37 100644
--- a/tests/test_mcp_server.py
+++ b/tests/test_mcp_server.py
@@ -77,6 +77,40 @@ def test_mcp_main_strips_leaked_pythonpath_from_env():
     assert "ENV_AFTER: None" in result.stderr, f"MCP server did not strip PYTHONPATH: {diag}"
 
 
+def test_install_shutdown_signal_handlers_routes_term_to_system_exit():
+    """SIGTERM/SIGHUP must raise SystemExit so atexit can release the lease (#2205)."""
+    import signal
+
+    from mempalace import mcp_server
+
+    previous = {}
+    for name in ("SIGTERM", "SIGHUP"):
+        sig = getattr(signal, name, None)
+        if sig is None:
+            continue
+        previous[sig] = signal.getsignal(sig)
+
+    try:
+        mcp_server._install_shutdown_signal_handlers()
+        term = signal.SIGTERM
+        handler = signal.getsignal(term)
+        assert callable(handler)
+        with pytest.raises(SystemExit) as exc_info:
+            handler(term, None)
+        assert exc_info.value.code == 0
+
+        sighup = getattr(signal, "SIGHUP", None)
+        if sighup is not None:
+            hup_handler = signal.getsignal(sighup)
+            assert callable(hup_handler)
+            with pytest.raises(SystemExit) as exc_info:
+                hup_handler(sighup, None)
+            assert exc_info.value.code == 0
+    finally:
+        for sig, old in previous.items():
+            signal.signal(sig, old)
+
+
 def _patch_mcp_server(monkeypatch, config, kg):
     """Patch the mcp_server module globals to use test fixtures."""
     from mempalace import mcp_server
diff --git a/tests/test_miner.py b/tests/test_miner.py
index 268a2e5d2c..037cd63d2f 100644
--- a/tests/test_miner.py
+++ b/tests/test_miner.py
@@ -1515,6 +1515,82 @@ def test_file_already_mined_detects_incomplete_multi_batch_remine():
         shutil.rmtree(tmpdir, ignore_errors=True)
 
 
+def test_process_file_cleans_partial_drawers_after_a_batch_upsert_failure(tmp_path, monkeypatch):
+    """A failed later batch must not leave mtime-stamped drawers that skip retry (#2122)."""
+    from mempalace import miner
+
+    class FailingCollection:
+        def __init__(self):
+            self.records = []
+            self.upsert_calls = 0
+            self.deleted_sources = []
+
+        def get(self, where=None, limit=None, offset=0, include=None):
+            records = self.records
+            if where and "source_file" in where:
+                records = [
+                    record
+                    for record in records
+                    if record["metadata"]["source_file"] == where["source_file"]
+                ]
+            page = records[offset : offset + (limit or len(records))]
+            return {
+                "ids": [record["id"] for record in page],
+                "metadatas": [record["metadata"] for record in page],
+            }
+
+        def delete(self, where=None):
+            source_file = where.get("source_file") if where else None
+            self.deleted_sources.append(source_file)
+            self.records = [
+                record
+                for record in self.records
+                if record["metadata"]["source_file"] != source_file
+            ]
+
+        def upsert(self, documents, ids, metadatas):
+            self.upsert_calls += 1
+            if self.upsert_calls == 2:
+                raise RuntimeError("simulated second-batch failure")
+            self.records.extend(
+                {"id": drawer_id, "metadata": metadata}
+                for drawer_id, metadata in zip(ids, metadatas)
+            )
+
+    class FakeClosets:
+        def __init__(self):
+            self.deleted_sources = []
+
+        def delete(self, where=None):
+            self.deleted_sources.append(where.get("source_file"))
+
+    source = tmp_path / "src.py"
+    source.write_text("print('hello')\n" * 20, encoding="utf-8")
+    chunks = [{"content": f"chunk {index} " * 20, "chunk_index": index} for index in range(3)]
+    collection = FailingCollection()
+    closets = FakeClosets()
+    monkeypatch.setattr(miner, "DRAWER_UPSERT_BATCH_SIZE", 2)
+    monkeypatch.setattr(miner, "chunk_text", lambda content, source_file, **kwargs: chunks)
+    monkeypatch.setattr(miner, "assert_no_collisions", lambda *args, **kwargs: None)
+
+    with pytest.raises(RuntimeError, match="second-batch failure"):
+        miner.process_file(
+            source,
+            tmp_path,
+            collection,
+            "wing",
+            [{"name": "general", "description": "General"}],
+            "agent",
+            False,
+            closets_col=closets,
+        )
+
+    assert collection.deleted_sources == [str(source), str(source)]
+    assert collection.records == []
+    assert closets.deleted_sources == [str(source)]
+    assert file_already_mined(collection, str(source), check_mtime=True) is False
+
+
 # ── normalize_version schema gate ───────────────────────────────────────
 #
 # When the normalization pipeline changes shape (e.g., strip_noise lands),
diff --git a/tests/test_non_regular_file_guards.py b/tests/test_non_regular_file_guards.py
index 9ea2703663..e982d9da71 100644
--- a/tests/test_non_regular_file_guards.py
+++ b/tests/test_non_regular_file_guards.py
@@ -211,8 +211,11 @@ def test_read_text_no_follow_still_reads_a_large_regular_file_whole(tmp_path):
     payload = "the quick brown fox jumps over the lazy dog\n" * 50_000
     regular = write_regular(tmp_path, "big.md", payload)
     with hard_timeout(TIMEOUT_SECONDS, "_read_text_no_follow on a 2 MB file"):
-        content = _read_text_no_follow(regular, tmp_path)
+        result = _read_text_no_follow(regular, tmp_path)
+    assert result is not None
+    content, mtime = result
     assert content == payload
+    assert mtime == os.path.getmtime(regular)
 
 
 @posix_only
@@ -614,8 +617,11 @@ def _fake_open(path, flags, *args, **kwargs):
 
     monkeypatch.setattr("mempalace.miner.os.open", _fake_open)
     with hard_timeout(TIMEOUT_SECONDS, "_read_text_no_follow under an injected EAGAIN"):
-        content = _read_text_no_follow(target, tmp_path)
+        result = _read_text_no_follow(target, tmp_path)
+    assert result is not None
+    content, mtime = result
     assert content == payload
+    assert mtime == os.path.getmtime(target)
     assert calls["n"] == 2
 
 

From 759b8f13cdd2ba8aa57098ab96a896c15424c169 Mon Sep 17 00:00:00 2001
From: Igor Lins e Silva <4753812+igorls@users.noreply.github.com>
Date: Tue, 11 Aug 2026 21:48:58 -0300
Subject: [PATCH 09/10] fix(convo): stamp chunk_total and clean partial
 multi-batch mines (#2183)

Port project-miner re-mine honesty to conversation ingest so an interrupted
transcript mine cannot permanently skip missing exchanges:

- stamp chunk_total on every convo drawer in the pass
- delete partial drawers for the source/extract_mode on upsert failure
- teach prefetch_mined_set the same completeness rule as file_already_mined
---
 CHANGELOG.md                   |   2 +
 mempalace/convo_miner.py       | 118 ++++++++++++++++++++-------------
 mempalace/palace.py            |  42 ++++++++++--
 tests/test_convo_miner.py      |  72 +++++++++++++++++++-
 tests/test_convo_miner_unit.py | 101 ++++++++++++++++++++++++++++
 5 files changed, 283 insertions(+), 52 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index ae8a90549d..e3fa82e312 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -12,6 +12,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 
 - **Ingest commands no longer hang on a named pipe.** `os.walk` and `glob` list a FIFO as an ordinary filename and MemPalace decides what to read from the suffix, so a pipe called `notes.md` in a mined directory wedged `mine` in the kernel forever: opening a FIFO for reading waits for a writer that never arrives, and the `S_ISREG` refusal written on the next line could never run. `mine --mode convos`, `sweep`, `init`, `compress` and `split` blocked the same way through their own readers. The four affected opens now pass `O_NONBLOCK`, which makes the existing type check reachable — a pipe is refused on its mode, with or without a live writer — and the discovery walks drop non-regular entries with a `SKIP: <name> (not a regular file)` line, so the readers that use a plain `open()` never see one. Regular files read back byte-identical; the one case where the flag is not inert, a reader breaking a write lease, re-checks the file type and retries without it rather than dropping the file. `mine --mode extract` was already immune through its zero-size gate. (#2221)
 - **Project re-mine no longer silently skips a partial or interrupted file.** Four related gaps in `process_file`: (1) multi-batch upserts now stamp every drawer with `chunk_total` so `file_already_mined` can tell "N of N committed" from "crashed after batch 1"; (2) `source_mtime` comes from the same `fstat` as the content read, so an append between read and a later re-stat cannot permanently hide the new tail; (3) a failed stale-drawer purge aborts the file instead of half-overwriting; (4) closets are purged even when the re-mine ends with zero drawers. A mid-file upsert failure also deletes the partial drawers and closets for that source before re-raising, so the next mine retries instead of treating the incomplete set as complete. (#2088, #2122, #2151)
+- **Conversation mine completeness matches the project path.** Convo drawers now stamp `chunk_total`; a mid-batch upsert failure deletes that source's partial drawers before re-raising; `prefetch_mined_set` omits incomplete groups so the bulk "already filed" skip cannot permanently strand missing exchanges from an interrupted transcript mine. (#2183)
+- **Stale chromadb System cache is cleared on palace reconnect.** After a peer or rebuild changes `chroma.sqlite3` on disk, both `mcp_server._get_client` and `ChromaBackend._client` drop chromadb's path-keyed `SharedSystemClient` cache before reopening — otherwise the stale in-memory HNSW segment is reused and can persist an outdated index over the peer's writes (index count going backwards). (#2002, #2028, #2026, #2032)
 - **MCP releases the palace writer lease on SIGTERM/SIGHUP.** The lease was only released via `atexit`, which CPython skips on those signals' default disposition. SSH disconnect (SIGHUP) and container/systemd stop (SIGTERM) therefore left `mine_palace_*.lock` naming a dead PID until a contender's liveness check reclaimed it. `main()` now installs handlers that exit through `sys.exit`, so the existing `atexit` release path runs. (#2205)
 
 ---
diff --git a/mempalace/convo_miner.py b/mempalace/convo_miner.py
index c9e4995f49..d6dc82d423 100644
--- a/mempalace/convo_miner.py
+++ b/mempalace/convo_miner.py
@@ -688,59 +688,85 @@ def _file_chunks_locked(
         # the embedding speedup without one huge Chroma/SQLite request. Keep
         # one filed_at per source file so all transcript drawers share an
         # ingest timestamp.
+        #
+        # Every drawer of this pass carries ``chunk_total`` so
+        # ``file_already_mined`` / ``prefetch_mined_set`` can tell a complete
+        # multi-batch mine from one that crashed mid-file (#2183). Without it
+        # a stable mtime + any surviving drawer permanently skips the file
+        # and the missing exchanges never come back.
         filed_at = datetime.now().isoformat()
         try:
             source_mtime = os.path.getmtime(source_file)
         except OSError:
             source_mtime = None
-        for batch_start in range(0, len(chunks), DRAWER_UPSERT_BATCH_SIZE):
-            batch_docs: list = []
-            batch_ids: list = []
-            batch_metas: list = []
-            for chunk in chunks[batch_start : batch_start + DRAWER_UPSERT_BATCH_SIZE]:
-                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 = make_convo_drawer_id(
-                    wing, chunk_room, source_file, extract_mode, chunk["chunk_index"]
-                )
-                batch_docs.append(chunk["content"])
-                batch_ids.append(drawer_id)
-                meta = {
-                    "wing": wing,
-                    "room": chunk_room,
-                    "hall": _detect_hall_cached(chunk["content"]),
-                    "source_file": source_file,
-                    "chunk_index": chunk["chunk_index"],
-                    "added_by": agent,
-                    "filed_at": filed_at,
-                    "entities": entities_metadata(chunk["content"]),
-                    "authored_at": authored_at if authored_at is not None else filed_at,
-                    "ingest_mode": "convos",
-                    "extract_mode": extract_mode,
-                    "normalize_version": NORMALIZE_VERSION,
-                    "id_recipe": ID_RECIPE,
-                }
-                if source_mtime is not None:
-                    meta["source_mtime"] = source_mtime
-                # Stamp content_hash only on chunk 0 so multi-conversation
-                # privacy-export hashes are not O(N²)-duplicated across every
-                # chunk row. ``prefetch_content_hashes`` still finds them —
-                # it scans all drawers and splits comma-joined hash fields.
-                if content_hash is not None and chunk.get("chunk_index", 0) == 0:
-                    meta["content_hash"] = content_hash
-                batch_metas.append(meta)
-            assert_no_collisions(list(zip(batch_ids, batch_metas)), collection)
+        chunk_total = len(chunks)
+        try:
+            for batch_start in range(0, len(chunks), DRAWER_UPSERT_BATCH_SIZE):
+                batch_docs: list = []
+                batch_ids: list = []
+                batch_metas: list = []
+                for chunk in chunks[batch_start : batch_start + DRAWER_UPSERT_BATCH_SIZE]:
+                    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 = make_convo_drawer_id(
+                        wing, chunk_room, source_file, extract_mode, chunk["chunk_index"]
+                    )
+                    batch_docs.append(chunk["content"])
+                    batch_ids.append(drawer_id)
+                    meta = {
+                        "wing": wing,
+                        "room": chunk_room,
+                        "hall": _detect_hall_cached(chunk["content"]),
+                        "source_file": source_file,
+                        "chunk_index": chunk["chunk_index"],
+                        "added_by": agent,
+                        "filed_at": filed_at,
+                        "entities": entities_metadata(chunk["content"]),
+                        "authored_at": authored_at if authored_at is not None else filed_at,
+                        "ingest_mode": "convos",
+                        "extract_mode": extract_mode,
+                        "normalize_version": NORMALIZE_VERSION,
+                        "id_recipe": ID_RECIPE,
+                        "chunk_total": chunk_total,
+                    }
+                    if source_mtime is not None:
+                        meta["source_mtime"] = source_mtime
+                    # Stamp content_hash only on chunk 0 so multi-conversation
+                    # privacy-export hashes are not O(N²)-duplicated across every
+                    # chunk row. ``prefetch_content_hashes`` still finds them —
+                    # it scans all drawers and splits comma-joined hash fields.
+                    if content_hash is not None and chunk.get("chunk_index", 0) == 0:
+                        meta["content_hash"] = content_hash
+                    batch_metas.append(meta)
+                assert_no_collisions(list(zip(batch_ids, batch_metas)), collection)
+                try:
+                    collection.upsert(
+                        documents=batch_docs,
+                        ids=batch_ids,
+                        metadatas=batch_metas,
+                    )
+                    drawers_added += len(batch_docs)
+                except Exception as e:
+                    if "already exists" not in str(e).lower():
+                        raise
+        except Exception:
+            # A successful earlier batch has the source's current mtime and
+            # chunk_total. Leaving those drawers behind would make the next
+            # run treat the incomplete set as fully filed (#2183 / #2122).
             try:
-                collection.upsert(
-                    documents=batch_docs,
-                    ids=batch_ids,
-                    metadatas=batch_metas,
+                delete_ids = _source_file_delete_ids(collection, source_file, extract_mode)
+                if delete_ids:
+                    collection.delete(ids=delete_ids)
+            except Exception:
+                logger.warning(
+                    "Failed to clean partial convo drawers after upsert error for %s",
+                    source_file,
+                    exc_info=True,
                 )
-                drawers_added += len(batch_docs)
-            except Exception as e:
-                if "already exists" not in str(e).lower():
-                    raise
+            raise
     return drawers_added, room_counts_delta, False
 
 
diff --git a/mempalace/palace.py b/mempalace/palace.py
index db2071b4a8..255d7518b7 100644
--- a/mempalace/palace.py
+++ b/mempalace/palace.py
@@ -1550,12 +1550,21 @@ def prefetch_mined_set(
     When extract_mode is set, mirrors file_already_mined(..., extract_mode=...)
     so conversation mines skip per extraction mode rather than per source file.
 
+    Completeness mirrors :func:`file_already_mined`'s ``chunk_total`` rule
+    (#2183): a source that only has a mid-file partial (surviving drawers
+    share the current mtime but are short of ``chunk_total``) is **omitted**
+    from the result so the bulk skip path re-mines instead of permanently
+    stranding the missing exchanges. Drawers with no ``chunk_total``
+    (legacy rows, registry sentinels) are trusted on their own, as before.
+
     The convo miner walks thousands of transcript files; per-file
     `collection.get(where={"source_file": X})` costs ~2s on a 150k-drawer
     palace, making a 2000-file sweep take >1h of pure skip-checking. This
     helper drops that to a single paginated scan plus O(1) lookups.
     """
-    mined: dict[str, Optional[float]] = {}
+    # Per source_file: per stored_mtime group → count + optional chunk_total.
+    # A source is only "mined" once some group is complete.
+    groups: dict[str, dict] = {}
     try:
         total = collection.count()
         offset = 0
@@ -1570,14 +1579,37 @@ def prefetch_mined_set(
                     continue
                 # Same default as file_already_mined: missing version == 1
                 version = meta.get("normalize_version", 1)
-                if version >= NORMALIZE_VERSION:
-                    stored_mtime = meta.get("source_mtime")
-                    mined[src] = float(stored_mtime) if stored_mtime is not None else None
+                if version < NORMALIZE_VERSION:
+                    continue
+                stored_mtime = meta.get("source_mtime")
+                mtime_key = float(stored_mtime) if stored_mtime is not None else None
+                entry = groups.setdefault(src, {}).setdefault(
+                    mtime_key, {"count": 0, "chunk_total": None}
+                )
+                entry["count"] += 1
+                chunk_total = meta.get("chunk_total")
+                if chunk_total is not None:
+                    try:
+                        entry["chunk_total"] = int(chunk_total)
+                    except (TypeError, ValueError):
+                        pass
             if not batch["ids"]:
                 break
             offset += len(batch["ids"])
     except Exception:
-        logger.warning("prefetch_mined_set: partial fetch, %d files loaded", len(mined))
+        logger.warning("prefetch_mined_set: partial fetch, %d source groups loaded", len(groups))
+
+    mined: dict[str, Optional[float]] = {}
+    for src, by_mtime in groups.items():
+        for mtime_key, entry in by_mtime.items():
+            chunk_total = entry["chunk_total"]
+            if chunk_total is None:
+                # Legacy / registry: no completion marker — trust membership.
+                mined[src] = mtime_key
+                break
+            if entry["count"] >= chunk_total:
+                mined[src] = mtime_key
+                break
     return mined
 
 
diff --git a/tests/test_convo_miner.py b/tests/test_convo_miner.py
index 3a069e97b9..b9bf8c9ce7 100644
--- a/tests/test_convo_miner.py
+++ b/tests/test_convo_miner.py
@@ -14,7 +14,12 @@
     _resolve_wing,
     mine_convos,
 )
-from mempalace.palace import MineAlreadyRunning, file_already_mined, prefetch_mined_set
+from mempalace.palace import (
+    NORMALIZE_VERSION,
+    MineAlreadyRunning,
+    file_already_mined,
+    prefetch_mined_set,
+)
 
 
 def test_convo_mining():
@@ -767,6 +772,71 @@ def test_prefetch_mined_set_none_for_drawer_without_stored_mtime():
         shutil.rmtree(tmpdir, ignore_errors=True)
 
 
+def test_prefetch_mined_set_omits_incomplete_chunk_total_group():
+    """Mid-file partials with chunk_total must not bulk-skip the source (#2183)."""
+    tmpdir = tempfile.mkdtemp()
+    try:
+        palace_path = os.path.join(tmpdir, "palace")
+        client = chromadb.PersistentClient(path=palace_path)
+        col = client.get_or_create_collection("mempalace_drawers")
+        mtime = 1_700_000_000.0
+        source = "/fake/session.jsonl"
+        # Only 2 of 3 expected chunks landed before a crash.
+        col.upsert(
+            ids=["d0", "d1"],
+            documents=["chunk 0", "chunk 1"],
+            metadatas=[
+                {
+                    "wing": "test",
+                    "room": "general",
+                    "source_file": source,
+                    "chunk_index": 0,
+                    "extract_mode": "exchange",
+                    "normalize_version": NORMALIZE_VERSION,
+                    "source_mtime": mtime,
+                    "chunk_total": 3,
+                },
+                {
+                    "wing": "test",
+                    "room": "general",
+                    "source_file": source,
+                    "chunk_index": 1,
+                    "extract_mode": "exchange",
+                    "normalize_version": NORMALIZE_VERSION,
+                    "source_mtime": mtime,
+                    "chunk_total": 3,
+                },
+            ],
+        )
+        mined = prefetch_mined_set(col, extract_mode="exchange")
+        assert source not in mined, (
+            "prefetch_mined_set treated 2/3 chunks as fully filed — the bulk "
+            "skip path would permanently strand the missing exchange (#2183)"
+        )
+
+        col.upsert(
+            ids=["d2"],
+            documents=["chunk 2"],
+            metadatas=[
+                {
+                    "wing": "test",
+                    "room": "general",
+                    "source_file": source,
+                    "chunk_index": 2,
+                    "extract_mode": "exchange",
+                    "normalize_version": NORMALIZE_VERSION,
+                    "source_mtime": mtime,
+                    "chunk_total": 3,
+                }
+            ],
+        )
+        mined = prefetch_mined_set(col, extract_mode="exchange")
+        assert source in mined
+        assert abs(mined[source] - mtime) < 0.001
+    finally:
+        shutil.rmtree(tmpdir, ignore_errors=True)
+
+
 def test_mine_convos_reprocesses_legacy_drawer_without_stored_mtime(capsys):
     """A file mined before source_mtime was tracked (simulated: drawer
     written directly, no source_mtime field) must be re-mined on the next
diff --git a/tests/test_convo_miner_unit.py b/tests/test_convo_miner_unit.py
index d6bb4c54f0..3545c791a5 100644
--- a/tests/test_convo_miner_unit.py
+++ b/tests/test_convo_miner_unit.py
@@ -718,6 +718,107 @@ def upsert(self, documents, ids, metadatas):
         assert drawers == 0
         assert skipped is True
 
+    def test_stamps_chunk_total_for_completion_check(self, monkeypatch):
+        """Every convo drawer of one pass must carry chunk_total (#2183)."""
+        import mempalace.convo_miner as convo_miner
+
+        class FakeCol:
+            def __init__(self):
+                self.metas = []
+
+            def delete(self, *args, **kwargs):
+                pass
+
+            def get(self, ids=None, include=None, **kwargs):
+                return {"ids": [], "metadatas": []}
+
+            def upsert(self, documents, ids, metadatas):
+                self.metas.extend(metadatas)
+
+        chunks = [{"content": f"chunk {i} " * 20, "chunk_index": i} for i in range(5)]
+        col = FakeCol()
+        monkeypatch.setattr(convo_miner, "DRAWER_UPSERT_BATCH_SIZE", 2)
+        monkeypatch.setattr(
+            convo_miner, "file_already_mined", lambda collection, source_file, **kwargs: False
+        )
+        monkeypatch.setattr(convo_miner, "mine_lock", lambda source_file: contextlib.nullcontext())
+        monkeypatch.setattr(convo_miner, "_detect_hall_cached", lambda content: "conversations")
+
+        _file_chunks_locked(col, "chat.txt", chunks, "wing", "general", "agent", "exchange")
+
+        assert len(col.metas) == 5
+        assert all(m.get("chunk_total") == 5 for m in col.metas), (
+            "not every convo chunk carries the pass's chunk_total — a mid-file "
+            "crash would leave mtime-stamped partials that skip forever (#2183)"
+        )
+
+    def test_cleans_partial_drawers_after_batch_upsert_failure(self, monkeypatch, tmp_path):
+        """A failed later batch must not leave mtime-stamped partials (#2183)."""
+        import mempalace.convo_miner as convo_miner
+
+        class FailingCol:
+            def __init__(self):
+                self.records = []
+                self.upsert_calls = 0
+                self.deleted_ids = []
+
+            def get(self, where=None, limit=None, offset=0, include=None, ids=None, **kwargs):
+                if ids is not None:
+                    return {"ids": [], "metadatas": []}
+                records = self.records
+                if where and "source_file" in where:
+                    records = [
+                        r
+                        for r in records
+                        if r["metadata"].get("source_file") == where["source_file"]
+                    ]
+                page = records[offset : offset + (limit or len(records))]
+                return {
+                    "ids": [r["id"] for r in page],
+                    "metadatas": [r["metadata"] for r in page],
+                }
+
+            def delete(self, ids=None, where=None, **kwargs):
+                if ids:
+                    self.deleted_ids.extend(ids)
+                    id_set = set(ids)
+                    self.records = [r for r in self.records if r["id"] not in id_set]
+                    return
+                if where and "source_file" in where:
+                    src = where["source_file"]
+                    self.records = [
+                        r for r in self.records if r["metadata"].get("source_file") != src
+                    ]
+
+            def upsert(self, documents, ids, metadatas):
+                self.upsert_calls += 1
+                if self.upsert_calls == 2:
+                    raise RuntimeError("simulated second-batch failure")
+                self.records.extend(
+                    {"id": drawer_id, "metadata": metadata}
+                    for drawer_id, metadata in zip(ids, metadatas)
+                )
+
+        source = tmp_path / "chat.txt"
+        source.write_text("content\n", encoding="utf-8")
+        chunks = [{"content": f"chunk {i} " * 20, "chunk_index": i} for i in range(3)]
+        col = FailingCol()
+        monkeypatch.setattr(convo_miner, "DRAWER_UPSERT_BATCH_SIZE", 2)
+        monkeypatch.setattr(
+            convo_miner, "file_already_mined", lambda collection, source_file, **kwargs: False
+        )
+        monkeypatch.setattr(convo_miner, "mine_lock", lambda source_file: contextlib.nullcontext())
+        monkeypatch.setattr(convo_miner, "_detect_hall_cached", lambda content: "conversations")
+
+        with pytest.raises(RuntimeError, match="second-batch failure"):
+            _file_chunks_locked(col, str(source), chunks, "wing", "general", "agent", "exchange")
+
+        assert col.records == [], (
+            "partial convo drawers survived a mid-file upsert failure — the "
+            "next mine would skip this incomplete file forever (#2183)"
+        )
+        assert col.deleted_ids, "cleanup did not delete the partial drawer ids"
+
 
 class TestSourceFileDeleteIds:
     """#104: the sweeper writes drawers with no extract_mode at all

From fc1431a66192010435da197f3d1f44e72e3bb89e Mon Sep 17 00:00:00 2001
From: Igor Lins e Silva <4753812+igorls@users.noreply.github.com>
Date: Wed, 12 Aug 2026 13:02:38 -0300
Subject: [PATCH 10/10] test(repair): release SharedSystemClient after seeding
 for Windows rename

In-place rebuild tests archive the palace directory after _seed_palace.
backend.close() alone left chromadb's path-keyed System holding files open
on Windows (WinError 5), so rebuild_from_sqlite aborted before the mocked
upsert path and test_rebuild_from_sqlite_raises_on_upsert_failure never
raised RebuildPartialError. Clear the shared cache and GC after close.
---
 tests/test_repair.py | 12 +++++++++++-
 1 file changed, 11 insertions(+), 1 deletion(-)

diff --git a/tests/test_repair.py b/tests/test_repair.py
index 2179e8443b..9338553d2f 100644
--- a/tests/test_repair.py
+++ b/tests/test_repair.py
@@ -1976,7 +1976,9 @@ def _seed_palace(palace_path, collection_name, rows):
 
     ``rows`` is a list of ``(id, document, metadata)`` tuples.
     """
-    from mempalace.backends.chroma import ChromaBackend
+    import gc
+
+    from mempalace.backends.chroma import ChromaBackend, _clear_chroma_system_cache
 
     backend = ChromaBackend()
     try:
@@ -1991,7 +1993,15 @@ def _seed_palace(palace_path, collection_name, rows):
         # caller proceeds. Without this, an in-place rebuild on Windows
         # fails with WinError 32 on data_level0.bin during the archive
         # rename (cf. PR #1310 test-windows job).
+        #
+        # Also drop the process-global SharedSystemClient cache: closing the
+        # backend releases our PersistentClient handle, but chromadb can keep
+        # the path-keyed System alive and on Windows that blocks renaming the
+        # palace directory (WinError 5 Access is denied). Seen on PR #2228
+        # ``test_rebuild_from_sqlite_raises_on_upsert_failure``.
         backend.close()
+        _clear_chroma_system_cache()
+        gc.collect()
 
 
 def test_extract_via_sqlite_returns_all_rows_with_metadata(tmp_path):