fix(cli): resuming a compressed session addresses its receipt owner, not the tip segment

After a local conversation has been compressed twice, a classic `hermes --resume <root>`
(or `hermes chat --resume <root> -q ...`) printed "Session <root> was compressed into
<seg3>; resuming the descendant" and then failed everything it sent. Every turn answered
`permission_denied` and never reached the provider. `/undo` and `/rollback` answered
`storage_unavailable`, and gateway.log read `Local session <seg3> receipt refused:
storage_unavailable at local_receipt`.

Cause: `HermesCLI._follow_compression_chain` (hermes_cli/cli_agent_setup_mixin.py), used by
`_preload_resumed_session` and `_load_resumed_history_late`, set `self.session_id` to
`resolve_resume_session_id(root)`, the physical tip segment, so that it could read the tip's
transcript. `GatewayAgent.session_id` then attached to the gateway under that segment id,
and every later `prompt.submit`, `session.mutate rewind` and `rollback.*` named the segment.
Canonical local receipts are keyed by the conversation's logical owner (its creation id),
so `authorize` -> `restore_local_session` -> `local_receipt(segment)` found no row. Mid-chat
`/resume` (`_resolve_resume_target`) had the same switch.

Fix: keep reading the transcript from the tip segment, but address the session by the id
the gateway answers for it. `_gateway_owned_id` calls `session.resume` on the segment and
takes its `stored_session_id`. For a canonical local segment the gateway already maps that
to the owner (`resolve_local_target`, LC-12 #139306). That is how Ink `/resume` and a raw
`session.resume` already work, and it is why both run turns on this same conversation.
The read id is held separately (`_resume_read_id`), so the history, the tip-only safety
guard and the Kanban guard still look at the segment that holds the messages. Off the
gateway store (unit-test SessionDBs), or when the attach is refused, nothing changes. The
gateway receipt check is untouched: a segment id is still refused `storage_unavailable`,
and the new test pins that.

The other option was teaching the gateway to map any segment id back to its lineage owner
on every verb. It was rejected because it widens the receipt ownership check on every
admission path. The resume-time mapping already exists, so the client should use it.

Test: tests/hermes_cli/test_resume_compressed_gateway_id.py drives the real CLI resume
resolution (both startup paths) on a `GatewaySessionStore` against a real session
authority and a real SQLite state.db with two published compression children. It then
authorizes `session:control` on the id the CLI kept, which runs the real
`restore_local_session`/`local_receipt` path with no mocks. It fails on c6961c62af
(`'seg-3' == 'local-…'`) and passes with the fix.

Live (loopback stub, tmux, classic CLI): after the fix, `--resume <root>` resumes as the
owner id. The turn after resume, the one-shot `-q` turn, the turn after `/undo` and the
turn after `/rollback` all reach the stub. `/undo` undoes a turn. `/rollback` lists the
checkpoint taken after both compressions, and `/rollback 1` restores the file and undoes
the chat turn. gateway.log has 0 `receipt refused` lines.

Live-pass finding (package 10)
Refs #134889
Refs #140719

hermes_cli/cli_agent_setup_mixin.py

@@ -113,6 +113,20 @@ def _compression_descendant(session_db, session_id):
     return resolved_id if resolved_id and resolved_id != session_id else None
 
 
+def _gateway_owned_id(session_db, session_id):
+    """The id the gateway answers for physical row *session_id*: ``session.resume``'s
+    ``stored_session_id``. A canonical local compression segment holds the transcript, but its creation
+    receipt lives on the conversation's logical owner, which every turn and mutation must name (as Ink
+    and ``session.resume`` do). Unchanged off the gateway store or when the resume is refused."""
+    from hermes_cli.gateway_session_store import GatewaySessionStore
+    if not isinstance(session_db, GatewaySessionStore):
+        return session_id
+    try:
+        return session_db.attach(session_id).get("stored_session_id") or session_id
+    except Exception:  # health: allow BLE001 -- a refused attach leaves main's own resume path to report it
+        return session_id
+
+
 def _user_display_text(content) -> str:
     """Recap text for a user row; multimodal lists become text parts + ``[image]`` markers."""
     if isinstance(content, list):
@@ -529,11 +543,19 @@ class CLIAgentSetupMixin:
     def _follow_compression_chain(self, session_meta, announce):
         """If the resumed id is an empty compression-chain head, announce and switch to
         the descendant holding the messages; returns the (possibly refreshed) meta."""
+        self._resume_read_id = self.session_id
         resolved_id = _compression_descendant(self._session_db, self.session_id)
         if resolved_id:
-            announce(resolved_id)
-            self.session_id = resolved_id
-            session_meta = self._session_db.get_session(self.session_id) or session_meta
+            session_meta = self._session_db.get_session(resolved_id) or session_meta
+            self._resume_read_id = resolved_id
+            owner = _gateway_owned_id(self._session_db, resolved_id)
+            if owner != resolved_id:
+                # The transcript is read from the tip segment; the session stays addressed by the
+                # id the gateway holds its receipt for (a segment id is refused storage_unavailable).
+                self.session_id = owner
+            else:
+                announce(resolved_id)
+                self.session_id = resolved_id
         return session_meta
 
     def _restore_session_state(self, session_meta, *, quiet: bool = False) -> None:
@@ -590,7 +612,7 @@ class CLIAgentSetupMixin:
                 t("cli.resume.cannot_resume", error=resume_limit_error),
                 f"[bold red]{_escape(t('cli.resume.cannot_resume_prefix'))}[/] {_escape(resume_limit_error)}")
             return False
