fix(cli): /compress commits on the gateway (GatewayAgent._compress_context → session.mutate compress)

Main's /compress path (compress_now → agent._compress_context) reached the GatewayAgent stand-in
and failed with "'GatewayAgent' object has no attribute '_compress_context'". The stand-in now
maps it to the gateway's `session.mutate compress` (focus / here [N] as partial+keep_last), then
re-reads the compacted transcript with session.resume and returns it, so main's unchanged
_manual_compress prints its own "✅ Compressed: N → M messages" report and installs the history.
The CLI never summarizes or writes locally; the verbatim tail is stamped as stored because the
gateway commit already kept it.

Test: tests/hermes_cli/test_gateway_agent.py::test_manual_compress_is_the_gateway_compress_mutation
(red on the merge head, green here). Harness step 09.

(cherry picked from commit 1ee2e1641c36dbdbebc1f1308c041e9f56b6d179)

hermes_cli/gateway_agent.py

@@ -231,6 +231,35 @@ class GatewayAgent:
             return False
         return (result or {}).get("status") in {"queued", "redirected"}
 
+    def _compress_context(self, messages, system_message=None, *, approx_tokens=None, focus_topic=None,
+                          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."""
+        from agent.context_compressor import _DB_PERSISTED_MARKER
+        from agent.conversation_compression import COMPACTION_STATUS
+        from hermes_cli.gateway_client import COMPRESS_RPC_TIMEOUT
+        payload = {}
+        if focus_topic:
+            payload["focus"] = focus_topic
+        if verbatim_tail:
+            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)
+        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"],
+                             operation="compress", payload=payload)
+        for row in verbatim_tail or ():
+            row[_DB_PERSISTED_MARKER] = True
+        after = self.connection.call("session.resume", session_id=self.session_id)
+        self._state().adopt_snapshot(after)
+        compressor = self.context_compressor
+        compressor.compression_count += 1
+        compressor.last_prompt_tokens = -1
+        self._turn_base_usage_anchor = None
+        return [{k: v for k, v in m.items() if k != "row_id"} for m in after.get("messages") or []], system_message
+
     def steer(self, text):
         return self._correct("steer", text)
 

tests/hermes_cli/test_gateway_agent.py

@@ -202,3 +202,28 @@ def test_agent_print_routes_to_print_fn(make_agent):
 def test_hard_interrupt_is_a_real_method():
     import inspect
     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 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