fix(gateway): open routeless stored chats read-only

Old cron, subagent, unbound API, WebUI and imported rows answered not_found.
The profile owner now opens them read-only; submit is refused with
session_read_only, archive, pin and leaf delete work, and deleting a row with
children is refused. The Claude Code and Codex importer goes through the
gateway's session.mutate import instead of writing state.db itself, and new
imports are cli sessions that resume writable.

Rows RT-30b, RT-48, RT-49, RT-54, RT-56.

Refs #143547

gateway/session_authority.py

@@ -201,9 +201,14 @@ class SessionAuthority:
                 restore_local_session(self, ref.session_id)
             else:
                 from gateway.session_route_recovery import restore_routed_session
-                if restore_routed_session(self, ref.session_id) is None:
+                from gateway.session_history_mount import mount_history
+                if (restore_routed_session(self, ref.session_id) is None
+                        and not mount_history(self, actor, ref.session_id)):
                     from gateway.session_api import restore_api_session
                     restore_api_session(self, ref.session_id)
+        from gateway.session_history_mount import is_history_mount, require_history_access
+        if is_history_mount(self.sessions[ref.session_id]):
+            require_history_access(self, actor, ref.session_id, capability)
         from gateway.config import Platform
         if self.sessions[ref.session_id].superseded and capability != 'session:read':
             # One gateway owns every session and never crosses them: the lane's route now belongs to

gateway/session_history_mount.py

@@ -0,0 +1,70 @@
+"""Read-only mount of stored history no receipt, route or binding restores (#143547 RT-30b/48/49).
+
+Pre-authority cron runs, subagent transcripts, API rows without a binding, and rows a third-party
+WebUI or the foreign-session importer wrote: before #106742 a viewer opened any of them straight
+from ``state.db``. Nothing can run them (no frozen policy, no lane, and adopting a WebUI row as
+writable would legitimise its second writer), so the owner mounts them for reading only. The mount
+is ``superseded``, so nothing route-keyed acts through it, and every capability but
+``session:read`` is refused by name. Archive, pin and delete still work as for any read-only mount
+(gateway/session_mutations.py), but only a leaf row deletes. Old local CLI/TUI/Desktop rows are not mounted
+here: resume adopts them as writable conversations (gateway/session_local_migration.py).
+"""
+import json
+
+from hermes_state_runtime import RuntimeStoreError
+
+# Never a lane key, so no lane publishing a conversation takes this mount over.
+_ROUTE_PREFIX = 'history:'
+
+
+def _reader(authority, actor, sid):
+    """The importer an import binding names, else the profile's owner: its native peer (the evidence
+    legacy adoption takes) or an operator. Knowing an old id is not ownership."""
+    from hermes_state_mutation_binding import history_binding
+    with authority.db._read_ctx() as conn:
+        binding = history_binding(conn, sid)
+    if binding is not None:
+        return binding == json.dumps([actor.profile_id, actor.subject], separators=(',', ':'))
+    return ('session:operator' in actor.capabilities
+            or getattr(authority, '_native_legacy_transports', {}).get(actor.transport_id) == actor)
+
+
+def mount_history(authority, actor, sid):
+    """Mount *sid* read-only for its owner; False when another loader owns it or the actor may not."""
+    from hermes_state_local import API_BINDING_PREFIX
+    from hermes_state_local_migration import LEGACY_LOCAL_SOURCES
+    row = authority.db.get_session(sid)
+    if (row['source'] in LEGACY_LOCAL_SOURCES | {'local'}
+            or row.get('profile_name') not in (None, '', authority.db._own_profile_name() or 'default')
+            or not _reader(authority, actor, sid)):
+        return False
+    with authority.db._read_ctx() as conn:
+        if conn.execute('SELECT 1 FROM state_meta WHERE key=?', (API_BINDING_PREFIX + sid,)).fetchone():
+            return False  # restore_api_session's
+    authority._require_admission_open()
+    from gateway.session_authority import LiveSession
+    authority.sessions.setdefault(sid, LiveSession(None, _ROUTE_PREFIX + sid, superseded=True, history_mount=True))
+    return True
+
+
+def is_history_mount(live):
+    return live.source is None and live.route.startswith(_ROUTE_PREFIX)
+
+
+def require_history_leaf(authority, sid):
+    """A delete takes a row's delegate subtree and detaches its other children, and a history mount
+    shows the owner neither, so it deletes only a leaf (as #140692 by @pnpwin did for subagents)."""
+    if not is_history_mount(authority.sessions[sid]):
+        return
+    with authority.db._read_ctx() as conn:
+        if conn.execute('SELECT 1 FROM sessions WHERE parent_session_id=? LIMIT 1', (sid,)).fetchone():
+            raise RuntimeStoreError('session_has_children', field='session_id',
+                                    detail='this stored conversation has child sessions a delete would take with it')
+
+
+def require_history_access(authority, actor, sid, capability):
+    if not _reader(authority, actor, sid):
+        raise RuntimeStoreError('not_found')
+    if capability != 'session:read':
+        raise RuntimeStoreError('session_read_only', field='session_id',
+                                detail='this stored conversation can be read but not continued; start a new chat')

