fix(compress): tell the user when /compress no-ops because another compression holds the lock

agent/conversation_compression.py

@@ -1282,6 +1282,12 @@ def _complete_compaction_lifecycle() -> None:
     _try_acquire_lock = None
     _lock_lookup_error: Optional[Exception] = None
     _legacy_session_db_without_lock_api = False
+    # Clear any stale lock-skip signal from a prior call so this call's
+    # outcome alone determines what callers see.  Without this an
+    # auto-compress lock-skip followed by a successful manual /compress
+    # would falsely report "Compression already in progress" and discard
+    # the compression results.
+    agent._compression_skipped_due_to_lock = None
     if _lock_db is not None:
         try:
             _legacy_session_db_without_lock_api = _lock_api_is_absent_on_session_db(
@@ -1369,6 +1375,11 @@ def _complete_compaction_lifecycle() -> None:
                 _lock_sid, existing,
             )
             _lock_holder = None  # don't release a lock we don't own
+            # Signal to callers that this no-op is due to a concurrent lock,
+            # not a genuine "nothing to compress" or aux-model failure.
+            # Manual /compress callers can surface a clear status message
+            # instead of the misleading "No changes from compression" text.
+            agent._compression_skipped_due_to_lock = existing or True
             # Surface to the user once — quiet for downstream auto-compress loops
             if getattr(agent, "_last_compression_lock_warning_sid", None) != _lock_sid:
                 agent._last_compression_lock_warning_sid = _lock_sid

agent/manual_compression_feedback.py

@@ -7,6 +7,36 @@
 from agent.redact import redact_sensitive_text
 
 
+def describe_compression_lock_skip(lock_signal: Any) -> str:
+    """User-facing text for a manual /compress skipped by the compression lock.
+
+    ``lock_signal`` is ``agent._compression_skipped_due_to_lock`` (or the
+    ``holder`` carried by the TUI's ``CompressionLockHeld``): a descriptive
+    holder string when another compressor CONFIRMED holds the lock, or
+    ``True``/``None`` when acquisition failed without a confirmed holder
+    (``hermes_state.try_acquire_compression_lock`` catches ``sqlite3.Error``
+    internally and returns ``False``, so a failed acquire is NOT proof that
+    another compression is running). The two cases must be worded
+    differently: claiming "already in progress" on an unconfirmed failure
+    misdirects the user when the real problem is a broken lock subsystem.
+    """
+    holder = (
+        lock_signal
+        if isinstance(lock_signal, str) and lock_signal.strip()
+        else None
+    )
+    if holder:
+        return (
+            f"⏳ Compression already in progress for this session "
+            f"(holder: {holder}). Please wait for it to finish."
+        )
+    return (
+        "⏳ Compression skipped: could not acquire this session's "
+        "compression lock. Another compression may still be running, or "
+        "the lock check failed — try again shortly."
+    )
+
+
 def summarize_manual_compression(
     before_messages: Sequence[dict[str, Any]],
     after_messages: Sequence[dict[str, Any]],

cli.py

@@ -10076,6 +10076,39 @@ def _manual_compress(self, cmd_original: str = ""):
                     force=True,
                     defer_context_engine_notification=True,
                 )
+
+                # If _compress_context returned unchanged because a
+                # concurrent compression lock is held, tell the user
+                # clearly instead of showing the misleading
+                # "No changes from compression" no-op text. The wording
+                # distinguishes a confirmed holder from an unconfirmed
+                # acquisition failure (describe_compression_lock_skip).
+                # Type-pinned check (is True / str): the flag's only real
+                # values are None/True/holder-string, and a bare getattr
+                # truthiness test is fooled by MagicMock auto-attributes on
+                # test-double agents (skill pitfall: MagicMock vs hasattr).
+                _lock_skip_signal = getattr(
+                    self.agent, "_compression_skipped_due_to_lock", None
+                )
+                if _lock_skip_signal is True or isinstance(_lock_skip_signal, str):
+                    from agent.manual_compression_feedback import (
+                        describe_compression_lock_skip,
+                    )
+                    print(
+                        "  "
+                        + describe_compression_lock_skip(
+                            self.agent._compression_skipped_due_to_lock
+                        )
+                    )
+                    self.agent._compression_skipped_due_to_lock = None
+                    # No boundary was committed on a lock-skip; discard the
+                    # deferred context-engine notification (exactly-once).
+                    finalize_context_engine_compression_notification(
+                        self.agent,
+                        committed=False,
+                    )
+                    return
+
                 if partial and tail:
                     compressed = rejoin_compressed_head_and_tail(compressed, tail)
                 self.conversation_history = compressed

