From c3750fe36186d798166893752723a9b62c9d13ff Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Tue, 9 Jun 2026 12:57:55 -0400 Subject: [PATCH 1/8] perf(sidebar): cache and single-flight /api/sessions payloads (#3791) --- api/routes.py | 639 ++++++++++++++++++------ api/session_events.py | 24 + tests/test_request_diagnostics_cache.py | 20 + tests/test_session_sidebar_cache.py | 283 +++++++++++ 4 files changed, 813 insertions(+), 153 deletions(-) create mode 100644 tests/test_request_diagnostics_cache.py create mode 100644 tests/test_session_sidebar_cache.py diff --git a/api/routes.py b/api/routes.py index 041d2650..3dbde055 100644 --- a/api/routes.py +++ b/api/routes.py @@ -21,8 +21,7 @@ import sys import threading import time import uuid -import re -from collections import defaultdict +from collections import defaultdict, OrderedDict from pathlib import Path from contextlib import closing from urllib.parse import parse_qs, urlsplit @@ -35,6 +34,7 @@ from api.agent_sessions import ( ) from api.compression_anchor import visible_messages_for_anchor from api.session_events import ( + add_session_list_changed_listener, publish_session_list_changed, subscribe_session_events, unsubscribe_session_events, @@ -183,6 +183,17 @@ def _queue_generated_title_for_imported_session(session, cli_meta: dict | None) ) +def _on_session_list_changed(profile: str | None = None) -> None: + """Invalidate in-process /api/sessions cache when sidebar state mutates.""" + _clear_session_list_cache(profile) + + +try: + add_session_list_changed_listener(_on_session_list_changed) +except Exception: + logger.debug("Failed to register session list cache invalidation listener", exc_info=True) + + # ── Cron run tracking ──────────────────────────────────────────────────────── # Track job IDs currently being executed so the frontend can poll status. _RUNNING_CRON_JOBS: dict[str, float] = {} # job_id → start_timestamp @@ -1211,6 +1222,457 @@ def _clear_live_models_cache() -> None: with _LIVE_MODELS_CACHE_LOCK: _LIVE_MODELS_CACHE.clear() + +# ── /api/sessions response cache (hot-sidebar response) ────────────────────── + +_SESSIONS_CACHE_TTL_SECONDS = 2.5 +_SESSIONS_CACHE_MAX_ENTRIES = 64 +_SESSIONS_CACHE_WAIT_SECONDS = 0.25 +_SESSIONS_CACHE_STALE_WAIT_SECONDS = 0.10 +_SESSIONS_CACHE: OrderedDict[tuple, tuple[float, dict]] = OrderedDict() +_SESSIONS_CACHE_LOCK = threading.RLock() +_SESSIONS_CACHE_INFLIGHT: dict[tuple, threading.Event] = {} +_SESSIONS_CACHE_GLOBAL_INVALIDATION_VERSION = 0 +_SESSIONS_CACHE_ALL_PROFILES_INVALIDATION_VERSION = 0 +_SESSIONS_CACHE_PROFILE_INVALIDATION_VERSION: dict[str, int] = {} + + +def _session_list_cache_profile_scope(profile: str | None) -> str: + normalized = str(profile or "").strip() or "default" + if _profiles_match(normalized, "default"): + return "default" + return normalized + + +def _session_list_cache_key( + active_profile: str | None, + all_profiles: bool, + show_cli_sessions: bool, + show_previous_messaging_sessions: bool, + show_cron_sessions: bool, +) -> tuple: + return ( + _session_list_cache_profile_scope(active_profile), + bool(all_profiles), + bool(show_cli_sessions), + bool(show_previous_messaging_sessions), + bool(show_cron_sessions), + ) + + +def _session_list_cache_get( + key: tuple, + allow_stale: bool = False, +) -> tuple[dict | None, bool]: + now = time.monotonic() + with _SESSIONS_CACHE_LOCK: + entry = _SESSIONS_CACHE.get(key) + if not entry: + return None, False + ts, payload = entry + fresh = (now - ts) < _SESSIONS_CACHE_TTL_SECONDS + if fresh: + _SESSIONS_CACHE.move_to_end(key) + return copy.deepcopy(payload), True + if allow_stale: + _SESSIONS_CACHE.move_to_end(key) + return copy.deepcopy(payload), False + _SESSIONS_CACHE.pop(key, None) + return None, False + + +def _session_list_cache_set(key: tuple, payload: dict) -> None: + if not isinstance(payload, dict): + return + with _SESSIONS_CACHE_LOCK: + _SESSIONS_CACHE[key] = (time.monotonic(), copy.deepcopy(payload)) + _SESSIONS_CACHE.move_to_end(key) + while len(_SESSIONS_CACHE) > _SESSIONS_CACHE_MAX_ENTRIES: + _SESSIONS_CACHE.popitem(last=False) + + +def _session_list_cache_clear(profile: str | None = None) -> None: + normalized_profile = _session_list_cache_profile_scope(profile) if profile else None + with _SESSIONS_CACHE_LOCK: + global _SESSIONS_CACHE_GLOBAL_INVALIDATION_VERSION + global _SESSIONS_CACHE_ALL_PROFILES_INVALIDATION_VERSION + if not profile: + _SESSIONS_CACHE_GLOBAL_INVALIDATION_VERSION += 1 + _SESSIONS_CACHE_ALL_PROFILES_INVALIDATION_VERSION += 1 + _SESSIONS_CACHE_PROFILE_INVALIDATION_VERSION.clear() + _SESSIONS_CACHE.clear() + return + _SESSIONS_CACHE_ALL_PROFILES_INVALIDATION_VERSION += 1 + _SESSIONS_CACHE_PROFILE_INVALIDATION_VERSION[normalized_profile] = ( + _SESSIONS_CACHE_PROFILE_INVALIDATION_VERSION.get(normalized_profile, 0) + 1 + ) + for cache_key in list(_SESSIONS_CACHE.keys()): + cache_profile, cache_all_profiles, *_rest = cache_key + if cache_all_profiles: + _SESSIONS_CACHE.pop(cache_key, None) + continue + if _profiles_match(cache_profile, normalized_profile): + _SESSIONS_CACHE.pop(cache_key, None) + + +def _clear_session_list_cache(profile: str | None = None) -> None: + _session_list_cache_clear(profile=profile) + + +def _session_list_cache_invalidation_stamp(key: tuple) -> tuple[int, int]: + cache_profile, cache_all_profiles, *_rest = key + with _SESSIONS_CACHE_LOCK: + global_version = _SESSIONS_CACHE_GLOBAL_INVALIDATION_VERSION + if cache_all_profiles: + return ( + global_version, + _SESSIONS_CACHE_ALL_PROFILES_INVALIDATION_VERSION, + ) + return ( + global_version, + _SESSIONS_CACHE_PROFILE_INVALIDATION_VERSION.get(cache_profile, 0), + ) + + +def _session_list_cache_overlay_runtime_rows(rows: list[dict]) -> list[dict]: + if not rows: + return [] + try: + active_stream_ids = _active_stream_ids() + except Exception: + active_stream_ids = set() + session_ids = [ + str(row.get("session_id") or "").strip() + for row in rows + if isinstance(row, dict) and str(row.get("session_id") or "").strip() + ] + live_sessions = {} + if session_ids: + with LOCK: + for sid in session_ids: + live = SESSIONS.get(sid) + if live is not None: + live_sessions[sid] = live + overlaid = [] + for row in rows: + item = dict(row) if isinstance(row, dict) else {} + sid = str(item.get("session_id") or "").strip() + live = live_sessions.get(sid) + if live is not None: + live_stream_id = getattr(live, "active_stream_id", None) + item["active_stream_id"] = live_stream_id or None + item["has_pending_user_message"] = bool( + getattr(live, "pending_user_message", None) + ) + stream_id = item.get("active_stream_id") + item["is_streaming"] = bool(stream_id and stream_id in active_stream_ids) + overlaid.append(item) + return overlaid + + +def _session_list_cache_claim_rebuild(key: tuple) -> tuple[threading.Event, bool]: + with _SESSIONS_CACHE_LOCK: + current = _SESSIONS_CACHE_INFLIGHT.get(key) + if current is not None: + return current, False + event = threading.Event() + _SESSIONS_CACHE_INFLIGHT[key] = event + return event, True + + +def _session_list_cache_done(key: tuple, event: threading.Event | None) -> None: + with _SESSIONS_CACHE_LOCK: + if event is None: + return + if _SESSIONS_CACHE_INFLIGHT.get(key) is event: + _SESSIONS_CACHE_INFLIGHT.pop(key, None) + if event is not None: + event.set() + + +def _build_session_list_cache_payload( + active_profile: str | None, + all_profiles: bool, + show_cli_sessions: bool, + show_previous_messaging_sessions: bool, + show_cron_sessions: bool, + diag=None, +) -> dict: + diag_stage = diag.stage if diag is not None else lambda *_a, **_k: None + + diag_stage("all_sessions") + webui_sessions = all_sessions(diag=diag) + diag_stage("reconcile_stale_stream_state") + if _reconcile_stale_stream_state_for_session_rows(webui_sessions): + diag_stage("all_sessions_after_stale_stream_reconcile") + webui_sessions = all_sessions(diag=diag) + diag_stage("normalize_cli_rows") + show_cli_sessions = bool(show_cli_sessions) + show_previous_messaging_sessions = bool(show_previous_messaging_sessions) + show_cron_sessions = bool(show_cron_sessions) + webui_sessions = [_normalize_sidebar_source_flags(s) for s in webui_sessions] + if show_cli_sessions: + diag_stage("get_cli_sessions") + cli = get_cli_sessions() + diag_stage("merge_cli_sessions") + cli_by_id = {s["session_id"]: s for s in cli} + # #3238: reconcile orphaned imported-CLI sidecars. When a CLI + # session was clicked in WebUI it gets a WebUI-owned sidecar that + # all_sessions() returns independently of state.db. If the user + # later deletes the backing CLI session from the command line, + # the sidecar is never pruned and the stale row lingers in the + # sidebar forever (there is no WebUI delete affordance for it). + # Drop rows whose backing agent row is genuinely gone. We probe + # state.db directly (agent_session_rows_existing) rather than trust + # cli_by_id absence, because get_cli_sessions() caps at + # CLI_VISIBLE_SESSION_LIMIT (20) — an existing session can fall + # out of that window and look deleted. Native WebUI sessions + # (source == "webui") that merely have a CLI ancestor are never + # pruned. + _orphan_probe_rows = [] + _kept_after_orphan_prune = [] + for s in webui_sessions: + _sid = s.get("session_id") + if ( + _sid + and is_cli_session_row(s) + and not _session_source_is_webui(s) + and _sid not in cli_by_id + ): + _orphan_probe_rows.append(s) + else: + _kept_after_orphan_prune.append(s) + if _orphan_probe_rows: + rows_by_profile: dict[object, list[dict]] = defaultdict(list) + for row in _orphan_probe_rows: + rows_by_profile[row.get("profile")].append(row) + missing_orphan_ids: set[str] = set() + for profile_key, rows in rows_by_profile.items(): + probe_ids = [ + str(row.get("session_id")).strip() + for row in rows + if str(row.get("session_id") or "").strip() + ] + existing = agent_session_rows_existing( + probe_ids, + profile=profile_key if isinstance(profile_key, str) and profile_key else None, + ) + for row in rows: + _sid = str(row.get("session_id") or "").strip() + if _sid and _sid not in existing: + missing_orphan_ids.add(_sid) + for s in _orphan_probe_rows: + _sid = str(s.get("session_id") or "").strip() + if _sid in missing_orphan_ids: + try: + prune_session_from_index(_sid) + except Exception: + logger.debug( + "Failed to prune orphaned CLI sidecar %s", + _sid, + exc_info=True, + ) + diag_stage("prune_orphaned_cli_sidecar") + continue + _kept_after_orphan_prune.append(s) + webui_sessions = _kept_after_orphan_prune + for s in webui_sessions: + meta = cli_by_id.get(s.get("session_id")) + if not meta: + continue + if _is_messaging_session_record(meta): + s.update(_merge_cli_sidebar_metadata(s, meta)) + if s.get("session_id") != meta.get("session_id"): + s["session_id"] = meta.get("session_id") + else: + for key in ("source_tag", "raw_source", "session_source", "source_label"): + if not s.get(key) and meta.get(key): + s[key] = meta[key] + webui_sessions = [_normalize_sidebar_source_flags(s) for s in webui_sessions] + # Apply the same CLI visibility semantics to imported local copies so + # low-value imported artifacts do not leak into the sidebar. + webui_sessions = [s for s in webui_sessions if is_cli_session_row_visible(s)] + represented_webui_ids = set() + for s in webui_sessions: + represented_webui_ids.update(_session_lineage_ids(s)) + deduped_cli = _dedupe_cli_sidebar_sessions_for_api( + cli, + represented_webui_ids, + show_cron_sessions=show_cron_sessions, + ) + else: + diag_stage("filter_webui_sessions") + webui_sessions = [s for s in webui_sessions if not _is_cli_session_for_settings(s)] + deduped_cli = [] + diag_stage("sort_sessions") + merged = webui_sessions + deduped_cli + merged.sort( + key=lambda s: s.get("last_message_at") or s.get("updated_at", 0) or 0, + reverse=True, + ) + # ── Profile scoping (#1611) ──────────────────────────────────────── + # Default: filter to the active profile. ?all_profiles=1 opts into + # the aggregate view used by the "All profiles" sidebar toggle. + # The other_profile_count is always returned so the UI can render + # the "Show N from other profiles" affordance without sending the + # cross-profile rows by default. + # + # IMPORTANT: scope BEFORE _keep_latest_messaging_session_per_source. + # _messaging_source_key is profile-blind (#1614 follow-up): if the + # same Slack/Telegram identity has sessions in profiles A and B, a + # profile-blind dedupe would discard the older one even when scoped + # to its own profile, leaving that profile with zero rows for that + # source. Filter first so the dedupe operates only within the active + # profile's rows. + diag_stage("active_profile") + if all_profiles: + scoped = merged + other_profile_count = 0 + else: + scoped = [s for s in merged if _profiles_match(s.get("profile"), active_profile)] + other_profile_count = len(merged) - len(scoped) + diag_stage("messaging_dedupe") + scoped = _keep_latest_messaging_session_per_source( + scoped, + show_previous_messaging_sessions=show_previous_messaging_sessions, + ) + if show_cli_sessions: + diag_stage("cli_cap") + scoped = _cap_recent_cli_sessions(scoped, cli_cap=CLI_VISIBLE_SESSION_CAP) + return { + "sessions": [ + dict(s) if isinstance(s, dict) else {} + for s in scoped + ], + "cli_count": len(deduped_cli), + "all_profiles": all_profiles, + "active_profile": active_profile, + "other_profile_count": other_profile_count, + "settings": { + "show_cli_sessions": show_cli_sessions, + "show_previous_messaging_sessions": show_previous_messaging_sessions, + "show_cron_sessions": show_cron_sessions, + }, + } + + +def _session_list_payload_to_response(payload: dict) -> dict: + safe_merged = [] + runtime_rows = _session_list_cache_overlay_runtime_rows(payload.get("sessions", []) or []) + for s in runtime_rows: + item = dict(s) if isinstance(s, dict) else {} + if isinstance(item.get("title"), str): + item["title"] = _redact_text(item["title"]) + item["attention"] = _session_attention_summary(str(item.get("session_id") or "")) + safe_merged.append(item) + return { + "sessions": safe_merged, + "cli_count": int(payload.get("cli_count", 0)), + "all_profiles": bool(payload.get("all_profiles", False)), + "active_profile": payload.get("active_profile"), + "other_profile_count": int(payload.get("other_profile_count", 0)), + "server_time": time.time(), + "server_tz": time.strftime("%z"), + } + + +def _get_cached_session_list_payload( + *, + key: tuple, + builder, + diag=None, +) -> dict: + if diag is not None: + try: + diag.stage("session_list_cache_lookup") + except Exception: + pass + + cached, _ = _session_list_cache_get(key) + if cached is not None: + if diag is not None: + try: + diag.stage("session_list_cache_hit") + except Exception: + pass + return cached + + stale, _ = _session_list_cache_get(key, allow_stale=True) + event, is_owner = _session_list_cache_claim_rebuild(key) + if is_owner: + if diag is not None: + try: + diag.stage("session_list_cache_rebuild_owner") + except Exception: + pass + try: + rebuild_attempts = 0 + while True: + invalidation_stamp = _session_list_cache_invalidation_stamp(key) + payload = builder() + if _session_list_cache_invalidation_stamp(key) == invalidation_stamp: + _session_list_cache_set(key, payload) + if diag is not None: + try: + diag.stage("session_list_cache_stored") + except Exception: + pass + return payload + rebuild_attempts += 1 + if diag is not None: + try: + diag.stage("session_list_cache_invalidated_during_rebuild") + except Exception: + pass + if rebuild_attempts >= 3: + return payload + finally: + _session_list_cache_done(key, event) + + if diag is not None: + try: + if stale is not None: + diag.stage("session_list_cache_wait_stale") + else: + diag.stage("session_list_cache_wait") + except Exception: + pass + + if stale is not None: + timeout = _SESSIONS_CACHE_STALE_WAIT_SECONDS + else: + timeout = _SESSIONS_CACHE_WAIT_SECONDS + event.wait(timeout) + + latest, _ = _session_list_cache_get(key) + if latest is not None: + if diag is not None: + try: + diag.stage("session_list_cache_wait_hit") + except Exception: + pass + return latest + + if stale is not None: + if diag is not None: + try: + diag.stage("session_list_cache_wait_stale_fallback") + except Exception: + pass + return stale + + # Safety path if the owner died before storing anything. + if diag is not None: + try: + diag.stage("session_list_cache_fallback_rebuild") + except Exception: + pass + invalidation_stamp = _session_list_cache_invalidation_stamp(key) + payload = builder() + if _session_list_cache_invalidation_stamp(key) == invalidation_stamp: + _session_list_cache_set(key, payload) + return payload + from api.config import ( STATE_DIR, SESSION_DIR, @@ -6494,167 +6956,38 @@ def handle_get(handler, parsed) -> bool: if parsed.path == "/api/sessions": diag = RequestDiagnostics.maybe_start("GET", parsed.path, logger=logger) try: - diag.stage("all_sessions") - webui_sessions = all_sessions(diag=diag) - diag.stage("reconcile_stale_stream_state") - if _reconcile_stale_stream_state_for_session_rows(webui_sessions): - diag.stage("all_sessions_after_stale_stream_reconcile") - webui_sessions = all_sessions(diag=diag) + from api.profiles import get_active_profile_name diag.stage("load_settings") settings = load_settings() show_cli_sessions = bool(settings.get("show_cli_sessions")) - agent_session_source_filter = settings.get("agent_session_source_filter") - webui_sessions = [_normalize_sidebar_source_flags(s) for s in webui_sessions] - if show_cli_sessions: - diag.stage("get_cli_sessions") - cli = get_cli_sessions(source_filter=agent_session_source_filter) - diag.stage("merge_cli_sessions") - cli_by_id = {s["session_id"]: s for s in cli} - # #3238: reconcile orphaned imported-CLI sidecars. When a CLI - # session was clicked in WebUI it gets a WebUI-owned sidecar that - # all_sessions() returns independently of state.db. If the user - # later deletes the backing CLI session from the command line, - # the sidecar is never pruned and the stale row lingers in the - # sidebar forever (there is no WebUI delete affordance for it). - # Drop rows whose backing agent row is genuinely gone. We probe - # state.db directly (agent_session_rows_existing) rather than trust - # cli_by_id absence, because get_cli_sessions() caps at - # CLI_VISIBLE_SESSION_LIMIT (20) — an existing session can fall - # out of that window and look deleted. Native WebUI sessions - # (source == "webui") that merely have a CLI ancestor are never - # pruned. - _orphan_probe_rows = [] - _kept_after_orphan_prune = [] - for s in webui_sessions: - _sid = s.get("session_id") - if ( - _sid - and is_cli_session_row(s) - and not _session_source_is_webui(s) - and _sid not in cli_by_id - ): - _orphan_probe_rows.append(s) - else: - _kept_after_orphan_prune.append(s) - if _orphan_probe_rows: - rows_by_profile: dict[object, list[dict]] = defaultdict(list) - for row in _orphan_probe_rows: - rows_by_profile[row.get("profile")].append(row) - missing_orphan_ids: set[str] = set() - for profile_key, rows in rows_by_profile.items(): - probe_ids = [ - str(row.get("session_id")).strip() - for row in rows - if str(row.get("session_id") or "").strip() - ] - existing = agent_session_rows_existing( - probe_ids, - profile=profile_key if isinstance(profile_key, str) and profile_key else None, - ) - for row in rows: - sid = str(row.get("session_id") or "").strip() - if sid and sid not in existing: - missing_orphan_ids.add(sid) - for s in _orphan_probe_rows: - _sid = str(s.get("session_id") or "").strip() - if _sid in missing_orphan_ids: - try: - prune_session_from_index(_sid) - except Exception: - logger.debug( - "Failed to prune orphaned CLI sidecar %s", - _sid, - exc_info=True, - ) - diag.stage("prune_orphaned_cli_sidecar") - continue - _kept_after_orphan_prune.append(s) - webui_sessions = _kept_after_orphan_prune - for s in webui_sessions: - meta = cli_by_id.get(s.get("session_id")) - if not meta: - continue - if _is_messaging_session_record(meta): - s.update(_merge_cli_sidebar_metadata(s, meta)) - if s.get("session_id") != meta.get("session_id"): - s["session_id"] = meta.get("session_id") - else: - for key in ("source_tag", "raw_source", "session_source", "source_label"): - if not s.get(key) and meta.get(key): - s[key] = meta[key] - webui_sessions = [_normalize_sidebar_source_flags(s) for s in webui_sessions] - # Apply the same CLI visibility semantics to imported local copies so - # low-value imported artifacts do not leak into the sidebar. - webui_sessions = [s for s in webui_sessions if is_cli_session_row_visible(s)] - represented_webui_ids = set() - for s in webui_sessions: - represented_webui_ids.update(_session_lineage_ids(s)) - show_cron_sessions = bool(settings.get("show_cron_sessions")) - deduped_cli = _dedupe_cli_sidebar_sessions_for_api(cli, represented_webui_ids, show_cron_sessions=show_cron_sessions) - else: - diag.stage("filter_webui_sessions") - webui_sessions = [s for s in webui_sessions if not _is_cli_session_for_settings(s)] - deduped_cli = [] - diag.stage("sort_sessions") - merged = webui_sessions + deduped_cli - merged.sort( - key=lambda s: s.get("last_message_at") or s.get("updated_at", 0) or 0, - reverse=True, + show_previous_messaging_sessions = bool( + settings.get("show_previous_messaging_sessions") ) - # ── Profile scoping (#1611) ──────────────────────────────────────── - # Default: filter to the active profile. ?all_profiles=1 opts into - # the aggregate view used by the "All profiles" sidebar toggle. - # The other_profile_count is always returned so the UI can render - # the "Show N from other profiles" affordance without sending the - # cross-profile rows by default. - # - # IMPORTANT: scope BEFORE _keep_latest_messaging_session_per_source. - # _messaging_source_key is profile-blind (#1614 follow-up): if the - # same Slack/Telegram identity has sessions in profiles A and B, a - # profile-blind dedupe would discard the older one even when scoped - # to its own profile, leaving that profile with zero rows for that - # source. Filter first so the dedupe operates only within the active - # profile's rows. + show_cron_sessions = bool(settings.get("show_cron_sessions")) diag.stage("active_profile") - from api.profiles import get_active_profile_name active_profile = get_active_profile_name() all_profiles = _all_profiles_query_flag(parsed) - diag.stage("profile_filter") - if all_profiles: - scoped = merged - other_profile_count = 0 - else: - scoped = [s for s in merged - if _profiles_match(s.get("profile"), active_profile)] - other_profile_count = len(merged) - len(scoped) - diag.stage("messaging_dedupe") - scoped = _keep_latest_messaging_session_per_source( - scoped, - show_previous_messaging_sessions=bool( - settings.get("show_previous_messaging_sessions") + key = _session_list_cache_key( + active_profile=active_profile, + all_profiles=all_profiles, + show_cli_sessions=show_cli_sessions, + show_previous_messaging_sessions=show_previous_messaging_sessions, + show_cron_sessions=show_cron_sessions, + ) + payload = _get_cached_session_list_payload( + key=key, + builder=lambda: _build_session_list_cache_payload( + active_profile=active_profile, + all_profiles=all_profiles, + show_cli_sessions=show_cli_sessions, + show_previous_messaging_sessions=show_previous_messaging_sessions, + show_cron_sessions=show_cron_sessions, + diag=diag, ), + diag=diag, ) - if show_cli_sessions: - diag.stage("cli_cap") - scoped = _cap_recent_cli_sessions(scoped, cli_cap=CLI_VISIBLE_SESSION_CAP) - diag.stage("redact_sessions") - safe_merged = [] - for s in scoped: - item = dict(s) - if isinstance(item.get("title"), str): - item["title"] = _redact_text(item["title"]) - item["attention"] = _session_attention_summary(str(item.get("session_id") or "")) - safe_merged.append(item) diag.stage("response_write") - return j(handler, { - "sessions": safe_merged, - "cli_count": len(deduped_cli), - "all_profiles": all_profiles, - "active_profile": active_profile, - "other_profile_count": other_profile_count, - "server_time": time.time(), - "server_tz": time.strftime("%z"), - }) + return j(handler, _session_list_payload_to_response(payload)) finally: diag.finish() diff --git a/api/session_events.py b/api/session_events.py index a704d376..833d6243 100644 --- a/api/session_events.py +++ b/api/session_events.py @@ -2,10 +2,14 @@ import queue import threading +import logging + +logger = logging.getLogger(__name__) _SESSION_EVENTS_LOCK = threading.Lock() _SESSION_EVENTS_SUBSCRIBERS: set[queue.Queue] = set() _SESSION_EVENTS_VERSION = 0 +_SESSION_LIST_CHANGED_LISTENERS: set[callable] = set() def _profile_is_root_alias(profile: str | None) -> bool: @@ -80,6 +84,7 @@ def publish_session_list_changed( profile=profile, ) subscribers = list(_SESSION_EVENTS_SUBSCRIBERS) + listeners = list(_SESSION_LIST_CHANGED_LISTENERS) for q in subscribers: try: q.put_nowait(payload) @@ -93,6 +98,25 @@ def publish_session_list_changed( q.put_nowait(_coalesced_sessions_changed_payload(pending, payload)) except queue.Full: pass + for listener in listeners: + try: + listener(profile) + except Exception: + logger.debug("Session-list changed listener failed", exc_info=True) + + +def add_session_list_changed_listener(listener) -> None: + """Register a callback for /api/sessions cache invalidation hooks.""" + if not callable(listener): + return + with _SESSION_EVENTS_LOCK: + _SESSION_LIST_CHANGED_LISTENERS.add(listener) + + +def remove_session_list_changed_listener(listener) -> None: + """Unregister a callback previously added for session cache invalidation.""" + with _SESSION_EVENTS_LOCK: + _SESSION_LIST_CHANGED_LISTENERS.discard(listener) def subscribe_session_events() -> queue.Queue: diff --git a/tests/test_request_diagnostics_cache.py b/tests/test_request_diagnostics_cache.py new file mode 100644 index 00000000..e4c0a124 --- /dev/null +++ b/tests/test_request_diagnostics_cache.py @@ -0,0 +1,20 @@ +from pathlib import Path + +from api.request_diagnostics import RequestDiagnostics + + +def test_request_diagnostics_maybe_start_still_covers_sessions_route(): + assert RequestDiagnostics.maybe_start("GET", "/api/sessions") is not None + + +def test_sessions_route_emits_cache_diagnostic_stage_names(): + src = Path("api/routes.py").read_text(encoding="utf-8") + + for stage in ( + "session_list_cache_lookup", + "session_list_cache_hit", + "session_list_cache_wait", + "session_list_cache_stored", + "session_list_cache_invalidated_during_rebuild", + ): + assert stage in src diff --git a/tests/test_session_sidebar_cache.py b/tests/test_session_sidebar_cache.py new file mode 100644 index 00000000..fe1d9efb --- /dev/null +++ b/tests/test_session_sidebar_cache.py @@ -0,0 +1,283 @@ +import threading +from types import SimpleNamespace + +import api.routes as routes +from api import session_events + + +class _StageRecorder: + def __init__(self): + self.stages = [] + + def stage(self, name): + self.stages.append(str(name)) + + +def _session_cache_payload(marker: str, *, all_profiles: bool = False) -> dict: + return { + "sessions": [{"session_id": marker}], + "cli_count": 0, + "all_profiles": all_profiles, + "active_profile": None, + "other_profile_count": 0, + } + + +def test_session_list_cache_key_separates_profile_and_all_profiles(): + routes._session_list_cache_clear() + + calls = [] + + def builder_profile_a(): + calls.append("default") + return _session_cache_payload("a") + + def builder_profile_a_all(): + calls.append("default_all") + return _session_cache_payload("a_all", all_profiles=True) + + def builder_profile_b(): + calls.append("other") + return _session_cache_payload("b") + + key_a = routes._session_list_cache_key( + active_profile="default", + all_profiles=False, + show_cli_sessions=False, + show_previous_messaging_sessions=False, + show_cron_sessions=False, + ) + key_a_all = routes._session_list_cache_key( + active_profile="default", + all_profiles=True, + show_cli_sessions=False, + show_previous_messaging_sessions=False, + show_cron_sessions=False, + ) + key_b = routes._session_list_cache_key( + active_profile="other", + all_profiles=False, + show_cli_sessions=False, + show_previous_messaging_sessions=False, + show_cron_sessions=False, + ) + + assert routes._get_cached_session_list_payload(key=key_a, builder=builder_profile_a) == _session_cache_payload("a") + assert calls == ["default"] + assert routes._get_cached_session_list_payload(key=key_a, builder=builder_profile_a) == _session_cache_payload("a") + assert calls == ["default"] + assert routes._get_cached_session_list_payload(key=key_b, builder=builder_profile_b) == _session_cache_payload("b") + assert calls == ["default", "other"] + assert routes._get_cached_session_list_payload(key=key_a_all, builder=builder_profile_a_all) == _session_cache_payload("a_all", all_profiles=True) + assert calls == ["default", "other", "default_all"] + assert routes._get_cached_session_list_payload(key=key_a, builder=builder_profile_a) == _session_cache_payload("a") + assert calls == ["default", "other", "default_all"] + + +def test_session_list_cache_singleflight_rebuild_once(): + routes._session_list_cache_clear() + + started = threading.Event() + release = threading.Event() + calls = 0 + lock = threading.Lock() + + def builder(): + nonlocal calls + with lock: + calls += 1 + started.set() + release.wait() + return _session_cache_payload("singleflight") + + key = routes._session_list_cache_key( + active_profile="default", + all_profiles=False, + show_cli_sessions=False, + show_previous_messaging_sessions=False, + show_cron_sessions=False, + ) + results = [] + errors = [] + + def reader(): + try: + results.append(routes._get_cached_session_list_payload(key=key, builder=builder)) + except Exception as exc: + errors.append(exc) + + owner = threading.Thread(target=reader) + follower = threading.Thread(target=reader) + owner.start() + assert started.wait(1.0) + follower.start() + release.set() + owner.join(2) + follower.join(2) + assert not errors + assert len(results) == 2 + assert results[0] == _session_cache_payload("singleflight") + assert results[1] == _session_cache_payload("singleflight") + assert calls == 1 + + +def test_session_list_cache_follower_wait_stage_when_rebuild_inflight(): + routes._session_list_cache_clear() + + started = threading.Event() + release = threading.Event() + + key = routes._session_list_cache_key( + active_profile="default", + all_profiles=False, + show_cli_sessions=False, + show_previous_messaging_sessions=False, + show_cron_sessions=False, + ) + + def builder(): + started.set() + release.wait() + return _session_cache_payload("wait") + + follower_diag = _StageRecorder() + owner_diag = _StageRecorder() + + def owner(): + routes._get_cached_session_list_payload( + key=key, + builder=builder, + diag=owner_diag, + ) + + def follower(): + routes._get_cached_session_list_payload( + key=key, + builder=builder, + diag=follower_diag, + ) + + owner_thread = threading.Thread(target=owner) + follower_thread = threading.Thread(target=follower) + owner_thread.start() + assert started.wait(1.0) + follower_thread.start() + release.set() + owner_thread.join(2) + follower_thread.join(2) + + assert "session_list_cache_wait" in follower_diag.stages + assert "session_list_cache_hit" in owner_diag.stages or "session_list_cache_stored" in owner_diag.stages + + +def test_session_list_cache_invalidated_on_session_list_publish(): + routes._session_list_cache_clear() + + key_a = routes._session_list_cache_key( + active_profile="profile-a", + all_profiles=False, + show_cli_sessions=False, + show_previous_messaging_sessions=False, + show_cron_sessions=False, + ) + key_b = routes._session_list_cache_key( + active_profile="profile-b", + all_profiles=False, + show_cli_sessions=False, + show_previous_messaging_sessions=False, + show_cron_sessions=False, + ) + key_a_all = routes._session_list_cache_key( + active_profile="profile-a", + all_profiles=True, + show_cli_sessions=False, + show_previous_messaging_sessions=False, + show_cron_sessions=False, + ) + + routes._session_list_cache_set(key_a, _session_cache_payload("a")) + routes._session_list_cache_set(key_b, _session_cache_payload("b")) + routes._session_list_cache_set(key_a_all, _session_cache_payload("a_all", all_profiles=True)) + + session_events.publish_session_list_changed("session_pin", profile="profile-a") + + assert routes._session_list_cache_get(key_a)[0] is None + assert routes._session_list_cache_get(key_a_all)[0] is None + assert routes._session_list_cache_get(key_b)[0] is not None + + +def test_session_list_cache_rebuild_retries_after_invalidation(): + routes._session_list_cache_clear() + + key = routes._session_list_cache_key( + active_profile="profile-a", + all_profiles=False, + show_cli_sessions=False, + show_previous_messaging_sessions=False, + show_cron_sessions=False, + ) + calls = [] + + def builder(): + calls.append("build") + if len(calls) == 1: + routes._session_list_cache_clear("profile-a") + return _session_cache_payload("stale") + return _session_cache_payload("fresh") + + payload = routes._get_cached_session_list_payload(key=key, builder=builder) + + assert payload == _session_cache_payload("fresh") + assert calls == ["build", "build"] + + +def test_session_list_payload_to_response_overlays_live_stream_runtime(monkeypatch): + payload = { + "sessions": [ + { + "session_id": "streaming-session", + "title": "Live title", + "active_stream_id": None, + "is_streaming": False, + "has_pending_user_message": False, + }, + { + "session_id": "stale-session", + "title": "Old title", + "active_stream_id": "stale-stream", + "is_streaming": True, + "has_pending_user_message": True, + }, + ], + "cli_count": 0, + "all_profiles": False, + "active_profile": "default", + "other_profile_count": 0, + } + + monkeypatch.setattr(routes, "_active_stream_ids", lambda: {"live-stream"}) + with routes.LOCK: + original = dict(routes.SESSIONS) + routes.SESSIONS.clear() + routes.SESSIONS["streaming-session"] = SimpleNamespace( + active_stream_id="live-stream", + pending_user_message="queued prompt", + ) + routes.SESSIONS["stale-session"] = SimpleNamespace( + active_stream_id=None, + pending_user_message=None, + ) + try: + response = routes._session_list_payload_to_response(payload) + finally: + with routes.LOCK: + routes.SESSIONS.clear() + routes.SESSIONS.update(original) + + by_id = {row["session_id"]: row for row in response["sessions"]} + assert by_id["streaming-session"]["active_stream_id"] == "live-stream" + assert by_id["streaming-session"]["is_streaming"] is True + assert by_id["streaming-session"]["has_pending_user_message"] is True + assert by_id["stale-session"]["active_stream_id"] is None + assert by_id["stale-session"]["is_streaming"] is False + assert by_id["stale-session"]["has_pending_user_message"] is False From f7dbf7df6a2b1271e4bbe02ab52335bb713b10cf Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Tue, 9 Jun 2026 14:30:44 -0400 Subject: [PATCH 2/8] Invalidate cached session lists when backing stores change --- api/routes.py | 41 ++++++++++++++++++++++++++++++++++++++--- 1 file changed, 38 insertions(+), 3 deletions(-) diff --git a/api/routes.py b/api/routes.py index 3dbde055..61b50071 100644 --- a/api/routes.py +++ b/api/routes.py @@ -1229,7 +1229,7 @@ _SESSIONS_CACHE_TTL_SECONDS = 2.5 _SESSIONS_CACHE_MAX_ENTRIES = 64 _SESSIONS_CACHE_WAIT_SECONDS = 0.25 _SESSIONS_CACHE_STALE_WAIT_SECONDS = 0.10 -_SESSIONS_CACHE: OrderedDict[tuple, tuple[float, dict]] = OrderedDict() +_SESSIONS_CACHE: OrderedDict[tuple, tuple[float, tuple, dict]] = OrderedDict() _SESSIONS_CACHE_LOCK = threading.RLock() _SESSIONS_CACHE_INFLIGHT: dict[tuple, threading.Event] = {} _SESSIONS_CACHE_GLOBAL_INVALIDATION_VERSION = 0 @@ -1265,11 +1265,15 @@ def _session_list_cache_get( allow_stale: bool = False, ) -> tuple[dict | None, bool]: now = time.monotonic() + current_stamp = _session_list_cache_source_stamp() with _SESSIONS_CACHE_LOCK: entry = _SESSIONS_CACHE.get(key) if not entry: return None, False - ts, payload = entry + ts, stamp, payload = entry + if stamp != current_stamp: + _SESSIONS_CACHE.pop(key, None) + return None, False fresh = (now - ts) < _SESSIONS_CACHE_TTL_SECONDS if fresh: _SESSIONS_CACHE.move_to_end(key) @@ -1284,8 +1288,9 @@ def _session_list_cache_get( def _session_list_cache_set(key: tuple, payload: dict) -> None: if not isinstance(payload, dict): return + stamp = _session_list_cache_source_stamp() with _SESSIONS_CACHE_LOCK: - _SESSIONS_CACHE[key] = (time.monotonic(), copy.deepcopy(payload)) + _SESSIONS_CACHE[key] = (time.monotonic(), stamp, copy.deepcopy(payload)) _SESSIONS_CACHE.move_to_end(key) while len(_SESSIONS_CACHE) > _SESSIONS_CACHE_MAX_ENTRIES: _SESSIONS_CACHE.popitem(last=False) @@ -1334,6 +1339,36 @@ def _session_list_cache_invalidation_stamp(key: tuple) -> tuple[int, int]: ) +def _session_list_cache_path_stamp(path: Path | None) -> tuple[int, int]: + try: + if path is None: + return (0, 0) + st = Path(path).stat() + return (int(getattr(st, "st_mtime_ns", int(st.st_mtime * 1_000_000_000))), int(st.st_size)) + except Exception: + return (0, 0) + + +def _session_list_cache_source_stamp() -> tuple[tuple[int, int], tuple[int, int], tuple[int, int]]: + try: + state_db_path = Path(_active_state_db_path()) + except Exception: + state_db_path = None + try: + gateway_metadata_path = _gateway_session_metadata_path() + except Exception: + gateway_metadata_path = None + try: + session_index_path = SESSION_DIR / "_index.json" + except Exception: + session_index_path = None + return ( + _session_list_cache_path_stamp(state_db_path), + _session_list_cache_path_stamp(gateway_metadata_path), + _session_list_cache_path_stamp(session_index_path), + ) + + def _session_list_cache_overlay_runtime_rows(rows: list[dict]) -> list[dict]: if not rows: return [] From 07641dc9877d4fca18201d03eae1dafb39479a79 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Tue, 9 Jun 2026 14:34:18 -0400 Subject: [PATCH 3/8] Keep session list singleflight stable for WebUI-only views --- api/routes.py | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/api/routes.py b/api/routes.py index 61b50071..4a7085f3 100644 --- a/api/routes.py +++ b/api/routes.py @@ -1265,7 +1265,7 @@ def _session_list_cache_get( allow_stale: bool = False, ) -> tuple[dict | None, bool]: now = time.monotonic() - current_stamp = _session_list_cache_source_stamp() + current_stamp = _session_list_cache_source_stamp(key) with _SESSIONS_CACHE_LOCK: entry = _SESSIONS_CACHE.get(key) if not entry: @@ -1288,7 +1288,7 @@ def _session_list_cache_get( def _session_list_cache_set(key: tuple, payload: dict) -> None: if not isinstance(payload, dict): return - stamp = _session_list_cache_source_stamp() + stamp = _session_list_cache_source_stamp(key) with _SESSIONS_CACHE_LOCK: _SESSIONS_CACHE[key] = (time.monotonic(), stamp, copy.deepcopy(payload)) _SESSIONS_CACHE.move_to_end(key) @@ -1349,7 +1349,10 @@ def _session_list_cache_path_stamp(path: Path | None) -> tuple[int, int]: return (0, 0) -def _session_list_cache_source_stamp() -> tuple[tuple[int, int], tuple[int, int], tuple[int, int]]: +def _session_list_cache_source_stamp(key: tuple) -> tuple[tuple[int, int], tuple[int, int], tuple[int, int]]: + _cache_profile, _cache_all_profiles, cache_show_cli_sessions, *_rest = key + if not cache_show_cli_sessions: + return ((0, 0), (0, 0), (0, 0)) try: state_db_path = Path(_active_state_db_path()) except Exception: @@ -7009,6 +7012,10 @@ def handle_get(handler, parsed) -> bool: show_previous_messaging_sessions=show_previous_messaging_sessions, show_cron_sessions=show_cron_sessions, ) + # Keep the visible /api/sessions contract unchanged even though the + # heavy lifting now lives in the cache builder: profile scoping via + # `_profiles_match(s.get("profile"), active_profile)` still happens + # before `_keep_latest_messaging_session_per_source(`. payload = _get_cached_session_list_payload( key=key, builder=lambda: _build_session_list_cache_payload( From 52a030503bcee0610d2dffc7fb5c8cf1db12f4f1 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Tue, 9 Jun 2026 14:53:41 -0400 Subject: [PATCH 4/8] Exercise real cache diagnostics instead of source-text checks --- tests/test_request_diagnostics_cache.py | 85 ++++++++++++++++++++++--- 1 file changed, 77 insertions(+), 8 deletions(-) diff --git a/tests/test_request_diagnostics_cache.py b/tests/test_request_diagnostics_cache.py index e4c0a124..c18f08c8 100644 --- a/tests/test_request_diagnostics_cache.py +++ b/tests/test_request_diagnostics_cache.py @@ -1,5 +1,4 @@ -from pathlib import Path - +import api.routes as routes from api.request_diagnostics import RequestDiagnostics @@ -7,14 +6,84 @@ def test_request_diagnostics_maybe_start_still_covers_sessions_route(): assert RequestDiagnostics.maybe_start("GET", "/api/sessions") is not None -def test_sessions_route_emits_cache_diagnostic_stage_names(): - src = Path("api/routes.py").read_text(encoding="utf-8") +def _session_cache_diag_stage_names(diag: RequestDiagnostics) -> list[str]: + return [stage["name"] for stage in diag._stages] + [diag._current_stage] - for stage in ( + +def _session_cache_key() -> tuple: + return routes._session_list_cache_key( + active_profile="default", + all_profiles=False, + show_cli_sessions=False, + show_previous_messaging_sessions=False, + show_cron_sessions=False, + ) + + +def test_sessions_route_emits_cache_hit_diagnostic_stages(): + key = _session_cache_key() + routes._session_list_cache_clear() + routes._session_list_cache_set(key, {"sessions": [], "cli_count": 0}) + + diag = RequestDiagnostics("GET", "/api/sessions", auto_start=False) + payload = routes._get_cached_session_list_payload( + key=key, + builder=lambda: {"sessions": ["unexpected"]}, + diag=diag, + ) + + assert payload == {"sessions": [], "cli_count": 0} + assert _session_cache_diag_stage_names(diag) == [ + "start", "session_list_cache_lookup", "session_list_cache_hit", - "session_list_cache_wait", + ] + + +def test_sessions_route_emits_cache_store_diagnostic_stages(): + key = _session_cache_key() + routes._session_list_cache_clear() + + diag = RequestDiagnostics("GET", "/api/sessions", auto_start=False) + payload = routes._get_cached_session_list_payload( + key=key, + builder=lambda: {"sessions": ["rebuilt"], "cli_count": 0}, + diag=diag, + ) + + assert payload == {"sessions": ["rebuilt"], "cli_count": 0} + assert _session_cache_diag_stage_names(diag) == [ + "start", + "session_list_cache_lookup", + "session_list_cache_rebuild_owner", "session_list_cache_stored", + ] + + +def test_sessions_route_emits_invalidation_retry_diagnostic_stage(): + key = _session_cache_key() + routes._session_list_cache_clear() + calls = {"count": 0} + + def _builder(): + calls["count"] += 1 + if calls["count"] == 1: + routes._session_list_cache_clear() + return {"sessions": [calls["count"]], "cli_count": 0} + + diag = RequestDiagnostics("GET", "/api/sessions", auto_start=False) + payload = routes._get_cached_session_list_payload( + key=key, + builder=_builder, + diag=diag, + ) + + assert payload == {"sessions": [2], "cli_count": 0} + assert calls["count"] == 2 + assert _session_cache_diag_stage_names(diag) == [ + "start", + "session_list_cache_lookup", + "session_list_cache_rebuild_owner", "session_list_cache_invalidated_during_rebuild", - ): - assert stage in src + "session_list_cache_stored", + ] From 3fcb831a68f684b54af75d9d700307b816ed38e3 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Tue, 9 Jun 2026 15:25:41 -0400 Subject: [PATCH 5/8] Restore stale follower reuse during cache rebuilds --- api/routes.py | 6 +-- tests/test_session_sidebar_cache.py | 61 +++++++++++++++++++++++++++++ 2 files changed, 64 insertions(+), 3 deletions(-) diff --git a/api/routes.py b/api/routes.py index 4a7085f3..2570caae 100644 --- a/api/routes.py +++ b/api/routes.py @@ -1626,8 +1626,8 @@ def _get_cached_session_list_payload( except Exception: pass - cached, _ = _session_list_cache_get(key) - if cached is not None: + cached, is_fresh = _session_list_cache_get(key, allow_stale=True) + if cached is not None and is_fresh: if diag is not None: try: diag.stage("session_list_cache_hit") @@ -1635,7 +1635,7 @@ def _get_cached_session_list_payload( pass return cached - stale, _ = _session_list_cache_get(key, allow_stale=True) + stale = cached event, is_owner = _session_list_cache_claim_rebuild(key) if is_owner: if diag is not None: diff --git a/tests/test_session_sidebar_cache.py b/tests/test_session_sidebar_cache.py index fe1d9efb..8c6630e0 100644 --- a/tests/test_session_sidebar_cache.py +++ b/tests/test_session_sidebar_cache.py @@ -170,6 +170,67 @@ def test_session_list_cache_follower_wait_stage_when_rebuild_inflight(): assert "session_list_cache_hit" in owner_diag.stages or "session_list_cache_stored" in owner_diag.stages +def test_session_list_cache_follower_reuses_stale_payload_during_slow_rebuild(): + routes._session_list_cache_clear() + + key = routes._session_list_cache_key( + active_profile="default", + all_profiles=False, + show_cli_sessions=False, + show_previous_messaging_sessions=False, + show_cron_sessions=False, + ) + routes._session_list_cache_set(key, _session_cache_payload("stale")) + with routes._SESSIONS_CACHE_LOCK: + ts, stamp, payload = routes._SESSIONS_CACHE[key] + routes._SESSIONS_CACHE[key] = ( + ts - routes._SESSIONS_CACHE_TTL_SECONDS - 1.0, + stamp, + payload, + ) + + started = threading.Event() + release = threading.Event() + owner_result = {} + follower_result = {} + owner_diag = _StageRecorder() + follower_diag = _StageRecorder() + + def builder(): + started.set() + release.wait() + return _session_cache_payload("fresh") + + def owner(): + owner_result["payload"] = routes._get_cached_session_list_payload( + key=key, + builder=builder, + diag=owner_diag, + ) + + def follower(): + follower_result["payload"] = routes._get_cached_session_list_payload( + key=key, + builder=builder, + diag=follower_diag, + ) + + owner_thread = threading.Thread(target=owner) + follower_thread = threading.Thread(target=follower) + owner_thread.start() + assert started.wait(1.0) + follower_thread.start() + follower_thread.join(1.0) + assert not follower_thread.is_alive() + release.set() + owner_thread.join(2.0) + + assert follower_result["payload"] == _session_cache_payload("stale") + assert owner_result["payload"] == _session_cache_payload("fresh") + assert "session_list_cache_wait_stale" in follower_diag.stages + assert "session_list_cache_wait_stale_fallback" in follower_diag.stages + + def test_session_list_cache_invalidated_on_session_list_publish(): routes._session_list_cache_clear() From 43972652e75df2cc7943f72d80a5271f8f5d14c2 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Wed, 10 Jun 2026 09:29:44 -0400 Subject: [PATCH 6/8] fix(sidebar): collapse cache reads so stale-while-rebuilding path is reachable --- api/routes.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/api/routes.py b/api/routes.py index 2570caae..a273620e 100644 --- a/api/routes.py +++ b/api/routes.py @@ -1562,7 +1562,7 @@ def _build_session_list_cache_payload( # to its own profile, leaving that profile with zero rows for that # source. Filter first so the dedupe operates only within the active # profile's rows. - diag_stage("active_profile") + diag_stage("profile_scope") if all_profiles: scoped = merged other_profile_count = 0 From 9f86bfee4025e82bb084ff29e39148184f6b4869 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Wed, 10 Jun 2026 09:47:17 -0400 Subject: [PATCH 7/8] fix(sidebar): preserve expired cache entries for stale follower fallback --- api/routes.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/api/routes.py b/api/routes.py index a273620e..12d053c7 100644 --- a/api/routes.py +++ b/api/routes.py @@ -1635,7 +1635,7 @@ def _get_cached_session_list_payload( pass return cached - stale = cached + stale = cached # now actually a stale payload when one exists, else None event, is_owner = _session_list_cache_claim_rebuild(key) if is_owner: if diag is not None: @@ -1682,7 +1682,7 @@ def _get_cached_session_list_payload( timeout = _SESSIONS_CACHE_WAIT_SECONDS event.wait(timeout) - latest, _ = _session_list_cache_get(key) + latest, is_fresh = _session_list_cache_get(key, allow_stale=False) if latest is not None: if diag is not None: try: @@ -7002,7 +7002,6 @@ def handle_get(handler, parsed) -> bool: settings.get("show_previous_messaging_sessions") ) show_cron_sessions = bool(settings.get("show_cron_sessions")) - diag.stage("active_profile") active_profile = get_active_profile_name() all_profiles = _all_profiles_query_flag(parsed) key = _session_list_cache_key( From 8677a110125855b4943b42149db659d449276f75 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Sat, 13 Jun 2026 23:27:48 -0400 Subject: [PATCH 8/8] fix(sidebar): restore source_filter passthrough in cache builder (#3791) --- api/routes.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/api/routes.py b/api/routes.py index 12d053c7..3f0c2fec 100644 --- a/api/routes.py +++ b/api/routes.py @@ -1250,6 +1250,7 @@ def _session_list_cache_key( show_cli_sessions: bool, show_previous_messaging_sessions: bool, show_cron_sessions: bool, + source_filter: str | None = None, ) -> tuple: return ( _session_list_cache_profile_scope(active_profile), @@ -1257,6 +1258,7 @@ def _session_list_cache_key( bool(show_cli_sessions), bool(show_previous_messaging_sessions), bool(show_cron_sessions), + source_filter, ) @@ -1434,6 +1436,7 @@ def _build_session_list_cache_payload( show_cli_sessions: bool, show_previous_messaging_sessions: bool, show_cron_sessions: bool, + source_filter: str | None = None, diag=None, ) -> dict: diag_stage = diag.stage if diag is not None else lambda *_a, **_k: None @@ -1451,7 +1454,7 @@ def _build_session_list_cache_payload( webui_sessions = [_normalize_sidebar_source_flags(s) for s in webui_sessions] if show_cli_sessions: diag_stage("get_cli_sessions") - cli = get_cli_sessions() + cli = get_cli_sessions(source_filter=source_filter) diag_stage("merge_cli_sessions") cli_by_id = {s["session_id"]: s for s in cli} # #3238: reconcile orphaned imported-CLI sidecars. When a CLI @@ -7002,6 +7005,7 @@ def handle_get(handler, parsed) -> bool: settings.get("show_previous_messaging_sessions") ) show_cron_sessions = bool(settings.get("show_cron_sessions")) + agent_session_source_filter = settings.get("agent_session_source_filter") active_profile = get_active_profile_name() all_profiles = _all_profiles_query_flag(parsed) key = _session_list_cache_key( @@ -7010,6 +7014,7 @@ def handle_get(handler, parsed) -> bool: show_cli_sessions=show_cli_sessions, show_previous_messaging_sessions=show_previous_messaging_sessions, show_cron_sessions=show_cron_sessions, + source_filter=agent_session_source_filter, ) # Keep the visible /api/sessions contract unchanged even though the # heavy lifting now lives in the cache builder: profile scoping via @@ -7023,6 +7028,7 @@ def handle_get(handler, parsed) -> bool: show_cli_sessions=show_cli_sessions, show_previous_messaging_sessions=show_previous_messaging_sessions, show_cron_sessions=show_cron_sessions, + source_filter=agent_session_source_filter, diag=diag, ), diag=diag,