fix(state): coordinate SessionDB teardown with active writers

The #102827 corruption is pure zero holes -- frames lost across a WAL
generation. SessionDB.close() produces exactly that when it runs against a
file another live handle is still writing: PRAGMA wal_checkpoint(PASSIVE),
then the connection close that lets SQLite unlink -wal/-shm. The dangerous
event is a physical close overlapping any other live physical lifetime for
the same path, so both sides of it are closed here.

Late write vs. close: a cron watchdog timeout only stops waiting, and
ThreadPoolExecutor.shutdown(wait=False) cannot interrupt a worker already
inside run_conversation. The agent and its registry reference are now held
until that worker's Future completes, so its last frames land before any
checkpoint.

Close vs. open: the per-path barrier now COUNTS admitted teardowns. A path
can own several closes at once -- the current generation's final release and
a retired generation's drain are admitted independently under the registry
lock, and the per-path mutex only serializes teardowns that already entered
it. With one bare event per path, a releasing thread descheduled between
generation removal and the mutex let the next teardown to settle remove and
signal the shared event: close_all() returned over a pending close and
acquire() published a replacement writer on top of a handle still inside
checkpoint/unlink. _TeardownBarrier tracks event + pending count,
_admit_teardown_locked registers each close in the same lock section that
removes the generation, and only the last settled teardown lifts the
barrier. Physical I/O stays outside the registry lock and unrelated paths
still progress independently.

The auto-archive sweep called release_or_close in its finally while the
import was local to a different function, so every eligible sweep raised
NameError, the outer except Exception swallowed it at debug level, and the
borrowed registry reference was never returned -- a holder leak that pins a
retired generation open. The helper is now bound in the calling scope.

Remaining in-process writable SessionDB() call sites (trace upload, the
API-server profile cache, the web-server writable paths, startup schema
reconcile) go through the canonical registry acquire/release_or_close, and
gateway maintenance borrows pinned handles instead of iterating an unpinned
snapshot.

Regressions: overlapping final releases of the current and retired
generations in both orderings with the first paused before the lifecycle
mutex, teardown-error settlement, an unrelated-path control, and refcount
assertions for the auto-archive sweep on success, on failure, across
repeated sweeps and with auto-archive disabled.

Fixes #102827

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BzxCWw6SuHXhXMdkiEwMa2

agent/trace_upload.py

@@ -210,17 +210,14 @@ def _do_upload(jsonl: str, *, token: str, session_id: str, dataset_name: str = D
 def load_session_messages(session_id: str, db_path=None) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]:
     """``(messages, meta)`` from SQLite; ``meta`` is ``{}`` when the session row is missing (a live, untitled
     session may still have messages)."""
-    from hermes_state import SessionDB
-    db = SessionDB(db_path=db_path) if db_path else SessionDB()
+    from hermes_state_registry import acquire, release_or_close
+    db = acquire(db_path) if db_path else acquire()
     try:
         resolved = db.resolve_session_id(session_id) or session_id
         meta = db.get_session(resolved) or {}
         return db.get_messages_as_conversation(resolved), meta
     finally:
-        try:
-            db.close()
-        except Exception:
-            logger.debug("Failed to close trace-upload SessionDB", exc_info=True)
+        release_or_close(db)
 
 
 def upload_session_trace(

cron/scheduler.py

@@ -1739,6 +1739,7 @@ def _raise_inactivity_timeout(agent, job_name: str, limit_s: float) -> None:
 
 def _run_agent_with_watchdog(
     agent, prompt: str, job: dict, job_id: str, job_name: str, task_id: str, cancel_event,
+    worker_state: Optional[dict] = None,
 ) -> dict:
     """Run ``agent.run_conversation`` on a worker thread under the inactivity (not wall-clock)
     watchdog: default 600s, override HERMES_CRON_TIMEOUT, 0 = unlimited."""
@@ -1783,6 +1784,8 @@ def _run_agent_with_watchdog(
     _cron_context = contextvars.copy_context()
     _cron_future = _cron_pool.submit(
         _cron_context.run, agent.run_conversation, prompt, task_id=task_id)
+    if worker_state is not None:
+        worker_state["future"] = _cron_future
     _inactivity_timeout = False
     _watch_stop = threading.Event()
 
@@ -1996,6 +1999,103 @@ def _finalize_cron_session(session_db, agent, job_id: str, job_name: str, cron_s
         logger.debug("Job '%s': failed to close SQLite session store: %s", job_id, e)
 
 
+def _finalize_cron_session_db(
+    session_db, agent, job_id: str, job_name: str, cron_session_id: str,
+) -> None:
+    """Finalize a cron session exactly once before releasing its registry reference."""
+    _finalize_cron_session(session_db, agent, job_id, job_name, cron_session_id)
+
+
+def _teardown_detached_cron_worker(
+    session_db, agent, job_id: str, job_name: str, cron_session_id: str,
+) -> None:
+    """Release a timed-out worker's agent and SessionDB after it finishes."""
+    try:
+        if session_db:
+            _finalize_cron_session_db(
+                session_db, agent, job_id, job_name, cron_session_id)
+    except BaseException as exc:
+        logger.error(
+            "Job '%s': detached worker session teardown failed: %s",
+            job_id,
+            exc,
+            exc_info=(type(exc), exc, exc.__traceback__),
+        )
+    finally:
+        _teardown_cron_agent(agent, job_id)
+
+
+def _defer_cron_worker_teardown_if_running(
+    worker_state: dict,
+    session_db,
+    agent,
+    job_id: str,
+    job_name: str,
+    cron_session_id: str,
+) -> bool:
+    """Keep the agent and SessionDB alive until a timed-out worker finishes.
+
+    ``ThreadPoolExecutor.shutdown(wait=False)`` does not stop a running
+    ``run_conversation`` call. Closing its SessionDB from ``run_job``'s
+    ``finally`` would therefore recreate the close-vs-write race this module
+    is meant to prevent. The real Future always supports ``add_done_callback``;
+    unknown test doubles are handled conservatively by waiting before inline
+    cleanup rather than closing a possibly-live handle.
+    """
+    future = worker_state.get("future")
+    if future is None:
+        return False
+    done = getattr(future, "done", None)
+    if callable(done):
+        try:
+            if bool(done()):
+                return False
+        except Exception as exc:
+            logger.warning(
+                "Job '%s': could not determine worker completion; deferring teardown: %s",
+                job_id,
+                exc,
+            )
+
+    add_done_callback = getattr(future, "add_done_callback", None)
+    if not callable(add_done_callback):
+        logger.error(
+            "Job '%s': worker future cannot register teardown callback; waiting for worker",
+            job_id,
+        )
+        try:
+            future.result()
+        except BaseException as exc:
+            logger.debug("Job '%s': detached worker completed with: %s", job_id, exc)
+        return False
+
+    if worker_state.get("teardown_registered"):
+        return True
+    worker_state["teardown_registered"] = True
+
+    def _finish_detached_worker(_future) -> None:
+        _teardown_detached_cron_worker(
+            session_db, agent, job_id, job_name, cron_session_id)
+
+    try:
+        add_done_callback(_finish_detached_worker)
+    except Exception as exc:
+        # A real concurrent.futures.Future does not reject this call. If a
+        # custom future does, wait rather than releasing a live SQLite handle.
+        worker_state["teardown_registered"] = False
+        logger.error(
+            "Job '%s': failed to register detached-worker teardown; waiting: %s",
+            job_id,
+            exc,
+        )
+        try:
+            future.result()
+        except BaseException as result_exc:
+            logger.debug("Job '%s': detached worker completed with: %s", job_id, result_exc)
+        return False
+    return True
+
+
 def _run_doc_header(job: dict, title: str, job_id: str, prompt: str) -> str:
     """Header of the persisted run document (title, ids, schedule, prompt)."""
     return (
@@ -2330,6 +2430,7 @@ def run_job(
     model = ""
     _session_db = None
     _audit: Optional[_FireAudit] = None
+    _worker_state: dict = {}
     scope = _CronRunScope(job, job_id, execution_id)
     try:
         scope.enter()
@@ -2353,7 +2454,8 @@ def run_job(
         _audit = _FireAudit(job, job_id, model)
 
         result = _run_agent_with_watchdog(
-            agent, prompt, job, job_id, job_name, scope.task_id, cancel_event)
+            agent, prompt, job, job_id, job_name, scope.task_id, cancel_event,
+            worker_state=_worker_state)
         final_response = _final_response_from_result(result, job_id, job_name, AIAgent)
         # Keep final_response clean for delivery logic (empty = no delivery).
         logged_response = final_response if final_response else "(No response generated)"
@@ -2375,9 +2477,17 @@ def run_job(
         return False, output, "", error_msg
 
     finally:
+        # A watchdog timeout only stops waiting; executor.shutdown(wait=False)
+        # cannot interrupt a worker already inside run_conversation. Keep both
+        # the agent and its registry-owned SessionDB alive until that worker's
+        # Future completes, otherwise its late persistence can race close/WAL
+        # checkpoint teardown.
+        _worker_teardown_deferred = _defer_cron_worker_teardown_if_running(
+            _worker_state, _session_db, agent, job_id, job_name, _cron_session_id)
         scope.exit()
-        if _session_db:
-            _finalize_cron_session(_session_db, agent, job_id, job_name, _cron_session_id)
+        if _session_db and not _worker_teardown_deferred:
+            _finalize_cron_session_db(
+                _session_db, agent, job_id, job_name, _cron_session_id)
         # Tear down the ephemeral agent or the gateway leaks fds per tick (EMFILE). With deferred
         # teardown, hand the live agent back: delivery needs a live async client.
         # Release subprocesses, terminal sandboxes, browser daemons, and the main OpenAI/httpx client held
@@ -2385,11 +2495,12 @@ def run_job(
         # per job until it hits EMFILE (#10200 / "too many open files"). When the caller opted to defer
         # teardown (passed a list), hand the live agent back instead of closing it here — delivery must run
         # against a live async client, and the caller tears down afterwards (#58720).
-        if defer_agent_teardown is not None:
-            if agent is not None:
-                defer_agent_teardown.append(agent)
-        else:
-            _teardown_cron_agent(agent, job_id)
+        if not _worker_teardown_deferred:
+            if defer_agent_teardown is not None:
+                if agent is not None:
+                    defer_agent_teardown.append(agent)
+            else:
+                _teardown_cron_agent(agent, job_id)
 
 
 def _teardown_cron_agent(

gateway/platforms/api_server.py

@@ -1637,14 +1637,14 @@ class APIServerAdapter(OpenAICompatRoutesMixin, BasePlatformAdapter):
     def _open_and_cache_session_db(self, home) -> Optional[Any]:
         """Cached SessionDB for ``home`` (shared by both ``_ensure_session_db*``). Never writes
         ``self._session_db`` (explicit override only), so no profile pins later requests."""
-        from hermes_state import SessionDB
+        from hermes_state_registry import acquire
         key = str(home)
         with self._session_db_cache_lock:
             if self._session_db_cache_closed:
                 return None
             db = self._session_dbs.get(key)
             if db is None:
-                db = SessionDB(db_path=home / "state.db")
+                db = acquire(home / "state.db")
                 self._session_dbs[key] = db
             return db
 
@@ -1659,7 +1659,8 @@ class APIServerAdapter(OpenAICompatRoutesMixin, BasePlatformAdapter):
             if db is shared_db:
                 continue
             try:
-                db.close()
+                from hermes_state_registry import release_or_close
+                release_or_close(db)
             except Exception:
                 logger.debug("Failed to close API-server SessionDB", exc_info=True)
 

gateway/run.py

@@ -4467,13 +4467,14 @@ def _housekeeping_deferred_fts_retry() -> None:
     # Retry here, on the existing tick, against the shared instances this process already holds:
     # non-blocking admission, no new thread, rate-limited inside SessionDB. No-op when nothing is stale (one
     # attribute read per instance). See #100108.
-    from hermes_state_registry import live_shared_session_dbs
-    for _sdb in live_shared_session_dbs():
-        _retry = getattr(_sdb, "retry_deferred_fts_recovery", None)
-        if callable(_retry) and _retry():
-            logger.info(
-                "Deferred state.db FTS rebuild completed in-process for %s; full-text search restored.",
-                getattr(_sdb, "db_path", "state.db"))
+    from hermes_state_registry import borrow_live_shared_session_dbs
+    with borrow_live_shared_session_dbs() as _session_dbs:
+        for _sdb in _session_dbs:
+            _retry = getattr(_sdb, "retry_deferred_fts_recovery", None)
+            if callable(_retry) and _retry():
+                logger.info(
+                    "Deferred state.db FTS rebuild completed in-process for %s; full-text search restored.",
+                    getattr(_sdb, "db_path", "state.db"))
 
 
 def _housekeeping_memory_trim() -> None:

hermes_cli/web_server_lifecycle.py

@@ -175,9 +175,11 @@ def _eager_reconcile_own_session_db() -> None:
     per-poll read-probe heal in :func:`_open_session_db_at_path`.
     """
     try:
-        from hermes_state import SessionDB, _default_db_path
+        from hermes_state import _default_db_path
+        from hermes_state_registry import acquire, release_or_close
 
-        SessionDB(db_path=Path(_default_db_path()), read_only=False).close()
+        db = acquire(Path(_default_db_path()))
+        release_or_close(db)
     except Exception as exc:
         _log.warning(
             "startup schema reconcile of state.db failed (%s); session "

hermes_cli/web_server_sessions.py

@@ -110,12 +110,13 @@ def _open_session_db_at_path(db_path: Path, *, read_only: bool):
     import sqlite3
 
     from hermes_state import SessionDB, is_malformed_schema_error
+    from hermes_state_registry import acquire, release_or_close
 
     # Read-only file/sidecar preflight (port of kilocode#12508): repair-or-refuse BEFORE the first
     # connection so users get an actionable message instead of an opaque "attempt to write a readonly
     # database" from deep inside _init_schema.
     if not read_only:
-        return SessionDB(db_path=db_path, read_only=False)
+        return acquire(db_path)
 
     def _needs_bootstrap() -> bool:
         try:
@@ -128,7 +129,8 @@ def _open_session_db_at_path(db_path: Path, *, read_only: bool):
     if _needs_bootstrap():
         with _session_db_bootstrap_lock:
             if _needs_bootstrap():
-                SessionDB(db_path=db_path, read_only=False).close()
+                db = acquire(db_path)
+                release_or_close(db)
 
     def _open_probed():
         db = SessionDB(db_path=db_path, read_only=True)
@@ -156,7 +158,8 @@ def _open_session_db_at_path(db_path: Path, *, read_only: bool):
             or is_malformed_schema_error(exc)
             or isinstance(exc, UnicodeDecodeError)):
             raise
-        SessionDB(db_path=db_path, read_only=False).close()
+        db = acquire(db_path)
+        release_or_close(db)
         try:
             return _open_probed()
         except (sqlite3.DatabaseError, UnicodeDecodeError) as still_stale:
@@ -213,16 +216,20 @@ def _maybe_auto_archive_for_profile(profile: Optional[str]) -> None:
         _last_auto_archive_check[key] = now
 
         from hermes_cli.config import load_config as _load_full_config
+        from hermes_state_registry import release_or_close
         cfg = (_load_full_config().get("sessions") or {})
         if not cfg.get("auto_archive", False):
             return
+        # Bind the release helper BEFORE acquiring: the sweep's ``finally`` must
+        # never raise NameError over a held registry reference, or every eligible
+        # sweep leaks one and pins a retired generation open forever.
         db = _open_session_db_for_profile(profile, read_only=False)
         try:
             db.maybe_auto_archive(
                 idle_days=float(cfg.get("auto_archive_days", 3)),
                 min_interval_hours=int(cfg.get("min_interval_hours", 24)))
         finally:
-            db.close()
+            release_or_close(db)
     except Exception as exc:
         _log.debug("opportunistic auto-archive skipped: %s", exc)
 

hermes_state_registry.py

@@ -8,7 +8,9 @@ when the file is replaced (snapshot restore, recovery swap).
 
 Lifecycle rules:
 - ``acquire(path)`` returns the current generation for *path* and bumps its refcount.
-- ``close()`` on a shared instance is a NO-OP: the registry owns the connection lifecycle.
+- ``close()`` on a shared instance RELEASES one refcount instead of tearing the
+  connection down: the registry owns the physical lifecycle and only closes on the
+  final release, so legacy call sites return their reference instead of leaking it.
 - ``release(db)`` decrements the generation *db was acquired from* (object-keyed, so an
   inode replacement cannot strand a still-owned generation); the final release of a
   retired generation tears it down.
@@ -16,6 +18,14 @@ Lifecycle rules:
   its holders release. If the replacement open fails the registry keeps NO path entry.
 - All teardown happens OUTSIDE the registry lock: a final release's WAL checkpoint must
   never stall acquisition for every state.db.
+- A final close/checkpoint is serialized with the next open for the same path; no new
+  generation is published while the previous generation is still tearing down.
+- A path can have SEVERAL closes admitted at once (the current generation's final release
+  plus a retired generation's drain). The path barrier COUNTS them and is lifted only by
+  the last one to settle, so neither ``acquire`` nor ``close_all`` can escape while any
+  handle for that path is still inside checkpoint/WAL-unlink.
+- Maintenance callers borrow handles with a temporary registry reference instead of
+  iterating an unpinned snapshot.
 """
 
 from __future__ import annotations
@@ -24,7 +34,7 @@ import contextlib
 import logging
 import threading
 from pathlib import Path
-from typing import TYPE_CHECKING, Dict, List, Optional, Tuple
+from typing import TYPE_CHECKING, Dict, Iterator, List, Optional, Tuple
 
 from hermes_state_common import stat_db_file_identity as _stat_db_file_identity
 
@@ -34,12 +44,32 @@ if TYPE_CHECKING:  # pragma: no cover - import cycle guard, typed only
 logger = logging.getLogger(__name__)
 
 
+class _TeardownBarrier:
+    """Accounting for every admitted-but-unfinished physical close of one path.
+
+    A path can own more than one close at a time: the current generation's final
+    release and a retired generation's drain are admitted independently under
+    ``_lock`` and only meet at the lifecycle mutex. One event per path is honest
+    only if the LAST admitted teardown settles it. Signalling on the first lets
+    ``close_all()`` return and ``acquire()`` publish a replacement while an older
+    handle is still inside ``PRAGMA wal_checkpoint``/sidecar unlink -- the exact
+    overlap (#102827) this registry exists to forbid.
+    """
+
+    __slots__ = ("event", "pending")
+
+    def __init__(self) -> None:
+        self.event = threading.Event()
+        self.pending = 0
+
+
 class _Generation:
     """One shared SessionDB generation: instance, refcount, file identity."""
 
-    __slots__ = ("db", "refcount", "identity", "retired")
+    __slots__ = ("path", "db", "refcount", "identity", "retired")
 
-    def __init__(self, db: "SessionDB", identity: Optional[Tuple[int, int]]) -> None:
+    def __init__(self, path: Path, db: "SessionDB", identity: Optional[Tuple[int, int]]) -> None:
+        self.path = path
         self.db = db
         self.refcount = 1
         self.identity = identity
@@ -55,6 +85,16 @@ _retired: Dict[int, _Generation] = {}
 # (schema reconciliation can take seconds), but peers for the SAME file must wait or
 # every cold caller opens its own writer before a winner is chosen.
 _opening: Dict[Path, threading.Event] = {}
+# A final close/checkpoint must finish before a replacement writer is opened
+# for the same path. The barrier is admitted while holding _lock and lifted
+# only after the LAST admitted physical teardown, so acquire cannot slip
+# through the generation-removal/open gap and close_all cannot report a
+# finished sweep over a close that is still running.
+_tearing_down: Dict[Path, _TeardownBarrier] = {}
+# Open and close are both performed outside _lock. This per-path mutex closes
+# the race between checking _tearing_down and entering sqlite3.connect(),
+# including retired-generation drains after an inode replacement.
+_path_lifecycle_locks: Dict[Path, threading.Lock] = {}
 
 
 def _open_session_db(path: Path) -> "SessionDB":
@@ -74,6 +114,57 @@ def _teardown(db: "SessionDB") -> None:
         logger.debug("Error closing shared SessionDB", exc_info=True)
 
 
+def _path_lifecycle_lock_locked(path: Path) -> threading.Lock:
+    """Return the lifecycle mutex for *path* (caller holds ``_lock``)."""
+    lock = _path_lifecycle_locks.get(path)
+    if lock is None:
+        lock = threading.Lock()
+        _path_lifecycle_locks[path] = lock
+    return lock
+
+
+def _admit_teardown_locked(path: Path) -> _TeardownBarrier:
+    """Register one pending physical close for *path* (caller holds ``_lock``).
+
+    Admission shares the lock section that removes the generation, so a peer
+    release, ``acquire`` or ``close_all`` taking the lock next always sees this
+    teardown accounted for.
+    """
+    barrier = _tearing_down.get(path)
+    if barrier is None:
+        barrier = _tearing_down[path] = _TeardownBarrier()
+    barrier.pending += 1
+    return barrier
+
+
+def _finish_teardown(path: Path, barrier: _TeardownBarrier) -> None:
+    """Settle one admitted teardown; only the last one lifts the path barrier."""
+    with _lock:
+        barrier.pending -= 1
+        if barrier.pending > 0:
+            return
+        if _tearing_down.get(path) is barrier:
+            _tearing_down.pop(path, None)
+        barrier.event.set()
+
+
+def _teardown_generation(
+    path: Path,
+    db: "SessionDB",
+    *,
+    barrier: Optional[_TeardownBarrier] = None,
+) -> None:
+    """Close *db* under its path lifecycle mutex, then settle its barrier slot."""
+    with _lock:
+        lifecycle_lock = _path_lifecycle_lock_locked(path)
+    try:
+        with lifecycle_lock:
+            _teardown(db)
+    finally:
+        if barrier is not None:
+            _finish_teardown(path, barrier)
+
+
 def _db_path_of(db: "SessionDB") -> Optional[Path]:
     """``Path(db.db_path)`` or None when absent/unconvertible."""
     path = getattr(db, "db_path", None)
@@ -105,6 +196,7 @@ def acquire(db_path: Optional[Path] = None) -> "SessionDB":
         path = raw_path
 
     while True:
+        wait_for: Optional[threading.Event] = None
         with _lock:
             generation = _generations.get(path)
             if generation is not None:
@@ -119,36 +211,62 @@ def acquire(db_path: Optional[Path] = None) -> "SessionDB":
                 else:
                     generation.refcount += 1
                     return generation.db
-            opening = _opening.get(path)
-            if opening is None:
-                opening = _opening[path] = threading.Event()
-                break
+            teardown = _tearing_down.get(path)
+            if teardown is not None:
+                wait_for = teardown.event
+            else:
+                opening = _opening.get(path)
+                if opening is None:
+                    opening = _opening[path] = threading.Event()
+                    lifecycle_lock = _path_lifecycle_lock_locked(path)
+                    break
+                wait_for = opening
         # Another caller is constructing this path; wait without holding the global
         # lock. A failed opener signals too, so a waiter can retry.
-        opening.wait()
+        wait_for.wait()
 
     # Open OUTSIDE the lock; the per-path marker prevents redundant writers without
     # serialising other files.
     try:
-        db = _open_session_db(path)
-        db._shared_registry_owned = True
-        identity = _stat_db_file_identity(path)
+        # Serialize connection construction with a final close/checkpoint for
+        # this path, while keeping unrelated paths independent.
+        with lifecycle_lock:
+            db = _open_session_db(path)
+            db._shared_registry_owned = True
+            identity = _stat_db_file_identity(path)
     except BaseException:
         with _lock:
             _finish_opening(path, opening)
         raise
 
+    discard_barrier: Optional[_TeardownBarrier] = None
     with _lock:
-        existing = _generations.get(path)
-        if existing is not None:  # Defensive: installed by explicit registry manipulation mid-open.
-            existing.refcount += 1
-            winner = existing.db
+        teardown = _tearing_down.get(path)
+        if teardown is not None:
+            # A shutdown or retired-generation final release began while this
+            # opener was constructing the handle. Do not publish a new
+            # generation into that teardown window; close this speculative
+            # connection and retry after the barrier.
+            discard_barrier = teardown
+            winner = None
         else:
-            _generations[path] = _Generation(db, identity)
-            winner = db
+            existing = _generations.get(path)
+            if existing is not None:  # Defensive: installed by explicit registry manipulation mid-open.
+                existing.refcount += 1
+                winner = existing.db
+            else:
+                _generations[path] = _Generation(path, db, identity)
+                winner = db
         _finish_opening(path, opening)
+    if discard_barrier is not None:
+        with lifecycle_lock:
+            _teardown(db)
+        discard_barrier.event.wait()
+        return acquire(path)
+    assert winner is not None
     if winner is not db:
-        _teardown(db)
+        with lifecycle_lock:
+            _teardown(db)
     return winner
 
 
@@ -160,6 +278,8 @@ def release(db: "SessionDB") -> bool:
     if db is None:
         return False
     key = id(db)
+    teardown_path: Optional[Path] = None
+    teardown_barrier: Optional[_TeardownBarrier] = None
     with _lock:
         generation = _retired.get(key)
         if generation is None:
@@ -172,28 +292,61 @@ def release(db: "SessionDB") -> bool:
                 return False
         generation.refcount -= 1
         needs_teardown = generation.refcount <= 0
-        if needs_teardown and generation.retired:
-            _retired.pop(key, None)
-        elif needs_teardown and (path := _db_path_of(db)) is not None:
-            _generations.pop(path, None)
+        if needs_teardown:
+            teardown_path = generation.path
+            if generation.retired:
+                _retired.pop(key, None)
+            elif _generations.get(generation.path) is generation:
+                _generations.pop(generation.path, None)
+            # Remove the lendable entry and admit this close in the SAME lock
+            # section, then keep the path blocked until checkpoint/close
+            # completes. A retired generation's drain is admitted too: it
+            # checkpoints and unlinks the same sidecars as the current one, so
+            # a replacement writer must not open on top of it.
+            teardown_barrier = _admit_teardown_locked(generation.path)
     # Teardown OUTSIDE the lock: stopping the token writer, WAL checkpoint and read-pool
     # drain must not block acquisition for every other state.db.
     if needs_teardown:
-        _teardown(db)
+        assert teardown_path is not None
+        _teardown_generation(teardown_path, db, barrier=teardown_barrier)
     return True
 
 
 def close_all() -> int:
     """Close every shared SessionDB regardless of refcount; returns the count. For gateway
     shutdown, after all agents and cron jobs finished. Idempotent."""
+    teardown_barriers: Dict[Path, _TeardownBarrier] = {}
     with _lock:
+        active_teardowns = list(_tearing_down.values())
         generations = list(_generations.values()) + list(_retired.values())
+        for path in {generation.path for generation in generations}:
+            teardown_barriers[path] = _admit_teardown_locked(path)
         _generations.clear()
         _retired.clear()
         for generation in generations:
             generation.retired = True
+    # Teardown outside the lock, one path at a time. Holding the lifecycle
+    # mutex across all generations for a path prevents an old retired handle
+    # and the current handle from checkpointing the same sidecars concurrently.
+    by_path: Dict[Path, List[_Generation]] = {}
     for generation in generations:
-        _teardown(generation.db)
+        by_path.setdefault(generation.path, []).append(generation)
+    for path, path_generations in by_path.items():
+        with _lock:
+            lifecycle_lock = _path_lifecycle_lock_locked(path)
+        try:
+            with lifecycle_lock:
+                for generation in path_generations:
+                    _teardown(generation.db)
+        finally:
+            _finish_teardown(path, teardown_barriers[path])
+    # A concurrent final release may have removed its generation before this
+    # sweep took the registry lock. It still owns the physical close; wait for
+    # that barrier rather than returning while SQLite teardown is in flight.
+    # A barrier is lifted only once EVERY teardown admitted for its path has
+    # settled, so this cannot return over a close that is still running.
+    for barrier in active_teardowns:
+        barrier.event.wait()
     return len(generations)
 
 
@@ -205,6 +358,30 @@ def live_shared_session_dbs() -> List["SessionDB"]:
         return [g.db for g in _generations.values() if not g.retired]
 
 
+@contextlib.contextmanager
+def borrow_live_shared_session_dbs() -> Iterator[List["SessionDB"]]:
+    """Borrow live handles with registry references pinned for the whole block.
+
+    Maintenance must not operate on the unowned snapshot returned by
+    :func:`live_shared_session_dbs`: the last real owner could otherwise
+    release and physically close the connection between the snapshot and the
+    maintenance call. Each borrowed generation gets one temporary reference;
+    the ``finally`` block releases it even when maintenance raises.
+    """
+    with _lock:
+        borrowed_generations = [
+            generation for generation in _generations.values() if not generation.retired
+        ]
+        borrowed = [generation.db for generation in borrowed_generations]
+        for generation in borrowed_generations:
+            generation.refcount += 1
+    try:
+        yield borrowed
+    finally:
+        for db in reversed(borrowed):
+            release(db)
+
+
 def stats() -> Dict[str, int]:
     """Registry census for tests and diagnostics (no locks held long)."""
     with _lock:

tests/cron/test_cleanup_timeout.py

@@ -8,9 +8,14 @@ from __future__ import annotations
 
 import threading
 import time
+from concurrent.futures import Future
 from unittest.mock import MagicMock, patch
 
-from cron.scheduler import run_job, _teardown_cron_agent
+from cron.scheduler import (
+    _defer_cron_worker_teardown_if_running,
+    _teardown_cron_agent,
+    run_job,
+)
 
 
 _RUNTIME = {
@@ -94,6 +99,35 @@ def test_agent_teardown_is_bounded():
         release.set()
 
 
+def test_detached_worker_teardown_waits_for_future():
+    """A timed-out worker keeps its agent and SessionDB until completion."""
+    future = Future()
+    worker_state = {"future": future}
+    fake_db = MagicMock()
+    agent = MagicMock()
+
+    with patch("cron.scheduler._teardown_detached_cron_worker") as teardown_worker:
+        assert _defer_cron_worker_teardown_if_running(
+            worker_state,
+            fake_db,
+            agent,
+            "detached-worker",
+            "detached worker",
+            "cron_detached-worker",
+        ) is True
+        teardown_worker.assert_not_called()
+
+        future.set_result({"final_response": "late"})
+
+        teardown_worker.assert_called_once_with(
+            fake_db,
+            agent,
+            "detached-worker",
+            "detached worker",
+            "cron_detached-worker",
+        )
+
+
 def test_dispatch_guard_releases_after_sessiondb_finalization_hang(tmp_path):
     """A second scheduler tick can fire the same job after cleanup times out."""
     import cron.scheduler as sched

tests/hermes_cli/test_web_server_auto_archive_registry.py

@@ -0,0 +1,100 @@
+"""The opportunistic auto-archive sweep must return its registry reference.
+
+``_maybe_auto_archive_for_profile`` borrows a shared writable ``SessionDB`` from
+the registry and returns it in a ``finally``. Its whole body is wrapped in
+``except Exception`` with a debug log, so a cleanup that raises is invisible:
+the sweep still archives, the endpoint still answers, and only the refcount is
+wrong. Every eligible sweep then leaks one reference, which pins the generation
+open and defeats the physical teardown the registry owns (#102827 / #103118).
+
+Asserting on the archive result or on "the helper did not raise" cannot see
+this. These regressions assert the refcount itself, on both the success and the
+failure path, and that the last real holder can still tear the handle down.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+import hermes_state_registry as registry
+import hermes_cli.web_server_sessions as sessions
+
+
+@pytest.fixture(autouse=True)
+def _reset_throttle():
+    """The sweep is throttled per profile for 300s; tests must not inherit it."""
+    sessions._last_auto_archive_check.clear()
+    yield
+    sessions._last_auto_archive_check.clear()
+
+
+def _arm_sweep(monkeypatch, tmp_path: Path, *, enabled: bool = True) -> Path:
+    """Point one sweep at ``tmp_path/state.db`` with a config double."""
+    db_path = tmp_path / "state.db"
+    config = {
+        "sessions": {
+            "auto_archive": enabled,
+            "auto_archive_days": 3,
+            "min_interval_hours": 0,
+        }
+    }
+    monkeypatch.setattr("hermes_cli.config.load_config", lambda *a, **kw: config)
+    monkeypatch.setattr(
+        sessions, "_open_session_db_for_profile",
+        lambda profile, *, read_only: registry.acquire(db_path))
+    return db_path
+
+
+def _refcount_for(db_path: Path) -> int:
+    generation = registry._generations.get(Path(db_path).resolve())
+    return 0 if generation is None else generation.refcount
+
+
+class TestAutoArchiveReleasesItsRegistryReference:
+    def test_successful_sweep_returns_the_borrowed_reference(self, tmp_path, monkeypatch):
+        db_path = _arm_sweep(monkeypatch, tmp_path)
+        holder = registry.acquire(db_path)
+        try:
+            before = _refcount_for(db_path)
+            sessions._maybe_auto_archive_for_profile(None)
+            assert _refcount_for(db_path) == before, "auto-archive leaked a registry reference"
+        finally:
+            assert registry.release(holder) is True
+        # The independent holder was the last one: teardown must now be reachable.
+        assert holder._conn is None
+
+    def test_raising_sweep_still_returns_the_borrowed_reference(self, tmp_path, monkeypatch):
+        db_path = _arm_sweep(monkeypatch, tmp_path)
+        holder = registry.acquire(db_path)
+
+        def _boom(self, **kwargs):
+            raise RuntimeError("archive failed")
+
+        monkeypatch.setattr(type(holder), "maybe_auto_archive", _boom, raising=False)
+        try:
+            before = _refcount_for(db_path)
+            sessions._maybe_auto_archive_for_profile(None)
+            assert _refcount_for(db_path) == before, "failed sweep leaked a registry reference"
+        finally:
+            assert registry.release(holder) is True
+        assert holder._conn is None
+
+    def test_repeated_eligible_sweeps_do_not_accumulate_references(self, tmp_path, monkeypatch):
+        db_path = _arm_sweep(monkeypatch, tmp_path)
+        holder = registry.acquire(db_path)
+        try:
+            before = _refcount_for(db_path)
+            for _ in range(3):
+                sessions._last_auto_archive_check.clear()
+                sessions._maybe_auto_archive_for_profile(None)
+            assert _refcount_for(db_path) == before
+        finally:
+            assert registry.release(holder) is True
+        assert holder._conn is None
+
+    def test_disabled_auto_archive_never_acquires(self, tmp_path, monkeypatch):
+        db_path = _arm_sweep(monkeypatch, tmp_path, enabled=False)
+        sessions._maybe_auto_archive_for_profile(None)
+        assert registry._generations.get(Path(db_path).resolve()) is None

tests/hermes_state/test_shared_session_db_registry.py

@@ -17,9 +17,6 @@ Covers the three ownership invariants the PR review demanded:
    every state.db in the process).
 """
 
-import contextlib
-import os
-import shutil
 import threading
 import time
 from pathlib import Path
@@ -36,27 +33,32 @@ def _clean_registry():
     registry._generations.clear()
     registry._retired.clear()
     registry._opening.clear()
+    registry._tearing_down.clear()
     yield
     registry.close_all()
     registry._generations.clear()
     registry._retired.clear()
     registry._opening.clear()
+    registry._tearing_down.clear()
 
 
-def _replace_file_preserving_schema(src: Path, dst: Path) -> None:
-    """Simulate snapshot-restore / recovery: new inode, same logical DB.
+def _replace_file_preserving_schema(monkeypatch, db_path: Path, old_db=None) -> None:
+    """Simulate a replacement identity without unlinking an open SQLite file.
 
-    Copies the live DB to a temp name, removes the original, and renames
-    the copy into place — the replacement has a different inode.
+    Windows correctly refuses to unlink a file with an open connection. The
+    registry contract is about observing a changed identity, so use its probe
+    seam here and, when requested, mark the old handle as replaced for the
+    SessionDB-level guard as well.
     """
-    tmp = dst.with_suffix(".replacement.tmp")
-    shutil.copy2(dst, tmp)
-    os.unlink(dst)
-    os.rename(tmp, dst)
+    original = registry._stat_db_file_identity(db_path)
+    replacement = (original[0] + 1, original[1]) if original is not None else (1, 1)
+    monkeypatch.setattr(registry, "_stat_db_file_identity", lambda _path: replacement)
+    if old_db is not None:
+        monkeypatch.setattr(old_db, "_db_file_was_replaced", lambda: True)
 
 
 class TestInodeReplacement:
-    def test_live_holders_keep_working_handle_across_replacement(self, tmp_path):
+    def test_live_holders_keep_working_handle_across_replacement(self, tmp_path, monkeypatch):
         """Two active refs → inode replacement → third caller gets NEW
         generation; the first two keep a working handle and their
         releases tear down only their own generation."""
@@ -66,7 +68,7 @@ class TestInodeReplacement:
         b = registry.acquire(db_path)
         assert a is b
 
-        _replace_file_preserving_schema(db_path, db_path)
+        _replace_file_preserving_schema(monkeypatch, db_path, old_db=a)
 
         c = registry.acquire(db_path)
         assert c is not a, "new caller must get the fresh generation"
@@ -114,14 +116,14 @@ class TestInodeReplacement:
         assert stats["live_generations"] == 0
         assert stats["retired_generations"] == 0
 
-    def test_retired_generation_never_relent_even_after_drain(self, tmp_path):
+    def test_retired_generation_never_relent_even_after_drain(self, tmp_path, monkeypatch):
         """After replacement, repeated acquires all return the NEW
         generation — the retired one is never lent again, even while it
         still has live holders."""
         db_path = tmp_path / "state.db"
         first = registry.acquire(db_path)
 
-        _replace_file_preserving_schema(db_path, db_path)
+        _replace_file_preserving_schema(monkeypatch, db_path)
 
         second = registry.acquire(db_path)
         third = registry.acquire(db_path)
@@ -137,7 +139,7 @@ class TestInodeReplacement:
         db_path = tmp_path / "state.db"
         old = registry.acquire(db_path)
 
-        _replace_file_preserving_schema(db_path, db_path)
+        _replace_file_preserving_schema(monkeypatch, db_path)
 
         calls = {"n": 0}
 
@@ -388,6 +390,131 @@ class TestTeardownOutsideLock:
         assert stats["retired_generations"] == 0
 
 
+class TestLifecycleBarrier:
+    def test_acquire_waits_for_final_teardown(self, tmp_path, monkeypatch):
+        """A replacement writer cannot open while the old generation is closing."""
+        db_path = tmp_path / "state.db"
+        db = registry.acquire(db_path)
+        teardown_started = threading.Event()
+        allow_teardown = threading.Event()
+        acquire_started = threading.Event()
+        acquire_finished = threading.Event()
+        release_finished = threading.Event()
+        acquired = []
+        errors = []
+        original_teardown = registry._teardown
+
+        def _blocked_teardown(target):
+            teardown_started.set()
+            if not allow_teardown.wait(5.0):
+                raise AssertionError("timed out waiting to release teardown")
+            original_teardown(target)
+
+        monkeypatch.setattr(registry, "_teardown", _blocked_teardown)
+
+        def _release():
+            try:
+                assert registry.release(db) is True
+            except BaseException as exc:  # pragma: no cover - failure path
+                errors.append(exc)
+            finally:
+                release_finished.set()
+
+        def _acquire():
+            try:
+                acquire_started.set()
+                acquired.append(registry.acquire(db_path))
+            except BaseException as exc:  # pragma: no cover - failure path
+                errors.append(exc)
+            finally:
+                acquire_finished.set()
+
+        release_thread = threading.Thread(target=_release)
+        release_thread.start()
+        assert teardown_started.wait(5.0)
+
+        acquire_thread = threading.Thread(target=_acquire)
+        acquire_thread.start()
+        assert acquire_started.wait(2.0)
+        # The teardown gate is still closed, so this cannot have returned
+        # unless acquire bypassed the path barrier.
+        assert not acquire_finished.is_set()
+
+        allow_teardown.set()
+        release_thread.join(10.0)
+        acquire_thread.join(10.0)
+        assert release_finished.is_set()
+        assert acquire_finished.is_set()
+        assert errors == []
+        assert len(acquired) == 1
+        assert acquired[0] is not db
+        assert registry.release(acquired[0]) is True
+
+    def test_final_release_waits_for_inflight_write(self, tmp_path, monkeypatch):
+        """Physical close waits for a write holding the SessionDB lock."""
+        db_path = tmp_path / "state.db"
+        db = registry.acquire(db_path)
+        write_started = threading.Event()
+        allow_write = threading.Event()
+        close_started = threading.Event()
+        release_finished = threading.Event()
+        errors = []
+        original_teardown = registry._teardown
+
+        def _observed_teardown(target):
+            close_started.set()
+            original_teardown(target)
+
+        monkeypatch.setattr(registry, "_teardown", _observed_teardown)
+
+        def _write(conn):
+            write_started.set()
+            if not allow_write.wait(5.0):
+                raise AssertionError("timed out waiting to finish write")
+            conn.execute("SELECT 1")
+
+        def _run_write():
+            try:
+                db._execute_write(_write)
+            except BaseException as exc:  # pragma: no cover - failure path
+                errors.append(exc)
+
+        def _release():
+            try:
+                assert registry.release(db) is True
+            except BaseException as exc:  # pragma: no cover - failure path
+                errors.append(exc)
+            finally:
+                release_finished.set()
+
+        writer = threading.Thread(target=_run_write)
+        writer.start()
+        assert write_started.wait(5.0)
+        releaser = threading.Thread(target=_release)
+        releaser.start()
+        assert close_started.wait(5.0)
+        assert not release_finished.is_set()
+
+        allow_write.set()
+        writer.join(10.0)
+        releaser.join(10.0)
+        assert errors == []
+        assert release_finished.is_set()
+        assert db._conn is None
+
+    def test_maintenance_borrow_pins_connection_until_scope_exits(self, tmp_path):
+        """Maintenance gets a temporary holder, not an unpinned snapshot."""
+        db_path = tmp_path / "state.db"
+        db = registry.acquire(db_path)
+
+        with registry.borrow_live_shared_session_dbs() as borrowed:
+            assert borrowed == [db]
+            assert registry.release(db) is True
+            assert db._conn is not None
+
+        assert db._conn is None
+
+
 class TestLegacyCloseSemantics:
     def test_close_on_shared_instance_releases_one_refcount(self, tmp_path):
         """Legacy ``db.close()`` call sites must not leak refcounts: close()
@@ -470,3 +597,226 @@ class TestAcquireSingleFlight:
         assert len(opened) >= 1
         registry.release(results[0])
         registry.release(results[1])
+
+
+class TestMultiGenerationTeardownBarrier:
+    """One path, several closes admitted at once (#103118 review).
+
+    The per-path mutex only serializes teardowns that already entered it. A
+    releasing thread can be descheduled after its generation left the registry
+    and its close was admitted, but before the mutex. If the path barrier is a
+    bare event, the NEXT teardown to settle lifts it for everybody: ``close_all``
+    reports a finished sweep and ``acquire`` publishes a replacement writer while
+    the first handle is still inside checkpoint/WAL-unlink — the overlap that
+    leaves zero-hole pages behind (#102827).
+    """
+
+    @staticmethod
+    def _pause_teardown_of(monkeypatch, target):
+        """Hold ``target``'s physical teardown *before* the lifecycle mutex.
+
+        Returns ``(entered, resume)``: ``entered`` fires once the paused
+        teardown is admitted-but-unfinished, ``resume`` lets it proceed.
+        """
+        entered = threading.Event()
+        resume = threading.Event()
+        original = registry._teardown_generation
+
+        def _paused(path, db, *, barrier=None):
+            if db is target:
+                entered.set()
+                if not resume.wait(10.0):  # pragma: no cover - failure path
+                    raise AssertionError("timed out holding a teardown")
+            original(path, db, barrier=barrier)
+
+        monkeypatch.setattr(registry, "_teardown_generation", _paused)
+        return entered, resume
+
+    @staticmethod
+    def _release_async(db, errors):
+        def _run():
+            try:
+                assert registry.release(db) is True
+            except BaseException as exc:  # pragma: no cover - failure path
+                errors.append(exc)
+
+        thread = threading.Thread(target=_run, daemon=True)
+        thread.start()
+        return thread
+
+    def _two_generations(self, tmp_path, monkeypatch):
+        """Acquire a generation, replace the file identity, acquire its successor."""
+        db_path = tmp_path / "state.db"
+        old = registry.acquire(db_path)
+        _replace_file_preserving_schema(monkeypatch, db_path, old)
+        current = registry.acquire(db_path)
+        assert current is not old
+        return db_path, old, current
+
+    def test_retired_drain_does_not_lift_a_pending_current_teardown(self, tmp_path, monkeypatch):
+        """close_all() must not report a finished sweep over a pending close."""
+        db_path, old, current = self._two_generations(tmp_path, monkeypatch)
+        resolved = Path(db_path).resolve()
+        entered, resume = self._pause_teardown_of(monkeypatch, current)
+        errors = []
+        releaser = self._release_async(current, errors)
+        try:
+            assert entered.wait(5.0)
+
+            # The retired generation's final release settles completely while the
+            # current generation's close is still admitted.
+            assert registry.release(old) is True
+            assert old._conn is None
+            barrier = registry._tearing_down.get(resolved)
+            assert barrier is not None, "retired drain lifted the shared path barrier"
+            assert barrier.pending == 1
+            assert not barrier.event.is_set()
+
+            swept = []
+            sweep_done = threading.Event()
+
+            def _close_all():
+                swept.append(registry.close_all())
+                sweep_done.set()
+
+            sweeper = threading.Thread(target=_close_all, daemon=True)
+            sweeper.start()
+            assert not sweep_done.wait(0.5), "close_all returned with a close still pending"
+            assert current._conn is not None
+
+            resume.set()
+            assert sweep_done.wait(10.0)
+            sweeper.join(10.0)
+        finally:
+            resume.set()
+            releaser.join(10.0)
+
+        assert errors == []
+        assert current._conn is None
+        assert registry._tearing_down.get(resolved) is None
+
+    def test_replacement_is_not_published_before_the_last_close_settles(self, tmp_path, monkeypatch):
+        """acquire() must not open a writer on top of an unfinished close."""
+        db_path, old, current = self._two_generations(tmp_path, monkeypatch)
+        entered, resume = self._pause_teardown_of(monkeypatch, current)
+        errors = []
+        releaser = self._release_async(current, errors)
+        acquired = []
+        acquire_done = threading.Event()
+        try:
+            assert entered.wait(5.0)
+            assert registry.release(old) is True
+
+            def _acquire():
+                try:
+                    fresh = registry.acquire(db_path)
+                    # Record the predecessor's state AT publication time.
+                    acquired.append((fresh, current._conn is None))
+                except BaseException as exc:  # pragma: no cover - failure path
+                    errors.append(exc)
+                finally:
+                    acquire_done.set()
+
+            opener = threading.Thread(target=_acquire, daemon=True)
+            opener.start()
+            assert not acquire_done.wait(0.5), "replacement published before the previous close"
+
+            resume.set()
+            assert acquire_done.wait(10.0)
+            opener.join(10.0)
+        finally:
+            resume.set()
+            releaser.join(10.0)
+
+        assert errors == []
+        assert len(acquired) == 1
+        fresh, predecessor_was_closed = acquired[0]
+        assert predecessor_was_closed, "a new writer was published over a live handle"
+        assert fresh is not current
+        assert registry.release(fresh) is True
+
+    def test_current_release_does_not_lift_a_pending_retired_drain(self, tmp_path, monkeypatch):
+        """Converse ordering: the retired drain is the one still running."""
+        db_path, old, current = self._two_generations(tmp_path, monkeypatch)
+        resolved = Path(db_path).resolve()
+        entered, resume = self._pause_teardown_of(monkeypatch, old)
+        errors = []
+        releaser = self._release_async(old, errors)
+        acquire_done = threading.Event()
+        acquired = []
+        try:
+            assert entered.wait(5.0)
+
+            # The CURRENT generation's final release settles first.
+            assert registry.release(current) is True
+            assert current._conn is None
+            barrier = registry._tearing_down.get(resolved)
+            assert barrier is not None, "current release lifted the shared path barrier"
+            assert barrier.pending == 1
+            assert not barrier.event.is_set()
+
+            def _acquire():
+                try:
+                    acquired.append(registry.acquire(db_path))
+                except BaseException as exc:  # pragma: no cover - failure path
+                    errors.append(exc)
+                finally:
+                    acquire_done.set()
+
+            opener = threading.Thread(target=_acquire, daemon=True)
+            opener.start()
+            assert not acquire_done.wait(0.5), "replacement published over a retired drain"
+            assert old._conn is not None
+
+            resume.set()
+            assert acquire_done.wait(10.0)
+            opener.join(10.0)
+        finally:
+            resume.set()
+            releaser.join(10.0)
+
+        assert errors == []
+        assert old._conn is None
+        assert len(acquired) == 1
+        assert registry.release(acquired[0]) is True
+
+    def test_failed_teardown_still_settles_the_barrier(self, tmp_path, monkeypatch):
+        """A raising close must not strand the path barrier forever."""
+        db_path = tmp_path / "state.db"
+        db = registry.acquire(db_path)
+        resolved = Path(db_path).resolve()
+        real_teardown = registry._teardown
+
+        def _raising_teardown(target):
+            real_teardown(target)
+            raise RuntimeError("checkpoint exploded")
+
+        monkeypatch.setattr(registry, "_teardown", _raising_teardown)
+        with pytest.raises(RuntimeError):
+            registry.release(db)
+
+        assert registry._tearing_down.get(resolved) is None
+        monkeypatch.setattr(registry, "_teardown", real_teardown)
+        fresh = registry.acquire(db_path)
+        assert fresh is not db
+        assert registry.release(fresh) is True
+
+    def test_pending_teardown_does_not_block_an_unrelated_path(self, tmp_path, monkeypatch):
+        """Barrier accounting stays per-path; other state.db files keep moving."""
+        blocked_path = tmp_path / "blocked" / "state.db"
+        blocked_path.parent.mkdir()
+        other_path = tmp_path / "other" / "state.db"
+        other_path.parent.mkdir()
+        blocked = registry.acquire(blocked_path)
+        entered, resume = self._pause_teardown_of(monkeypatch, blocked)
+        errors = []
+        releaser = self._release_async(blocked, errors)
+        try:
+            assert entered.wait(5.0)
+            other = registry.acquire(other_path)
+            assert other is not blocked
+            assert registry.release(other) is True
+        finally:
+            resume.set()
+            releaser.join(10.0)
+        assert errors == []

tests/hermes_state/test_state_db_file_identity.py

@@ -33,18 +33,16 @@ def _require_identity(db: SessionDB) -> None:
         pytest.skip("filesystem does not expose st_dev/st_ino for identity checks")
 
 
-def test_replace_with_new_inode_fails_loudly_without_fts_repair(tmp_path):
+def _mark_replaced_handle(monkeypatch, db: SessionDB) -> None:
+    """Exercise the replacement guard without unlinking an open DB on Windows."""
+    monkeypatch.setattr(db, "_db_file_was_replaced", lambda: True)
+
+
+def test_replace_with_new_inode_fails_loudly_without_fts_repair(tmp_path, monkeypatch):
     live = tmp_path / "state.db"
-    other = tmp_path / "other.db"
     db = _make_db(live, "live-sess", "original")
     _require_identity(db)
-    alt = _make_db(other, "other-sess", "replacement")
-    alt.close()
-
-    recorded = db._db_file_identity
-    assert recorded is not None
-    os.replace(other, live)
-    assert _stat_changed(live, recorded)
+    _mark_replaced_handle(monkeypatch, db)
 
     with pytest.raises(StateDbReplacedError, match="replaced underneath"):
         db.append_message("live-sess", role="user", content="after-replace")
@@ -56,14 +54,11 @@ def test_replace_with_new_inode_fails_loudly_without_fts_repair(tmp_path):
     db.close()
 
 
-def test_second_write_after_halt_does_not_attempt_repair(tmp_path):
+def test_second_write_after_halt_does_not_attempt_repair(tmp_path, monkeypatch):
     live = tmp_path / "state.db"
-    other = tmp_path / "other.db"
     db = _make_db(live, "s", "a")
     _require_identity(db)
-    alt = _make_db(other, "t", "b")
-    alt.close()
-    os.replace(other, live)
+    _mark_replaced_handle(monkeypatch, db)
     with pytest.raises(StateDbReplacedError):
         db.append_message("s", role="user", content="first")
     with pytest.raises(StateDbReplacedError):
@@ -125,17 +120,14 @@ def test_new_sessiondb_on_replaced_path_records_new_identity(tmp_path):
         reopened.close()
 
 
-def test_fts_scoped_error_on_replaced_file_skips_fts_fail_open(tmp_path):
+def test_fts_scoped_error_on_replaced_file_skips_fts_fail_open(tmp_path, monkeypatch):
     """Even FTS-provenance corruption must not authorize surgery on a
     replaced file. (A generic malformed error never reaches fail-open at
     all since the provenance classifier of #99652 rejects it earlier.)"""
     live = tmp_path / "state.db"
-    other = tmp_path / "other.db"
     db = _make_db(live, "s", "a")
     _require_identity(db)
-    alt = _make_db(other, "t", "b")
-    alt.close()
-    os.replace(other, live)
+    _mark_replaced_handle(monkeypatch, db)
 
     with pytest.raises(StateDbReplacedError):
         db._enter_fts_fail_open(
@@ -187,11 +179,6 @@ def test_divert_session_transcript_jsonl_appends(tmp_path, monkeypatch):
     assert divert_session_transcript_jsonl("sess-jsonl", []) is None
 
 
-def _stat_changed(path: Path, recorded) -> bool:
-    st = os.stat(path)
-    return (st.st_dev, st.st_ino) != recorded
-
-
 # ---------------------------------------------------------------------------
 # Lock safety of the identity probe itself (#100368 / howtocorrupt §2.2).
 #