fix(gateway): mount a messaging row its lane moved past read-only

TUI-8's cold mount rebuilt a historic messaging row on its lane's CURRENT
route after /new. Everything route-keyed then acted on the newer
conversation: authority.agent and _runs_own_agent returned the successor's
running agent (a Stop on the old row interrupted it), a turn admitted on the
old row ran through _handle_message(source) and landed in the successor, and
viewer close / delete interrupted, refused and evicted the successor's route.

One gateway owns every session and never crosses them, so such a row mounts
read-only (LiveSession.superseded): authorize refuses every capability except
session:read with the named reason session_superseded, agent() and
_runs_own_agent() return nothing, and viewer close and delete leave the route
alone. A private route was rejected: no adapter delivers to it and the old
row's turns would fork the lane. A cold mount later superseded by its lane
publishing another conversation goes read-only too; the lane publishing the
row itself clears it.

Greptile: a messaging compression continuation mounted under its physical id
while the lane's turns publish to the lineage root, so its viewer missed every
event. restore_routed_session no longer mounts a continuation, and
session.resume maps it to the lineage root (as LC-12 does for local rows) and
answers with that id.

Review finding on #140183 (independent review)
Review finding on #140183 (Greptile)
Refs #134889 (TUI-8)
Refs #134889 (RT-30)

gateway/session_authority.py

@@ -60,6 +60,14 @@ class LiveSession:
     settling: dict = field(default_factory=dict)
     # A viewer attached while a guarded deferred pick awaits consent: the next boundary asks again.
     reask_pick: bool = False
+    # A historic messaging row mounted for reading after its lane moved on (``/new``): ``route`` is
+    # the lane's CURRENT key, which names another conversation, so nothing route-keyed (agent, Stop,
+    # turns, eviction) may act through it (gateway/session_route_recovery.py). The lane publishing
+    # this row again clears it.
+    superseded: bool = False
+    # Mounted by that cold path (not published by its lane): the lane publishing ANOTHER conversation
+    # on the same route later supersedes it too.
+    history_mount: bool = False
 
     def __post_init__(self):
         self.controls = PendingControls(self.event_stream)
@@ -144,6 +152,11 @@ class SessionAuthority:
                     from gateway.session_api import restore_api_session
                     restore_api_session(self, ref.session_id)
         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
+            # its newer conversation, so a write here would land there. Named, so a viewer can say so.
+            raise RuntimeStoreError('session_superseded', field='session_id',
+                                    detail='this conversation\'s chat has moved on to a newer session; it is read-only')
         source = self.sessions[ref.session_id].source
         if (source is not None and source.platform == Platform.LOCAL
                 and source.user_id != actor.subject and 'session:operator' not in actor.capabilities):
@@ -197,12 +210,24 @@ class SessionAuthority:
 
     def _publish_route(self, source, sid, route):
         self._require_admission_open()
-        self.sessions.setdefault(sid, LiveSession(source, route))
+        live = self.sessions.get(sid)
+        if live is None:
+            live = self.sessions[sid] = LiveSession(source, route)
+            # The lane moved on to *sid*: a cold history mount still on its route is now another
+            # conversation's, so it goes read-only (gateway/session_route_recovery.py).
+            for other in self.sessions.values():
+                if other.history_mount and other.route == route and other is not live:
+                    other.superseded = True
+        # The lane itself names *sid* (again): a historic mount becomes its live session.
+        live.superseded = live.history_mount = False
         return SessionRef(self.profile_id, sid)
 
     def agent(self, ref):
+        live = self.sessions[ref.session_id]
+        if live.superseded:
+            return None  # the route's agent is the newer conversation's
         lookup = getattr(self.runner, '_resident_agent_for', None) or self.runner._cached_agent_for
-        return lookup(self.sessions[ref.session_id].route)
+        return lookup(live.route)
 
     def _pending_rows(self, ref):
         """Pending admissions as the event stream presents them: a settlement committed but not yet
@@ -726,8 +751,11 @@ class SessionAuthority:
         """The route's running slot holds a real agent (not the claim's pending sentinel), so the
         Stop above reached the turn's own agent, not a cached one it may still replace."""
         from gateway.run import _AGENT_PENDING_SENTINEL
+        live = self.sessions[ref.session_id]
+        if live.superseded:
+            return False
         running = getattr(self.runner, '_running_agents', None) or {}
-        agent = running.get(self.sessions[ref.session_id].route)
+        agent = running.get(live.route)
         return agent is not None and agent is not _AGENT_PENDING_SENTINEL
 
     def steer_agent(self, ref, generation, text):

gateway/session_controls.py