-        restored = self._session_db.get_messages_as_conversation(self.session_id, repair_alternation=True)
+        restored = self._session_db.get_messages_as_conversation(self._resume_read_id, repair_alternation=True)
         if restored:
             restored = [m for m in restored if m.get("role") != "session_meta"]
             self.conversation_history = restored
@@ -761,7 +783,8 @@ class CLIAgentSetupMixin:
             safety_check = getattr(self._session_db, "assert_resume_safe", None)
             if not callable(safety_check):
                 return None
-            safety_check(self.session_id, **({"tip_only": True} if tip_only else {}))
+            safety_check(getattr(self, "_resume_read_id", None) or self.session_id,
+                         **({"tip_only": True} if tip_only else {}))
         except SessionResumeTooLargeError as exc:
             return str(exc)
         except Exception as exc:
@@ -790,7 +813,7 @@ class CLIAgentSetupMixin:
         # an ordinary CLI resume would be a live writer the board cannot observe. Refuse AND
         # fall back to a fresh session id, so this process never appends to the worker row.
         from hermes_cli.kanban_resume_guard import kanban_resume_refusal
-        if (kanban_refusal := kanban_resume_refusal(self._session_db, self.session_id)):
+        if (kanban_refusal := kanban_resume_refusal(self._session_db, self._resume_read_id)):
             self._console_print(f"[bold red]{_escape(kanban_refusal)}[/]")
             self._resumed = False
             from hermes_state_ids import new_session_id
@@ -801,7 +824,7 @@ class CLIAgentSetupMixin:
             self._resume_history_error = resume_limit_error
             self._console_print(f"[bold red]{_escape(t('cli.resume.cannot_resume_prefix'))}[/] {resume_limit_error}")
             return False
