fix(cli): /compress prints the gateway's report and counts only a commit

Symptom: in the classic CLI, a /compress whose summary call fails printed
"✅ Compressed: 8 → 8 messages" with no reason, and the status bar and
/usage counted one more compression, though nothing was compressed.

Root cause: GatewayAgent._compress_context (4a4114ac5a, the classic CLI's
move onto session.mutate compress) threw away the session.mutate result.
That result carries the owner's own report (aborted, refused, fallback,
reason) and status "noop" when nothing was committed. The owner runs its
compressor with abort_on_summary_failure=True, so a failed summary comes
back as "noop" with aborted set. compress_now then rebuilt the report from
the CLI's _CompressorView, which has none of those flags, and the stand-in
added to compression_count on every call.

Change: keep the result's summary on the agent for compress_now, which
uses it in place of a locally derived one. A "noop" answer returns the
input history unchanged: no re-read, no verbatim-tail stamp, no count.

Why: the gateway owns the session and its compressor, so its report is the
only one that knows why a compression did not happen; the client reads it
instead of guessing from a view that cannot see it.

agent/conversation_compression_manual.py

@@ -146,7 +146,12 @@ def compress_now(
     if tail and not all(row.get(_DB_PERSISTED_MARKER) is True for row in tail_rows):
         compressed = rejoin_compressed_head_and_tail(compressed, tail)
     after_tokens = estimate_request_tokens(agent, compressed)
-    summary = summarize_manual_compression(before, compressed, before_tokens, after_tokens, compression_state=compressor)
+    # A gateway-backed agent compresses on its owner, whose compressor holds the abort/refusal/reason
+    # flags: the owner's report is the result. Type-pinned (dict) for the MagicMock reason above.
+    summary = getattr(agent, "compression_summary", None)
+    if not isinstance(summary, dict):
+        summary = summarize_manual_compression(before, compressed, before_tokens, after_tokens,
+                                               compression_state=compressor)
     return CompressResult("compressed", before, list(compressed), before_tokens, after_tokens, request, summary=summary)
 
 

hermes_cli/gateway_agent.py

@@ -166,6 +166,8 @@ class GatewayAgent:
         # todo snapshot (``_set_todos``), read by the pet's done-plan beat.
         from tools.todo_tool import TodoStore
         self._todo_store = TodoStore()
+        # Part of the _compress_context seam: it sets the owner's report here and compress_now reads it.
+        self.compression_summary = None
         self._generation = None
         self._admission = None
         self._turn = None  # a turn a later one replaced stops taking frames
@@ -593,7 +595,10 @@ class GatewayAgent:
                           force=True, verbatim_tail=None, **_kw):
         """Main's manual ``/compress`` seam (``compress_now``) as ``session.mutate compress``: the
         gateway summarizes and commits; the returned transcript is the gateway's compacted history.
-        ``verbatim_tail`` rows are stamped as stored, so ``compress_now`` does not rejoin them."""
+        ``verbatim_tail`` rows are stamped as stored, so ``compress_now`` does not rejoin them.
+        The gateway's own report (``summary``: flags and note (with the reason)) is kept for ``compress_now``;
+        a ``noop`` answer (a failed summary or nothing removable) committed nothing, so the history,
+        the tail and the compression count stay as they were."""
         from agent.context_compressor import _DB_PERSISTED_MARKER
         from agent.conversation_compression import COMPACTION_STATUS
         from hermes_cli.gateway_client import COMPRESS_RPC_TIMEOUT
@@ -604,11 +609,16 @@ class GatewayAgent:
             payload["partial"] = True
             payload["keep_last"] = max(1, sum(1 for m in verbatim_tail if m.get("role") == "user"))
         self._emit_print(COMPACTION_STATUS)
+        self.compression_summary = None
         snapshot = self.connection.call("session.resume", session_id=self.session_id)
-        self.connection.call("session.mutate", timeout=COMPRESS_RPC_TIMEOUT, session_id=self.session_id,
-                             request_id=uuid.uuid4().hex, expected_revision=snapshot["revision"],
-                             expected_generation=snapshot["execution_generation"],
-                             operation="compress", payload=payload)
+        result = self.connection.call("session.mutate", timeout=COMPRESS_RPC_TIMEOUT, session_id=self.session_id,
+                                      request_id=uuid.uuid4().hex, expected_revision=snapshot["revision"],
+                                      expected_generation=snapshot["execution_generation"],
+                                      operation="compress", payload=payload) or {}
+        if isinstance(result.get("summary"), dict):
+            self.compression_summary = result["summary"]
+        if result.get("status") == "noop":
+            return list(messages), system_message
         for row in verbatim_tail or ():
             row[_DB_PERSISTED_MARKER] = True
         after = self.connection.call("session.resume", session_id=self.session_id)

tests/hermes_cli/test_gateway_agent.py

@@ -209,32 +209,6 @@ def test_hard_interrupt_is_a_real_method():
     assert inspect.getattr_static(GatewayAgent, "hard_interrupt")
 
 
-def test_manual_compress_is_the_gateway_compress_mutation(make_agent):
-    """Main's /compress (compress_now -> agent._compress_context) commits on the gateway and adopts
-    the compacted transcript; the CLI prints main's \"Compressed: N -> M\" report from it."""
-    from agent.conversation_compression_manual import CompressRequest, compress_now
-    agent, client = make_agent([])
-    compacted = [{"role": "user", "content": "summary", "row_id": 1}, {"role": "assistant", "content": "ok", "row_id": 2}]
-    original = client.rpc
-
-    async def rpc(method, *, _timeout=None, **params):
-        if method == "session.resume":
-            client.calls.append((method, params))
-            done = any(m == "session.mutate" for m, _ in client.calls)
-            return {"revision": 9, "execution_generation": 7, "messages": compacted if done else []}
-        return await original(method, _timeout=_timeout, **params)
-    client.rpc = rpc
-    history = [{"role": "user" if i % 2 == 0 else "assistant", "content": f"m{i}"} for i in range(6)]
-    result = compress_now(agent, history, CompressRequest(), task_id=SID)
-    mutate = dict(client.calls)["session.mutate"]
-    assert mutate["operation"] == "compress" and mutate["expected_revision"] == 9 and mutate["payload"] == {}
-    assert mutate["expected_generation"] == 7
-    assert result.status == "compressed"
-    assert result.after_messages == [{"role": "user", "content": "summary"}, {"role": "assistant", "content": "ok"}]
-    assert result.summary["headline"] == "Compressed: 6 → 2 messages"
-    assert agent.context_compressor.compression_count == 1
-
-
 SESSION_USAGE = {"input_tokens": 500, "output_tokens": 80, "cache_read_tokens": 0, "cache_write_tokens": 0,
                  "prompt_tokens": 500, "completion_tokens": 80, "total_tokens": 580, "reasoning_tokens": 0,
                  "api_calls": 5, "last_prompt_tokens": 229, "context_tokens": 229, "context_length": 128000,

tests/hermes_cli/test_gateway_agent_compress.py

@@ -0,0 +1,94 @@
+"""Classic CLI ``/compress`` through GatewayAgent: main's ``compress_now`` is the gateway's ``session.mutate
+compress``, and the CLI prints the owner's report (no live gateway)."""
+from contextlib import nullcontext
+from types import SimpleNamespace
+
+import pytest
+
+from tests.hermes_cli import test_gateway_agent
+from tests.hermes_cli.test_gateway_agent import SID
+
+make_agent = test_gateway_agent.make_agent  # the shared fixture, bound here so pytest finds it
+
+COMPACTED = [{"role": "user", "content": "summary", "row_id": 1}, {"role": "assistant", "content": "ok", "row_id": 2}]
+
+
+def six_rows():
+    return [{"role": "user" if i % 2 == 0 else "assistant", "content": f"m{i}"} for i in range(6)]
+
+
+def _compress_with_mutate_reply(make_agent, reply=None):
+    """Run main's ``compress_now`` against a scripted gateway whose ``session.resume`` answers the compacted
+    transcript once ``session.mutate`` (answering ``reply``, or the fake's default) has run."""
+    from agent.conversation_compression_manual import CompressRequest, compress_now
+    agent, client = make_agent([])
+    if reply is not None:
+        client.mutate_replies.append(reply)
+    original = client.rpc
+
+    async def rpc(method, *, _timeout=None, **params):
+        if method == "session.resume":
+            client.calls.append((method, params))
+            done = any(m == "session.mutate" for m, _ in client.calls)
+            return {"revision": 9, "execution_generation": 7, "messages": COMPACTED if done else []}
+        return await original(method, _timeout=_timeout, **params)
+    client.rpc = rpc
+    return agent, client, compress_now(agent, six_rows(), CompressRequest(), task_id=SID)
+
+
+def test_manual_compress_is_the_gateway_compress_mutation(make_agent):
+    """Main's /compress (compress_now -> agent._compress_context) commits on the gateway and adopts
+    the compacted transcript; the CLI prints main's \"Compressed: N -> M\" report from it."""
+    agent, client, result = _compress_with_mutate_reply(make_agent)
+    mutate = dict(client.calls)["session.mutate"]
+    assert mutate["operation"] == "compress" and mutate["expected_revision"] == 9 and mutate["payload"] == {}
+    assert mutate["expected_generation"] == 7
+    assert result.status == "compressed"
+    assert result.after_messages == [{"role": "user", "content": "summary"}, {"role": "assistant", "content": "ok"}]
+    assert result.summary["headline"] == "Compressed: 6 → 2 messages"
+    assert agent.context_compressor.compression_count == 1
+
+
+FAILED_SUMMARY = {"_last_compress_aborted": True, "_last_summary_error": "provider exploded"}
+
+
+@pytest.mark.parametrize("command, flags, shown", [
+    ("/compress", FAILED_SUMMARY, "⚠️ Compression aborted: 6 messages preserved"),
+    ("/compress here 1", FAILED_SUMMARY, "⚠️ Compression aborted: 6 messages preserved"),
+    ("/compress", {}, "🗜️ No changes from compression: 6 messages"),
+])
+def test_cli_compress_prints_the_gateways_noop_report_and_changes_nothing(make_agent, capsys, command, flags, shown):
+    """The owner's ``noop`` answer (a failed summary, or nothing removable) committed nothing: the CLI
+    prints the owner's report, keeps its history (an aborted ``here N`` tail is not doubled or lost),
+    counts no compression and leaves the status-bar estimate alone."""
+    from agent.manual_compression_feedback import summarize_manual_compression
+    from cli import HermesCLI
+    agent, client = make_agent([])
+    history = six_rows()
+    report = summarize_manual_compression(history, history, 90, 90, compression_state=SimpleNamespace(**flags))
+    client.mutate_replies.append({"session_id": SID, "operation": "compress", "status": "noop", "summary": report})
+    agent.context_compressor.last_prompt_tokens = 229
+    cli = HermesCLI.__new__(HermesCLI)
+    cli.conversation_history, cli.agent, cli.session_id = six_rows(), agent, SID
+    cli._busy_command = lambda *_a, **_k: nullcontext()
+    cli._manual_compress(command)
+    out = capsys.readouterr().out
+    assert shown in out and "✅" not in out
+    assert ("Reason: provider exploded" in out) is bool(flags)
+    assert cli.conversation_history == history
+    assert agent.context_compressor.compression_count == 0
+    assert agent.context_compressor.last_prompt_tokens == 229
+    assert [m for m, _ in client.calls].count("session.resume") == 1  # nothing committed: nothing adopted
+
+
+def test_manual_compress_reports_the_gateways_committed_summary(make_agent):
+    """A committed compression with a fallback summary keeps the owner's flags and reason, which the
+    CLI's compressor view cannot know."""
+    fallback = {"noop": False, "aborted": False, "refused_would_grow": False, "fallback_used": True,
+                "headline": "Compressed with fallback: 6 → 2 messages", "token_line": "Approx request size: ~90 → ~30 tokens",
+                "note": "Summary generation failed; Hermes used limited fallback context and removed 4 message(s)."}
+    agent, _, result = _compress_with_mutate_reply(
+        make_agent, {"session_id": SID, "operation": "compress", "revision": 10, "summary": fallback})
+    assert result.summary == fallback
+    assert len(result.after_messages) == 2
+    assert agent.context_compressor.compression_count == 1