@@ -514,6 +514,11 @@ class AuthorityConnection:
             if row and row['source'] in LEGACY_LOCAL_SOURCES | {Platform.LOCAL.value}:
                 ref = resolve_local_target(self.authority, self.actor,
                     self.authority.db.get_compression_tip(ref.session_id) or ref.session_id)
+            elif row:
+                # A messaging compression continuation attaches to its lineage root, the session
+                # its lane's turns stream to; the reply names that root.
+                from gateway.session_route_recovery import messaging_owner_ref
+                ref = messaging_owner_ref(self.authority, ref)
         if params.get('editor') is not None:
             self.authority.authorize(self.actor, ref, 'session:control')
             from gateway.session_local_mcp import resume_editor_mcp

gateway/session_mutations.py

@@ -310,7 +310,7 @@ def retire_live_sessions(authority, session_ids):
         # A closed session's refusal of new delegations (session_viewer_close) goes with its live state.
         allow_for_session((authority.profile_id, sid))
         candidate = authority.sessions.pop(sid, None)
-        if candidate is not None:
+        if candidate is not None and not candidate.superseded:  # its route is the newer session's
             evict = getattr(authority.runner, '_evict_cached_agent', None)
             if callable(evict):
                 evict(candidate.route)

gateway/session_route_recovery.py

@@ -12,7 +12,13 @@ every cold row of this profile that carries a ``session_key`` and a messaging-pl
   else its columns, rebuild the source.
 
 Either way the source is revived through ``_restored_source``, so delivery stays pinned to the
