From 8cd49c496fe7d6f6cfd82bfb623f84308d437015 Mon Sep 17 00:00:00 2001 From: 3ASiC Date: Thu, 16 Jul 2026 23:48:09 +0800 Subject: [PATCH] fix(compression): reclaim locks from dead processes --- hermes_state.py | 80 +++++++++++++++++--- tests/test_hermes_state_compression_locks.py | 49 ++++++++++++ 2 files changed, 117 insertions(+), 12 deletions(-) diff --git a/hermes_state.py b/hermes_state.py index a4f57d32e74d..a4d9b9429acc 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -33,6 +33,36 @@ from typing import Any, Callable, Dict, List, Optional, Tuple, TypeVar logger = logging.getLogger(__name__) +_COMPRESSION_LOCK_HOLDER_PID_RE = re.compile(r"(?:^|:)pid=(\d+)(?::|$)") + + +def _compression_lock_holder_process_is_dead(holder: str) -> bool: + """Return True only when a structured lock holder's local PID is gone. + + Compression locks are stored in a host-local SQLite database and holder + IDs created by ``conversation_compression`` start with ``pid=``. A + process killed during gateway shutdown cannot release its lease, so waiting + for the full TTL makes every new turn repeatedly attempt compaction. Reclaim + only when the kernel proves that PID no longer exists; legacy/unstructured + holders and permission errors remain protected until normal TTL expiry. + """ + match = _COMPRESSION_LOCK_HOLDER_PID_RE.search(holder or "") + if match is None: + return False + try: + pid = int(match.group(1)) + except (TypeError, ValueError): + return False + if pid <= 0: + return False + try: + os.kill(pid, 0) + except ProcessLookupError: + return True + except (PermissionError, OSError): + return False + return False + def _scrub_surrogates(value: Any) -> Any: """Replace lone surrogates when *value* is text; pass anything else through. @@ -4152,10 +4182,10 @@ class SessionDB: MUST NOT proceed with compression in that case (its rotation would race against the holder's, splitting the session lineage). - Expired locks (``expires_at < now``) are reclaimed transparently: - the stale row is deleted and the new holder acquires it. This - prevents a crashed compressor from permanently blocking the - session. + Expired locks (``expires_at < now``) are reclaimed transparently. + Structured holders whose local ``pid=`` no longer exists are reclaimed + immediately, so a gateway killed during compression does not stall the + replacement process for the full lease TTL. Implementation: single-transaction DELETE-expired + INSERT-or-IGNORE, followed by a SELECT to confirm we got the row. SQLite serialises @@ -4167,12 +4197,29 @@ class SessionDB: expires_at = now + ttl_seconds def _do(conn): - # First: reclaim any expired lock for this session_id. - conn.execute( - "DELETE FROM compression_locks " - "WHERE session_id = ? AND expires_at < ?", - (session_id, now), - ) + reclaimed_holder = None + row = conn.execute( + "SELECT holder, expires_at FROM compression_locks " + "WHERE session_id = ?", + (session_id,), + ).fetchone() + if row is not None: + current_holder = ( + row["holder"] if isinstance(row, sqlite3.Row) else row[0] + ) + current_expires_at = ( + row["expires_at"] if isinstance(row, sqlite3.Row) else row[1] + ) + if ( + current_expires_at < now + or _compression_lock_holder_process_is_dead(current_holder) + ): + conn.execute( + "DELETE FROM compression_locks " + "WHERE session_id = ? AND holder = ?", + (session_id, current_holder), + ) + reclaimed_holder = current_holder # Then: try to insert. INSERT OR IGNORE returns no rowcount # difference — verify ownership via SELECT. conn.execute( @@ -4185,12 +4232,21 @@ class SessionDB: "SELECT holder FROM compression_locks WHERE session_id = ?", (session_id,), ).fetchone() - return row is not None and ( + acquired = row is not None and ( row["holder"] if isinstance(row, sqlite3.Row) else row[0] ) == holder + return acquired, reclaimed_holder try: - return bool(self._execute_write(_do)) + acquired, reclaimed_holder = self._execute_write(_do) + if reclaimed_holder: + logger.warning( + "Reclaimed stale compression lock for session=%s " + "(holder=%s)", + session_id, + reclaimed_holder, + ) + return bool(acquired) except sqlite3.Error as exc: logger.warning( "try_acquire_compression_lock(%s) failed: %s", diff --git a/tests/test_hermes_state_compression_locks.py b/tests/test_hermes_state_compression_locks.py index 4e44d92a4457..2655a75c05d5 100644 --- a/tests/test_hermes_state_compression_locks.py +++ b/tests/test_hermes_state_compression_locks.py @@ -12,12 +12,14 @@ diagnostic accessor) — not the wiring into compression. from __future__ import annotations +import os import threading import time from pathlib import Path import pytest +import hermes_state from hermes_state import SessionDB @@ -99,6 +101,53 @@ def test_non_expired_lock_is_held(db: SessionDB) -> None: assert db.try_acquire_compression_lock("sess1", "holder2") is False +def test_non_expired_lock_from_dead_pid_is_reclaimed( + db: SessionDB, monkeypatch: pytest.MonkeyPatch +) -> None: + dead_holder = "pid=424242:tid=1:agent=abc:nonce=deadbeef" + assert db.try_acquire_compression_lock( + "sess1", dead_holder, ttl_seconds=300 + ) is True + + def process_is_gone(pid: int, signal: int) -> None: + assert pid == 424242 + assert signal == 0 + raise ProcessLookupError + + monkeypatch.setattr(hermes_state.os, "kill", process_is_gone) + + assert db.try_acquire_compression_lock( + "sess1", "pid=525252:tid=2:agent=def:nonce=fresh", ttl_seconds=300 + ) is True + + +def test_non_expired_lock_from_live_pid_is_not_reclaimed(db: SessionDB) -> None: + live_holder = f"pid={os.getpid()}:tid=1:agent=abc:nonce=live" + assert db.try_acquire_compression_lock( + "sess1", live_holder, ttl_seconds=300 + ) is True + assert db.try_acquire_compression_lock( + "sess1", "pid=525252:tid=2:agent=def:nonce=other", ttl_seconds=300 + ) is False + + +def test_unstructured_holder_waits_for_ttl( + db: SessionDB, monkeypatch: pytest.MonkeyPatch +) -> None: + assert db.try_acquire_compression_lock( + "sess1", "legacy_holder", ttl_seconds=300 + ) is True + kill = monkeypatch.setattr( + hermes_state.os, + "kill", + lambda *_args: pytest.fail("unstructured holder must not probe a PID"), + ) + assert kill is None + assert db.try_acquire_compression_lock( + "sess1", "pid=525252:tid=2:agent=def:nonce=other", ttl_seconds=300 + ) is False + + # ---------------------------------------------------------------------- # Empty / invalid input # ----------------------------------------------------------------------