fix(gateway): every route change moves the chat's old conversation off the chat, the context-exhaustion auto-reset included

Review of #144840 (F1): /new, /resume and /branch detached the conversation a chat moved past, but
the compression-exhaustion auto-reset (reset_and_rehome_messages) and a CLI /handoff rotated the
route without it, so the app's next prompt on the old conversation ran in the new one and its reply
reached the chat.

The session store now announces each conversation a route comes to name (reset_session,
switch_session, a get_or_create transition: SessionStore.route_named), and the owner wires that once
(session_detached.watch_routes) to the owning authority's name_lane, on its loop, ahead of the
writer's own resumption. The per-command _name_lane calls are gone. The exhaustion reset's fresh
session takes the chat's source as it was, not the old conversation's detached one.
This commit is contained in:
Teknium committed 2026-10-27 17:04:02 -07:00
1 parent 226c9bffd8
commit 58eeea143e
7 files changed
+115 -15

No files matched your search

+7
View File
@@ -851,6 +851,9 @@ class SessionStore(
# session_id -> owning key for ids proven but not yet published in ``_entries`` (a
# compression child row is written before its reroute is published).
self._session_owner_hints: dict[str, str] = {}
# Called with (key, session id, origin) once a route names another conversation; the owner
# wires it so a conversation its chat moved past detaches at once (gateway/session_detached.py).
self.route_named = None
from gateway.session_db_recovery import RecoverableHandleCache
self._db_handle_cache = RecoverableHandleCache(
@@ -996,6 +999,8 @@ class SessionStore(
finally:
if create_kwargs:
self._row_settled(create_kwargs["session_id"])
if observed is None or decision.entry.session_id != observed.session_id:
self._announce_route(session_key, decision.entry)
return decision.entry
def _apply_route_checks(
@@ -1183,6 +1188,7 @@ class SessionStore(
)
finally:
self._row_settled(session_id)
self._announce_route(session_key, new_entry)
return new_entry
def _replace_route_locked(self, session_key, old_entry, session_id, now, **fields) -> SessionEntry:
@@ -1328,6 +1334,7 @@ class SessionStore(
)
finally:
self._row_settled(target_session_id)
self._announce_route(session_key, new_entry)
return new_entry
def list_sessions(self, active_minutes: Optional[int] = None) -> list[SessionEntry]:
+2
View File
@@ -1488,6 +1488,8 @@ async def initialize_session_authority(runner, *, profile_id, instance_id, db=No
if epochs is None:
epochs = store._local_authority_epochs = {}
epochs[db_key] = epoch
from gateway.session_detached import watch_routes
watch_routes(runner, store)
from gateway.session_local_recovery import recover_local_sessions
recover_local_sessions(authority)
return authority
+9 -3
View File
@@ -539,6 +539,7 @@ async def move_route_wakes(runner, session_key, old_session_id):
from functools import partial
from gateway.session_authority import LiveSession
from gateway.session_contract import Principal
from gateway.session_detached import lane_source
from gateway.session_runtime_workers import tracked_write
from hermes_state_runtime import list_session_admissions
old = SessionRef(authority.profile_id, authority.logical_owner(old_session_id))
@@ -561,7 +562,9 @@ async def move_route_wakes(runner, session_key, old_session_id):
continue # claimed meanwhile: it runs where it is
payload = deepcopy(row['payload'])
payload['native_text_v1']['automation']['owner'] = new.session_id
authority.sessions.setdefault(new.session_id, LiveSession(authority.sessions[old.session_id].source, session_key))
# The old conversation left the route at its rotation (gateway/session_detached.py).
authority.sessions.setdefault(new.session_id, LiveSession(lane_source(authority.sessions[old.session_id].source),
session_key))
await tracked_write(authority, partial(
admit_session_input, authority.db, epoch=authority.epoch, principal_id=row['principal_id'],
session_id=new.session_id, request_id=row['request_id'], payload=payload,
@@ -632,6 +635,9 @@ async def reset_and_rehome_messages(runner, session_key, old_session_id):
from gateway.session_runtime_workers import tracked_write
from hermes_state_runtime import rehome_queued_inputs
from gateway.session_detached import lane_source
chat_source = lane_source(live.source) # the rotation detaches the old conversation off the route
def rotate_and_move():
entry = runner.session_store.reset_session(session_key)
if entry is None:
@@ -640,7 +646,7 @@ async def reset_and_rehome_messages(runner, session_key, old_session_id):
if new == old:
return entry, None, []
if authority.db.get_session(new.session_id) is None:
authority.db.create_session(new.session_id, source=live.source.platform.value)
authority.db.create_session(new.session_id, source=chat_source.platform.value)
# An admission that reserved the old route before this rotation commits after it.
authority.rotated_owners[old.session_id] = session_key
return entry, new, rehome_queued_inputs(
@@ -652,7 +658,7 @@ async def reset_and_rehome_messages(runner, session_key, old_session_id):
new = result[1]
if new is None:
return result, None
fresh = authority.sessions.setdefault(new.session_id, LiveSession(live.source, session_key))
fresh = authority.sessions.setdefault(new.session_id, LiveSession(chat_source, session_key))
await fresh.claim_gate.acquire()
return result, fresh
+51
View File
@@ -19,9 +19,14 @@ replies reach its viewers only, never the chat, which now gets only its new conv
The key carries no platform slot, so a reader that parses a chat out of a session key
(``gateway.run._parse_session_key``) finds none and delivers nowhere. The chat naming the
conversation again (``/resume``) moves it back onto the chat's route (``SessionAuthority.name_lane``).
Every route change reaches ``name_lane`` from one place: the session store announces each conversation a
route comes to name (``SessionStore.route_named``: ``/new``, ``/resume``, ``/branch``, ``/handoff``,
the context-exhaustion auto-reset, a reset policy, a topic binding heal), wired by ``watch_routes``.
"""
from __future__ import annotations
import asyncio
from datetime import datetime, timedelta
_SLOT = 'detached'
@@ -57,6 +62,52 @@ def detach(authority, live, session_id) -> None:
live.source = replace_source(live.source, detached_route=live.route)
def lane_source(source):
"""*source* as its chat's own, a detached copy's pin dropped: what a conversation that takes the
chat's route over starts from (the route moved first, so the one it replaced is detached by now)."""
if detached_route_of(source) is None:
return source
from gateway.session_identity import replace_source
return replace_source(source, detached_route=None)
def _route_authority(runner, store, route):
"""The authority owning *route*: its key names the profile, whatever scope the writer runs in."""
from gateway.session_authorities import authority_for_home
from hermes_constants import hermes_home_key
registry = getattr(runner, 'session_authorities', None)
if store._named_profile_for_key(route) is None: # the launch profile's own key
if registry is None:
return getattr(runner, 'session_authority', None)
return next((a for a in registry.serving() if hermes_home_key(a.profile_id) == registry.launch_key), None)
home = store._profile_home_for_key(route)
return authority_for_home(runner, home) if home is not None else None
def watch_routes(runner, store) -> None:
"""Wire *store*'s route announcements to the owning authority's ``name_lane``. A store write off
the owner loop (a worker thread) hands the move to the loop, ahead of the writer's own resumption,
so nothing on the loop sees the new route before the conversation it replaced has left it."""
loop = asyncio.get_running_loop()
def route_named(route, session_id, origin):
authority = _route_authority(runner, store, route)
if authority is None or origin is None:
return
def name():
authority.name_lane(route, authority.logical_owner(session_id), origin)
try:
on_loop = asyncio.get_running_loop() is loop
except RuntimeError:
on_loop = False
if on_loop:
name()
elif not loop.is_closed():
loop.call_soon_threadsafe(name)
store.route_named = route_named
def recover_detached_sessions(authority, pending):
"""Boot: schedule the queued input of conversations their chats moved past (an app prompt
accepted during a restart's after-turn wait). The routing index names only each chat's current
+7
View File
@@ -412,6 +412,13 @@ class SessionRecoveryMixin:
create_kwargs.get("session_id"), session_key, during, e),
)
def _announce_route(self, session_key: str, entry) -> None:
"""*session_key* now names *entry*'s conversation: tell the owner (``route_named``), so a
live conversation still on that route leaves it before anything else can run there."""
listener = getattr(self, "route_named", None)
if listener is not None and entry is not None:
listener(session_key, entry.session_id, entry.origin)
def _row_settled(self, session_id: str) -> None:
"""The route transition that published *session_id* returned (its state.db half done or
failed): from here a missing or ended row is real again (see ``_is_session_ended_in_db``)."""
-12
View File
@@ -176,14 +176,6 @@ class GatewaySessionCommandsMixin:
await self.hooks.emit("session:end", dict(hook_payload))
await self.hooks.emit("session:reset", dict(hook_payload))
def _name_lane(self, source, session_key: str, session_id: str) -> None:
"""The chat now names another conversation: a view still open on its previous one detaches
at once, not at the chat's next message (``SessionAuthority.name_lane``)."""
from gateway.session_authorities import active_authority
authority = active_authority(self)
if authority is not None:
authority.name_lane(session_key, authority.logical_owner(session_id), source)
async def _handle_discard_command(self, event: MessageEvent) -> str:
"""``/discard``: drop this chat's turn lost in a gateway restart so its queue runs again."""
from gateway.session_authorities import active_authority
@@ -235,8 +227,6 @@ class GatewaySessionCommandsMixin:
async with route_owner_lock(self, _old_sid):
new_entry = await self.async_session_store.reset_session(session_key)
await move_route_wakes(self, session_key, _old_sid) # main #114456: a parked wake survives /new
if new_entry:
self._name_lane(source, session_key, new_entry.session_id)
await self._fire_session_reset_hooks(source, session_key, _old_sid,
new_entry.session_id if new_entry else None)
# Scoped to the profile serving this source so a multiplexed /new banner reports the
@@ -1106,7 +1096,6 @@ class GatewaySessionCommandsMixin:
return t("gateway.resume.switch_failed")
if local_owner is not None and authority is not None: # a Desktop still viewing it follows it
follow_hand_off(authority, source, session_key, authority.logical_owner(target_id), local_owner)
self._name_lane(source, session_key, target_id)
# Conversation boundary: all conversation-scoped state + security state in one funnel call.
# Conversation boundary: clear ALL conversation-scoped per-session state (model/reasoning overrides
# #10702, one-turn restores, model notes, last-resolved cache #58403, /queue overflow) + security
@@ -1287,7 +1276,6 @@ class GatewaySessionCommandsMixin:
new_entry = await self.async_session_store.switch_session(dest_key, new_session_id)
if not new_entry:
return t("gateway.branch.switch_failed")
self._name_lane(dest_source, dest_key, new_session_id)
self._clear_session_boundary_security_state(dest_key)
self._evict_cached_agent(dest_key)
msg_count = len([m for m in history if m.get("role") == "user"])
@@ -0,0 +1,39 @@
"""The lifecycle of a conversation its chat moved past (review of #144840 at 03d04918dde).
Teknium's ruling: once a messaging chat names another conversation, nothing the old one produces
reaches the chat, and the old one stays fully usable from the apps. Real GatewayRunner, DiscordAdapter
(``send`` recorded), SessionAuthority, drain and SessionStore over a real SessionDB; the agent run is
the stand-in of ``test_authority_lane_moves_on``.
"""
import pytest
from tests.gateway.test_authority_lane_moves_on import _app_turn, _chat, _gateway
from tests.waiting import async_wait_until
@pytest.mark.asyncio
async def test_the_chats_context_exhaustion_reset_moves_the_old_conversation_off_the_chat(monkeypatch):
"""F1: the auto-reset after context exhaustion rotates the chat to B without a /new. The app's
next prompt on A runs in A, and the chat gets nothing of it."""
from gateway.session_results import execution_result
runner, adapter, authority, source, sent, ran, _gates = await _gateway(monkeypatch)
lane = runner.session_store._generate_session_key(source)
await _chat(adapter, source, sent, 'hello')
ref_a = authority.register(source)
prepare = runner._hmwa_prepare_turn
async def exhausting(event, source, session_entry, session_key, *args):
if event.text != 'too big':
return await prepare(event, source, session_entry, session_key, *args)
execution_result.get()['result'] = {'final_response': 'Context too large', 'messages': []}
entry = await runner._hmwa_rotate_exhausted_session(session_entry, session_key, source)
return 'Context too large', entry
monkeypatch.setattr(runner, '_hmwa_prepare_turn', exhausting)
await _chat(adapter, source, sent, 'too big')
await async_wait_until(lambda: runner.session_store.peek_session_id(lane) != ref_a.session_id,
what='the route to rotate')
chat_replies = len(sent)
assert await _app_turn(authority, ref_a, 'app prompt on A') == ('terminal', 'completed')
assert ran[-1][:2] == ('app prompt on A', ref_a.session_id) and ran[-1][2] != lane
assert sent[chat_replies:] == [], f'the chat got the old conversation\'s reply: {sent[chat_replies:]}'