fix(tui_gateway): profile sessions never read or write through the launch store

Three wrong-store paths in the TUI/Desktop backend under multiplexing:

- insights.get was scoped=True and then called _get_db(): before the launch
  handle was pinned (#102526, #108074) a scoped first touch bound the
  process-wide handle to the requested profile's state.db; even pinned, the
  RPC answered for the launch profile whatever `profile` said. Route it
  through _profile_db so it counts the requested profile's sessions.

- prompt.background side agents were handed the launch handle, so a
  named-profile Bot Chat's bg_* rows appeared in the default profile's
  session list and were missing from the profile's own history. Inherit the
  parent agent's dedicated _session_db.

- session_lifecycle's gateway-owned-source guard and the notification
  poller's compression-tip resolver looked the session up in the launch
  store, where a named-profile row does not exist: the guard was dead and
  a compression-rotated profile session lost every post-compression
  delegation/background completion (the fail-closed owner gate never
  matched the compressed parent's key). Both now use the session's own
  store via _session_db(session).

Addresses #102157, #102526.

tests/tui_gateway/test_launch_db_home_override_race.py

@@ -44,3 +44,60 @@ def test_get_db_first_touch_under_foreign_override_uses_launch_path(launch_db_en
         assert server._get_db() is db
     finally:
         reset_hermes_home_override(token)
+
+
+def test_insights_get_reads_the_requested_profile_store_not_the_launch_handle(launch_db_env, monkeypatch, tmp_path):
+    """``insights.get {profile}`` was ``scoped=True`` then ``_get_db()``: a scoped first touch pinned
+    the launch handle to the foreign store. It must count the requested profile's sessions through
+    ``_profile_db`` and leave the launch handle on the launch home."""
+    launch_home, _foreign = launch_db_env
+    profiles_root = tmp_path / "profiles"
+    work = profiles_root / "work"
+    work.mkdir(parents=True)
+    monkeypatch.setattr("hermes_cli.profiles.get_profile_dir", lambda name: profiles_root / name)
+    monkeypatch.setattr(server, "_canonical_profile_request", lambda name: name or None)
+
+    seeded = registry.acquire(work / "state.db")
+    seeded.create_session("work-only", source="tui", model="m")
+    registry.release(seeded)
+
+    result = server._methods["insights.get"]("rid", {"profile": "work", "days": 30})
+
+    assert result["result"]["sessions"] == 1
+    assert server._get_db().db_path.resolve() == (launch_home / "state.db").resolve()
+    assert server._get_db().get_session("work-only") is None
+
+
+def test_background_side_agent_persists_into_the_parent_agent_store(launch_db_env, monkeypatch):
+    """``prompt.background`` side agents write ``bg_*`` rows next to their parent's transcript: a
+    named-profile chat's parent holds a dedicated profile handle, and handing the launch handle
+    instead made those rows show up in the default profile's history."""
+    import types
+
+    parent_db = object()
+    agent = types.SimpleNamespace(model="m", provider="p", _fallback_chain=[], _session_db=parent_db)
+    monkeypatch.setattr(server, "_load_cfg", lambda: {"max_turns": 25})
+    monkeypatch.setattr(server, "_load_enabled_toolsets", lambda *_a, **_kw: ["file"])
+
+    assert server._background_agent_kwargs(agent, "bg_1")["session_db"] is parent_db
+
+
+def test_notification_owner_gate_resolves_rotated_key_in_the_session_profile_store(launch_db_env, tmp_path):
+    """A compression-rotated NAMED-PROFILE session must still claim events keyed by its compressed
+    parent: the lineage lives in ``profiles/<x>/state.db``, which the launch handle cannot see, so
+    the fail-closed owner gate silently dropped every post-compression notification."""
+    profile_home = tmp_path / "profiles" / "work"
+    profile_home.mkdir(parents=True)
+    db = registry.acquire(profile_home / "state.db")
+    db.create_session("parent", source="tui", model="m")
+    db.append_message("parent", "user", "hello")
+    db.end_session("parent", "compression")
+    db.create_session("child", source="tui", model="m", parent_session_id="parent")
+    db.append_message("child", "user", "later")
+    registry.release(db)
+
+    session = {"profile_home": str(profile_home), "session_key": "child", "agent": None}
+    evt = {"type": "async_delegation", "session_key": "parent"}
+
+    assert server._session_owns_notification_event("ui1", session, evt) is True
+    assert server._get_db().get_session("parent") is None  # never looked up through the launch store

tui_gateway/agent_callbacks.py

@@ -298,7 +298,9 @@ def _background_agent_kwargs(agent, task_id: str) -> dict:
         "reasoning_config": g("reasoning_config") or _load_reasoning_config(str(g("model", "") or "")),
         "service_tier": g("service_tier") or _load_service_tier(),
         "request_overrides": dict(g("request_overrides", {}) or {}),
-        "platform": "tui", "session_db": _get_db(), "fallback_model": fallback}
+        # The side agent persists into the PARENT's store: a named-profile chat's ``bg_*`` rows
+        # belong to that profile's state.db, not the launch handle.
+        "platform": "tui", "session_db": getattr(agent, "_session_db", None) or _get_db(), "fallback_model": fallback}
 
 
 def _ephemeral_preview_agent_kwargs(agent, task_id: str) -> dict:

tui_gateway/methods_tools.py

@@ -870,13 +870,16 @@ def _(rid, params: dict) -> dict:
 
 
 # ─── Insights / rollback / browser / config ──────────────────────────────────
-@_rpc("insights.get", 5017)
+@_scoped_rpc("insights.get", 5017)
 def _(rid, params: dict) -> dict:
     days = params.get("days", 30)
-    if (db := _get_db()) is None:
-        return _db_unavailable_error(rid, code=5017)
-    cutoff = time.time() - days * 86400
-    rows = [s for s in db.list_sessions_rich(limit=500, compact_rows=True) if (s.get("started_at") or 0) >= cutoff]
+    # ``profile`` selects that profile's store; the launch handle is never the fallback for a
+    # scoped call (a foreign first touch used to pin the process-wide handle, #102526).
+    with _profile_db(params) as db:
+        if db is None:
+            return _db_unavailable_error(rid, code=5017)
+        cutoff = time.time() - days * 86400
+        rows = [s for s in db.list_sessions_rich(limit=500, compact_rows=True) if (s.get("started_at") or 0) >= cutoff]
     return _ok(rid, {"days": days, "sessions": len(rows), "messages": sum(s.get("message_count", 0) for s in rows)})
 
 

tui_gateway/session_lifecycle.py

@@ -438,8 +438,9 @@ def _session_has_active_delegations(sid: str, session: dict | None = None) -> bo
     if session_id:
         # Only when this session may end its durable row by key — never for gateway-originated sessions (TUI is a
         # viewer there). Unknown DB state -> assume ownership.
-        with contextlib.suppress(Exception):
-            db = _get_db()
+        # The row lives in the session's OWN store (a named-profile session's row is invisible to
+        # the launch handle, which would leave this guard permanently dead).
+        with contextlib.suppress(Exception), _session_db(session) as db:
             if db is not None and _is_gateway_owned_source((db.get_session(session_id) or {}).get("source", "")):
                 owned_session_key = ""
     if not own_sid and not owned_session_key:

tui_gateway/session_notifications.py

@@ -35,11 +35,13 @@ def _notif_live_session_matches(keys, exclude: dict | None = None) -> bool:
         False)
 
 
