fix(gateway): default-profile rows and /undo eviction stay on the profile that owns them

Two multiplex key/store mismatches in the gateway session layer:

- SessionStore._db_for_key resolved legacy agent:main keys to the AMBIENT
  store. Named-profile keys were already pinned to profiles/<x>/state.db
  (#97309), but a default-profile session touched inside a secondary's
  runtime scope (a coalesced async-delegation drain, a cron mirror into a
  default chat, a scoped watcher) was written into the secondary's state.db
  — the profile_name='<A>' row inside B's store reported in #102157, after
  which the routing index and row disagree and the #54878 self-heal drops a
  live conversation. Default keys now resolve to _routing_home/state.db, the
  launch home already pinned for the routing index. Single-profile gateways
  (multiplex off) keep the ambient store byte-for-byte.

- /undo evicted the cached agent under build_session_key(source) —
  agent:main:… — while the cache is keyed by the profile namespace, so on a
  secondary profile the eviction missed and the next turn reused an agent
  whose in-memory history still held the undone turns. Use
  _session_key_for_source like /reset and /retry.

Addresses #102157.

gateway/session_persistence.py

@@ -156,7 +156,16 @@ class SessionPersistenceMixin:
             return pinned
         profile = self._named_profile_for_key(session_key)
         if profile is None:
-            return self._db
+            # Default-profile (``agent:main``) rows belong to the launch home, not to whichever
+            # profile's scope happens to be active: a scoped drain tick or cron mirror touching a
+            # default chat used to write its rows into the secondary's store (#102157's picture).
+            routing_home = getattr(self, "_routing_home", None)
+            if routing_home is None or not getattr(self.config, "multiplex_profiles", False):
+                return self._db
+            try:
+                return self._open_session_db_for_active_scope(db_path=routing_home / "state.db")
+            except Exception:
+                return None
         home = self._profile_home_for_key(session_key)
         if home is None:
             # Falling back to the ambient store would split ONE session identity across two

gateway/slash_commands_session.py

@@ -443,7 +443,9 @@ class GatewaySessionCommandsMixin:
             return t("gateway.undo.nothing")
         session_entry.last_prompt_tokens = 0  # transcript was truncated
         try:
-            self._evict_cached_agent(build_session_key(source))
+            # The cache is keyed by the profile-namespaced key; a bare build_session_key(source)
+            # yields ``agent:main:…`` and misses for every secondary profile.
+            self._evict_cached_agent(self._session_key_for_source(source))
         except Exception as e:
             logger.debug("undo: cached-agent eviction skipped: %s", e)
         target_text = result["target_text"]

tests/gateway/test_multiplex_session_db_profile_scope.py

@@ -773,3 +773,20 @@ def test_crash_marker_from_a_secondary_profile_survives_restart(multiplex_homes)
     assert recovered.resume_pending is True
     assert recovered.resume_reason == "restart_interrupted"
     assert recovered.active_turn_token is None
+
+
+def test_default_namespace_rows_stay_in_launch_store_under_secondary_scope(multiplex_homes):
+    """``agent:main`` rows belong to the launch home even while a secondary profile's scope is
+    active. A scoped background tick (async-delegation drain, cron mirror) that touches a
+    default-profile chat used to persist it into the secondary's ``state.db`` — the
+    ``profile_name='<A>'`` row inside B's store from #102157."""
+    root, profile = multiplex_homes
+    store = _multiplex_store(root)
+
+    scope = set_hermes_home_override(str(profile))
+    try:
+        db = store._db_for_key("agent:main:telegram:dm:1")
+    finally:
+        reset_hermes_home_override(scope)
+
+    assert Path(db.db_path) == root / "state.db"

tests/gateway/test_undo_rewind_session.py

@@ -144,3 +144,40 @@ def test_rewind_fails_closed_when_new_turn_lands_after_id_snapshot(
         ("a3-from-other-process", 1),
     ]
     sibling.close()
+
+
+@pytest.mark.asyncio
+async def test_undo_evicts_cached_agent_under_profile_namespaced_key():
+    """/undo must evict under the key the cache is keyed by. On a multiplexed gateway that is
+    ``agent:<profile>:…``; a bare ``build_session_key(source)`` yields ``agent:main:…``, the
+    eviction misses and the next turn reuses an agent still holding the undone turns."""
+    from gateway.platforms.base import Platform, SessionSource
+    from gateway.platforms.event import MessageEvent, MessageType
+    from gateway.run import GatewayRunner
+    from gateway.session import build_session_key
+
+    class _Entry:
+        session_id = "sid"
+        last_prompt_tokens = 0
+
+    class _Store:
+        async def get_or_create_session(self, source):
+            return _Entry()
+
+        async def rewind_session(self, sid, n):
+            return {"target_text": "q3", "turns_undone": 1, "rewound_count": 2}
+
+    source = SessionSource(platform=Platform.TELEGRAM, chat_id="123", chat_type="dm", user_id="u1")
+    source.profile = "work"
+    evicted = []
+    runner = object.__new__(GatewayRunner)
+    runner.config = GatewayConfig(multiplex_profiles=True)
+    runner.session_store = _Store()
+    runner._async_session_store = _Store()
+    runner._async_session_store._store = runner.session_store
+    runner._evict_cached_agent = evicted.append
+    runner._session_key_for_source = lambda s: build_session_key(s, profile=s.profile)
+
+    await runner._handle_undo_command(MessageEvent(text="/undo", message_type=MessageType.TEXT, source=source))
+
+    assert evicted == ["agent:work:telegram:dm:123"]