fix(compress): classify unconfirmed lock-acquire failures and cover all manual-compress surfaces

Follow-up to the salvaged #57634 commits:

- agent/manual_compression_feedback.py: new describe_compression_lock_skip()
  — single source of truth for lock-skip wording. A descriptive holder
  string means another compressor CONFIRMED holds the lock ('already in
  progress (holder: ...)'); True/None means acquisition failed without a
  confirmed holder (hermes_state.try_acquire_compression_lock catches
  sqlite3.Error internally and returns False), so the message says
  'could not acquire ... the lock check failed' instead of falsely
  claiming a concurrent compression is running.
- cli.py, gateway/slash_commands.py, tui_gateway/server.py (all three
  in-process consumers: session.compress RPC, command.dispatch compress
  branch, slash.exec mirror) now route through the shared helper.
- tui_gateway/server.py command.dispatch compress branch: catch
  CompressionLockHeld explicitly — it previously fell into the generic
  'compress failed' error handler.
- Deferred-notify contract (#69324): lock-skip discards the pending
  context-engine notification (committed=False) in _compress_session_history
  and the CLI path before returning.
- tests: lock-skip wording pins per surface, VISIBLE_COMPRESSION_MESSAGES
  noise-filter carve-outs for both wordings, MagicMock signal opt-outs for
  sibling tests added on main after the original PR.

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_manual_compress.py

@@ -85,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()
@@ -246,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()
@@ -281,10 +285,13 @@ def test_manual_compress_no_sync_when_session_id_unchanged():
     assert shell._pending_title == "keep me"
 
 
-def test_manual_compress_shows_lock_in_progress_when_skipped_due_to_lock(capsys):
-    """When _compress_context skips due to a concurrent compression lock,
-    _manual_compress must print a clear "already in progress" message,
-    NOT show the misleading "No changes from compression" no-op text."""
+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
@@ -306,7 +313,10 @@ def test_manual_compress_shows_lock_in_progress_when_skipped_due_to_lock(capsys)
         shell._manual_compress()
 
     output = capsys.readouterr().out
-    assert "Compression already in progress" in output
+    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

tests/gateway/test_compress_command.py

@@ -115,6 +115,8 @@ async def test_compress_command_works_when_auto_compaction_disabled():
     agent_instance.context_compressor.has_content_to_compress.return_value = True
     agent_instance.session_id = "sess-1"
     agent_instance._compress_context.return_value = (compressed, "")
+    # Explicit non-lock-skip: MagicMock getattr would return a truthy mock.
+    agent_instance._compression_skipped_due_to_lock = False
 
     def _estimate(messages, **_kwargs):
         return 100 if messages == history else 60

tests/gateway/test_telegram_noise_filter.py

@@ -82,6 +82,18 @@ VISIBLE_COMPRESSION_MESSAGES = [
         "⚠ Compression returned an empty transcript. No session split was "
         "performed; conversation continues unchanged."
     ),
+    # Manual /compress lock-skip feedback (issue #57631): both the
+    # confirmed-holder and unconfirmed-acquire wordings must reach the user.
+    (
+        "⏳ Compression already in progress for this session "
+        "(holder: pid=12345:tid=7:agent=1:nonce=ab). Please wait for it to "
+        "finish."
+    ),
+    (
+        "⏳ Compression skipped: could not acquire this session's "
+        "compression lock. Another compression may still be running, or "
+        "the lock check failed — try again shortly."
+    ),
 ]
 
 

tests/test_tui_gateway_server.py

@@ -5982,6 +5982,8 @@ def test_compress_session_history_passes_force():
     agent.context_compressor = None  # keep _get_usage on the simple path
     compressed = [{"role": "user", "content": "summary"}]
     agent._compress_context.return_value = (compressed, "")
+    # Explicit non-lock-skip: MagicMock getattr would return a truthy mock.
+    agent._compression_skipped_due_to_lock = False
     session = _session(
         agent=agent,
         history=[
@@ -6012,6 +6014,8 @@ def test_compress_session_history_works_when_auto_compaction_disabled():
     agent.context_compressor = None  # keep _get_usage on the simple path
     compressed = [{"role": "user", "content": "summary"}]
     agent._compress_context.return_value = (compressed, "")
+    # Explicit non-lock-skip: MagicMock getattr would return a truthy mock.
+    agent._compression_skipped_due_to_lock = False
     session = _session(
         agent=agent,
         history=[

tests/tui_gateway/test_compress_lock_skip.py

@@ -1,37 +1,64 @@
-"""Tests for TUI gateway /compress lock-hold signalling."""
+"""Tests for TUI gateway /compress lock-hold signalling.
+
+Covers the shared _compress_session_history choke point plus ALL THREE
+in-process manual-compress consumers (session.compress RPC, the
+command.dispatch compress branch, and the slash.exec mirror via
+_mirror_slash_side_effects) — each must surface the lock-skip reason
+instead of the misleading "No changes from compression" no-op text or a
+generic "compress failed" error.
+"""
+import threading
 from unittest.mock import MagicMock, patch
 
 import pytest
 
 
+def _make_history():
+    return [
+        {"role": "user", "content": "a"},
+        {"role": "assistant", "content": "b"},
+        {"role": "user", "content": "c"},
+        {"role": "assistant", "content": "d"},
+    ]
+
+
+def _make_lock_skip_agent(signal):
+    """Agent double whose _compress_context no-ops and sets the lock signal."""
+    agent = MagicMock()
+    agent._cached_system_prompt = ""
+    agent.tools = None
+    agent.session_id = "sess-lock"
+    # Deferred-notify contract: no pending notification exists.
+    agent._pending_context_engine_compression_notification = None
+
+    def _fake_compress(msgs=None, *_args, **_kwargs):
+        agent._compression_skipped_due_to_lock = signal
+        return (list(msgs) if msgs is not None else _make_history(), "")
+
+    agent._compress_context.side_effect = _fake_compress
+    return agent
+
+
+def _make_session(agent, history):
+    return {
+        "agent": agent,
+        "history_lock": threading.Lock(),
+        "history": history,
+        "history_version": 1,
+        "running": False,
+        "session_key": "sess-lock",
+    }
+
+
 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,
-    }
+    history = _make_history()
+    agent = _make_lock_skip_agent("pid=99999:tid=1:agent=1:nonce=abc")
+    session = _make_session(agent, history)
 
     with (
         patch(
@@ -42,6 +69,9 @@ def test_compress_session_history_raises_on_lock_skip():
         _compress_session_history(session)
 
     assert exc_info.value.holder == "pid=99999:tid=1:agent=1:nonce=abc"
+    # The history must be untouched by the lock-skip.
+    assert session["history"] == history
+    assert session["history_version"] == 1
 
 
 def test_compress_session_history_clears_signal_after_raise():
@@ -49,28 +79,9 @@ def test_compress_session_history_clears_signal_after_raise():
     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,
-    }
+    history = _make_history()
+    agent = _make_lock_skip_agent(True)
+    session = _make_session(agent, history)
 
     with (
         patch(
@@ -82,3 +93,147 @@ def test_compress_session_history_clears_signal_after_raise():
 
     # Signal must be cleared after the raise.
     assert agent._compression_skipped_due_to_lock is None
+
+
+def test_compress_session_history_unconfirmed_signal_yields_none_holder():
+    """signal=True (acquisition failed, holder unconfirmed — e.g. SQLite
+    error made try_acquire return False) must map to holder=None, NOT a
+    fabricated holder, so downstream wording doesn't claim a concurrent
+    compression is definitely running."""
+    from tui_gateway.server import _compress_session_history, CompressionLockHeld
+
+    agent = _make_lock_skip_agent(True)
+    session = _make_session(agent, _make_history())
+
+    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 is None
+
+
+# ── Consumer 1: session.compress RPC ───────────────────────────────────
+
+
+def test_session_compress_rpc_returns_lock_held_payload():
+    from tui_gateway import server
+
+    agent = _make_lock_skip_agent("pid=4242")
+    session = _make_session(agent, _make_history())
+    sid = "sid-lock-rpc"
+    server._sessions[sid] = session
+    try:
+        with (
+            patch.object(server, "_sess", return_value=(session, None)),
+            patch.object(server, "_session_uses_compute_host", return_value=False),
+            patch.object(server, "_status_update"),
+            patch(
+                "agent.model_metadata.estimate_request_tokens_rough",
+                return_value=100,
+            ),
+        ):
+            r = server._methods["session.compress"]("r1", {"session_id": sid})
+    finally:
+        server._sessions.pop(sid, None)
+
+    result = r["result"]
+    assert result["lock_held"] is True
+    assert result["compressed"] is False
+    assert "Compression already in progress" in result["message"]
+    assert "pid=4242" in result["message"]
+    assert "No changes from compression" not in result["message"]
+
+
+# ── Consumer 2: command.dispatch compress branch ───────────────────────
+
+
+def test_command_dispatch_compress_reports_lock_skip_as_output_not_error():
+    """CompressionLockHeld must NOT fall through to the generic
+    'compress failed' error handler — it's a clean no-op with feedback."""
+    from tui_gateway import server
+
+    agent = _make_lock_skip_agent("pid=5150")
+    session = _make_session(agent, _make_history())
+    sid = "sid-lock-dispatch"
+    server._sessions[sid] = session
+    try:
+        with (
+            patch.object(server, "_session_uses_compute_host", return_value=False),
+            patch(
+                "agent.model_metadata.estimate_request_tokens_rough",
+                return_value=100,
+            ),
+        ):
+            r = server._methods["command.dispatch"](
+                "r2", {"name": "compress", "arg": "", "session_id": sid}
+            )
+    finally:
+        server._sessions.pop(sid, None)
+
+    assert "error" not in r, f"lock-skip surfaced as error: {r.get('error')}"
+    output = r["result"]["output"]
+    assert r["result"]["type"] == "exec"
+    assert "Compression already in progress" in output
+    assert "pid=5150" in output
+    assert "compress failed" not in output
+    assert "No changes from compression" not in output
+
+
+# ── Consumer 3: slash.exec mirror (_mirror_slash_side_effects) ─────────
+
+
+def test_mirror_slash_side_effects_reports_lock_skip():
+    from tui_gateway import server
+
+    agent = _make_lock_skip_agent("pid=6161")
+    session = _make_session(agent, _make_history())
+    sid = "sid-lock-mirror"
+    server._sessions[sid] = session
+    try:
+        with (
+            patch.object(server, "_sync_session_key_after_compress"),
+            patch.object(server, "_emit"),
+            patch(
+                "agent.model_metadata.estimate_request_tokens_rough",
+                return_value=100,
+            ),
+        ):
+            output = server._mirror_slash_side_effects(sid, session, "/compress")
+    finally:
+        server._sessions.pop(sid, None)
+
+    assert "Compression already in progress" in output
+    assert "pid=6161" in output
+    assert "No changes from compression" not in output
+    assert "live session sync failed" not in output
+
+
+def test_mirror_slash_side_effects_unconfirmed_lock_skip_wording():
+    """signal=True (no confirmed holder) must use the 'could not acquire'
+    wording rather than claiming another compression is running."""
+    from tui_gateway import server
+
+    agent = _make_lock_skip_agent(True)
+    session = _make_session(agent, _make_history())
+    sid = "sid-lock-mirror-unconfirmed"
+    server._sessions[sid] = session
+    try:
+        with (
+            patch.object(server, "_sync_session_key_after_compress"),
+            patch.object(server, "_emit"),
+            patch(
+                "agent.model_metadata.estimate_request_tokens_rough",
+                return_value=100,
+            ),
+        ):
+            output = server._mirror_slash_side_effects(sid, session, "/compress")
+    finally:
+        server._sessions.pop(sid, None)
+
+    assert "Compression skipped" in output
+    assert "could not acquire" in output
+    assert "already in progress" not in output