hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

refactor(state): fold the speculative-open retry into acquire's loop; keep real inode-swap tests

Browse Source 
- acquire(): the discard-and-recurse path becomes one more iteration of the existing wait loop;
  the two inline 'with lifecycle_lock: _teardown(db)' copies reuse _teardown_generation, and the
  type-narrowing asserts go away with the recursion. release() reads generation.path directly.
- Restore the real os.replace inode swaps in test_state_db_file_identity.py and the registry tests:
  those files carry no windows_only marker so they never run on Windows, and the monkeypatched
  predicate stopped exercising the stat->identity->halt path anywhere.
- Drop the auto-archive change and its 4 tests: on main the sweep gets a bare SessionDB and
  db.close() already releases a registry-shared handle, so the described NameError leak only
  existed on this branch's earlier head. trace_upload: acquire(None) already defaults.
- Trim the barrier tests to the invariant pair (retired drain must not lift a pending current
  teardown; replacement not published before the last close settles) plus the raising-close
  settlement; comments say the WHY once.
This commit is contained in:
kshitijk4poor
2026-09-06 17:04:07 +00:00
committed by kshitij
parent 40488a4e54
commit b114641c88
6 changed files with 85 additions and 369 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
agent/trace_upload.py
View File

@@ -211,7 +211,7 @@ def load_session_messages(session_id: str, db_path=None) -> Tuple[List[Dict[str,
"""``(messages, meta)`` from SQLite; ``meta`` is ``{}`` when the session row is missing (a live, untitled
session may still have messages)."""
from hermes_state_registry import acquire, release_or_close
db = acquire(db_path) if db_path else acquire()
db = acquire(db_path or None)
try:
resolved = db.resolve_session_id(session_id) or session_id
meta = db.get_session(resolved) or {}

6
hermes_cli/web_server_sessions.py
View File

@@ -216,20 +216,16 @@ 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:
release_or_close(db)
db.close()
except Exception as exc:
_log.debug("opportunistic auto-archive skipped: %s", exc)

101
hermes_state_registry.py
View File

@@ -219,55 +219,47 @@ def acquire(db_path: Optional[Path] = None) -> "SessionDB":
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.
wait_for.wait()
else:
wait_for = opening
if wait_for is not None:
# Another caller is constructing or closing this path; wait without holding the
# global lock. A failed opener signals too, so a waiter can retry.
wait_for.wait()
continue
# Open OUTSIDE the registry lock; the per-path marker prevents redundant writers without
# serialising other files, the lifecycle mutex keeps the open off a same-path close.
try:
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
# Open OUTSIDE the lock; the per-path marker prevents redundant writers without
# serialising other files.
try:
# 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:
teardown = _tearing_down.get(path)
if teardown is None:
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)
raise
discard_barrier: Optional[_TeardownBarrier] = None
with _lock:
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:
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:
with lifecycle_lock:
_teardown(db)
return winner
# A shutdown or retired-generation final release was admitted while this opener was
# constructing: never publish into that window — discard the speculative handle and
# go round again once the barrier lifts.
_teardown_generation(path, db)
teardown.event.wait()
continue
if winner is not db:
_teardown_generation(path, db)
return winner
def release(db: "SessionDB") -> bool:
@@ -278,7 +270,6 @@ 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)
@@ -293,22 +284,17 @@ def release(db: "SessionDB") -> bool:
generation.refcount -= 1
needs_teardown = generation.refcount <= 0
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.
# 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:
assert teardown_path is not None
_teardown_generation(teardown_path, db, barrier=teardown_barrier)
_teardown_generation(generation.path, db, barrier=teardown_barrier)
return True
@@ -340,11 +326,8 @@ def close_all() -> int:
_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.
# A final release that removed its generation before this sweep took _lock still owns
# its physical close; wait for it rather than return over a running teardown.
for barrier in active_teardowns:
barrier.event.wait()
return len(generations)

100
tests/hermes_cli/test_web_server_auto_archive_registry.py
View File

@@ -1,100 +0,0 @@
"""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

210
tests/hermes_state/test_shared_session_db_registry.py
View File

@@ -17,6 +17,8 @@ Covers the three ownership invariants the PR review demanded:
every state.db in the process).
"""
import os
import shutil
import threading
import time
from pathlib import Path
@@ -34,6 +36,7 @@ def _clean_registry():
registry._retired.clear()
registry._opening.clear()
registry._tearing_down.clear()
registry._path_lifecycle_locks.clear()
yield
registry.close_all()
registry._generations.clear()
@@ -42,23 +45,20 @@ def _clean_registry():
registry._tearing_down.clear()
def _replace_file_preserving_schema(monkeypatch, db_path: Path, old_db=None) -> None:
"""Simulate a replacement identity without unlinking an open SQLite file.
def _replace_file_preserving_schema(src: Path, dst: Path) -> None:
"""Simulate snapshot-restore / recovery: new inode, same logical DB.
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.
Copies the live DB to a temp name, removes the original, and renames
the copy into place — the replacement has a different inode.
"""
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)
tmp = dst.with_suffix(".replacement.tmp")
shutil.copy2(dst, tmp)
os.unlink(dst)
os.rename(tmp, dst)
class TestInodeReplacement:
def test_live_holders_keep_working_handle_across_replacement(self, tmp_path, monkeypatch):
def test_live_holders_keep_working_handle_across_replacement(self, tmp_path):
"""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."""
@@ -68,7 +68,7 @@ class TestInodeReplacement:
b = registry.acquire(db_path)
assert a is b
_replace_file_preserving_schema(monkeypatch, db_path, old_db=a)
_replace_file_preserving_schema(db_path, db_path)
c = registry.acquire(db_path)
assert c is not a, "new caller must get the fresh generation"
@@ -116,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, monkeypatch):
def test_retired_generation_never_relent_even_after_drain(self, tmp_path):
"""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(monkeypatch, db_path)
_replace_file_preserving_schema(db_path, db_path)
second = registry.acquire(db_path)
third = registry.acquire(db_path)
@@ -139,7 +139,7 @@ class TestInodeReplacement:
db_path = tmp_path / "state.db"
old = registry.acquire(db_path)
_replace_file_preserving_schema(monkeypatch, db_path)
_replace_file_preserving_schema(db_path, db_path)
calls = {"n": 0}
@@ -391,117 +391,6 @@ class TestTeardownOutsideLock:
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"
@@ -648,7 +537,7 @@ class TestMultiGenerationTeardownBarrier:
"""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)
_replace_file_preserving_schema(db_path, db_path)
current = registry.acquire(db_path)
assert current is not old
return db_path, old, current
@@ -735,51 +624,6 @@ class TestMultiGenerationTeardownBarrier:
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"
@@ -800,23 +644,3 @@ class TestMultiGenerationTeardownBarrier:
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 == []

35
tests/hermes_state/test_state_db_file_identity.py
View File

@@ -33,16 +33,18 @@ def _require_identity(db: SessionDB) -> None:
pytest.skip("filesystem does not expose st_dev/st_ino for identity checks")
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):
def test_replace_with_new_inode_fails_loudly_without_fts_repair(tmp_path):
live = tmp_path / "state.db"
other = tmp_path / "other.db"
db = _make_db(live, "live-sess", "original")
_require_identity(db)
_mark_replaced_handle(monkeypatch, 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)
with pytest.raises(StateDbReplacedError, match="replaced underneath"):
db.append_message("live-sess", role="user", content="after-replace")
@@ -54,11 +56,14 @@ def test_replace_with_new_inode_fails_loudly_without_fts_repair(tmp_path, monkey
db.close()
def test_second_write_after_halt_does_not_attempt_repair(tmp_path, monkeypatch):
def test_second_write_after_halt_does_not_attempt_repair(tmp_path):
live = tmp_path / "state.db"
other = tmp_path / "other.db"
db = _make_db(live, "s", "a")
_require_identity(db)
_mark_replaced_handle(monkeypatch, db)
alt = _make_db(other, "t", "b")
alt.close()
os.replace(other, live)
with pytest.raises(StateDbReplacedError):
db.append_message("s", role="user", content="first")
with pytest.raises(StateDbReplacedError):
@@ -120,14 +125,17 @@ 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, monkeypatch):
def test_fts_scoped_error_on_replaced_file_skips_fts_fail_open(tmp_path):
"""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)
_mark_replaced_handle(monkeypatch, db)
alt = _make_db(other, "t", "b")
alt.close()
os.replace(other, live)
with pytest.raises(StateDbReplacedError):
db._enter_fts_fail_open(
@@ -179,6 +187,11 @@ 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).
#