-persisted receiving bot. Rows without a ``session_key`` (subagents) have nothing to rebuild a lane
+persisted receiving bot. A row the lane moved past mounts READ-ONLY (``LiveSession.superseded``): its
+only route is the lane's current key, which names the newer conversation, so a Stop, an agent lookup
+or a turn through it would act on that conversation. One gateway owns every session and never
+crosses them, so writes are refused by name (``session_superseded``) rather than given a private
+route that no adapter would deliver to (#140183 review F2). A compression continuation is never
+mounted under its physical id: the lane's turns publish to the lineage root (``logical_owner``), so
+``session.resume`` maps it to that root, as normal admission does. Rows without a ``session_key`` (subagents) have nothing to rebuild a lane
 from. LOCAL rows (cron, CLI, TUI) restore from their private receipt and API rows from their binding
 (``restore_api_session``); neither is ever mounted here. The mount semantics are the first inbound's
 ``_publish_route``: a row that ended for a recoverable reason (``agent_close``, ``ws_orphan_reap``,
@@ -63,6 +69,14 @@ def _row_source(row, platform, profile):
     return source
 
 
+def _names_another_conversation(authority, sid, route):
+    """The lane's current routing entry names a conversation other than *sid* (``/new``, reset)."""
+    store = authority.runner.session_store
+    with store._lock:
+        entry = store._entry_locked(route)
+    return entry is not None and authority.logical_owner(entry.session_id) != sid
+
+
 def _row_entry(row, platform, route, profile):
     """A routing entry rebuilt from the row alone (the route no longer vouches for it)."""
     from gateway.session import SessionEntry
@@ -83,19 +97,32 @@ def restore_routed_session(authority, sid):
     row = authority.db.get_session(sid)
     route = row.get('session_key') if row is not None else None
     platform = _messaging_platform(row['source']) if route else None
-    if platform is None:
-        return None
+    if platform is None or authority.logical_owner(sid) != sid:
+        return None  # a compression continuation: its lineage root is the mount (session.resume maps it)
     store = authority.runner.session_store
     registry = getattr(authority.runner, 'session_authorities', None)
     profile = store._named_profile_for_key(route)
     if profile != (registry.profile_name(authority) if registry is not None else None):
         return None
-    entry = _routed_entry(authority, sid, route) or _row_entry(row, platform, route, profile)
+    routed = _routed_entry(authority, sid, route)
+    entry = routed or _row_entry(row, platform, route, profile)
     if entry is None:
         return None
+    superseded = routed is None and _names_another_conversation(authority, sid, route)
     authority._require_admission_open()
     # Re-pin the persisted receiving bot so delivery after a restart goes through it or nowhere,
     # never the default bot by heuristic (gateway/AGENTS.md § Identity survives the process).
     source = authority.runner._restored_source(entry)
-    authority.sessions.setdefault(sid, LiveSession(source, route))
+    authority.sessions.setdefault(sid, LiveSession(source, route, superseded=superseded, history_mount=True))
     return SessionRef(authority.profile_id, sid)
+
+
+def messaging_owner_ref(authority, ref):
+    """The mount a messaging row resumes on: its compression lineage root, where the lane's turns
+    publish their events (``logical_owner``). Any other row keeps *ref*."""
+    row = authority.db.get_session(ref.session_id)
+    if row is None or not row.get('session_key') or _messaging_platform(row['source']) is None:
+        return ref
+    from gateway.session_contract import SessionRef
+    owner = authority.logical_owner(ref.session_id)
+    return ref if owner == ref.session_id else SessionRef(ref.profile_id, owner)

gateway/session_viewer_close.py

@@ -54,6 +54,8 @@ async def close_viewer_session(authority, session_id, reason):
     live = authority.sessions[session_id]
     if live.subscribers:
         return  # a viewer attached while the detach was off the loop: nothing closes or ends
+    if live.superseded:
+        return  # a read-only historic mount: its route's delegations and agent are the newer session's
     from tools.async_delegation import interrupt_for_session, refuse_for_session
     route = live.route
     parents = tuple(dict.fromkeys((session_id, _physical_target(authority, session_id))))

tests/gateway/test_authority_cold_route_recovery.py

@@ -257,3 +257,140 @@ async def test_a_draining_gateway_mounts_nothing(tmp_path):
     with pytest.raises(RuntimeStoreError, match='runtime_draining'):
         cold.authority.authorize(cold.actor, _ref('lane-drain'), 'session:read')
     assert 'lane-drain' not in cold.authority.sessions
+
+
+async def _superseded(tmp_path):
+    """``old`` was the lane's conversation; ``/new`` moved the lane on to a successor."""
+    _seed(tmp_path, 'old', _source(chat_name='from-row'))
+    cold = await _cold(tmp_path)
+    successor = cold.store.get_or_create_session(_source(), force_new=True)
+    assert successor.session_id != 'old'
+    return cold, successor
+
+
+@pytest.mark.asyncio
+async def test_a_row_its_route_moved_past_is_read_only(tmp_path):
+    """#140183 review F2: the old row's only route is the lane's CURRENT key, so a Stop, an agent
+    lookup or a turn on it reached the successor conversation. One gateway owns every session and
+    never crosses them: the historic row reads (its own transcript) and refuses every write by name."""
+    from hermes_state_runtime import RuntimeStoreError
+    from gateway.session_contract import Principal, Submission
+    cold, _successor = await _superseded(tmp_path)
+    cold.authority.authorize(cold.actor, _ref('old'), 'session:read')
+    caps = frozenset({'session:read', 'session:submit', 'session:control', 'session:approve',
+                      'session:respond'})
+    writer = Principal('viewer', 'fixture', caps, 'socket')
+    for capability in sorted(caps - {'session:read'}):
+        with pytest.raises(RuntimeStoreError, match='session_superseded'):
+            cold.authority.authorize(writer, _ref('old'), capability)
+    # The successor's running agent sits under the shared route; the old row never resolves it.
+    route = cold.authority.sessions['old'].route
+    cold.authority.runner._running_agents = {route: object()}
+    cold.authority.runner._cached_agent_for = lambda key: cold.authority.runner._running_agents.get(key)
+    assert cold.authority.agent(_ref('old')) is None
+    assert cold.authority._runs_own_agent(_ref('old')) is False
+    with pytest.raises(RuntimeStoreError, match='session_superseded'):
+        await cold.authority.interrupt(writer, _ref('old'), 0)
+    with pytest.raises(RuntimeStoreError, match='session_superseded'):
+        await cold.authority.submit(writer, Submission('r1', _ref('old'), {'text': 'hi'}, 'queue'))
+    assert cold.authority.db.get_session('old')['message_count'] == 0
+
+
+@pytest.mark.asyncio
+async def test_closing_or_deleting_a_superseded_row_leaves_the_successor_route_alone(tmp_path, monkeypatch):
+    """Viewer close interrupts/refuses delegations and evicts by route; delete evicts by route. On a
+    superseded mount that route is the successor's, so neither touches it."""
+    import tools.async_delegation as delegation
+    from gateway.session_mutations import retire_live_sessions
+    from gateway.session_viewer_close import close_viewer_session
+    cold, _successor = await _superseded(tmp_path)
+    cold.authority.authorize(cold.actor, _ref('old'), 'session:read')
+    touched = []
+    monkeypatch.setattr(delegation, 'interrupt_for_session', lambda **kw: touched.append(('interrupt', kw)))
+    monkeypatch.setattr(delegation, 'refuse_for_session', lambda *a, **kw: touched.append(('refuse', kw)))
+    cold.authority.runner._evict_cached_agent = lambda key, **kw: touched.append(('evict', key))
+    await close_viewer_session(cold.authority, 'old', 'tui_close')
+    retire_live_sessions(cold.authority, ['old'])
+    assert touched == []
+    assert 'old' not in cold.authority.sessions
+
+
+@pytest.mark.asyncio
+async def test_the_lane_reopening_the_row_makes_it_live_again(tmp_path):
+    """A read-only mount is only about the lane naming another conversation: once the lane's own
+    inbound publishes this row again (recovery reopened it), it is the live conversation."""
+    cold, _successor = await _superseded(tmp_path)
+    cold.authority.authorize(cold.actor, _ref('old'), 'session:read')
+    live = cold.authority.sessions['old']
+    cold.authority._publish_route(live.source, 'old', live.route)
+    assert cold.authority.sessions['old'] is live and live.superseded is False
+
+
+@pytest.mark.asyncio
+async def test_a_row_whose_lane_has_no_entry_mounts_live(tmp_path):
+    """Nothing names another conversation: the row is what the lane's next inbound publishes."""
+    from gateway.session_contract import Principal
+    _seed(tmp_path, 'unrouted', _source(), routed=False)
+    cold = await _cold(tmp_path)
+    writer = Principal('viewer', 'fixture', frozenset({'session:read', 'session:submit'}), 'socket')
+    cold.authority.authorize(writer, _ref('unrouted'), 'session:submit')
+    assert cold.authority.sessions['unrouted'].superseded is False
+
+
+def _compressed_lane(tmp_path):
+    """A messaging conversation compressed once: root ``lane-root`` ended ``compression``, its
+    continuation ``lane-tip`` copies the route columns; the lane's entry names the tip."""
+    route = _seed(tmp_path, 'lane-root', _source(), routed=False)
+    runner = _Runner(tmp_path)
+    db = runner._session_db
+    source = _source()
+    db.create_session('lane-tip', source='telegram', session_key=route, chat_id=source.chat_id,
+                      user_id=source.user_id, chat_type='dm', origin_json=json.dumps(source.to_dict()),
+                      parent_session_id='lane-root')
+    db.end_session('lane-root', 'compression')
+    from gateway.session import SessionEntry
+    now = datetime.now()
+    entry = SessionEntry(route, 'lane-tip', now, now, origin=source, platform=source.platform)
+    runner.session_store._routing_db.save_gateway_routing_entry(
+        route, json.dumps(entry.to_dict()), scope=runner.session_store._routing_scope())
+    return route
+
+
+@pytest.mark.asyncio
+async def test_a_compression_child_is_never_mounted_under_its_physical_id(tmp_path):
+    """Greptile on #140183: the lane's turns publish to the lineage root (``logical_owner``), so a
+    viewer mounted on the child missed every event. The physical id is not a mount of its own."""
+    _compressed_lane(tmp_path)
+    await _refused(tmp_path, 'lane-tip')
+
+
+@pytest.mark.asyncio
+async def test_resume_of_a_compression_child_attaches_to_its_logical_owner(tmp_path):
+    from gateway.session_controls import AuthorityConnection
+    _compressed_lane(tmp_path)
+    cold = await _cold(tmp_path)
+    connection = AuthorityConnection(cold.authority, object(), {
+        'provider': 'local', 'user_id': 'viewer', 'profile_id': 'fixture', 'instance_id': 'cold',
+        'capabilities': ['session:read']})
+    resumed = await connection.dispatch({'id': 1, 'method': 'session.resume',
+                                         'params': {'session_id': 'lane-tip'}})
+    assert resumed.get('result', {}).get('session_id') == 'lane-root', resumed
+    assert 'lane-tip' not in cold.authority.sessions
+    assert cold.authority.sessions['lane-root'].superseded is False
+
+
+@pytest.mark.asyncio
+async def test_a_live_mount_goes_read_only_when_its_lane_publishes_another_conversation(tmp_path):
+    """Mounted while nothing named another conversation, then the lane's next inbound starts a fresh
+    session on the same route: from then on the old mount's route is the newer session's."""
+    from hermes_state_runtime import RuntimeStoreError
+    from gateway.session_contract import Principal
+    _seed(tmp_path, 'unrouted', _source(), routed=False)
+    cold = await _cold(tmp_path)
+    writer = Principal('viewer', 'fixture', frozenset({'session:read', 'session:submit'}), 'socket')
+    cold.authority.authorize(writer, _ref('unrouted'), 'session:submit')
+    fresh = cold.store.get_or_create_session(_source(), force_new=True)
+    cold.authority._publish_route(_source(), fresh.session_id, fresh.session_key)
+    with pytest.raises(RuntimeStoreError, match='session_superseded'):
+        cold.authority.authorize(writer, _ref('unrouted'), 'session:submit')
+    cold.authority.authorize(writer, _ref(fresh.session_id), 'session:submit')