gateway/slash_commands.py

@@ -3563,6 +3563,22 @@ async def _handle_compress_command(self, event: MessageEvent) -> str:
                         defer_context_engine_notification=True,
                     )
                 )
+
+                # If _compress_context returned unchanged because a
+                # concurrent compression lock is held, tell the user
+                # clearly instead of showing the misleading
+                # "No changes from compression" no-op text. The wording
+                # distinguishes a confirmed holder from an unconfirmed
+                # acquisition failure (describe_compression_lock_skip).
+                # The deferred context-engine notification is discarded by
+                # the finally block below (finalize committed=False).
+                _lock_skipped = getattr(tmp_agent, "_compression_skipped_due_to_lock", None)
+                if _lock_skipped is True or isinstance(_lock_skipped, str):
+                    from agent.manual_compression_feedback import (
+                        describe_compression_lock_skip,
+                    )
+                    return describe_compression_lock_skip(_lock_skipped)
+
                 if partial and tail:
                     compressed = rejoin_compressed_head_and_tail(compressed, tail)
 

tests/agent/test_compress_signal_leak.py

@@ -0,0 +1,60 @@
+"""Invariant: stale signal leak between consecutive compress_context calls."""
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+def test_signal_cleared_on_entry_between_calls(monkeypatch):
+    """Call 1: lock held → signal set. Call 2: lock available → signal must
+    be None after call 2 because the entry code cleared it. This prevents
+    a prior auto-compress lock-skip from causing a subsequent successful
+    manual /compress to falsely report 'Compression already in progress'."""
+    from agent.conversation_compression import compress_context
+
+    agent = MagicMock()
+    agent._cached_system_prompt = ""
+    agent.tools = None
+    agent._memory_manager = None
+    agent._build_system_prompt = MagicMock(return_value="sys prompt")
+    agent._emit_warning = MagicMock()
+
+    msgs = [
+        {"role": "user", "content": "a"}, {"role": "assistant", "content": "b"},
+        {"role": "user", "content": "c"}, {"role": "assistant", "content": "d"},
+    ]
+
+    monkeypatch.setattr(
+        "agent.conversation_compression._compression_lock_holder",
+        lambda a: "pid=test:holder",
+    )
+
+    # --- Call 1: lock held ---
+    db1 = MagicMock()
+    db1.try_acquire_compression_lock.return_value = False
+    db1.get_compression_lock_holder.return_value = "pid=holder1"
+    agent._session_db = db1
+
+    compress_context(agent, msgs, "", approx_tokens=100, force=True)
+
+    # Call 1: signal set (lock was held).
+    assert agent._compression_skipped_due_to_lock is not None
+
+    # --- Call 2: lock available, compressor succeeds ---
+    db2 = MagicMock()
+    db2.try_acquire_compression_lock.return_value = True
+    agent._session_db = db2
+    agent.context_compressor = MagicMock()
+    compressed = [
+        {"role": "user", "content": "[summary]"},
+        {"role": "assistant", "content": "ok"},
+    ]
+    agent.context_compressor.compress = MagicMock(return_value=compressed)
+    agent.context_compressor.compression_count = 0
+    agent.context_compressor.last_compression_rough_tokens = 0
+
+    compress_context(agent, msgs, "", approx_tokens=100, force=True)
+
+    # Call 2: signal must be None — entry code cleared stale Call 1 signal.
+    assert agent._compression_skipped_due_to_lock is None, (
+        "stale signal from lock-skip call 1 leaked into successful call 2"
+    )

tests/agent/test_manual_compression_feedback.py

@@ -2,7 +2,10 @@
 
 from types import SimpleNamespace
 
-from agent.manual_compression_feedback import summarize_manual_compression
+from agent.manual_compression_feedback import (
+    describe_compression_lock_skip,
+    summarize_manual_compression,
+)
 
 
 def _messages(count: int) -> list[dict[str, str]]:
@@ -82,3 +85,27 @@ def test_fallback_compression_reports_dropped_message_count():
     assert feedback["headline"] == "Compressed with fallback: 12 → 4 messages"
     assert "removed 8 message(s)" in feedback["note"]
     assert "invalid response" in feedback["note"]
+
+
+def test_lock_skip_with_confirmed_holder_names_it():
+    """A descriptive holder string means another compressor CONFIRMED holds
+    the lock — say so and name the holder."""
+    text = describe_compression_lock_skip("pid=12345:tid=7:agent=1:nonce=ab")
+
+    assert "Compression already in progress" in text
+    assert "pid=12345:tid=7:agent=1:nonce=ab" in text
+    assert "wait for it to finish" in text
+
+
+def test_lock_skip_without_confirmed_holder_does_not_claim_concurrency():
+    """signal=True / None / '' / whitespace: acquisition failed but the
+    holder is unconfirmed (hermes_state.try_acquire_compression_lock catches
+    sqlite3.Error internally and returns False). The message must not assert
+    another compression is definitely running."""
+    for signal in (True, None, "", "   "):
+        text = describe_compression_lock_skip(signal)
+
+        assert "already in progress" not in text, f"signal={signal!r}"
+        assert "Compression skipped" in text
+        assert "could not acquire" in text
+        assert "try again" in text

tests/cli/test_compress_here.py

@@ -26,6 +26,7 @@ def _wire_agent(shell, compressed_head):
     shell.agent.session_id = None
     shell.agent.tools = None
     shell.agent._compress_context.return_value = (compressed_head, "")
+    shell.agent._compression_skipped_due_to_lock = False
 
 
 def test_compress_here_compresses_head_only(capsys):

tests/cli/test_manual_compress.py

@@ -53,6 +53,9 @@ def test_manual_compress_reports_noop_without_success_banner(capsys):
     shell.agent.tools = None
     shell.agent.session_id = shell.session_id  # no-op compression: no split
     shell.agent._compress_context.return_value = (list(history), "")
+    # Explicitly signal this is NOT a lock-skip to avoid MagicMock
+    # getattr returning a truthy mock for unset attributes.
+    shell.agent._compression_skipped_due_to_lock = False
 
     def _estimate(messages, **_kwargs):
         assert messages == history
@@ -82,6 +85,8 @@ def test_manual_compress_reports_aborted_summary_without_success_banner(capsys):
         "Provider 'opencode-zen' is set in config.yaml but no API key was found."
     )
     shell.agent._compress_context.return_value = (list(history), "")
+    # Explicit non-lock-skip: MagicMock getattr would return a truthy mock.
+    shell.agent._compression_skipped_due_to_lock = False
 
     with patch("agent.model_metadata.estimate_request_tokens_rough", return_value=100):
         shell._manual_compress()
@@ -108,6 +113,7 @@ def test_manual_compress_explains_when_token_estimate_rises(capsys):
     shell.agent.tools = None
     shell.agent.session_id = shell.session_id  # no-op: no split
     shell.agent._compress_context.return_value = (compressed, "")
+    shell.agent._compression_skipped_due_to_lock = False
 
     def _estimate(messages, **_kwargs):
         if messages == history:
@@ -152,6 +158,7 @@ def _fake_compress(*args, **kwargs):
         shell.agent.session_id = new_child_id
         return (compressed, "")
     shell.agent._compress_context.side_effect = _fake_compress
+    shell.agent._compression_skipped_due_to_lock = False
     shell.agent.session_id = old_id  # starts in sync
     shell._pending_title = "stale title"
 
@@ -194,6 +201,7 @@ def _fake_compress(*args, **kwargs):
         return (compressed, "")
 
     shell.agent._compress_context.side_effect = _fake_compress
+    shell.agent._compression_skipped_due_to_lock = False
 
     with patch("agent.model_metadata.estimate_messages_tokens_rough", return_value=100):
         shell._manual_compress()