-        restored, display_history = self._session_db.get_resume_conversations(self.session_id)
+        restored, display_history = self._session_db.get_resume_conversations(self._resume_read_id)
         accent_color = _accent_hex()
         if not restored:
             self._console_print(

hermes_cli/cli_commands_mixin.py

@@ -31,7 +31,7 @@ from hermes_state_ids import new_session_id as mint_session_id
 from agent.i18n import t
 from agent.message_metadata import message_identity
 from agent.turn_context import extract_api_content_sidecar
-from hermes_cli.cli_agent_setup_mixin import _retire_agent
+from hermes_cli.cli_agent_setup_mixin import _gateway_owned_id, _retire_agent
 from hermes_cli.cli_commands_session_tools import (
     CLICommandsSessionToolsMixin, _TTYBuf, _accent, _accent_line, _command_arg, _cp, _dim,
     _dim_line, _gt, _lines, _pr, _probe, _rollback_kept, _save, _say_block, _shlex_args, _t, _tn)
@@ -994,6 +994,7 @@ class CLICommandsMixin(CLICommandsSessionToolsMixin):
         from hermes_cli.kanban_resume_guard import kanban_resume_refusal
         if (kanban_refusal := kanban_resume_refusal(self._session_db, target_id)):
             return _cp(f"  {kanban_refusal}")
+        read_id, target_id = target_id, _gateway_owned_id(self._session_db, target_id)  # the receipt owner
         if target_id == self.session_id:
             return _cp(f"  {_t('resume.already_on')}")
         old_session_id = self.session_id
@@ -1002,7 +1003,7 @@ class CLICommandsMixin(CLICommandsSessionToolsMixin):
         _sync_process_session_id(target_id)
         # One lineage SELECT, two projections: model_history is alternation-repaired for live
         # replay (heals a durable user;user once); display_history is verbatim (as startup --resume).
-        model_history, display_history = self._session_db.get_resume_conversations(target_id)
+        model_history, display_history = self._session_db.get_resume_conversations(read_id)
         self.conversation_history = _without_session_meta(model_history)
         self._resume_display_history = _without_session_meta(display_history)
         with suppress(Exception):  # re-open the target session so it's not marked as ended
@@ -1031,9 +1032,8 @@ class CLICommandsMixin(CLICommandsSessionToolsMixin):
         self._restore_session_model(session_meta)
 
     def _resolve_resume_target(self, target: str):
-        """``(session_id, meta)`` for a numbered selection, title, or id; None after printing why
-        it could not be resolved. An empty compression-chain head redirects to the descendant
-        that actually holds the transcript."""
+        """``(session_id, meta)`` for a numbered selection, title, or id (an empty compression-chain head
+        redirects to the descendant holding the transcript); None after printing why it was not resolved."""
         if target.isdigit():
             sessions = self._list_recent_sessions(limit=10)
             index = int(target)

tests/hermes_cli/test_resume_compressed_gateway_id.py

@@ -0,0 +1,139 @@
+"""Classic CLI ``--resume <root>`` of a twice-compressed local conversation addresses the gateway by
+the id it owns a receipt for (live-pass finding, package 10).
+
+The CLI reads the transcript from the compression tip (a physical segment), but the creation receipt
+lives on the logical owner. Ink and ``session.resume`` already answer with that owner; the CLI used to
+keep the segment id it read from, so every turn answered ``permission_denied`` and ``/undo`` /
+``/rollback`` answered ``storage_unavailable`` (``local_receipt`` found no receipt for the segment).
+
+Real CLI resume resolution (``_preload_resumed_session`` / ``_load_resumed_history_late`` on a
+``GatewaySessionStore``) against a real session authority and its real SQLite receipt path.
+"""
+import asyncio
+import io
+import types
+from types import SimpleNamespace
+
+import pytest
+
+
+def _runner(tmp_path):
+    from gateway.config import GatewayConfig
+    from gateway.session import SessionStore
+    store = SessionStore(tmp_path / 'sessions', GatewayConfig())
+    runner = SimpleNamespace(session_store=store, _session_db=store._db, adapters={}, _draining=False,
+                             _cached_agent_for=lambda route: None)
+    runner._adapter_for_source = lambda source: runner.adapters.get(source.platform)
+    return runner
+
+
+def _compress(db, parent, child, source):
+    history = [{'role': 'user', 'content': f'retained {child}'}, {'role': 'assistant', 'content': 'history'}]
+    assert db.try_acquire_compression_lock(parent, 'fixture')
+    try:
+        db.publish_compression_child(parent_session_id=parent, child_session_id=child,
+                                     source=source, messages=history, compression_lock_holder='fixture')
+    finally:
+        db.release_compression_lock(parent, 'fixture')
+
+
+async def _compressed(tmp_path):
+    from gateway.session_authority import initialize_session_authority
+    from gateway.session_contract import Principal
+    from gateway.session_local import create_local_session
+    authority = await initialize_session_authority(_runner(tmp_path), profile_id='fixture', instance_id='first')
+    actor = Principal('owner', 'fixture', frozenset({'session:create', 'session:read'}), 'socket')
+    ref = create_local_session(authority, actor, dict(request_id='row10', source='cli', cwd=str(tmp_path),
+                                                      model='frozen', toolsets=[]))
+    live = authority.sessions[ref.session_id]
+    store = authority.runner.session_store
+    for parent, child in ((ref.session_id, 'seg-2'), ('seg-2', 'seg-3')):
+        store._record_gateway_session_peer(parent, live.route, live.source)
+        _compress(authority.db, parent, child, authority.db.get_session(parent)['source'])
+        store._record_gateway_session_peer(child, live.route, live.source)
+    return authority, ref.session_id
+
+
+def _gateway_store(loop, authority, tmp_path):
+    """A ``GatewaySessionStore`` whose verbs run on the real authority (no socket) and whose reads use
+    the real read-only state.db reader, as the classic CLI's does."""
+    from gateway.session_controls import AuthorityConnection
+    from hermes_cli.gateway_client import GatewayClientError
+    from hermes_cli.gateway_connection import SessionState
+    from hermes_cli.gateway_session_store import GatewaySessionStore
+    from hermes_state import SessionDB
+    states = {}
+    connection = types.SimpleNamespace(final=False, state=lambda sid: states.setdefault(sid, SessionState(sid)))
+    gateway = AuthorityConnection(authority, object(), {
+        'provider': 'local', 'user_id': 'owner', 'profile_id': 'fixture', 'instance_id': authority.instance_id,
+        'capabilities': ['session:create', 'session:read', 'session:control']})
+    calls = []
+
+    def call(method, **params):
+        calls.append((method, params))
+        reply = loop.run_until_complete(gateway.dispatch({'id': len(calls), 'method': method, 'params': params}))
+        if 'error' in reply:
+            raise GatewayClientError(reply['error']['message'])
+        return reply['result']
+
+    store = GatewaySessionStore(connection)
+    store.call = call
+    store._reader = SessionDB(db_path=authority.db.db_path, read_only=True)
+    return store, calls
+
+
+def _cli(store, session_id):
+    from cli import HermesCLI
+    cli = HermesCLI.__new__(HermesCLI)
+    cli._resumed, cli._session_db, cli.session_id = True, store, session_id
+    cli.session_start = None
+    cli.conversation_history = []
+    cli.tool_progress_mode = 'off'
+    cli._console_print = lambda *a, **k: None
+    cli._restore_session_state = lambda *a, **k: None
+    return cli
+
+
+@pytest.fixture
+def rig(tmp_path, monkeypatch):
+    from gateway import run
+    monkeypatch.setattr(run, '_load_gateway_config', lambda: {'platform_toolsets': {'cli': []}})
+    monkeypatch.setenv('HERMES_HOME', str(tmp_path))
+    loop = asyncio.new_event_loop()
+    try:
+        authority, owner = loop.run_until_complete(_compressed(tmp_path))
+        store, calls = _gateway_store(loop, authority, tmp_path)
+        # _ensure_gateway_session: the CLI opens the gateway session for the --resume target first.
+        assert store.attach(owner)['stored_session_id'] == owner
+        yield SimpleNamespace(loop=loop, authority=authority, owner=owner, store=store, calls=calls)
+    finally:
+        loop.close()
+
+
+def _gateway_accepts(rig, session_id):
+    """The admission/mutation gate every turn, /undo and /rollback reach: ``authorize`` restores the
+    local session from its stored receipt (``restore_local_session`` -> ``local_receipt``)."""
+    from gateway.session_contract import Principal, SessionRef
+    actor = Principal('owner', 'fixture', frozenset({'session:read', 'session:control'}), 'socket')
+    rig.authority.authorize(actor, SessionRef('fixture', session_id), 'session:control')
+
+
+@pytest.mark.parametrize('path', ['_preload_resumed_session', '_load_resumed_history_late'])
+def test_resume_of_a_compressed_root_addresses_the_receipt_owner(rig, path, monkeypatch):
+    monkeypatch.setattr('sys.stderr', io.StringIO())
+    cli = _cli(rig.store, rig.owner)
+    assert getattr(cli, path)() is True
+    # History comes from the live tip segment (what main shows) ...
+    assert [m['content'] for m in cli.conversation_history] == ['retained seg-3', 'history']
+    # ... but the CLI keeps addressing the gateway by the id it owns a receipt for, as Ink and
+    # session.resume do. The segment id fails the receipt check with storage_unavailable.
+    assert cli.session_id == rig.owner
+    _gateway_accepts(rig, cli.session_id)
+
+
+def test_the_tip_segment_alone_still_has_no_receipt(rig):
+    """The receipt check is not weakened: addressing the physical segment is still refused by name."""
+    from hermes_state_runtime import RuntimeStoreError
+    with pytest.raises(RuntimeStoreError) as refused:
+        _gateway_accepts(rig, 'seg-3')
+    assert refused.value.reason == 'storage_unavailable'