diff --git a/gateway/session.py b/gateway/session.py index fbc9a4e738..216c09ab89 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -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]: diff --git a/gateway/session_authority.py b/gateway/session_authority.py index 3077cd64fc..df00fe2718 100644 --- a/gateway/session_authority.py +++ b/gateway/session_authority.py @@ -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 diff --git a/gateway/session_automation.py b/gateway/session_automation.py index f47a6b7b4f..93d94fcce6 100644 --- a/gateway/session_automation.py +++ b/gateway/session_automation.py @@ -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 diff --git a/gateway/session_detached.py b/gateway/session_detached.py index c422bb8dc6..b67a8491d2 100644 --- a/gateway/session_detached.py +++ b/gateway/session_detached.py @@ -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 diff --git a/gateway/session_recovery.py b/gateway/session_recovery.py index 005f1cb887..f358039bbb 100644 --- a/gateway/session_recovery.py +++ b/gateway/session_recovery.py @@ -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``).""" diff --git a/gateway/slash_commands_session.py b/gateway/slash_commands_session.py index 33d0e6b830..02c2f7c6e7 100644 --- a/gateway/slash_commands_session.py +++ b/gateway/slash_commands_session.py @@ -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"]) diff --git a/tests/gateway/test_detached_conversation_lifecycle.py b/tests/gateway/test_detached_conversation_lifecycle.py new file mode 100644 index 0000000000..e97906cbc8 --- /dev/null +++ b/tests/gateway/test_detached_conversation_lifecycle.py @@ -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:]}'