gateway/session_mutations.py

@@ -57,12 +57,17 @@ def _authorize_control(authority, actor, ref, params, owner=None):
                 require_history_claim(authority, conn, actor, ref.session_id)
                 cold_history = authorize_history(conn, actor, ref.session_id)
         if not cold_history:
-            if params.get('operation') == 'delete':
-                # The read gate mounts a cold row first, so the check below sees what it is.
+            operation = params.get('operation')
+            if operation in {'delete', 'archive', 'sidebar'}:
+                # The read gate mounts a cold row first, so the checks below see what it is.
                 authority.authorize(actor, ref, 'session:read')
-            if not (params.get('operation') == 'delete' and authority.sessions[ref.session_id].superseded):
+            if operation == 'delete':
+                from gateway.session_history_mount import require_history_leaf
+                require_history_leaf(authority, ref.session_id)
+            if not (operation in {'delete', 'archive', 'sidebar'} and authority.sessions[ref.session_id].superseded):
                 # An old conversation its chat moved past is read-only, but deleting it retires its
-                # own lineage only (retirement never evicts a read-only mount's route).
+                # own lineage only (retirement never evicts a read-only mount's route), and archiving
+                # or pinning it flags that lineage's rows only, never the route's newer conversation.
                 authority.authorize(actor, ref, 'session:control')
     return cold_history
 

hermes_cli/foreign_sessions.py

@@ -8,6 +8,7 @@ import json
 import os
 import re
 import sys
+import uuid
 from dataclasses import dataclass
 from datetime import datetime
 from pathlib import Path
@@ -228,10 +229,13 @@ def _list_sessions(source: str, root: Optional[Path]) -> list[ForeignSession]:
     return results
 
 
-def import_foreign_session(source: str, path, db=None) -> str:
-    """Import one foreign session into the Hermes SessionDB; returns the new Hermes session id.
+def import_foreign_session(source: str, path) -> str:
+    """Import one foreign session through the gateway's ``session.mutate import`` (the gateway is the
+    only state.db writer, #106742); returns the new Hermes session id. The row lands as an ordinary
+    ``cli`` session bound to this importer, so ``hermes --resume`` continues it writable.
 
-    Raises ``ValueError`` on unknown source or a session with no usable conversation turns."""
+    Raises ``ValueError`` on unknown source or a session with no usable conversation turns, and
+    ``GatewayClientError`` (a ``ValueError``) when the gateway refuses the import."""
     source = (source or "").strip().lower().lstrip("@")
     if source not in _SOURCE_LABELS:
         raise ValueError(f"Unknown foreign session source: {source!r}")
