mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-31 19:16:29 +00:00
0f56e20d23
Root-cause fixes from the flake audit (session-DB mining + repo sweep): Event-based sync instead of sleep-sync: - title_generator: mock sets threading.Event, wait(10) replaces sleep(0.3) hoping the daemon thread got scheduled - docker zombie_reaping / profile_gateway: poll-for-state helpers replace fixed 1-3s sleeps (s6 transitions + SIGCHLD reaping are async) - process_registry tree test: select()-bounded readline replaces an unbounded blocking read (parent wedge now fails THIS test with a clear message instead of an opaque rc=124 file kill); SIGTERM grace 1s->2s (the 1s partition window mid-interpreter-startup is how a child PID escaped the live-system guard in CI) Timeout raises (loaded 8-way-sliced runners see ~5s scheduling floors; all of these complete in ms-to-1s when healthy so the raises cost nothing on green runs): - subprocess/thread waits <= 2s raised to 10-15s across mcp_tool, mcp_circuit_breaker, mcp_reconnect_retry_reset, mcp_parked_self_probe, mcp_cancelled_error_propagation, registry, clarify_gateway, interrupt, voice_cli_integration, docker_environment, session_store_lock_io, planned_stop_watcher, cli_interrupt_subagent, thread_scoped_output (joins now also assert not is_alive() so stragglers fail loudly) - wall-clock discrimination ceilings loosened where the guarded hang is 10x larger: local_background_child_hang 4s->10s, interrupt_cleanup setup 5s->20s + pgid-exit 30s->60s, mcp_stability grandchild spinup 5s->15s, protocol/gil-starvation fast-handler 0.5s->2s, iso_certify_seam 1.5s->5s, wait_for_mcp_discovery 0.1s->1s - narrow assertion windows widened: honcho first-turn wait 0.4..0.65 -> 0.25..2.0 (property is bounded-not-hung, not an exact wall-clock); compression fork-lock TTL 1s->3s (12 refresh chances per lease); compression-lock expiry margins symmetric (ttl 0.05->0.5, sleep 1.0) - telegram hung-DNS bound 1.0->1.4 (fake hang is 1.5s — must stay under)
149 lines
5.4 KiB
Python
149 lines
5.4 KiB
Python
"""Tests for ``SessionDB`` compression-lock primitives.
|
|
|
|
These cover the atomic per-session lock that prevents two compression
|
|
paths from racing on the same ``session_id`` and producing orphan child
|
|
sessions (Damien's "parent → two orphan children" repro shape, see
|
|
``tests/agent/test_compression_concurrent_fork.py`` for the
|
|
behavioural regression test).
|
|
|
|
Focus here: the lock primitives themselves (acquire, release, TTL,
|
|
diagnostic accessor) — not the wiring into compression.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from hermes_state import SessionDB
|
|
|
|
|
|
@pytest.fixture
|
|
def db(tmp_path: Path) -> SessionDB:
|
|
return SessionDB(tmp_path / "state.db")
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Single-holder semantics
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_acquire_succeeds_when_unlocked(db: SessionDB) -> None:
|
|
assert db.try_acquire_compression_lock("sess1", "holder1") is True
|
|
assert db.get_compression_lock_holder("sess1") == "holder1"
|
|
|
|
|
|
def test_acquire_blocks_second_holder(db: SessionDB) -> None:
|
|
assert db.try_acquire_compression_lock("sess1", "holder1") is True
|
|
assert db.try_acquire_compression_lock("sess1", "holder2") is False
|
|
# First holder still owns it
|
|
assert db.get_compression_lock_holder("sess1") == "holder1"
|
|
|
|
|
|
def test_release_allows_reacquire(db: SessionDB) -> None:
|
|
db.try_acquire_compression_lock("sess1", "holder1")
|
|
db.release_compression_lock("sess1", "holder1")
|
|
assert db.get_compression_lock_holder("sess1") is None
|
|
assert db.try_acquire_compression_lock("sess1", "holder2") is True
|
|
|
|
|
|
def test_release_with_wrong_holder_is_noop(db: SessionDB) -> None:
|
|
db.try_acquire_compression_lock("sess1", "holder1")
|
|
# Late-returning compressor must not release a lock it doesn't own
|
|
db.release_compression_lock("sess1", "holder_other")
|
|
assert db.get_compression_lock_holder("sess1") == "holder1"
|
|
|
|
|
|
def test_release_when_unlocked_is_noop(db: SessionDB) -> None:
|
|
# No exception, no state change
|
|
db.release_compression_lock("never_locked", "holder1")
|
|
assert db.get_compression_lock_holder("never_locked") is None
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Per-session isolation
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_locks_are_per_session(db: SessionDB) -> None:
|
|
assert db.try_acquire_compression_lock("sess1", "holder1") is True
|
|
# Different session: independent lock
|
|
assert db.try_acquire_compression_lock("sess2", "holder2") is True
|
|
assert db.get_compression_lock_holder("sess1") == "holder1"
|
|
assert db.get_compression_lock_holder("sess2") == "holder2"
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# TTL / expiry recovery
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_expired_lock_is_reclaimable(db: SessionDB) -> None:
|
|
"""A crashed compressor must not permanently block the session."""
|
|
# Acquire with a very short TTL
|
|
db.try_acquire_compression_lock("sess1", "crashed_holder", ttl_seconds=0.5)
|
|
time.sleep(1.0)
|
|
# Holder check honours expiry
|
|
assert db.get_compression_lock_holder("sess1") is None
|
|
# New holder can claim it
|
|
assert db.try_acquire_compression_lock("sess1", "fresh_holder") is True
|
|
assert db.get_compression_lock_holder("sess1") == "fresh_holder"
|
|
|
|
|
|
def test_non_expired_lock_is_held(db: SessionDB) -> None:
|
|
db.try_acquire_compression_lock("sess1", "holder1", ttl_seconds=60)
|
|
# Immediately after, still held
|
|
assert db.try_acquire_compression_lock("sess1", "holder2") is False
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Empty / invalid input
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_acquire_empty_session_id_returns_false(db: SessionDB) -> None:
|
|
assert db.try_acquire_compression_lock("", "holder1") is False
|
|
|
|
|
|
def test_release_empty_session_id_is_noop(db: SessionDB) -> None:
|
|
# No exception
|
|
db.release_compression_lock("", "holder1")
|
|
|
|
|
|
def test_holder_empty_session_id_returns_none(db: SessionDB) -> None:
|
|
assert db.get_compression_lock_holder("") is None
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Concurrency: real threads racing on the same session_id
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
def test_concurrent_acquire_only_one_winner(db: SessionDB) -> None:
|
|
"""Damien's race shape: N threads call acquire on the same session_id;
|
|
exactly one must win, the rest must be cleanly rejected."""
|
|
results: list[bool] = []
|
|
barrier = threading.Barrier(8)
|
|
lock = threading.Lock()
|
|
|
|
def try_acquire(idx: int) -> None:
|
|
holder = f"thread_{idx}"
|
|
barrier.wait() # synchronize start
|
|
got = db.try_acquire_compression_lock("contended_session", holder)
|
|
with lock:
|
|
results.append(got)
|
|
|
|
threads = [threading.Thread(target=try_acquire, args=(i,)) for i in range(8)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
# Exactly one thread acquired
|
|
assert sum(1 for r in results if r is True) == 1
|
|
assert sum(1 for r in results if r is False) == 7
|
|
# The single winner still owns it
|
|
assert db.get_compression_lock_holder("contended_session") is not None
|