-def _notif_resolve_event_key(evt_key: str) -> str:
-    """Resolve a compression-rotated session key to its continuation tip (or itself)."""
+def _notif_resolve_event_key(evt_key: str, session: dict | None = None) -> str:
+    """Resolve a compression-rotated session key to its continuation tip (or itself). Looked up in
+    ``session``'s own store: a named-profile session's lineage lives in ``profiles/<x>/state.db``,
+    where the launch handle cannot see it."""
     try:
-        db = _get_db()
-        return (db.resolve_resume_session_id(evt_key) if db is not None else evt_key) or evt_key
+        with _session_db(session or {}) as db:
+            return (db.resolve_resume_session_id(evt_key) if db is not None else evt_key) or evt_key
     except Exception:
         return evt_key
 
@@ -62,7 +64,7 @@ def _notification_event_belongs_elsewhere(sid: str, session: dict, evt: dict) ->
     # Compression can rotate AIAgent.session_id while the detached child is still running: map the event's original
     # key to its continuation tip so it reaches the live session instead of becoming an orphan any poller may consume.
     # A live continuation wins over the compressed parent, else a stale parent tab could consume the event first.
-    resolved_key = _notif_resolve_event_key(evt_key)
+    resolved_key = _notif_resolve_event_key(evt_key, session)
     if resolved_key != evt_key:
         if resolved_key in current_keys:
             return False
@@ -82,7 +84,7 @@ def _session_owns_notification_event(sid: str, session: dict, evt: dict) -> bool
         return True
     evt_key = str(evt.get("session_key") or "")
     current_keys = _notif_current_keys(sid, session)
-    return bool(evt_key) and (evt_key in current_keys or _notif_resolve_event_key(evt_key) in current_keys)
+    return bool(evt_key) and (evt_key in current_keys or _notif_resolve_event_key(evt_key, session) in current_keys)
 
 
 def _notification_event_requires_owner(evt: dict) -> bool: