perf(sessions): batch sidebar subagent source lookup (#7231)

Reuse the authoritative batched state.db sidebar metadata to classify delegated subagents before route filtering. Keep direct lookups for single-session recovery and mutation guards.

Assisted-by: Hermes Agent:gpt-5.6-sol
Assisted-by: Codex:gpt-5.6-luna

Co-authored-by: nesquena-hermes <nesquena+hermes@gmail.com>

api/models.py

@@ -5950,6 +5950,9 @@ def _read_state_db_sidebar_overrides(
         import sqlite3
     except ImportError:
         return {}
+    ids = list(wanted)
+    chunk_size = 500
+    overrides: dict[str, dict] = {}
     try:
         with closing(open_state_db_readonly(db_path)) as conn:
             conn.row_factory = sqlite3.Row
@@ -5975,10 +5978,7 @@ def _read_state_db_sidebar_overrides(
                 messages_has_timestamp = 'timestamp' in message_cols
                 messages_has_title_fields = {'session_id', 'role', 'content', 'timestamp'}.issubset(message_cols)
 
-            overrides: dict[str, dict] = {}
             delegated_title_ids: set[str] = set()
-            ids = list(wanted)
-            chunk_size = 500
             for i in range(0, len(ids), chunk_size):
                 chunk = ids[i:i + chunk_size]
                 placeholders = ','.join('?' * len(chunk))
@@ -6072,7 +6072,38 @@ def _read_state_db_sidebar_overrides(
                             overrides.setdefault(sid, {})['_state_db_display_title'] = display_title
             return overrides
     except Exception:
-        return {}
+        missing_source_ids = [
+            sid for sid in ids
+            if not overrides.get(sid, {}).get('_state_db_source')
+        ]
+        if not missing_source_ids:
+            return overrides
+        try:
+            with closing(open_state_db_readonly(db_path)) as conn:
+                conn.row_factory = sqlite3.Row
+                cur = conn.cursor()
+                for i in range(0, len(missing_source_ids), chunk_size):
+                    chunk = missing_source_ids[i:i + chunk_size]
+                    placeholders = ','.join('?' * len(chunk))
+                    cur.execute(
+                        f"SELECT id, source FROM sessions WHERE id IN ({placeholders})",
+                        chunk,
+                    )
+                    for row in cur.fetchall():
+                        sid = str(row['id'])
+                        state_source = str(row['source'] or '').strip().lower()
+                        if not state_source:
+                            continue
+                        entry = overrides.setdefault(sid, {})
+                        entry['_state_db_source'] = state_source
+                        source_meta = normalize_agent_session_source(state_source)
+                        entry['_state_db_source_tag'] = state_source
+                        entry['_state_db_raw_source'] = source_meta.get('raw_source')
+                        entry['_state_db_session_source'] = source_meta.get('session_source')
+                        entry['_state_db_source_label'] = source_meta.get('source_label')
+                return overrides
+        except Exception:
+            return overrides
 
 
 def _apply_sidebar_state_db_overrides(sessions: list[dict]) -> None:
@@ -6125,12 +6156,14 @@ def _apply_sidebar_state_db_override_metadata(sessions: list[dict], metadata: di
         state_db_message_count = entry.pop('_state_db_message_count', None)
         state_db_last_message_at = entry.pop('_state_db_last_message_at', None)
         state_db_display_title = entry.pop('_state_db_display_title', None)
-        if state_db_source == 'webui':
+        if state_db_source in ('webui', 'subagent'):
             session['source_tag'] = state_db_source_tag
             session['raw_source'] = state_db_raw_source
             session['session_source'] = state_db_session_source
             session['source_label'] = state_db_source_label
             session['is_cli_session'] = False
+            if state_db_source == 'subagent':
+                session['read_only'] = True
         # Overlay the real state.db message count for WebUI-owned rows AND for
         # delegated subagent children (#5308). A subagent child
         # (state_db_source == 'subagent') is backed by the delegate runner's
@@ -6141,9 +6174,9 @@ def _apply_sidebar_state_db_override_metadata(sessions: list[dict], metadata: di
         # session vanishes from the sidebar entirely (regression seam behind
         # #5308, same state.db-blind-metadata root as the #5307 transcript
         # recovery). The count overlay keeps the same conservative
-        # anti-resurrection guard used for WebUI rows. The source-tag / title
-        # reassignment above stays WebUI-only — a subagent child keeps its
-        # subagent classification.
+        # anti-resurrection guard used for WebUI rows. Source metadata is
+        # authoritative for WebUI rows and delegated subagent children;
+        # foreign CLI sources retain their existing flags.
         if state_db_source in ('webui', 'subagent'):
             try:
                 current_count = max(0, int(session.get('message_count') or 0))

api/routes.py

@@ -2567,9 +2567,8 @@ def _build_session_list_cache_payload(
     diag_stage("visible_lineage_metadata")
     _enrich_sidebar_lineage_metadata(scoped)
     # Delegated subagent children (#5307) are view-only, owned by the delegate
-    # runner. Coerce their sidebar rows to read_only=True + is_cli_session=False
-    # so the UI never offers delete / edit / truncate / pin affordances on them
-    # (defense-in-depth is also enforced server-side on the mutation routes).
+    # runner. The model-layer batch overlay above has already applied the
+    # authoritative source and view-only flags before this route runs.
     def _coerce_subagent_rows(_rows):
         for _r in _rows:
             if not isinstance(_r, dict):
@@ -2578,16 +2577,7 @@ def _build_session_list_cache_payload(
                 str(_r.get("source_tag") or _r.get("raw_source")
                     or _r.get("session_source") or _r.get("source") or "").strip().lower()
             )
-            _is_sa = _src == "subagent"
-            # A stale index row can say webui/fork while state.db records the
-            # row as source='subagent' (the child shares the parent's lineage).
-            # For rows not already read-only, confirm via the state.db source so
-            # a delegated child can't surface as a writable/CLI sidebar row.
-            if not _is_sa and not _r.get("read_only"):
-                _sid = str(_r.get("session_id") or "").strip()
-                if _sid and _is_subagent_child_session_id(_sid):
-                    _is_sa = True
-            if _is_sa:
+            if _src == "subagent":
                 _r["read_only"] = True
                 _r["is_cli_session"] = False
     _coerce_subagent_rows(scoped)

tests/test_webui_state_db_reconciliation.py

@@ -132,6 +132,19 @@ def _large_timestamped_sidecar_messages(count=500):
     ]
 
 
+def _state_db_source_metadata(source):
+    from api.agent_sessions import normalize_agent_session_source
+
+    source_meta = normalize_agent_session_source(source)
+    return {
+        "_state_db_source": source,
+        "_state_db_source_tag": source,
+        "_state_db_raw_source": source_meta["raw_source"],
+        "_state_db_session_source": source_meta["session_source"],
+        "_state_db_source_label": source_meta["source_label"],
+    }
+
+
 def test_sidebar_state_db_overlay_preserves_numeric_actual_count():
     import api.models as models
 
@@ -162,6 +175,243 @@ def test_sidebar_state_db_overlay_preserves_numeric_actual_count():
     assert sessions[0]["actual_message_count"] == 5
 
 
+def test_sidebar_state_db_overlay_reclassifies_authoritative_subagent_rows():
+    import api.models as models
+
+    def sidebar_row(sid, source=None, *, session_source=None, source_label=None, is_cli_session=True):
+        row = {"session_id": sid, "is_cli_session": is_cli_session, "read_only": False}
+        if source:
+            row.update(
+                source_tag=source,
+                raw_source=source,
+                session_source=session_source,
+                source_label=source_label,
+            )
+        return row
+
+    sessions = [
+        sidebar_row(
+            "row-01", "webui", session_source="webui", source_label="WebUI", is_cli_session=False
+        ),
+        sidebar_row("row-02", "fork", session_source="other", source_label="Fork"),
+        sidebar_row("row-03"),
+        sidebar_row("row-04", "tui", session_source="cli", source_label="TUI"),
+    ]
+    subagent_metadata = {
+        sid: _state_db_source_metadata("subagent")
+        for sid in ("row-01", "row-02", "row-03")
+    }
+    subagent_metadata["row-04"] = _state_db_source_metadata("tui")
+
+    models._apply_sidebar_state_db_override_metadata(sessions, subagent_metadata)
+
+    observed = [
+        (
+            session.get("source_tag"), session.get("raw_source"),
+            session.get("session_source"), session.get("source_label"),
+            session["is_cli_session"], session["read_only"],
+        )
+        for session in sessions
+    ]
+    assert observed[:3] == [("subagent", "subagent", "other", "Subagent", False, True)] * 3
+    assert observed[3] == ("tui", "tui", "cli", "TUI", True, False)
+
+
+def test_api_sessions_bulk_uses_batched_subagent_metadata_without_row_probes(
+    monkeypatch, tmp_path
+):
+    import api.models as models
+    import api.routes as routes
+
+    db_path = tmp_path / "state.db"
+    stale_rows = [
+        {
+            "session_id": f"row-{index:02d}",
+            "source_tag": "webui",
+            "raw_source": "webui",
+            "session_source": "webui",
+            "source_label": "WebUI",
+        }
+        for index in range(32)
+    ]
+    with sqlite3.connect(db_path) as conn:
+        conn.execute("CREATE TABLE sessions (id TEXT PRIMARY KEY, source TEXT)")
+        conn.executemany(
+            "INSERT INTO sessions (id, source) VALUES (?, 'subagent')",
+            ((row["session_id"],) for row in stale_rows),
+        )
+    monkeypatch.setattr(models, "_active_state_db_path", lambda: db_path)
+
+    def fake_all_sessions(**_kwargs):
+        rows = [
+            dict(row, is_cli_session=True, read_only=False, message_count=1)
+            for row in stale_rows
+        ]
+        models._apply_sidebar_state_db_overrides(rows)
+        return rows
+
+    monkeypatch.setattr(routes, "all_sessions", fake_all_sessions)
+    monkeypatch.setattr(routes, "_enrich_sidebar_lineage_metadata", lambda _rows: None)
+    monkeypatch.setattr(routes, "_reconcile_stale_stream_state_for_session_rows", lambda _rows: False)
+    monkeypatch.setattr(routes, "_prune_orphaned_webui_zero_message_sessions", lambda rows, **_kwargs: list(rows))
+    single_row_probes = []
+
+    def record_single_row_probe(sid):
+        single_row_probes.append(sid)
+        return True
+
+    monkeypatch.setattr(routes, "_is_subagent_child_session_id", record_single_row_probe)
+
+    payload = routes._build_session_list_cache_payload(
+        active_profile="default",
+        all_profiles=False,
+        show_cli_sessions=False,
+        show_previous_messaging_sessions=False,
+        show_cron_sessions=False,
+        include_archived=False,
+    )
+
+    rows = {row["session_id"]: row for row in payload["sessions"]}
+    assert single_row_probes == []
+    assert set(rows) == {row["session_id"] for row in stale_rows}
+    for row in rows.values():
+        assert (
+            row["source_tag"],
+            row["raw_source"],
+            row["session_source"],
+            row["source_label"],
+        ) == ("subagent", "subagent", "other", "Subagent")
+        assert row["read_only"] is True
+        assert row["is_cli_session"] is False
+
+
+def test_sidebar_override_reader_uses_one_connection_and_500_id_chunks(
+    monkeypatch, tmp_path
+):
+    import api.models as models
+
+    db_path = tmp_path / "state.db"
+    session_ids = [f"row-{index:04d}" for index in range(1001)]
+    with sqlite3.connect(db_path) as conn:
+        conn.execute(
+            "CREATE TABLE sessions (id TEXT PRIMARY KEY, source TEXT)"
+        )
+        conn.executemany(
+            "INSERT INTO sessions (id, source) VALUES (?, 'subagent')",
+            ((sid,) for sid in session_ids),
+        )
+
+    connections = []
+    session_queries = []
+    real_open = models.open_state_db_readonly
+
+    def tracked_open(path):
+        connection = real_open(path)
+        connections.append(connection)
+
+        def trace(statement):
+            if "FROM sessions s" in statement and "s.id IN" in statement:
+                session_queries.append(statement)
+
+        connection.set_trace_callback(trace)
+        return connection
+
+    monkeypatch.setattr(models, "open_state_db_readonly", tracked_open)
+    overrides = models._read_state_db_sidebar_overrides(db_path, set(session_ids))
+
+    assert len(connections) == 1
+    assert len(session_queries) == 3
+    assert set(overrides) == set(session_ids)
+    assert overrides[session_ids[0]] == _state_db_source_metadata("subagent")
+
+
+def test_sidebar_override_reader_recovers_sources_after_rich_reader_error(
+    monkeypatch, tmp_path
+):
+    import api.models as models
+
+    db_path = tmp_path / "state.db"
+    session_ids = [f"row-{index:04d}" for index in range(1001)]
+    with sqlite3.connect(db_path) as conn:
+        conn.execute("CREATE TABLE sessions (id TEXT PRIMARY KEY, source TEXT)")
+        conn.executemany(
+            "INSERT INTO sessions (id, source) VALUES (?, 'subagent')",
+            ((sid,) for sid in session_ids),
+        )
+
+    class FailingCursor:
+        def __init__(self, cursor):
+            self._cursor = cursor
+
+        def execute(self, statement, parameters=()):
+            if "sqlite_master" in statement:
+                raise sqlite3.OperationalError("controlled optional schema failure")
+            self._cursor.execute(statement, parameters)
+            return self
+
+        def __getattr__(self, name):
+            return getattr(self._cursor, name)
+
+    first_closed = []
+
+    class FailingConnection:
+        def __init__(self, connection):
+            self._connection = connection
+
+        @property
+        def row_factory(self):
+            return self._connection.row_factory
+
+        @row_factory.setter
+        def row_factory(self, value):
+            self._connection.row_factory = value
+
+        def cursor(self):
+            return FailingCursor(self._connection.cursor())
+
+        def close(self):
+            first_closed.append(True)
+            self._connection.close()
+
+    connections = []
+    source_only_queries = []
+    second_connection = None
+    real_open = models.open_state_db_readonly
+
+    def tracked_open(path):
+        nonlocal second_connection
+        connection = real_open(path)
+        connections.append(connection)
+        if len(connections) == 1:
+            return FailingConnection(connection)
+        second_connection = connection
+
+        def trace(statement):
+            normalized = " ".join(statement.split()).lower()
+            if normalized.startswith("select id, source from sessions where id in ("):
+                source_only_queries.append(statement)
+
+        connection.set_trace_callback(trace)
+        return connection
+
+    monkeypatch.setattr(models, "open_state_db_readonly", tracked_open)
+    overrides = models._read_state_db_sidebar_overrides(db_path, set(session_ids))
+
+    assert len(connections) == 2
+    assert first_closed == [True]
+    assert second_connection is not None
+    with pytest.raises(sqlite3.ProgrammingError):
+        second_connection.execute("SELECT 1")
+    assert len(source_only_queries) == 3
+    assert [
+        statement.split("IN (", 1)[1].split(")", 1)[0].count(",") + 1
+        for statement in source_only_queries
+    ] == [500, 500, 1]
+    assert set(overrides) == set(session_ids)
+    assert overrides[session_ids[0]] == _state_db_source_metadata("subagent")
+    assert overrides[session_ids[-1]] == _state_db_source_metadata("subagent")
+
+
 def test_sidebar_state_db_overlay_counts_subagent_child_5308():
     """#5308: a delegated subagent child whose stale sidecar reports
     message_count == 0 must receive its real state.db message count so the
@@ -191,7 +441,7 @@ def test_sidebar_state_db_overlay_counts_subagent_child_5308():
         sessions,
         {
             sid: {
-                "_state_db_source": "subagent",
+                **_state_db_source_metadata("subagent"),
                 "_state_db_message_count": 6,
                 "_state_db_last_message_at": 2002.0,
             }
@@ -203,10 +453,11 @@ def test_sidebar_state_db_overlay_counts_subagent_child_5308():
     assert sessions[0]["message_count"] == 6
     assert sessions[0]["actual_message_count"] == 6
     assert sessions[0]["last_message_at"] == 2002.0
-    # Subagent classification is preserved (source-tag reassignment stays
-    # WebUI-only) — the child does not get re-tagged as a webui session.
+    # Subagent classification is authoritative and view-only — the child does
+    # not get re-tagged as a WebUI or writable CLI session.
     assert sessions[0]["source_tag"] == "subagent"
-    assert sessions[0].get("is_cli_session") is not False
+    assert sessions[0]["is_cli_session"] is False
+    assert sessions[0]["read_only"] is True
 
 
 def test_sidebar_state_db_overlay_does_not_count_foreign_cli_source_5308():