fix(sessions): bundled platforms and automation labels stay read-only; adopt scan stays fast

Review folds on #147802:
- client_label also refuses bundled platform plugins (teams, irc, line, ...) and
  curator/batch/homeassistant, which the enum-only check let adopt as writable.
- hermes sessions adopt pre-filters out refused labels (subagent/cron bulk) instead
  of resolving every routeless row (0.9s -> 6.6s on an 18k-session store).
- handed_off_ref accepts an adopted phone chat, whose created_source keeps the app label.

gateway/session_route_recovery.py

@@ -149,10 +149,12 @@ def handed_off_ref(authority, session_id):
     authority mounts. Its title stays the conversation's, so ``/resume <title>`` and the Bots roster's
     registry lookup open it on that lane (#143286). None for any other row."""
     from gateway.session_contract import SessionRef
-    from hermes_state_local_migration import LEGACY_LOCAL_SOURCES
+    from hermes_state_local_migration import LEGACY_LOCAL_SOURCES, client_label
     ref = messaging_owner_ref(authority, SessionRef(authority.profile_id, session_id))
     row = authority.db.get_session(ref.session_id) or {}
-    if row.get('created_source') not in LEGACY_LOCAL_SOURCES or not mountable(authority, ref.session_id):
+    created = row.get('created_source')
+    # An adopted phone-app chat keeps the app's label as its created_source (bind_legacy_target).
+    if (created not in LEGACY_LOCAL_SOURCES and not client_label(created)) or not mountable(authority, ref.session_id):
         return None
     return ref
 

hermes_cli/sessions_adopt.py

@@ -25,17 +25,21 @@ def adoptable_sessions(path) -> list[dict]:
     offered again, which makes a second run a no-op."""
     from hermes_state import SessionDB
     from hermes_state_local import POLICY_PREFIX
-    from hermes_state_local_migration import LEGACY_PREFIX, validate_legacy_row
+    from hermes_state_local_migration import LEGACY_LOCAL_SOURCES, LEGACY_PREFIX, not_client_labels, validate_legacy_row
     from hermes_state_runtime import RuntimeStoreError
 
     db = SessionDB(db_path=path, read_only=True)
     try:
+        # Local surfaces' labels, or any label a client app chose (not a platform, automation or
+        # foreign writer): validate_legacy_row judges each row; the filter keeps cron/subagent bulk out.
+        local = tuple(sorted(LEGACY_LOCAL_SOURCES))
+        refused = tuple(sorted(not_client_labels()))
         with db._read_ctx() as conn:
-            # Every routeless row: validate_legacy_row judges the label (local surfaces and phone apps'
-            # own labels adopt; automation and foreign-writer rows do not).
             ids = [row[0] for row in conn.execute(
-                "SELECT id FROM sessions WHERE ifnull(session_key,'')='' AND ifnull(chat_id,'')=''"
-                " AND ifnull(origin_json,'')='' ORDER BY started_at, id")]
+                f"SELECT id FROM sessions WHERE source IN ({','.join('?' * len(local))})"
+                f" OR (source NOT IN ({','.join('?' * len(refused))}) AND ifnull(session_key,'')=''"
+                " AND ifnull(chat_id,'')='' AND ifnull(origin_json,'')='') ORDER BY started_at, id",
+                local + refused)]
             bound = {row[0] for row in conn.execute(
                 'SELECT key FROM state_meta WHERE key LIKE ? OR key LIKE ?', (POLICY_PREFIX + '%', LEGACY_PREFIX + '%'))}
         found, seen = [], set()

hermes_state_local_migration.py

@@ -1,4 +1,5 @@
 """Transactional legacy binding; never a second transcript writer."""
+import functools
 import json
 import time
 
@@ -15,11 +16,32 @@ LEGACY_PREFIX = 'gateway.local_legacy.v1:'
 LEGACY_LOCAL_SOURCES = frozenset({'cli', 'tui', 'gui', 'desktop', 'oneshot', 'tool', 'acp'})
 # Legacy labels that are not creation sources of their own (session_policy.SURFACES).
 LEGACY_CREATION_SOURCES = {'desktop': 'gui'}
-# Labels a routeless row can carry that never name a chat app: automation and recovery rows, and a
-# third-party WebUI that runs its own agent on this state.db (adopting its rows would legitimise a
-# second writer, so they stay read-only: gateway/session_history_mount.py).
+# Labels a routeless row can carry that never name a chat app: automation and recovery rows, a
+# retired platform, and a third-party WebUI that runs its own agent on this state.db (adopting its rows
+# would legitimise a second writer, so they stay read-only: gateway/session_history_mount.py).
 NOT_CLIENT_LABELS = frozenset({'cron', 'subagent', 'kanban', 'bot_room', 'a2a', 'webhook', 'api_server',
-                               'import', 'recovered', 'webui', 'local', 'unknown'})
+                               'import', 'recovered', 'curator', 'batch', 'homeassistant', 'webui',
+                               'local', 'unknown'})
+
+
+@functools.lru_cache(maxsize=1)
+def _messaging_labels():
+    """Every platform Hermes ships: the enum and the bundled platform plugins. A runtime-registered
+    catalog plugin (hermes-mobile's ``mobile``) is not in it: its rows are routed, which excludes them."""
+    from gateway.config import Platform
+    bundled, aliases = Platform._scan_bundled_plugin_platforms()
+    return frozenset({platform.value for platform in Platform} | set(bundled) | set(aliases))
+
+
+def not_client_labels():
+    """Labels :func:`client_label` refuses, for queries that pre-filter rows."""
+    return LEGACY_LOCAL_SOURCES | NOT_CLIENT_LABELS | _messaging_labels()
+
+
+def client_label(source):
+    """Whether *source* is a client app's own label (``mobile``, ``ios``): not a local surface, a
+    shipped messaging platform, or an automation / foreign-writer label."""
+    return isinstance(source, str) and bool(source.strip()) and source not in not_client_labels()
 
 
 def client_history_row(row):
@@ -29,12 +51,7 @@ def client_history_row(row):
     in through the dashboard login left routeless rows labelled ``mobile``, ``ios`` and the like. They
     are the owner's Desktop-style chats: adopted as ``gui``. A routed row is never one, which keeps a
     plugin platform sharing the label (hermes-mobile registers ``mobile``) on its messaging route."""
-    source = row.get('source')
-    if (not isinstance(source, str) or not source.strip() or source in LEGACY_LOCAL_SOURCES
-            or source in NOT_CLIENT_LABELS or any(row.get(k) for k in ('session_key', 'chat_id', 'origin_json'))):
-        return False
-    from gateway.config import Platform
-    return source not in {platform.value for platform in Platform}
+    return client_label(row.get('source')) and not any(row.get(k) for k in ('session_key', 'chat_id', 'origin_json'))
 
 
 def legacy_history_row(row):

tests/gateway/test_legacy_client_label_adoption.py

@@ -40,7 +40,8 @@ async def test_a_phone_apps_own_label_resumes_writable_and_a_webui_row_stays_rea
     store, desktop = await _owner(tmp_path, monkeypatch, {'gui': []})
     db = store._db
     for sid, source, user_id in (('phone-mobile', 'mobile', None), ('phone-ios', 'ios', 'basic:admin'),
-                                 ('webui-row', 'webui', None)):
+                                 ('webui-row', 'webui', None), ('teams-row', 'teams', None),
+                                 ('curator-row', 'curator', None)):
         db.create_session(sid, source=source, model='m', cwd=str(tmp_path), user_id=user_id)
         db.append_message(sid, role='user', content='hi from ' + sid)
         db.end_session(sid, 'ws_orphan_reap')
@@ -55,11 +56,13 @@ async def test_a_phone_apps_own_label_resumes_writable_and_a_webui_row_stays_rea
         assert db.get_session(sid)['source'] == 'desktop'
         again = await desktop.dispatch({'id': 20 + i, 'method': 'session.resume', 'params': {'session_id': sid}})
         assert again.get('result', {}).get('session_id') == sid, again
-    await desktop.dispatch({'id': 30, 'method': 'session.resume', 'params': {'session_id': 'webui-row'}})
-    sent = await desktop.dispatch({'id': 31, 'method': 'prompt.submit',
-                                   'params': {'session_id': 'webui-row', 'text': 'go on', 'submission_id': 's-w'}})
-    assert (sent.get('error', {}).get('data') or {}).get('reason') == 'session_read_only', sent
-    assert db.get_session('webui-row')['source'] == 'webui'
+    # A foreign writer's rows, a bundled plugin platform's label and automation stay read-only.
+    for i, sid in enumerate(('webui-row', 'teams-row', 'curator-row')):
+        await desktop.dispatch({'id': 30 + i, 'method': 'session.resume', 'params': {'session_id': sid}})
+        sent = await desktop.dispatch({'id': 40 + i, 'method': 'prompt.submit',
+                                       'params': {'session_id': sid, 'text': 'go on', 'submission_id': 's-' + sid}})
+        assert (sent.get('error', {}).get('data') or {}).get('reason') == 'session_read_only', (sid, sent)
+        assert db.get_session(sid)['source'] == sid.split('-')[0]
 
 
 @pytest.mark.asyncio