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

fix(sessions): keep turn lease across inherited-marker compressions

Browse Source 
Presence-only _delegate_from/_branched_from checks stopped the lease walk on
continuations that copied a delegate's model_config, so the first refresh
after rotation missed the parent-key lease and hard-interrupted. A failed
get_session probe also skipped acquire entirely. Walk the lineage inside
the write transaction and treat a probe error as contended, not a fresh
session.
This commit is contained in:
fangliquan
2026-08-12 08:42:57 +00:00
committed by kshitij
parent 5e2be43fd4
commit c21efeeb52
5 changed files with 489 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_state.py
+90 -31
View File
@@ -5740,25 +5740,56 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
session_id, exc,
)
def _session_turn_lease_key(self, session_id: str) -> str:
"""Return the stable serialization key for every compression segment."""
def _session_turn_lease_key_on_conn(self, conn, session_id: str) -> str:
"""Walk compression parents on ``conn`` to the conversation lease key.
Must run on the same connection as the lease INSERT/UPDATE/DELETE.
A prior ``get_session`` failure must not compute a child id that the
later write then persists: refresh would walk to the parent and
fail-close. Markers bind to ``parent_session_id`` (same contract as
``_NON_CONTINUATION_CHILD_FILTER_SQL``). Lock errors propagate so
``_execute_write`` / ``acquire_session_turn_lease`` can retry.
"""
if not session_id:
return session_id
try:
current = self.get_session(session_id)
seen = {session_id}
while current and self._is_compression_child_row(current):
parent_id = current.get("parent_session_id")
if not parent_id or parent_id in seen:
break
parent = self.get_session(parent_id)
if not parent:
break
seen.add(parent_id)
current = parent
return str(current.get("id") or session_id) if current else session_id
except Exception:
def _row(sid: str):
row = conn.execute(
"SELECT id, parent_session_id, source, model_config, end_reason "
"FROM sessions WHERE id = ?",
(sid,),
).fetchone()
return dict(row) if row else None
current = _row(session_id)
seen = {session_id}
while current:
parent_id = current.get("parent_session_id")
if (
not parent_id
or parent_id in seen
or self._is_explicit_fork_child_row(current)
):
break
parent = _row(parent_id)
if not parent or parent.get("end_reason") != "compression":
break
seen.add(parent_id)
current = parent
return str(current.get("id") or session_id) if current else session_id
def _session_turn_lease_key(self, session_id: str) -> str:
"""Return the stable serialization key for every compression segment.
Acquire/refresh/release resolve this inside their write transaction.
This helper is for tests and diagnostics; it does not swallow lock
errors (a swallowed walk plus a later successful write was the
fail-open that replayed the post-rotation refresh miss).
"""
if not session_id:
return session_id
with self._read_ctx() as conn:
return self._session_turn_lease_key_on_conn(conn, session_id)
def try_acquire_session_turn_lease(
self,
@@ -5766,21 +5797,23 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
holder: str,
*,
ttl_seconds: float = 300.0,
patience_s: Optional[float] = None,
) -> bool:
"""Atomically acquire the cross-process turn lease for a conversation.
Compression rotates a session into child segments, so the durable key
is the lineage root rather than the current segment id. Expired leases
and leases whose structured local holder PID is known dead are reclaimed
in the same write transaction as acquisition.
is the lineage root rather than the current segment id. The walk and
INSERT share one write transaction. Expired leases and leases whose
structured local holder PID is known dead are reclaimed in that same
transaction.
"""
if not session_id or not holder:
return False
conversation_id = self._session_turn_lease_key(session_id)
now = time.time()
expires_at = now + max(0.1, float(ttl_seconds))
def _do(conn):
conversation_id = self._session_turn_lease_key_on_conn(conn, session_id)
row = conn.execute(
"SELECT holder, expires_at FROM session_turn_leases "
"WHERE conversation_id = ?",
@@ -5809,7 +5842,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
).fetchone()
return owner is not None and owner["holder"] == holder
return bool(self._execute_write(_do))
return bool(self._execute_write(_do, patience_s=patience_s))
def acquire_session_turn_lease(
self,
@@ -5822,6 +5855,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
on_wait=None,
wait_notice_interval_seconds: float = 15.0,
should_abort=None,
acquire_patience_s: float = 0.5,
) -> bool:
"""Wait for a cross-process turn lease without holding a SQLite lock.
@@ -5848,10 +5882,20 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
"session turn lease should_abort callback failed",
exc_info=True,
)
if self.try_acquire_session_turn_lease(
session_id, holder, ttl_seconds=ttl_seconds
):
return True
try:
if self.try_acquire_session_turn_lease(
session_id,
holder,
ttl_seconds=ttl_seconds,
patience_s=acquire_patience_s,
):
return True
except sqlite3.Error as exc:
# Long holder transactions (compression publish, large
# flushes) can exhaust a single write-patience budget.
# Keep polling until wait_seconds or should_abort.
if classify_persistence_error(exc) != "locked":
raise
now = time.monotonic()
remaining = deadline - now
if remaining <= 0:
@@ -5883,10 +5927,10 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
"""Extend a turn lease only while ``holder`` still owns it."""
if not session_id or not holder:
return False
conversation_id = self._session_turn_lease_key(session_id)
expires_at = time.time() + max(0.1, float(ttl_seconds))
def _do(conn):
conversation_id = self._session_turn_lease_key_on_conn(conn, session_id)
cursor = conn.execute(
"UPDATE session_turn_leases SET expires_at = ? "
"WHERE conversation_id = ? AND holder = ?",
@@ -5900,9 +5944,9 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
"""Release a turn lease iff ``holder`` still owns it; idempotent."""
if not session_id or not holder:
return
conversation_id = self._session_turn_lease_key(session_id)
def _do(conn):
conversation_id = self._session_turn_lease_key_on_conn(conn, session_id)
conn.execute(
"DELETE FROM session_turn_leases "
"WHERE conversation_id = ? AND holder = ?",
@@ -10041,6 +10085,17 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
# =========================================================================
def _is_explicit_fork_child_row(self, session: Dict[str, Any]) -> bool:
"""True when ``session`` is a branch, delegate, or tool child of its parent.
Markers only count as a fork when they point at ``parent_session_id``.
Compression copies ``model_config`` onto the continuation
(``publish_compression_child`` callers pass
``agent._session_init_model_config``), so a delegate's continuation
carries ``_delegate_from=<the delegate's own parent>``. Presence-only
matching would treat that real continuation as a fork — the same
misclassification ``_NON_CONTINUATION_CHILD_FILTER_SQL`` already
avoids by binding both markers to the queried parent.
"""
if session.get("source") == "tool":
return True
raw = session.get("model_config")
@@ -10050,10 +10105,14 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
cfg = json.loads(raw) if isinstance(raw, str) else raw
except (TypeError, json.JSONDecodeError):
return False
return isinstance(cfg, dict) and (
cfg.get("_branched_from") is not None
or cfg.get("_delegate_from") is not None
)
if not isinstance(cfg, dict):
return False
parent_id = session.get("parent_session_id")
branched = cfg.get("_branched_from")
delegated = cfg.get("_delegate_from")
if parent_id:
return branched == parent_id or delegated == parent_id
return branched is not None or delegated is not None
def _is_compression_child_row(self, child: Dict[str, Any]) -> bool:
parent_id = child.get("parent_session_id")
run_agent.py
+38 -4
View File
@@ -8087,10 +8087,19 @@ class AIAgent:
try:
_durable_session_exists = _turn_db.get_session(session_id) is not None
except Exception:
logger.debug(
"Could not check durable session before turn lease",
# A locked / non-WAL read is not proof the row is absent.
# Treating probe failure as "fresh session" skipped the
# lease this block exists to take and ran fail-open on
# the exact contention point (#84234). Acquire (or fail
# closed if acquire itself cannot) rather than start
# load/run/flush unsynchronized. get_session returns
# None — it does not raise — when the row is missing.
logger.warning(
"Could not check durable session before turn lease; "
"will acquire rather than run without serialization",
exc_info=True,
)
_durable_session_exists = True
if (
_turn_db is not None
and session_id
@@ -8146,13 +8155,33 @@ class AIAgent:
session_id,
)
relay_outcome = "cancelled"
return {
"final_response": "",
interrupt_msg = (
"Stopped waiting for another Hermes process on "
"this session. Your message was not processed."
)
interrupt_result = {
"final_response": interrupt_msg,
"messages": list(conversation_history or []),
"api_calls": 0,
"completed": False,
"interrupted": True,
}
interrupt_message = getattr(
self, "_interrupt_message", None
)
if interrupt_message:
interrupt_result["interrupt_message"] = (
interrupt_message
)
# Conversation-loop finalizer never runs on this
# early return. Clear so a cached agent cannot
# fail-close the next turn as interrupted.
try:
self.clear_interrupt()
except Exception:
self._interrupt_requested = False
self._interrupt_message = None
return interrupt_result
# Fail closed like gateway TurnLeaseTimeoutError: do not
# enter load/run/flush, and surface a resend notice instead
# of a bare TimeoutError that looks like a hang.
@@ -8218,6 +8247,11 @@ class AIAgent:
durable_turn_lease,
ttl_seconds=_lease_ttl,
):
# finally sets the stop event then releases.
# A late holder-fenced miss after that join
# timeout must not hard-interrupt the next turn.
if durable_turn_lease_stop.is_set():
return
logger.error(
"Lost session turn lease while turn is active: %s",
getattr(self, "session_id", None) or session_id,
tests/run_agent/test_cross_process_turn_lease.py
+132 -1
View File
@@ -2,6 +2,8 @@
from __future__ import annotations
import sqlite3
import threading
import time
from run_agent import AIAgent
@@ -53,6 +55,11 @@ def _agent_with_db(db, *, session_id="stale-parent", platform="desktop"):
agent.log_prefix = ""
agent._vprint = lambda *a, **k: None
agent.status_callback = None
agent._interrupt_requested = False
agent._interrupt_message = None
agent._pending_redirect = None
agent._execution_thread_id = None
agent._interrupt_thread_signal_pending = False
return agent
@@ -113,6 +120,43 @@ def test_run_conversation_acquires_then_reloads_latest_tip(monkeypatch):
)
def test_run_conversation_acquires_lease_when_session_probe_raises(monkeypatch):
"""A locked / non-WAL get_session must not skip the durable lease."""
db = _DB()
def locked_get_session(_session_id):
raise sqlite3.OperationalError("database is locked")
db.get_session = locked_get_session
agent = _agent_with_db(db)
observed = {}
def fake_run(_agent, _message, _system, history, *_args, **_kwargs):
observed["history"] = history
observed["session_id"] = _agent.session_id
return {"final_response": "ok", "messages": history, "failed": False}
monkeypatch.setattr("agent.conversation_loop.run_conversation", fake_run)
result = AIAgent.run_conversation(
agent,
"new message",
conversation_history=[{"role": "user", "content": "stale"}],
)
assert result["final_response"] == "ok"
assert observed == {
"history": [{"role": "user", "content": "durable latest"}],
"session_id": "compressed-tip",
}
assert [event[0] for event in db.events] == [
"acquire",
"resolve",
"reload",
"release",
]
def test_fresh_session_keeps_caller_seed_without_durable_lease(monkeypatch):
db = _DB(session_exists=False)
agent = _agent_with_db(db, session_id="fresh", platform="subagent")
@@ -173,13 +217,13 @@ def test_run_conversation_lease_timeout_returns_resend_notice(monkeypatch):
def test_run_conversation_lease_wait_honors_interrupt(monkeypatch):
db = _DB()
agent = _agent_with_db(db)
agent._interrupt_requested = False
def acquire_with_abort(session_id, holder, **kwargs):
db.events.append(("acquire", session_id, holder))
should_abort = kwargs.get("should_abort")
assert callable(should_abort)
agent._interrupt_requested = True
agent._interrupt_message = "follow-up while waiting"
assert should_abort()
return False
@@ -197,8 +241,51 @@ def test_run_conversation_lease_wait_honors_interrupt(monkeypatch):
assert result.get("interrupted") is True
assert result.get("failed") is not True
assert result.get("final_response")
assert "not processed" in result["final_response"]
assert result.get("interrupt_message") == "follow-up while waiting"
assert "session_turn_lease_timeout" not in str(result.get("error", ""))
assert [event[0] for event in db.events] == ["acquire"]
assert agent._interrupt_requested is False
assert agent._interrupt_message is None
def test_run_conversation_second_turn_after_lease_wait_abort(monkeypatch):
db = _DB()
agent = _agent_with_db(db)
turns = {"n": 0}
def acquire_then_succeed(session_id, holder, **kwargs):
db.events.append(("acquire", session_id, holder))
should_abort = kwargs.get("should_abort")
if turns["n"] == 0:
agent._interrupt_requested = True
agent._interrupt_message = "follow-up while waiting"
assert should_abort()
return False
assert not should_abort()
return True
db.acquire_session_turn_lease = acquire_then_succeed
def fake_run(_agent, _message, _system, history, *_args, **_kwargs):
return {"final_response": "ok", "messages": history, "failed": False}
monkeypatch.setattr("agent.conversation_loop.run_conversation", fake_run)
first = AIAgent.run_conversation(
agent,
"new message",
conversation_history=[{"role": "user", "content": "stale"}],
)
assert first.get("interrupted") is True
turns["n"] = 1
second = AIAgent.run_conversation(
agent,
"follow-up",
conversation_history=[{"role": "user", "content": "stale"}],
)
assert second["final_response"] == "ok"
assert agent._interrupt_requested is False
def test_run_conversation_interrupts_when_lease_refresh_lost(monkeypatch):
@@ -248,3 +335,47 @@ def test_run_conversation_interrupts_when_lease_refresh_lost(monkeypatch):
assert interrupt_calls
assert interrupt_calls[0][1] is True
assert "lease lost" in str(interrupt_calls[0][0]).lower()
def test_late_refresh_miss_after_release_does_not_interrupt(monkeypatch):
db = _DB()
agent = _agent_with_db(db)
agent._session_turn_lease_refresh_interval = 0.01
released = threading.Event()
interrupt_calls = []
def track_interrupt(message=None, hard_cancel=False):
interrupt_calls.append((message, hard_cancel))
agent._interrupt_requested = True
agent.interrupt = track_interrupt
def refresh_after_release(session_id, holder, **kwargs):
released.wait(timeout=2.0)
return False
db.refresh_session_turn_lease = refresh_after_release
orig_release = db.release_session_turn_lease
def release_and_signal(session_id, holder):
orig_release(session_id, holder)
released.set()
db.release_session_turn_lease = release_and_signal
def fake_run(_agent, _message, _system, history, *_args, **_kwargs):
time.sleep(0.03)
return {"final_response": "ok", "messages": history, "failed": False}
monkeypatch.setattr("agent.conversation_loop.run_conversation", fake_run)
result = AIAgent.run_conversation(
agent,
"new message",
conversation_history=[{"role": "user", "content": "seed"}],
)
time.sleep(0.05)
assert result["final_response"] == "ok"
assert interrupt_calls == []
assert agent._interrupt_requested is False
tests/state/test_compression_lineage_guard.py
+22
View File
@@ -136,6 +136,28 @@ def test_find_live_child_returns_continuation_with_foreign_markers(
assert child["id"] == "inherited-continuation"
def test_compression_lineage_includes_continuation_with_foreign_markers(
db: SessionDB,
) -> None:
"""Lineage walk uses the same parent-bound marker rule as orphan recovery."""
_compression_parent(db, "delegate-session-3")
db.create_session(
"inherited-tip",
source="subagent",
parent_session_id="delegate-session-3",
model_config={"_delegate_from": "some-original-parent"},
)
assert db.get_compression_lineage("inherited-tip") == [
"delegate-session-3",
"inherited-tip",
]
assert db.get_compression_lineage("delegate-session-3") == [
"delegate-session-3",
"inherited-tip",
]
def test_reopen_orphaned_compression_session_fails_closed_with_active_lease(
db: SessionDB,
) -> None:
tests/state/test_session_turn_lease.py
+207
View File
@@ -3,6 +3,7 @@
from __future__ import annotations
import os
import sqlite3
import threading
import time
from types import SimpleNamespace
@@ -91,6 +92,166 @@ def test_turn_lease_does_not_serialize_delegate_child_with_parent(tmp_path):
)
def test_turn_lease_walks_compression_child_that_inherited_fork_markers(tmp_path):
"""Inherited ``_delegate_from`` / ``_branched_from`` must not stop the walk.
``publish_compression_child`` copies ``model_config`` verbatim, so a
delegate or branch continuation carries a marker pointing at some other
session. Presence-only fork detection would key the child separately:
the holder still owns the parent-key lease, but the first refresh after
rotation looks up the child id and fail-closes with a hard interrupt.
"""
db = SessionDB(tmp_path / "state.db")
db.create_session("original-parent", source="test")
db.create_session(
"delegate",
source="delegate",
parent_session_id="original-parent",
model_config={"_delegate_from": "original-parent"},
)
db.end_session("delegate", "compression")
db.create_session(
"delegate-continuation",
source="delegate",
parent_session_id="delegate",
model_config={"_delegate_from": "original-parent"},
)
db.create_session(
"branch",
source="test",
parent_session_id="original-parent",
model_config={"_branched_from": "original-parent"},
)
db.end_session("branch", "compression")
db.create_session(
"branch-continuation",
source="test",
parent_session_id="branch",
model_config={"_branched_from": "original-parent"},
)
assert db._session_turn_lease_key("delegate-continuation") == "delegate"
assert db._session_turn_lease_key("branch-continuation") == "branch"
delegate_holder = f"pid={os.getpid()}:turn=delegate"
assert db.try_acquire_session_turn_lease(
"delegate", delegate_holder, ttl_seconds=5
)
assert not db.try_acquire_session_turn_lease(
"delegate-continuation",
f"pid={os.getpid()}:turn=delegate-child",
ttl_seconds=5,
)
assert db.refresh_session_turn_lease(
"delegate-continuation", delegate_holder, ttl_seconds=5
)
branch_holder = f"pid={os.getpid()}:turn=branch"
assert db.try_acquire_session_turn_lease(
"branch", branch_holder, ttl_seconds=5
)
assert not db.try_acquire_session_turn_lease(
"branch-continuation",
f"pid={os.getpid()}:turn=branch-child",
ttl_seconds=5,
)
assert db.refresh_session_turn_lease(
"branch-continuation", branch_holder, ttl_seconds=5
)
original_holder = f"pid={os.getpid()}:turn=original"
assert db.try_acquire_session_turn_lease(
"original-parent", original_holder, ttl_seconds=5
)
db.release_session_turn_lease("delegate-continuation", delegate_holder)
db.release_session_turn_lease("branch-continuation", branch_holder)
db.release_session_turn_lease("original-parent", original_holder)
def test_turn_lease_write_txn_does_not_trust_fail_open_key_helper(
tmp_path, monkeypatch: pytest.MonkeyPatch
):
"""Acquire/refresh/release walk inside the write txn.
The old helper swallowed get_session failures and returned the child id.
P2 then proceeded to acquire; the write succeeded under that child key
and the first working refresh walked to the parent and hard-interrupted.
Poisoning the outer helper must not change the conversation key.
"""
db = SessionDB(tmp_path / "state.db")
db.create_session(
"delegate",
source="delegate",
model_config={"_delegate_from": "original-parent"},
)
db.end_session("delegate", "compression")
db.create_session(
"delegate-continuation",
source="delegate",
parent_session_id="delegate",
model_config={"_delegate_from": "original-parent"},
)
monkeypatch.setattr(db, "_session_turn_lease_key", lambda sid: sid)
holder = f"pid={os.getpid()}:turn=delegate"
assert db.try_acquire_session_turn_lease(
"delegate", holder, ttl_seconds=5
)
assert not db.try_acquire_session_turn_lease(
"delegate-continuation",
f"pid={os.getpid()}:turn=child",
ttl_seconds=5,
)
assert db.refresh_session_turn_lease(
"delegate-continuation", holder, ttl_seconds=5
)
db.release_session_turn_lease("delegate-continuation", holder)
assert db.try_acquire_session_turn_lease(
"delegate", f"pid={os.getpid()}:turn=next", ttl_seconds=5
)
def test_turn_lease_retries_locked_in_txn_key_walk(
tmp_path, monkeypatch: pytest.MonkeyPatch
):
"""A locked lineage walk must retry, not INSERT under the child id."""
db = SessionDB(tmp_path / "state.db")
db.create_session(
"delegate",
source="delegate",
model_config={"_delegate_from": "original-parent"},
)
db.end_session("delegate", "compression")
db.create_session(
"delegate-continuation",
source="delegate",
parent_session_id="delegate",
model_config={"_delegate_from": "original-parent"},
)
attempts = {"n": 0}
original = db._session_turn_lease_key_on_conn
def flaky_walk(conn, session_id):
attempts["n"] += 1
if attempts["n"] == 1:
raise sqlite3.OperationalError("database is locked")
return original(conn, session_id)
monkeypatch.setattr(db, "_session_turn_lease_key_on_conn", flaky_walk)
holder = f"pid={os.getpid()}:turn=delegate"
assert db.try_acquire_session_turn_lease(
"delegate-continuation", holder, ttl_seconds=5
)
assert attempts["n"] >= 2
monkeypatch.setattr(db, "_session_turn_lease_key_on_conn", original)
assert not db.try_acquire_session_turn_lease(
"delegate", f"pid={os.getpid()}:turn=other", ttl_seconds=5
)
assert db.refresh_session_turn_lease("delegate", holder, ttl_seconds=5)
db.release_session_turn_lease("delegate-continuation", holder)
def test_turn_lease_refresh_and_release_are_owner_fenced(tmp_path):
db = SessionDB(tmp_path / "state.db")
db.create_session("shared", source="test")
@@ -203,6 +364,52 @@ def test_acquire_turn_lease_honors_should_abort(tmp_path):
first.release_session_turn_lease("shared", first_holder)
def test_acquire_turn_lease_retries_sqlite_lock(tmp_path, monkeypatch):
"""Write-lock exhaustion is contended, not a hard abort of the wait."""
db = SessionDB(tmp_path / "state.db")
db.create_session("shared", source="test")
holder = f"pid={os.getpid()}:turn=waiter"
attempts = {"n": 0}
original = db.try_acquire_session_turn_lease
def flaky_acquire(*args, **kwargs):
attempts["n"] += 1
if attempts["n"] == 1:
raise sqlite3.OperationalError(
"database is locked (another Hermes process held the "
"state.db write lock for over 20s)"
)
return original(*args, **kwargs)
monkeypatch.setattr(db, "try_acquire_session_turn_lease", flaky_acquire)
assert db.acquire_session_turn_lease(
"shared",
holder,
wait_seconds=2,
poll_interval_seconds=0.02,
acquire_patience_s=0.05,
)
assert attempts["n"] >= 2
db.release_session_turn_lease("shared", holder)
def test_acquire_turn_lease_reraises_non_lock_sqlite_error(tmp_path, monkeypatch):
db = SessionDB(tmp_path / "state.db")
db.create_session("shared", source="test")
def disk_full(*args, **kwargs):
raise sqlite3.OperationalError("database or disk is full")
monkeypatch.setattr(db, "try_acquire_session_turn_lease", disk_full)
with pytest.raises(sqlite3.OperationalError, match="disk is full"):
db.acquire_session_turn_lease(
"shared",
f"pid={os.getpid()}:turn=waiter",
wait_seconds=1,
poll_interval_seconds=0.02,
)
def test_non_expired_turn_lease_from_dead_pid_is_reclaimed(
tmp_path, monkeypatch: pytest.MonkeyPatch
) -> None: