From a48315e3223cbd6a7893391f59f5fbafc45f9c5d Mon Sep 17 00:00:00 2001 From: Jan-Stefan Janetzky Date: Tue, 14 Jul 2026 12:05:43 +0000 Subject: [PATCH] fix(compression): guard lock refresher startup --- agent/conversation_compression.py | 19 ++++++----- .../agent/test_compression_concurrent_fork.py | 34 +++++++++++++++++++ 2 files changed, 44 insertions(+), 9 deletions(-) diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index 8177f23dd120..2cc764740251 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -935,15 +935,6 @@ def compress_context( if not _existing_sp: _existing_sp = agent._build_system_prompt(system_message) return messages, _existing_sp - if _lock_holder is not None: - _lock_refresher = _CompressionLockLeaseRefresher( - _lock_db, - _lock_sid, - _lock_holder, - _lock_ttl, - _lock_refresh_interval, - ).start() - def _release_lock() -> None: """Release the lock keyed on the OLD session_id (before rotation).""" if _lock_refresher is not None: @@ -1007,6 +998,16 @@ def compress_context( return messages, existing_prompt try: + if _lock_holder is not None: + _lock_refresher = _CompressionLockLeaseRefresher( + _lock_db, + _lock_sid, + _lock_holder, + _lock_ttl, + _lock_refresh_interval, + ) + _lock_refresher.start() + # Notify external memory provider before compression discards context. # The provider's on_pre_compress() may return a string of insights it # wants surfaced inside the compression summary; capture and forward it diff --git a/tests/agent/test_compression_concurrent_fork.py b/tests/agent/test_compression_concurrent_fork.py index 14dcce6da5fc..2c66c210583d 100644 --- a/tests/agent/test_compression_concurrent_fork.py +++ b/tests/agent/test_compression_concurrent_fork.py @@ -523,6 +523,40 @@ def test_internal_typeerror_stops_lock_refresher_without_retry(tmp_path: Path, m assert db.try_acquire_compression_lock(parent_sid, "probe", ttl_seconds=1.0) is True +def test_lease_refresher_start_exception_releases_lock(tmp_path: Path, monkeypatch) -> None: + """A failed refresher start must not strand the lock until its TTL.""" + refreshers = [] + + class FailingLeaseRefresher: + def __init__(self, *_args, **_kwargs): + self.stopped = False + refreshers.append(self) + + def start(self): + raise RuntimeError("cannot start lock refresher") + + def stop(self): + self.stopped = True + + monkeypatch.setattr( + "agent.conversation_compression._CompressionLockLeaseRefresher", + FailingLeaseRefresher, + ) + + db = SessionDB(db_path=tmp_path / "state.db") + parent_sid = "REFRESHER_START_EXCEPTION_TEST" + db.create_session(parent_sid, source="discord") + agent = _build_agent_with_db(db, parent_sid) + messages = [{"role": "user", "content": f"m{i}"} for i in range(20)] + + with pytest.raises(RuntimeError, match="cannot start lock refresher"): + agent._compress_context(messages, "sys", approx_tokens=120_000) + + assert db.get_compression_lock_holder(parent_sid) is None + assert len(refreshers) == 1 + assert refreshers[0].stopped is True + + def test_signature_introspection_exception_releases_lock_and_refresher( tmp_path: Path, monkeypatch ) -> None: