diff --git a/gateway/run.py b/gateway/run.py index 8fdb0fd0796a..9afea42c3228 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -13053,7 +13053,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew "auxiliary.compression model configuration." ) try: - _adapter = self.adapters.get(source.platform) + _adapter = self._adapter_for_source(source) if _adapter and source.chat_id: await _adapter.send( source.chat_id, diff --git a/tests/agent/test_compression_concurrent_fork.py b/tests/agent/test_compression_concurrent_fork.py index 02068ff81d85..2e4624d1bdac 100644 --- a/tests/agent/test_compression_concurrent_fork.py +++ b/tests/agent/test_compression_concurrent_fork.py @@ -407,6 +407,77 @@ def test_cancelled_commit_fence_blocks_late_session_db_compaction( assert db.get_compression_lock_holder(session_id) is None +def test_fence_cancelled_compression_leaves_lock_reacquirable(tmp_path: Path) -> None: + """A fence-cancelled attempt must not poison the per-session lock. + + Lock-release verification for the hygiene-timeout path: after the gateway + times out and cancels a hygiene compression at the commit fence, the very + next attempt on the same session (e.g. the user running ``/compress``) + must acquire the compression lock and commit normally. A leaked lock here + would silently block every future compaction for the session until TTL + expiry. + """ + from agent.conversation_compression import CompressionCommitFence + + db = SessionDB(db_path=tmp_path / "state.db") + session_id = "HYGIENE_LOCK_REACQUIRE" + db.create_session(session_id, source="telegram") + + agent = _build_agent_with_db(db, session_id) + agent.compression_in_place = True + agent._cached_system_prompt = "sys" + summary_started = threading.Event() + release_summary = threading.Event() + + def _slow_summary(*_args, **_kwargs): + summary_started.set() + assert release_summary.wait(timeout=5) + return [ + {"role": "user", "content": "[CONTEXT COMPACTION] summary"}, + {"role": "user", "content": "tail"}, + ] + + agent.context_compressor.compress.side_effect = _slow_summary + messages = [{"role": "user", "content": f"m{i}"} for i in range(20)] + fence = CompressionCommitFence() + result = {} + + def _run_compression() -> None: + result["value"] = agent._compress_context( + messages, + "sys", + approx_tokens=120_000, + commit_fence=fence, + ) + + worker = threading.Thread(target=_run_compression, name="fenced-hygiene") + worker.start() + assert summary_started.wait(timeout=2) + assert fence.cancel_before_commit() is True + release_summary.set() + worker.join(timeout=5) + assert not worker.is_alive() + + # Cancelled attempt: no mutation, and — the invariant under test — the + # per-session compression lock is fully released. + assert result["value"][0] is messages + assert db.get_compression_lock_holder(session_id) is None + + # The NEXT attempt (no fence — a manual /compress retry) must be able to + # acquire the lock and commit an in-place compaction normally. + agent.context_compressor.compress.side_effect = lambda *_a, **_kw: [ + {"role": "user", "content": "[CONTEXT COMPACTION] retry summary"}, + {"role": "user", "content": "tail"}, + ] + retried, _sp = agent._compress_context(messages, "sys", approx_tokens=120_000) + + assert retried is not messages + assert len(retried) < len(messages) + assert agent.session_id == session_id # in-place: same session id + assert agent._last_compaction_in_place is True + assert db.get_compression_lock_holder(session_id) is None + + def test_commit_fence_waits_for_an_active_commit() -> None: """A timeout that loses the fence race cannot overlap the live turn.""" from agent.conversation_compression import CompressionCommitFence