@@ -210,6 +218,7 @@ def test_manual_compress_does_not_flush_full_history_when_session_id_unchanged()
     shell.agent._cached_system_prompt = ""
     shell.agent.session_id = shell.session_id
     shell.agent._compress_context.return_value = (list(history), "")
+    shell.agent._compression_skipped_due_to_lock = False
 
     with patch("agent.model_metadata.estimate_messages_tokens_rough", return_value=100):
         shell._manual_compress()
@@ -239,6 +248,8 @@ def test_manual_compress_runs_when_auto_compaction_disabled(capsys):
     shell.agent.tools = None
     shell.agent.session_id = shell.session_id
     shell.agent._compress_context.return_value = (compressed, "")
+    # Explicit non-lock-skip: MagicMock getattr would return a truthy mock.
+    shell.agent._compression_skipped_due_to_lock = False
 
     with patch("agent.model_metadata.estimate_request_tokens_rough", return_value=100):
         shell._manual_compress()
@@ -264,10 +275,75 @@ def test_manual_compress_no_sync_when_session_id_unchanged():
     shell.agent.tools = None
     shell.agent.session_id = shell.session_id
     shell.agent._compress_context.return_value = (list(history), "")
+    shell.agent._compression_skipped_due_to_lock = False
     shell._pending_title = "keep me"
 
     with patch("agent.model_metadata.estimate_request_tokens_rough", return_value=100):
         shell._manual_compress()
 
     # No split → pending title untouched.
     assert shell._pending_title == "keep me"
+
+
+def test_manual_compress_shows_lock_skip_without_confirmed_holder(capsys):
+    """When _compress_context skips due to the compression lock WITHOUT a
+    confirmed holder (signal=True — acquisition failed but
+    get_compression_lock_holder returned nothing, e.g. a SQLite error made
+    try_acquire return False), _manual_compress must print the unconfirmed
+    lock-skip wording, NOT claim another compression is definitely running,
+    and NOT show the misleading "No changes from compression" no-op text."""
+    shell = _make_cli()
+    history = _make_history()
+    shell.conversation_history = history
+    shell.agent = MagicMock()
+    shell.agent.compression_enabled = True
+    shell.agent._cached_system_prompt = ""
+    shell.agent.tools = None
+    shell.agent.session_id = shell.session_id
+
+    # Simulate _compress_context setting the lock-skip signal and
+    # returning unchanged messages.
+    def _fake_compress(*args, **kwargs):
+        shell.agent._compression_skipped_due_to_lock = True
+        return (list(history), "")
+
+    shell.agent._compress_context.side_effect = _fake_compress
+
+    with patch("agent.model_metadata.estimate_request_tokens_rough", return_value=100):
+        shell._manual_compress()
+
+    output = capsys.readouterr().out
+    assert "Compression skipped" in output
+    assert "could not acquire" in output
+    # No confirmed holder → must not assert one is running.
+    assert "already in progress" not in output
+    assert "No changes from compression" not in output
+    # Signal should be cleared after use.
+    assert shell.agent._compression_skipped_due_to_lock is None
+
+
+def test_manual_compress_shows_lock_in_progress_with_holder(capsys):
+    """When the lock holder is a descriptive string, include it in the
+    status message so the user knows which process to investigate."""
+    shell = _make_cli()
+    history = _make_history()
+    shell.conversation_history = history
+    shell.agent = MagicMock()
+    shell.agent.compression_enabled = True
+    shell.agent._cached_system_prompt = ""
+    shell.agent.tools = None
+    shell.agent.session_id = shell.session_id
+
+    def _fake_compress(*args, **kwargs):
+        shell.agent._compression_skipped_due_to_lock = "pid=12345"
+        return (list(history), "")
+
+    shell.agent._compress_context.side_effect = _fake_compress
+
+    with patch("agent.model_metadata.estimate_request_tokens_rough", return_value=100):
+        shell._manual_compress()
+
+    output = capsys.readouterr().out
+    assert "Compression already in progress" in output
+    assert "pid=12345" in output
+    assert "No changes from compression" not in output