diff --git a/agent/manual_compression_feedback.py b/agent/manual_compression_feedback.py index 91ada6ec4909..b2e12d683423 100644 --- a/agent/manual_compression_feedback.py +++ b/agent/manual_compression_feedback.py @@ -7,6 +7,36 @@ from typing import Any, Sequence 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]], diff --git a/cli.py b/cli.py index ab2aac80f90c..f89adec37803 100644 --- a/cli.py +++ b/cli.py @@ -10180,17 +10180,26 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): # 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. + # "No changes from compression" no-op text. The wording + # distinguishes a confirmed holder from an unconfirmed + # acquisition failure (describe_compression_lock_skip). if getattr(self.agent, "_compression_skipped_due_to_lock", None): - holder = self.agent._compression_skipped_due_to_lock - if isinstance(holder, str): - print(f" ⏳ Compression already in progress for this " - f"session (holder: {holder}). " - f"Please wait for it to finish.") - else: - print(f" ⏳ Compression already in progress for this " - f"session. Please wait for it to finish.") + 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: diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py index 0817b7fafdf3..d10baa56f8d7 100644 --- a/gateway/slash_commands.py +++ b/gateway/slash_commands.py @@ -3567,15 +3567,17 @@ class GatewaySlashCommandsMixin: # 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. + # "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: - holder = _lock_skipped if isinstance(_lock_skipped, str) else None - holder_clause = f" (holder: {holder})" if holder else "" - return ( - f"⏳ Compression already in progress for this session" - f"{holder_clause}. Please wait for it to finish." + 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) diff --git a/tests/agent/test_manual_compression_feedback.py b/tests/agent/test_manual_compression_feedback.py index b8a80da7c1a8..9944589868d6 100644 --- a/tests/agent/test_manual_compression_feedback.py +++ b/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 diff --git a/tests/cli/test_manual_compress.py b/tests/cli/test_manual_compress.py index 3279c78b5c6e..969ec963fc3e 100644 --- a/tests/cli/test_manual_compress.py +++ b/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 diff --git a/tests/gateway/test_compress_command.py b/tests/gateway/test_compress_command.py index a85589617220..c734a02e6b4d 100644 --- a/tests/gateway/test_compress_command.py +++ b/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 diff --git a/tests/gateway/test_telegram_noise_filter.py b/tests/gateway/test_telegram_noise_filter.py index d1b99c6a3efb..dbbfc543a249 100644 --- a/tests/gateway/test_telegram_noise_filter.py +++ b/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." + ), ] diff --git a/tests/test_tui_gateway_server.py b/tests/test_tui_gateway_server.py index 540ea37ed221..c1c794473c1f 100644 --- a/tests/test_tui_gateway_server.py +++ b/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=[ diff --git a/tests/tui_gateway/test_compress_lock_skip.py b/tests/tui_gateway/test_compress_lock_skip.py index 42687771726c..c915dc964099 100644 --- a/tests/tui_gateway/test_compress_lock_skip.py +++ b/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 diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 3ffeffc8bbd3..cc0d8a1e5dd0 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -9372,11 +9372,13 @@ def _(rid, params: dict) -> dict: _status_update(sid, "ready") except CompressionLockHeld as e: _status_update(sid, "ready") - holder_msg = f" (holder: {e.holder})" if e.holder else "" + from agent.manual_compression_feedback import ( + describe_compression_lock_skip, + ) 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." + "message": describe_compression_lock_skip(e.holder), }) except Exception as e: finalize_context_engine_compression_notification( @@ -14482,6 +14484,19 @@ def _(rid, params: dict) -> dict: ), }, ) + except CompressionLockHeld as e: + # Lock-skip is a clean no-op, not a failure: report it as + # normal command output (matching the slash-mirror and + # session.compress RPC), never as a "compress failed" error. + # _compress_session_history already discarded the deferred + # context-engine notification before raising. + from agent.manual_compression_feedback import ( + describe_compression_lock_skip, + ) + return _ok( + rid, + {"type": "exec", "output": describe_compression_lock_skip(e.holder)}, + ) except Exception as exc: finalize_context_engine_compression_notification( session["agent"], @@ -15543,8 +15558,10 @@ def _mirror_slash_side_effects(sid: str, session: dict, command: str) -> str: 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." + from agent.manual_compression_feedback import ( + describe_compression_lock_skip, + ) + return describe_compression_lock_skip(e.holder) _sync_session_key_after_compress(sid, session) with session["history_lock"]: