From c21efeeb52f53dc40b18d14f7d21cd99d2e204a6 Mon Sep 17 00:00:00 2001 From: fangliquan Date: Wed, 12 Aug 2026 08:42:57 +0000 Subject: [PATCH] fix(sessions): keep turn lease across inherited-marker compressions 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. --- hermes_state.py | 121 +++++++--- run_agent.py | 42 +++- .../test_cross_process_turn_lease.py | 133 ++++++++++- tests/state/test_compression_lineage_guard.py | 22 ++ tests/state/test_session_turn_lease.py | 207 ++++++++++++++++++ 5 files changed, 489 insertions(+), 36 deletions(-) diff --git a/hermes_state.py b/hermes_state.py index d9f00d555d..73055194a1 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -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=``. 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") diff --git a/run_agent.py b/run_agent.py index 00b473d775..7d894c715a 100644 --- a/run_agent.py +++ b/run_agent.py @@ -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, diff --git a/tests/run_agent/test_cross_process_turn_lease.py b/tests/run_agent/test_cross_process_turn_lease.py index 88b9314761..7b60cefb80 100644 --- a/tests/run_agent/test_cross_process_turn_lease.py +++ b/tests/run_agent/test_cross_process_turn_lease.py @@ -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 diff --git a/tests/state/test_compression_lineage_guard.py b/tests/state/test_compression_lineage_guard.py index ef071c7974..ec8d2da182 100644 --- a/tests/state/test_compression_lineage_guard.py +++ b/tests/state/test_compression_lineage_guard.py @@ -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: diff --git a/tests/state/test_session_turn_lease.py b/tests/state/test_session_turn_lease.py index 0031342f42..9a49aff3bd 100644 --- a/tests/state/test_session_turn_lease.py +++ b/tests/state/test_session_turn_lease.py @@ -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: