From 23cd99026849f4a861d87a97fde52dcaacedcab2 Mon Sep 17 00:00:00 2001 From: 686f6c61 <6115107+686f6c61@users.noreply.github.com> Date: Tue, 29 Sep 2026 19:13:10 +0200 Subject: [PATCH] fix(agent): stop a locked lease refresh once its lifetime runs out A SQLite lock is still a missed tick, but only while the next attempt lands before the row expiry. Past that, acquisition can hand the session to someone else while this turn is still active. --- agent/turn_facade_lease.py | 30 +++++++++++++--- tests/agent/test_turn_facade_lease.py | 51 +++++++++++++++++++++++++++ 2 files changed, 76 insertions(+), 5 deletions(-) diff --git a/agent/turn_facade_lease.py b/agent/turn_facade_lease.py index 68cee326e4..96be1e6b4b 100644 --- a/agent/turn_facade_lease.py +++ b/agent/turn_facade_lease.py @@ -10,6 +10,7 @@ bodies run on per-handle workers), not per-turn threads. import logging import os import threading +import time from contextlib import nullcontext from dataclasses import dataclass from typing import Any, Dict, List, Optional @@ -39,6 +40,8 @@ class DurableTurnLease: self.db = db self.session_id = session_id # id at admission; release always targets this row self.holder = holder + # Wall clock, same as the row's expires_at. A failed refresh must not move it. + self._authority_deadline = time.time() + LEASE_TTL_SECONDS self.stop = threading.Event() self.refresh_interval = float(getattr(agent, "_session_turn_lease_refresh_interval", 60.0)) self._lock = threading.Lock() @@ -183,8 +186,14 @@ class DurableTurnLease: def refresh_tick(self): """One periodic renewal (every ``refresh_interval`` via the shared scheduler). A real - miss (rowcount 0) or a non-lock error interrupts the turn. A SQLite lock is only a - missed tick: the timer stays armed. Returning False stops the timer. + miss (rowcount 0) or a non-lock error interrupts the turn. A SQLite lock is a missed + tick only while the next attempt still lands before ``_authority_deadline``. Returning + False stops the timer. + + Acquisition reclaims a row once ``expires_at <= now``, even if the old process is + alive. Holder-qualified release does not block that path, so a run of lock errors + must not keep this turn active up to and past the deadline. A successful renewal + is the only thing that moves the deadline. The holder-qualified UPDATE fences a late refresher from a successor lease. The façade's finally sets ``stop`` before releasing, so a holder-fenced miss observed after stop is not @@ -195,6 +204,7 @@ class DurableTurnLease: if self.db.refresh_session_turn_lease( self._current_session_id(), self.holder, ttl_seconds=LEASE_TTL_SECONDS ): + self._authority_deadline = time.time() + LEASE_TTL_SECONDS return None if self.stop.is_set(): return False @@ -205,12 +215,22 @@ class DurableTurnLease: except Exception as exc: if self.stop.is_set(): return False - # A lock is contention, not a lost holder. The row is still ours - # (release is holder-qualified), so a convoy must not kill the turn. - # Keep the timer: the next tick retries, and one miss fits inside the TTL. from hermes_state_errors import is_sqlite_lock_error if is_sqlite_lock_error(exc): + now = time.time() + # The scheduler will not try again until one interval from now. If that + # attempt would meet or pass the row expiry, stop before a successor + # can claim it. + if now + self.refresh_interval >= self._authority_deadline: + logger.warning( + "Session turn lease refresh stayed locked through its lifetime: %s", + self._current_session_id(), + ) + self._interrupt_turn( + "Session turn lease could not be refreshed; stopping to protect the transcript." + ) + return False logger.warning( "Session turn lease refresh hit a SQLite lock; will retry: %s", self._current_session_id(), diff --git a/tests/agent/test_turn_facade_lease.py b/tests/agent/test_turn_facade_lease.py index 99f5664c9f..17997e1893 100644 --- a/tests/agent/test_turn_facade_lease.py +++ b/tests/agent/test_turn_facade_lease.py @@ -1,6 +1,7 @@ """Unit tests for agent.turn_facade_lease (admission + lease bracket).""" import sqlite3 import threading +import time from types import SimpleNamespace from agent.turn_facade_lease import ( @@ -145,6 +146,56 @@ def test_refresh_tick_sqlite_lock_keeps_the_turn(): assert lease.stop.is_set() is False +def test_repeated_lock_misses_do_not_outlive_the_lease(monkeypatch): + now = [1000.0] + monkeypatch.setattr(time, "time", lambda: now[0]) + db = _Db() + + def locked(session_id, holder, **kwargs): + raise sqlite3.OperationalError("database is locked") + + db.refresh_session_turn_lease = locked + lease, calls = _active_lease(db) + result = None + for elapsed in (60, 120, 180, 240, 300): + now[0] = 1000.0 + elapsed + result = lease.refresh_tick() + if result is False: + break + assert result is False + assert calls == [ + "Session turn lease could not be refreshed; stopping to protect the transcript." + ] + assert now[0] <= 1000.0 + LEASE_TTL_SECONDS + + +def test_one_lock_then_renewal_opens_a_fresh_window(monkeypatch): + now = [1000.0] + monkeypatch.setattr(time, "time", lambda: now[0]) + db = _Db() + locked = {"on": True} + + def refresh(session_id, holder, **kwargs): + if locked["on"]: + raise sqlite3.OperationalError("database is locked") + return True + + db.refresh_session_turn_lease = refresh + lease, calls = _active_lease(db) + now[0] = 1060.0 + assert lease.refresh_tick() is None and calls == [] + locked["on"] = False + now[0] = 1120.0 + assert lease.refresh_tick() is None and calls == [] + locked["on"] = True + # Original admission deadline was 1300. The renewal moved it to 1420. + now[0] = 1300.0 + assert lease.refresh_tick() is None and calls == [] + now[0] = 1360.0 + assert lease.refresh_tick() is False + assert calls + + def test_refresh_tick_real_loss_still_interrupts(): db = _Db() db.refresh_session_turn_lease = lambda session_id, holder, **kwargs: False