fix(state): refcount guard locks shared by several handles in one process

Two SessionDB handles on one state.db in one process see the same
descriptors; the first release must not drop the range the second still
needs. Test covers the same-process sibling and the foreign-process close.

hermes_state_lockguard.py

@@ -22,6 +22,7 @@ import logging
 import os
 import struct
 import sys
+import threading
 from typing import Dict, Optional, Tuple
 
 logger = logging.getLogger("hermes_state")
@@ -51,6 +52,11 @@ _FLOCK_FORMAT = "@qqihh" if sys.platform == "darwin" or "bsd" in sys.platform el
 Identity = Tuple[int, int]
 Held = Dict[int, Identity]  # fd -> (st_dev, st_ino) it referenced when locked
 
+# Several handles in one process share the same inodes (and see each other's descriptors), so a
+# lock on a given (fd, inode) is reference-counted: only the last holder unlocks it.
+_LOCK = threading.Lock()
+_REFS: Dict[Tuple[int, Identity], int] = {}
+
 
 def supported() -> bool:
     return _F_OFD_SETLK is not None
@@ -117,11 +123,13 @@ def hold(db_path, held: Optional[Held] = None) -> Held:
         if ident is not None:
             wanted[ident] = rng
     try:
-        for fd, ident, (start, length) in _own_fds_for(wanted):
-            if held.get(fd) == ident:
-                continue
-            if _ofd_lock(fd, _F_RDLCK, start, length):
-                held[fd] = ident
+        with _LOCK:
+            for fd, ident, (start, length) in _own_fds_for(wanted):
+                if held.get(fd) == ident:
+                    continue
+                if _REFS.get((fd, ident)) or _ofd_lock(fd, _F_RDLCK, start, length):
+                    held[fd] = ident
+                    _REFS[(fd, ident)] = _REFS.get((fd, ident), 0) + 1
     except OSError:
         logger.debug("WAL lock guard unavailable for %s", base, exc_info=True)
     return held
@@ -135,12 +143,18 @@ def release(held: Held) -> None:
     replace never pairs with a stale WAL."""
     if not supported():
         return
-    for fd, ident in list(held.items()):
-        try:
-            st = os.fstat(fd)
-            if (st.st_dev, st.st_ino) == ident:
-                _ofd_lock(fd, _F_UNLCK, _SHARED_FIRST, _SHARED_SIZE)
-                _ofd_lock(fd, _F_UNLCK, _SHM_DMS_BYTE, 1)
-        except OSError:
-            pass
-    held.clear()
+    with _LOCK:
+        for fd, ident in list(held.items()):
+            remaining = _REFS.get((fd, ident), 1) - 1
+            if remaining > 0:
+                _REFS[(fd, ident)] = remaining
+                continue
+            _REFS.pop((fd, ident), None)
+            try:
+                st = os.fstat(fd)
+                if (st.st_dev, st.st_ino) == ident:
+                    _ofd_lock(fd, _F_UNLCK, _SHARED_FIRST, _SHARED_SIZE)
+                    _ofd_lock(fd, _F_UNLCK, _SHM_DMS_BYTE, 1)
+            except OSError:
+                pass
+        held.clear()

tests/hermes_state/test_wal_lock_guard.py

@@ -5,7 +5,7 @@ SQLite guards a WAL generation with per-PROCESS POSIX locks, so any raw ``open()
 last-connection close anywhere then checkpoints and unlinks ``-wal``/``-shm`` and the holder
 sticky-halts with ``DeletedWalGenerationError`` (#109727, #110042, #110276, Desktop "chat fails
 after update"). ``hermes_state_lockguard`` re-holds the same ranges as OFD locks that a stray
-close cannot cancel. Linux-only: the assertions read kernel truth from ``/proc``.
+close cannot cancel. Linux-only: the scenario needs ``/proc``-visible POSIX lock semantics.
 """
 
 import os
@@ -29,7 +29,7 @@ def _foreign_open_close(db_path: Path) -> None:
                     "c.execute('select count(*) from messages').fetchone(); c.close()"], check=True)
 
 
-def test_stray_in_process_close_does_not_let_a_sibling_unlink_the_wal(tmp_path, monkeypatch):
+def test_holder_survives_stray_close_then_sibling_closes(tmp_path, monkeypatch):
     pin_wal(monkeypatch)
     db = make_db(tmp_path / "state.db", "s", "seed")
     wal = require_wal(db)
@@ -37,9 +37,14 @@ def test_stray_in_process_close_does_not_let_a_sibling_unlink_the_wal(tmp_path,
     try:
         for name in ("state.db", "state.db-shm"):  # the §2.2 bug, e.g. a raw header probe
             os.close(os.open(tmp_path / name, os.O_RDONLY))
+        # Same-process second handle (Desktop backend / Herder tab shape) closing...
+        sibling = SessionDB(db_path=db.db_path)
+        sibling.append_message("s", role="user", content="tab")
+        sibling.close()
+        # ...and a foreign process (cron worker / one-shot CLI shape) closing.
         _foreign_open_close(db.db_path)
-        assert wal.exists() and wal.stat().st_ino == wal_inode, "sibling close unlinked the live WAL"
+        assert wal.exists() and wal.stat().st_ino == wal_inode, "a sibling close unlinked the live WAL"
         db.append_message("s", role="user", content="after")  # would raise DeletedWalGenerationError
     finally:
         db.close()
-    assert sqlite3.connect(db.db_path).execute("SELECT COUNT(*) FROM messages").fetchone()[0] == 2
+    assert sqlite3.connect(db.db_path).execute("SELECT COUNT(*) FROM messages").fetchone()[0] == 3