@@ -245,24 +249,21 @@ def import_foreign_session(source: str, path, db=None) -> str:
     first_user = _first_user_line([(t["role"], t["content"]) for t in turns]) or path.stem
     if len(first_user) > _TITLE_MAX:
         first_user = first_user[: _TITLE_MAX - 1] + "…"
-    tool = _SOURCE_DB_NAMES[source]
-    owns_db = db is None
-    if owns_db:
-        from hermes_state_registry import acquire
-        db = acquire()  # the CLI resume that follows acquires this same handle
-    try:
-        session_id = new_session_id()
-        origin = {"imported_from": {"tool": tool, "path": str(path), "foreign_session_id": parsed.get("session_id")}}
-        db.create_session(session_id, source=tool, cwd=parsed.get("cwd"), origin_json=json.dumps(origin))
-        for turn in turns:
-            db.append_message(session_id, turn["role"], turn["content"])
-        with contextlib.suppress(Exception):  # title is cosmetic; the import itself succeeded
-            db.set_session_title(session_id, f"Imported from {_SOURCE_LABELS[source]}: {first_user}")
-        return session_id
-    finally:
-        if owns_db:
-            with contextlib.suppress(Exception):
-                db.close()
+    from hermes_cli.gateway_client import GatewayClientError
+    from hermes_cli.gateway_session_store import session_store
+    store = session_store()
+    session_id = new_session_id()
+    cwd = parsed.get("cwd")
+    # Resume refuses a folder missing here (another machine's, or deleted), as the Desktop import does.
+    row = {"id": session_id, "source": "cli", "cwd": cwd if cwd and Path(cwd).is_dir() else None, "messages": turns}
+    receipt = store.call("session.mutate", session_id=session_id, request_id=uuid.uuid4().hex, expected_revision=0,
+                         operation="import", payload={"sessions": [row]})
+    # Titles are UNIQUE (a re-import's is taken), so the title is a separate, cosmetic rename.
+    with contextlib.suppress(GatewayClientError):
+        store.call("session.mutate", session_id=session_id, request_id=uuid.uuid4().hex,
+                   expected_revision=receipt["revision"], operation="rename",
+                   payload={"title": f"Imported from {_SOURCE_LABELS[source]}: {first_user}"})
+    return session_id
 
 
 def gather_foreign_sessions(source: Optional[str] = None, *, claude_root: Optional[Path] = None,
@@ -304,7 +305,7 @@ def pick_foreign_session(source: Optional[str] = None, *, limit: int = 25) -> Op
     return None
 
 
-def run_sessions_import(args, db=None) -> Optional[str]:
+def run_sessions_import(args) -> Optional[str]:
     """`hermes sessions import` entry point. Returns new session id or None."""
     source = getattr(args, "from_source", None)
     path = getattr(args, "path", None)
@@ -328,7 +329,7 @@ def run_sessions_import(args, db=None) -> Optional[str]:
             return None
         source, chosen_path = picked.source, picked.path
     try:
-        session_id = import_foreign_session(source, chosen_path, db=db)
+        session_id = import_foreign_session(source, chosen_path)
     except ValueError as e:
         print(f"Error: {e}")
         return None

tests/gateway/test_stored_history_mount.py

@@ -0,0 +1,117 @@
+"""Old stored chats no loader restores open read-only for their owner (#143547 RT-30b, RT-48, RT-49, RT-54).
+
+Pre-update cron runs, subagent transcripts, API rows without a binding, WebUI rows and rows the
+foreign-session importer wrote answered ``not_found`` on ``session.resume`` and on delete. They
+now mount read-only: the profile's owner reads and deletes them, a submit is refused by name, and
+a caller that is not the owner still finds nothing.
+"""
+import json
+from types import SimpleNamespace
+
+import pytest
+
+from gateway.session_contract import SessionRef
+from hermes_state_runtime import RuntimeStoreError
+
+ROWS = {
+    'cron_job1_20261001_120000': ('cron', {}),
+    'sub-1': ('subagent', {}),
+    'api-1': ('api_server', {'session_key': 'agent:main:api_server:dm:api-1', 'chat_id': 'api-1'}),
+    'webui-1': ('webui', {'session_key': 'webui-1', 'origin_json': json.dumps({'platform': 'webui'})}),
+    'cc-1': ('claude-code', {'origin_json': json.dumps({'imported_from': {'tool': 'claude-code', 'path': '/x'}})}),
+    'hook-1': ('webhook', {}),
+}
+
+
+async def _connect(tmp_path, monkeypatch, sid, **identity):
+    """A fresh gateway over a state.db holding *sid*, and a native Desktop connection to it."""
+    from gateway import run
+    from gateway.config import GatewayConfig
+    from gateway.session import SessionStore
+    from gateway.session_authority import initialize_session_authority
+    from gateway.session_controls import AuthorityConnection
+    monkeypatch.setattr(run, '_load_gateway_config', lambda: {'model': {'default': 'm1', 'provider': 'openrouter'}})
+    store = SessionStore(tmp_path / 'sessions', GatewayConfig())
+    db = store._db
+    source, columns = ROWS[sid]
+    db.create_session(sid, source=source, model='m-old')
+    db.append_message(sid, role='user', content='q')
+    db.append_message(sid, role='assistant', content='a')
+    for column, value in columns.items():
+        db._write_sql(f'UPDATE sessions SET {column}=? WHERE id=?', (value, sid))
+    runner = SimpleNamespace(session_store=store, _session_db=db, adapters={}, _draining=False,
+                             _adapter_for_source=lambda source: None, _cached_agent_for=lambda route: None)
+    authority = await initialize_session_authority(runner, profile_id=str(tmp_path), instance_id='i')
+    native = {'user_id': 'owner', 'provider': 'local', 'profile_id': str(tmp_path), 'instance_id': 'i',
+              'capabilities': ['session:create', 'session:read', 'session:submit', 'session:control'],
+              'native_bootstrap': True}
+    return AuthorityConnection(authority, object(), {**native, **identity}), SessionRef(str(tmp_path), sid)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('sid', sorted(ROWS))
+async def test_old_stored_history_opens_read_only_for_its_owner(tmp_path, monkeypatch, sid):
+    conn, ref = await _connect(tmp_path, monkeypatch, sid)
+
+    opened = await conn.resume(ref, {})
+
+    assert [m['content'] for m in opened['messages']] == ['q', 'a']
+    with pytest.raises(RuntimeStoreError, match='session_read_only'):
+        await conn.submit(ref, {'text': 'more', 'submission_id': 's1'})
+    assert conn.authority.db.get_session(sid)['message_count'] == 2
+
+
+@pytest.mark.asyncio
+async def test_a_caller_that_is_not_the_owner_still_finds_nothing(tmp_path, monkeypatch):
+    """A remote login is not evidence of ownership, before or after the owner opened the row."""
+    from gateway.session_contract import Principal
+    conn, ref = await _connect(tmp_path, monkeypatch, 'cron_job1_20261001_120000')
+    login = Principal('basic:alice', ref.profile_id, frozenset({'session:read'}), 'web-socket')
+    with pytest.raises(RuntimeStoreError, match='not_found'):
+        conn.authority.authorize(login, ref, 'session:read')
+    assert ref.session_id not in conn.authority.sessions
+    await conn.resume(ref, {})
+    with pytest.raises(RuntimeStoreError, match='not_found'):
+        conn.authority.authorize(login, ref, 'session:read')
+
+
+@pytest.mark.asyncio
+async def test_the_owner_deletes_a_completed_subagent_history(tmp_path, monkeypatch):
+    conn, ref = await _connect(tmp_path, monkeypatch, 'sub-1')
+    row = conn.authority.db.get_session('sub-1')
+    await conn.mutate(ref, {'session_id': 'sub-1', 'request_id': 'delete-1', 'expected_revision': row['runtime_revision'],
+                            'expected_generation': row['runtime_generation'], 'operation': 'delete', 'payload': {}})
+    assert conn.authority.db.get_session('sub-1') is None
+
+
+async def _mutate(conn, ref, request_id, operation, payload):
+    row = conn.authority.db.get_session(ref.session_id)
+    return await conn.mutate(ref, {'session_id': ref.session_id, 'request_id': request_id, 'operation': operation,
+                                   'payload': payload, 'expected_revision': row['runtime_revision'],
+                                   'expected_generation': row['runtime_generation']})
+
+
+@pytest.mark.asyncio
+async def test_the_owner_archives_unarchives_and_pins_old_stored_history(tmp_path, monkeypatch):
+    """RT-54/RT-56: these answered ``session_read_only`` on the mount, as a moved-on chat's old
+    conversation answered ``session_superseded``."""
+    conn, ref = await _connect(tmp_path, monkeypatch, 'cron_job1_20261001_120000')
+    await _mutate(conn, ref, 'archive', 'archive', {'archived': True})
+    assert conn.authority.db.get_session(ref.session_id)['archived'] == 1
+    await _mutate(conn, ref, 'unarchive', 'archive', {'archived': False})
+    await _mutate(conn, ref, 'pin', 'sidebar', {'pinned': True})
+    row = conn.authority.db.get_session(ref.session_id)
+    assert (row['archived'], row['pinned']) == (0, 1)
+
+
+@pytest.mark.asyncio
+async def test_a_stored_history_with_child_sessions_is_not_deleted(tmp_path, monkeypatch):
+    """The delete took the run's delegate child with it and detached any other child, unseen."""
+    conn, ref = await _connect(tmp_path, monkeypatch, 'cron_job1_20261001_120000')
+    db = conn.authority.db
+    db.create_session('sub-child', source='subagent', model='m-old')
+    db._write_sql('UPDATE sessions SET parent_session_id=?, model_config=? WHERE id=?',
+                  (ref.session_id, json.dumps({'_delegate_from': ref.session_id}), 'sub-child'))
+    with pytest.raises(RuntimeStoreError, match='session_has_children'):
+        await _mutate(conn, ref, 'delete', 'delete', {})
+    assert db.get_session('sub-child')['parent_session_id'] == ref.session_id

tests/hermes_cli/test_foreign_sessions.py

@@ -1,11 +1,13 @@
 """Tests for hermes_cli.foreign_sessions — Claude Code / Codex CLI import.
 
-Fixture JSONL is synthesized inline (tmp_path); the SessionDB is opened
-against a temp path so nothing touches the real HERMES_HOME store.
+Fixture JSONL is synthesized inline (tmp_path); imports reach a gateway owner over a temp
+state.db, so nothing touches the real HERMES_HOME store.
 """
 
+import asyncio
 import json
 from pathlib import Path
+from types import SimpleNamespace
 
 import pytest
 
@@ -96,12 +98,39 @@ def _write_codex_fixture(tmp_path, extra_lines=None):
     return f
 
 
+def _route_store(monkeypatch, mutate):
+    """Answer the importer's gateway calls with *mutate* (``session.mutate`` params -> receipt)."""
+    from hermes_cli.gateway_client import GatewayClientError
+    from hermes_state_runtime import RuntimeStoreError
+    calls = []
+
+    def call(method, **params):
+        calls.append((method, params["operation"]))
+        try:
+            return mutate(params)
+        except (RuntimeStoreError, ValueError) as exc:  # a refusal crosses the wire as a bounded code
+            raise GatewayClientError(getattr(exc, "reason", "invalid_params")) from exc
+
+    monkeypatch.setattr("hermes_cli.gateway_session_store.session_store", lambda: SimpleNamespace(call=call))
+    return calls
+
+
 @pytest.fixture
-def session_db(tmp_path):
+def gateway(tmp_path, monkeypatch):
+    """A gateway owner over a temp state.db that the importer's ``session.mutate`` calls reach."""
+    from gateway.session_authority import SessionAuthority
+    from gateway.session_contract import Principal, SessionRef
+    from gateway.session_mutations import mutate_session
     from hermes_state import SessionDB
+    from hermes_state_runtime import begin_runtime_epoch
 
     db = SessionDB(db_path=tmp_path / "state.db")
-    yield db
+    authority = SessionAuthority(SimpleNamespace(_draining=False), profile_id="p", instance_id="owner",
+                                 db=db, epoch=begin_runtime_epoch(db, instance_id="owner"))
+    actor = Principal("owner", "p", frozenset({"session:read", "session:control", "session:create"}), "cli")
+    calls = _route_store(monkeypatch, lambda params: asyncio.run(
+        mutate_session(authority, actor, SessionRef("p", params["session_id"]), params)))
+    yield SimpleNamespace(db=db, calls=calls)
     db.close()
 
 
@@ -217,52 +246,103 @@ def test_blank_env_overrides_fall_back_to_home(tmp_path, monkeypatch):
     assert {s.source for s in both} == {"claude", "codex"}
 
 
-# ── import into SessionDB ────────────────────────────────────────────────
+# ── import through the gateway ───────────────────────────────────────────
 
 
-def test_import_claude_session(tmp_path, session_db):
+def test_import_claude_session(tmp_path, gateway):
     f = _write_claude_fixture(tmp_path)
-    session_id = import_foreign_session("claude", f, db=session_db)
-    row = session_db.get_session(session_id)
+    session_id = import_foreign_session("claude", f)
+    row = gateway.db.get_session(session_id)
     assert row is not None
-    assert row["source"] == "claude-code"
-    assert row["cwd"] == "/home/user/proj"
+    assert row["source"] == "cli"
+    assert row["origin_json"] is None
+    assert row["cwd"] is None  # /home/user/proj is not a folder here, and resume would refuse it
     assert row["message_count"] == 4
-    title = session_db.get_session_title(session_id)
+    title = gateway.db.get_session_title(session_id)
     assert title.startswith("Imported from Claude Code: ")
     assert "Please fix the flaky test" in title
-    messages = session_db.get_messages(session_id)
+    messages = gateway.db.get_messages(session_id)
     assert len(messages) == 4
     _assert_alternating(messages)
-    origin = json.loads(row["origin_json"])
-    assert origin["imported_from"]["tool"] == "claude-code"
-    assert origin["imported_from"]["path"] == str(f)
+    # A re-import keeps its conversation; only the taken (UNIQUE) title is dropped.
+    again = import_foreign_session("claude", f)
+    assert gateway.db.get_session(again)["message_count"] == 4
+    assert gateway.db.get_session_title(again) is None
 
 
-def test_import_codex_session(tmp_path, session_db):
+def test_import_codex_session(tmp_path, gateway):
     f = _write_codex_fixture(tmp_path)
-    session_id = import_foreign_session("@codex", f, db=session_db)
-    row = session_db.get_session(session_id)
+    f.write_text(f.read_text(encoding="utf-8").replace("/home/user/repo", str(tmp_path)), encoding="utf-8")
+    session_id = import_foreign_session("@codex", f)
+    row = gateway.db.get_session(session_id)
     assert row is not None
-    assert row["source"] == "codex-cli"
+    assert row["source"] == "cli"
+    assert row["cwd"] == str(tmp_path)
     assert row["message_count"] == 4
-    title = session_db.get_session_title(session_id)
+    title = gateway.db.get_session_title(session_id)
     assert title.startswith("Imported from Codex CLI: ")
-    messages = session_db.get_messages(session_id)
+    messages = gateway.db.get_messages(session_id)
     _assert_alternating(messages)
-    # resumable: resolve_session_id round-trips
-    assert session_db.resolve_session_id(session_id) == session_id
 
 
-def test_import_rejects_bad_input(tmp_path, session_db):
+def test_import_rejects_bad_input(tmp_path, gateway):
     with pytest.raises(ValueError, match="Unknown foreign session source"):
-        import_foreign_session("gemini", tmp_path / "x.jsonl", db=session_db)
+        import_foreign_session("gemini", tmp_path / "x.jsonl")
     with pytest.raises(ValueError, match="not found"):
-        import_foreign_session("claude", tmp_path / "missing.jsonl", db=session_db)
+        import_foreign_session("claude", tmp_path / "missing.jsonl")
     empty = tmp_path / "empty.jsonl"
     empty.write_text("not json\n", encoding="utf-8")
     with pytest.raises(ValueError, match="No user/assistant conversation"):
-        import_foreign_session("claude", empty, db=session_db)
+        import_foreign_session("claude", empty)
+    assert gateway.calls == []
+
+
+def test_import_is_written_by_the_gateway_not_this_process(tmp_path, gateway, monkeypatch):
+    """The gateway is the only state.db writer (#106742): the importer only sends ``session.mutate``."""
+    from hermes_state import SessionDB
+    from hermes_state_mutation_binding import history_binding
+
+    def refused(*_a, **_k):
+        raise AssertionError("the importer wrote state.db itself")
+    monkeypatch.setattr(SessionDB, "create_session", refused)
+    monkeypatch.setattr(SessionDB, "append_message", refused)
+
+    session_id = import_foreign_session("claude", _write_claude_fixture(tmp_path))
+
+    assert gateway.calls == [("session.mutate", "import"), ("session.mutate", "rename")]
+    with gateway.db._read_ctx() as conn:  # the import receipt bound the row to its importer
+        assert history_binding(conn, session_id) == json.dumps(["p", "owner"], separators=(",", ":"))
+
+
+@pytest.mark.asyncio
+async def test_imported_chat_resumes_writable_as_cli(tmp_path, monkeypatch):
+    """``hermes --resume @claude`` continues the import as an ordinary CLI session, not a read-only mount."""
+    from gateway import run
+    from gateway.config import GatewayConfig
+    from gateway.session import SessionStore
+    from gateway.session_authority import initialize_session_authority
+    from gateway.session_contract import SessionRef
+    from gateway.session_controls import AuthorityConnection
+    monkeypatch.setattr(run, "_load_gateway_config", lambda: {"model": {"default": "m1", "provider": "openrouter"}})
+    store = SessionStore(tmp_path / "sessions", GatewayConfig())
+    runner = SimpleNamespace(session_store=store, _session_db=store._db, adapters={}, _draining=False,
+                             _adapter_for_source=lambda source: None, _cached_agent_for=lambda route: None)
+    authority = await initialize_session_authority(runner, profile_id=str(tmp_path), instance_id="i")
+    conn = AuthorityConnection(authority, object(), {
+        "user_id": "owner", "provider": "local", "profile_id": str(tmp_path), "instance_id": "i",
+        "capabilities": ["session:create", "session:read", "session:submit", "session:control"],
+        "native_bootstrap": True})
+    loop = asyncio.get_running_loop()
+    _route_store(monkeypatch, lambda params: asyncio.run_coroutine_threadsafe(
+        conn.mutate(SessionRef(str(tmp_path), params["session_id"]), params), loop).result())
+
+    session_id = await asyncio.to_thread(import_foreign_session, "claude", _write_claude_fixture(tmp_path))
+    ref = SessionRef(str(tmp_path), session_id)
+    opened = await conn.resume(ref, {})
+
+    assert len(opened["messages"]) == 4
+    assert not authority.sessions[session_id].superseded
+    authority.authorize(conn.actor, ref, "session:submit")
 
 
 def test_leading_assistant_gets_single_stub(tmp_path):