fix(agent): route repeated-compression warning through _emit_status (#36908)

The 'Session compressed N times — accuracy may degrade' warning went
through _vprint (CLI stdout only), so the Ink TUI / Telegram / Discord
never saw it — unlike the two other compression warnings in the same
module, which route through _emit_status (and store _compression_warning
for late-bound gateway status_callback replay).

Set agent._compression_warning + call agent._emit_status() for this
warning too, matching the sibling pattern. _emit_status still _vprints
for the CLI, so CLI output is unchanged; TUI / gateway surfaces now
receive it via status_callback (and replay_compression_warning can
re-deliver it once a late-bound gateway callback is wired).

Co-authored-by: liuhao1024 <sunsky.lau@gmail.com>

agent/conversation_compression.py

@@ -719,14 +719,20 @@ def compress_context(
     except Exception as _me_err:
         logger.debug("memory manager on_session_switch (compression): %s", _me_err)
 
-    # Warn on repeated compressions (quality degrades with each pass)
+    # Warn on repeated compressions (quality degrades with each pass).
+    # Route through _emit_status (like the other compression warnings above)
+    # so the warning reaches the TUI / Telegram / Discord via status_callback,
+    # not just CLI stdout. _emit_status still _vprints for the CLI, and
+    # storing it on _compression_warning lets replay_compression_warning
+    # re-deliver it once a late-bound gateway status_callback is wired (#36908).
     _cc = agent.context_compressor.compression_count
     if _cc >= 2:
-        agent._vprint(
+        _cc_msg = (
             f"{agent.log_prefix}⚠️  Session compressed {_cc} times — "
-            f"accuracy may degrade. Consider /new to start fresh.",
-            force=True,
+            f"accuracy may degrade. Consider /new to start fresh."
         )
+        agent._compression_warning = _cc_msg
+        agent._emit_status(_cc_msg)
 
     # Emit session:compress event so hooks (e.g. MemPalace sync) can ingest
     # the completed old session before its details are lost. In in-place mode

tests/agent/test_compression_count_warning_36908.py

@@ -0,0 +1,87 @@
+"""Regression for #36908: the repeated-compression warning must reach the
+TUI / gateway, not just CLI stdout.
+
+When a session is compressed >= 2 times, ``compress_context`` warns that
+accuracy may degrade. That warning used to go through ``_vprint`` (stdout
+only), so the Ink TUI / Telegram / Discord never saw it — unlike the two
+other compression warnings in the same module, which route through
+``_emit_status`` (and store ``_compression_warning`` for late-bound
+gateway replay). This pins the warning onto the gateway-aware channel.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from hermes_state import SessionDB
+
+
+def _build_agent_with_db(db: SessionDB, session_id: str, compression_count: int):
+    with patch.dict(os.environ, {"OPENROUTER_API_KEY": "test-key"}):
+        from run_agent import AIAgent
+
+        agent = AIAgent(
+            api_key="test-key",
+            base_url="https://openrouter.ai/api/v1",
+            model="test/model",
+            quiet_mode=True,
+            session_db=db,
+            session_id=session_id,
+            skip_context_files=True,
+            skip_memory=True,
+        )
+
+    compressor = MagicMock()
+    compressor.compress.return_value = [
+        {"role": "user", "content": "[CONTEXT COMPACTION] summary"},
+        {"role": "user", "content": "tail"},
+    ]
+    compressor.compression_count = compression_count
+    compressor.last_prompt_tokens = 0
+    compressor.last_completion_tokens = 0
+    compressor._last_summary_error = None
+    compressor._last_compress_aborted = False
+    compressor._last_aux_model_failure_model = None
+    compressor._last_aux_model_failure_error = None
+    agent.context_compressor = compressor
+    return agent
+
+
+def test_repeated_compression_warning_routed_through_emit_status(tmp_path: Path) -> None:
+    db = SessionDB(db_path=tmp_path / "state.db")
+    sid = "PARENT_36908"
+    db.create_session(sid, source="cli")
+
+    # compression_count == 2 → the "compressed N times" warning should fire.
+    agent = _build_agent_with_db(db, sid, compression_count=2)
+
+    emitted: list[str] = []
+    agent._emit_status = lambda message: emitted.append(message)
+
+    messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
+    agent._compress_context(messages, "sys", approx_tokens=120_000)
+
+    # The warning reached the gateway-aware channel...
+    assert any("compressed 2 times" in m.lower() for m in emitted), (
+        f"repeated-compression warning not emitted via _emit_status: {emitted}"
+    )
+    # ...and was stored for late-bound gateway status_callback replay.
+    assert "compressed 2 times" in (getattr(agent, "_compression_warning", "") or "").lower()
+
+
+def test_no_warning_below_threshold(tmp_path: Path) -> None:
+    db = SessionDB(db_path=tmp_path / "state.db")
+    sid = "PARENT_36908_ONCE"
+    db.create_session(sid, source="cli")
+
+    # compression_count == 1 → no repeated-compression warning.
+    agent = _build_agent_with_db(db, sid, compression_count=1)
+    emitted: list[str] = []
+    agent._emit_status = lambda message: emitted.append(message)
+
+    messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
+    agent._compress_context(messages, "sys", approx_tokens=120_000)
+
+    assert not any("compressed" in m.lower() and "times" in m.lower() for m in emitted)