fix(gateway): route hygiene-timeout warning via profile-aware adapter lookup + verify lock reacquire after fence cancel

- gateway/run.py: use _adapter_for_source(source) instead of the raw
  adapters.get(source.platform) map so the compression-timeout warning
  respects transport provenance, relay ingress, and multiplexed profiles
  (matches every other user-facing send in the hygiene block).
- tests: add a lock-release verification regression — a fence-cancelled
  hygiene compression must leave the per-session compression lock free so
  the next attempt (manual /compress retry) acquires it and commits
  normally.

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,

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