fix(gateway): a turn finishing during /new no longer hands the chat back to its old conversation

SessionStore publishes a route transition in two halves: the new entry under _lock, then the
state.db row (insert for /new and a fresh route, reopen for /resume) after releasing it. Anything
that resolved the route in between saw the named conversation's row missing or still ended and
took the #54878 stale-route self-heal: drop the entry and recover the newest open row, which is
the conversation the user just left. A Desktop/Webapp turn finishing on a chat's previous
conversation does exactly that (its post-turn bookkeeping calls get_or_create_session), so after
/new the chat's route named A again and its next message could land there.

The store now remembers the ids it published whose row is not live yet, and
_is_session_ended_in_db does not call those ended. Every reader of that check (the routing
self-heal, the reaped-run eviction, the cached-agent liveness check) gets the fix.

gateway/session.py

@@ -805,6 +805,9 @@ class SessionStore(
         self._fast_persisted_entries: dict[str, tuple[int, str]] = {}
         self._inflight_lock = threading.Lock()
         self._inflight_sessions: dict[str, _SessionFlight] = {}
+        # Ids a route names before their state.db row is live: a route transition publishes under
+        # _lock and inserts/reopens the row after releasing it. Guarded by _lock.
+        self._rows_pending: set[str] = set()
         # An unscoped legacy Slack key is claimed once per process (two workspaces must not both
         # revive one session).
         self._legacy_slack_claim_lock = threading.Lock()
@@ -1058,6 +1061,7 @@ class SessionStore(
             current = self._entries.get(session_key)
             if current is None or (force_new and current is observed):
                 self._entries[session_key] = current = candidate
+                self._rows_pending.add(session_id)
         decision.entry = current
         decision.needs_save = True
         if current is not candidate:
@@ -1173,6 +1177,7 @@ class SessionStore(
             transport_profile=old_entry.transport_profile, **fields,
         )
         self._entries[session_key] = new_entry
+        self._rows_pending.add(session_id)
         self._save()
         return new_entry
 
@@ -1287,23 +1292,26 @@ class SessionStore(
                 ),
             )
 
-        if self._db_for_key(session_key) and old_entry.session_id:
-            self._promote_session_reset(
-                session_key, old_entry.session_id, "session_switch",
-                log=lambda e: logger.debug("Session DB end_session failed: %s", e),
-            )
-        if self._db_for_key(session_key):
-            # An explicit /resume/handoff/branch repoint is real user activity: the row is
-            # reopened (a repoint onto a finalized row must resume it, not write into a dead row).
-            # Mount-time reads (TUI session.resume) no longer reopen (#85303) — only this path does.
-            self._reopen_session_row(
-                session_key, target_session_id, log_prefix="Session DB reopen_session failed"
-            )
-            self._record_gateway_session_peer(
-                target_session_id, session_key, new_entry.origin,
-                display_name=new_entry.display_name, include_compression_ancestors=True,
-                transport_profile=new_entry.transport_profile,
-            )
+        try:
+            if self._db_for_key(session_key) and old_entry.session_id:
+                self._promote_session_reset(
+                    session_key, old_entry.session_id, "session_switch",
+                    log=lambda e: logger.debug("Session DB end_session failed: %s", e),
+                )
+            if self._db_for_key(session_key):
+                # An explicit /resume/handoff/branch repoint is real user activity: the row is
+                # reopened (a repoint onto a finalized row must resume it, not write into a dead row).
+                # Mount-time reads (TUI session.resume) no longer reopen (#85303) — only this path does.
+                self._reopen_session_row(
+                    session_key, target_session_id, log_prefix="Session DB reopen_session failed"
+                )
+                self._record_gateway_session_peer(
+                    target_session_id, session_key, new_entry.origin,
+                    display_name=new_entry.display_name, include_compression_ancestors=True,
+                    transport_profile=new_entry.transport_profile,
+                )
+        finally:
+            self._row_settled(target_session_id)
         return new_entry
 
     def list_sessions(self, active_minutes: Optional[int] = None) -> list[SessionEntry]:

gateway/session_lifecycle.py

@@ -80,6 +80,12 @@ class SessionLifecycleMixin:
         after the self-heal re-homes the key, the id has no owner and would resolve to the launch
         store, which never holds a routed profile's rows (#118862).
         """
+        with self._lock:
+            if session_id in self._rows_pending:
+                # A route transition published this id and has not inserted/reopened its row yet
+                # (a /new racing a turn that resolves the route as it finishes): healing it here
+                # would hand the route back to the conversation the user just left.
+                return False
         db = self._db_for_key(session_key) if session_key else self._db_for_session_id(session_id)
         if not db or not session_id:
             return False

gateway/session_recovery.py

@@ -389,26 +389,36 @@ class SessionRecoveryMixin:
         explicit reset boundary (with the specific reason so state.db is auditable, e.g.
         ``suspended`` vs plain ``session_reset``), then INSERT the new row + routing
         peer. Both best-effort: failures are warned and self-healed by the next peer refresh."""
-        if end_session_id:
-            from hermes_cli.observability.relay_shared_metrics import close_session_run
+        try:
+            if end_session_id:
+                from hermes_cli.observability.relay_shared_metrics import close_session_run
 
-            close_session_run(end_session_id)
-        if self._db_for_key(session_key) and end_session_id:
-            self._promote_session_reset(
-                session_key, end_session_id, end_reason,
-                log=lambda e: logger.warning(
-                    "Failed to end predecessor session row %s for %s%s: %s — the old row remains "
-                    "open and may win restart recovery until the next successful peer refresh",
-                    end_session_id, session_key, during, e),
-            )
-        if self._db_for_key(session_key) and create_kwargs:
-            self._create_session_row(
-                session_key, create_kwargs, origin, display_name,
-                log=lambda e: logger.warning(
-                    "Failed to create session row %s for %s%s: %s — deferring to the "
-                    "self-healing peer refresh on the next turn",
-                    create_kwargs.get("session_id"), session_key, during, e),
-            )
+                close_session_run(end_session_id)
+            if self._db_for_key(session_key) and end_session_id:
+                self._promote_session_reset(
+                    session_key, end_session_id, end_reason,
+                    log=lambda e: logger.warning(
+                        "Failed to end predecessor session row %s for %s%s: %s — the old row remains "
+                        "open and may win restart recovery until the next successful peer refresh",
+                        end_session_id, session_key, during, e),
+                )
+            if self._db_for_key(session_key) and create_kwargs:
+                self._create_session_row(
+                    session_key, create_kwargs, origin, display_name,
+                    log=lambda e: logger.warning(
+                        "Failed to create session row %s for %s%s: %s — deferring to the "
+                        "self-healing peer refresh on the next turn",
+                        create_kwargs.get("session_id"), session_key, during, e),
+                )
+        finally:
+            if create_kwargs:
+                self._row_settled(create_kwargs["session_id"])
+
+    def _row_settled(self, session_id: str) -> None:
+        """The route transition that published *session_id* finished its state.db half: from here
+        a missing or ended row is real again (see ``_is_session_ended_in_db``)."""
+        with self._lock:
+            self._rows_pending.discard(session_id)
 
     @staticmethod
     def _session_create_kwargs(

tests/gateway/test_route_transition_racing_turn.py

@@ -0,0 +1,137 @@
+"""A chat's /new (or /resume) wins over a turn that finishes on its previous conversation.
+
+A Desktop turn runs on a Telegram chat's conversation A when the platform user sends ``/new``. The
+reset publishes the chat's new conversation B on its route, then writes state.db (A ended, B's row
+inserted). The finishing turn resolves the same route for its post-turn bookkeeping; if that lands
+between the two halves, B has no row yet, and treating it as deleted hands the route back to A, so
+the chat's next message continues the conversation the user just left. ``/resume`` has the same
+window: the resumed row is reopened only after the route names it.
+
+Real GatewayRunner, SessionAuthority, SessionStore and SessionDB; the commands run through their own
+handlers. The window is held open at the real SQL statement that ends the predecessor row.
+"""
+import asyncio
+import threading
+import time
+import weakref
+from pathlib import Path
+
+import pytest
+
+from gateway.config import GatewayConfig, Platform
+from gateway.platforms.event import MessageEvent, MessageType
+from gateway.run import GatewayRunner
+from gateway.run_runtime import initialize_gateway_runtime
+from gateway.runtime_ownership import process_ownership
+from gateway.session import SessionSource
+from gateway.session_ingress_context import register_transport_home
+
+
+class _Adapter:
+    pass
+
+
+@pytest.fixture
+def served(tmp_path, monkeypatch):
+    root = tmp_path / '.hermes'
+    root.mkdir(mode=0o700)
+    monkeypatch.setattr(Path, 'home', lambda: tmp_path)
+    monkeypatch.setenv('HERMES_HOME', str(root))
+    import hermes_state
+    monkeypatch.setattr(hermes_state, 'DEFAULT_DB_PATH', hermes_state._IMPORT_DEFAULT_DB_PATH)
+    config = GatewayConfig()
+    config._runtime_profile_homes = (('default', root),)
+    runner = GatewayRunner(config)
+    process_ownership.reserve([root])
+    yield runner, root
+    for served_authority in list(getattr(runner, 'session_authorities', []) or []):
+        served_authority.db.close()
+    process_ownership.release(root)
+
+
+async def _start(runner, root):
+    await initialize_gateway_runtime(runner)
+    register_transport_home(runner, None, root)
+    adapter = _Adapter()
+    runner.adapters = {Platform.TELEGRAM: adapter}
+    runner._is_user_authorized_for_source = lambda *a, **k: True
+    from gateway.session_authorities import active_authority
+    return runner, active_authority(runner), adapter
+
+
+def _chat(adapter):
+    source = SessionSource(platform=Platform.TELEGRAM, chat_id='4242', chat_type='dm', user_id='u1',
+                           user_name='probe')
+    source._transport_adapter_ref = weakref.ref(adapter)
+    return source
+
+
+def _command(runner, source, text):
+    event = MessageEvent(text=text, message_type=MessageType.TEXT, source=source, message_id='m-' + text)
+    handler = runner._handle_reset_command if text.startswith('/new') else runner._handle_resume_command
+    return handler(event)
+
+
+class _HoldPredecessorEnd:
+    """Parks the route transition at the SQL that ends the route's previous row: the route already
+    names the new conversation, and nothing of it is in state.db yet."""
+
+    def __init__(self, db):
+        self.conn, self.reached, self.release = db._conn, threading.Event(), threading.Event()
+
+    def _trace(self, sql):
+        if 'SET ended_at' in sql and not self.reached.is_set():
+            self.reached.set()
+            self.release.wait(30)
+
+    def __enter__(self):
+        self.conn.set_trace_callback(self._trace)
+        return self
+
+    def __exit__(self, *exc):
+        self.release.set()
+        self.conn.set_trace_callback(None)
+
+
+async def _until(predicate, what):
+    deadline = time.monotonic() + 10
+    while not predicate():
+        assert time.monotonic() < deadline, what
+        await asyncio.sleep(0.01)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('command', ['/new', '/resume First chat'])
+async def test_a_turn_finishing_mid_transition_leaves_the_chat_on_the_conversation_it_named(served, command):
+    runner, authority, adapter = await _start(*served)
+    store, chat = runner.session_store, _chat(adapter)
+    key = runner._session_key_for_source(chat)
+    a = authority.register(chat).session_id  # the chat's first message
+    authority.db.set_session_title(a, 'First chat')
+    if command.startswith('/resume'):
+        await _command(runner, chat, '/new')
+        authority.register(chat)  # the chat talks in its second conversation, then resumes the first
+    before = store.peek_session_id(key)
+
+    with _HoldPredecessorEnd(authority.db) as hold:
+        transition = asyncio.create_task(_command(runner, chat, command))
+        await _until(hold.reached.is_set, 'the transition reaches state.db')
+        named = store.lookup_by_session_key(key)
+        assert named.session_id not in (None, before)
+        stamp = named.updated_at
+        # The Desktop turn on the previous conversation returns: its post-turn bookkeeping resolves
+        # the chat's route while the transition is still writing state.db.
+        finishing = asyncio.create_task(runner._run_post_turn_hooks(
+            agent_result={'final_response': 'the answer on the old conversation'}, source=chat,
+            is_internal=False))
+        await _until(lambda: store.peek_session_id(key) != named.session_id or named.updated_at != stamp,
+                     'the finishing turn resolved the route')
+        hold.release.set()
+        await transition
+        await finishing
+
+    label = {named.session_id: 'the conversation the command named', before: 'the one it left'}.get
+    assert label(store.peek_session_id(key), 'a fresh one') == 'the conversation the command named'
+    # Where the chat's next message lands.
+    assert label(authority.register(chat).session_id, 'a fresh one') == 'the conversation the command named'
+    assert authority.db.get_session(before)['end_reason'] is not None