fix(tui): show lock-hold reason when /compress no-ops

tests/tui_gateway/test_compress_lock_skip.py

@@ -0,0 +1,84 @@
+"""Tests for TUI gateway /compress lock-hold signalling."""
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+def test_compress_session_history_raises_on_lock_skip():
+    """When _compression_skipped_due_to_lock is set on the agent,
+    _compress_session_history must raise CompressionLockHeld with
+    the holder string so callers can surface a clear message."""
+    from tui_gateway.server import _compress_session_history, CompressionLockHeld
+
+    history = [
+        {"role": "user", "content": "a"},
+        {"role": "assistant", "content": "b"},
+        {"role": "user", "content": "c"},
+        {"role": "assistant", "content": "d"},
+    ]
+    agent = MagicMock()
+    agent._cached_system_prompt = ""
+    agent.tools = None
+    agent._compression_skipped_due_to_lock = "pid=99999:tid=1:agent=1:nonce=abc"
+
+    def _fake_compress(msgs=None, *_args, **_kwargs):
+        return (msgs or history, "")
+
+    agent._compress_context.side_effect = _fake_compress
+
+    session = {
+        "agent": agent,
+        "history_lock": MagicMock(),
+        "history": history,
+        "history_version": 1,
+    }
+
+    with (
+        patch(
+            "agent.model_metadata.estimate_request_tokens_rough", return_value=100
+        ),
+        pytest.raises(CompressionLockHeld) as exc_info,
+    ):
+        _compress_session_history(session)
+
+    assert exc_info.value.holder == "pid=99999:tid=1:agent=1:nonce=abc"
+
+
+def test_compress_session_history_clears_signal_after_raise():
+    """The signal attribute must be cleared when the exception is raised
+    so stale signals don't leak into subsequent operations."""
+    from tui_gateway.server import _compress_session_history, CompressionLockHeld
+
+    history = [
+        {"role": "user", "content": "a"},
+        {"role": "assistant", "content": "b"},
+        {"role": "user", "content": "c"},
+        {"role": "assistant", "content": "d"},
+    ]
+    agent = MagicMock()
+    agent._cached_system_prompt = ""
+    agent.tools = None
+    agent._compression_skipped_due_to_lock = True
+
+    def _fake_compress(msgs=None, *_args, **_kwargs):
+        return (msgs or history, "")
+
+    agent._compress_context.side_effect = _fake_compress
+
+    session = {
+        "agent": agent,
+        "history_lock": MagicMock(),
+        "history": history,
+        "history_version": 1,
+    }
+
+    with (
+        patch(
+            "agent.model_metadata.estimate_request_tokens_rough", return_value=100
+        ),
+        pytest.raises(CompressionLockHeld),
+    ):
+        _compress_session_history(session)
+
+    # Signal must be cleared after the raise.
+    assert agent._compression_skipped_due_to_lock is None

tui_gateway/server.py

@@ -3642,6 +3642,14 @@ def _sync_agent_model_with_config(sid: str, session: dict) -> None:
         )
 
 
+class CompressionLockHeld(Exception):
+    """Raised by _compress_session_history when compression skipped due
+    to a concurrent lock on the session's compression_locks row."""
+    def __init__(self, holder: str | None = None):
+        self.holder = holder
+        super().__init__(f"Compression lock held: {holder or 'unknown'}")
+
+
 def _compress_session_history(
     session: dict,
     focus_topic: str | None = None,
@@ -3734,6 +3742,22 @@ def _compress_session_history(
             committed=False,
         )
         raise
+    # If _compress_context returned unchanged because a concurrent
+    # compression lock is held, raise so callers can surface a clear
+    # message instead of the misleading "No changes from compression" text.
+    _lock_skipped = getattr(agent, "_compression_skipped_due_to_lock", None)
+    if _lock_skipped:
+        agent._compression_skipped_due_to_lock = None
+        # No boundary was committed on a lock-skip; discard any pending
+        # deferred context-engine notification (exactly-once, no-op safe).
+        finalize_context_engine_compression_notification(
+            agent,
+            committed=False,
+        )
+        raise CompressionLockHeld(
+            _lock_skipped if isinstance(_lock_skipped, str) else None
+        )
+
     if partial and tail:
         compressed = rejoin_compressed_head_and_tail(compressed, tail)
     with session["history_lock"]:
@@ -9346,6 +9370,14 @@ def _(rid, params: dict) -> dict:
             # reverts to neutral whether compaction succeeded, was a
             # no-op, or raised.
             _status_update(sid, "ready")
+    except CompressionLockHeld as e:
+        _status_update(sid, "ready")
+        holder_msg = f" (holder: {e.holder})" if e.holder else ""
+        return _ok(rid, {
+            "compressed": False,
+            "lock_held": True,
+            "message": f"Compression already in progress for this session{holder_msg}. Please wait for it to finish."
+        })
     except Exception as e:
         finalize_context_engine_compression_notification(
             session["agent"],
@@ -15508,7 +15540,11 @@ def _mirror_slash_side_effects(sid: str, session: dict, command: str) -> str:
             # (the choke point shared by all three manual-compress routes)
             # parses the boundary-aware forms (here [N], up to here, --keep N)
             # and does the partial head/tail split there (#35533).
-            _compress_session_history(session, arg)
+            try:
+                _compress_session_history(session, arg)
+            except CompressionLockHeld as e:
+                holder_msg = f" (holder: {e.holder})" if e.holder else ""
+                return f"⏳ Compression already in progress for this session{holder_msg}. Please wait for it to finish."
             _sync_session_key_after_compress(sid, session)
 
             with session["history_lock"]: