stage #4680 (nesquena-hermes): freeze /api/sessions cache during streaming (fixes #4672) (#4690)

Freezes the session-list cache source-stamp on the SET of active stream ids (not per-write
state) so mid-stream sidebar polls hold the cache instead of rebuilding all_sessions() every
poll and contending for the global LOCK the streaming worker holds (the ~2 tok/s + 15s
/api/sessions cause). Structural mutations still invalidate via publish_session_list_changed;
only a streaming session's own title/message_count lags <=TTL. routes.py + 109-line test.
99 session-list/sidebar/cache tests pass.

Co-authored-by: nesquena-hermes <agent@nesquena-hermes>
Co-authored-by: nesquena-hermes <nesquena-hermes@users.noreply.github.com>
This commit is contained in:
authored and GitHub committed 2026-06-22 02:11:04 -07:00
1 parent 12ff6e807d
commit a66913ac26
3 files changed
+185 -9

No files matched your search

+6
View File
@@ -3,6 +3,12 @@
## [Unreleased]
## [v0.51.584] — 2026-06-22 — Release UQ (freeze /api/sessions cache during streaming)
### Fixed
- **`/api/sessions` no longer crawls (and drags streaming to a few tokens/sec) during an active chat turn.** While a turn streamed, every message-row write advanced the session-list cache's source fingerprint, so each sidebar poll popped the cache and forced a full `all_sessions()` rebuild — which then contended for the same global lock the streaming worker holds, producing multi-second (occasionally ~15s) `/api/sessions` latencies and ~2 tok/s output. The cache source stamp is now frozen on the *set* of actively-streaming sessions (not per-write state), so it holds steady mid-stream and rebuilds only at the normal TTL cadence; structural sidebar changes (new/deleted/renamed sessions, attention, cron completion) still invalidate immediately, and a streaming session's own title/message-count refreshes the moment the turn ends. (#4680, fixes #4672)
## [v0.51.583] — 2026-06-22 — Release UP (no legacy interim toggle in anchor Worklog)
### Fixed
+70 -9
View File
@@ -1699,10 +1699,80 @@ def _session_list_cache_path_stamp(path: Path | None) -> tuple[int, int]:
return (0, 0)
def _session_list_cache_streaming_freeze_marker():
"""Return a hold-down marker while any session is actively streaming, else None.
During an active chat turn the gateway/CLI writes message rows to state.db
continuously. Each write advances the WAL stat and the content fingerprint
(``MAX(rowid)`` of ``messages``) that ``_session_list_cache_source_stamp``
folds in, so the source stamp changes on essentially every ``/api/sessions``
poll — popping the cache and forcing a full ``all_sessions()`` rebuild
mid-stream. That rebuild then contends for the global ``LOCK`` the streaming
worker holds while writing, which is what drags token output down to
~2 tok/s and produces the multi-second (and occasional ~15s) ``/api/sessions``
latencies in issue #4672.
The marker is keyed only on the *set* of active stream ids, not on any
per-write state, so:
* while the same turn(s) stream, the marker is constant → the cache holds
steady and rebuilds are bounded to the TTL cadence (one per
``_SESSIONS_CACHE_TTL_SECONDS``) instead of one per poll;
* the instant a stream starts or stops, the active set changes → the
marker changes → the cache re-validates and the just-finished turn's
final title/message_count is picked up immediately.
Structural sidebar mutations (new/deleted/renamed/imported sessions,
attention, cron completion) do NOT rely on this stamp — they invalidate the
cache directly through the ``publish_session_list_changed`` listener — so the
only thing that can lag under the hold-down is a streaming session's own
title/message_count, which already tolerates a <=TTL refresh delay.
"""
try:
active = _active_stream_ids()
except Exception:
return None
if not active:
return None
try:
return ("streaming", tuple(sorted(str(x) for x in active)))
except Exception:
return ("streaming",)
def _session_list_cache_source_stamp(key: tuple) -> tuple[tuple[int, int], tuple[int, int], tuple[int, int], tuple[int, int], tuple[int, int], object, 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), (0, 0), (0, 0), None, 0)
try:
settings_file = SETTINGS_FILE
except Exception:
settings_file = None
try:
from api.config import _SETTINGS_WRITE_VERSION
swv = _SETTINGS_WRITE_VERSION
except Exception:
swv = 0
# Streaming hold-down (#4672): while a turn is in flight, collapse the
# volatile state.db-derived components (db/WAL stat, gateway metadata, index
# stat, content fingerprint) to a marker that only changes when a stream
# starts or stops. This stops per-token message writes from busting the
# cache and triggering LOCK-contending rebuilds on every poll. The TTL still
# forces a periodic rebuild so the streaming session's own count/title stay
# fresh within the TTL window, and settings_file + the settings write
# version stay live so user-initiated sidebar/setting toggles invalidate
# immediately. Skipping the fingerprint's SQLite connect here also makes the
# streaming-path stamp strictly cheaper than the idle path.
streaming_marker = _session_list_cache_streaming_freeze_marker()
if streaming_marker is not None:
return (
streaming_marker,
streaming_marker,
streaming_marker,
streaming_marker,
_session_list_cache_path_stamp(settings_file),
streaming_marker,
swv,
)
try:
state_db_path = Path(_active_state_db_path())
except Exception:
@@ -1719,15 +1789,6 @@ def _session_list_cache_source_stamp(key: tuple) -> tuple[tuple[int, int], tuple
session_index_path = SESSION_DIR / "_index.json"
except Exception:
session_index_path = None
try:
settings_file = SETTINGS_FILE
except Exception:
settings_file = None
try:
from api.config import _SETTINGS_WRITE_VERSION
swv = _SETTINGS_WRITE_VERSION
except Exception:
swv = 0
return (
_session_list_cache_path_stamp(state_db_path),
_session_list_cache_path_stamp(state_db_wal_path),
+109
View File
@@ -406,3 +406,112 @@ def test_session_list_payload_to_response_overlays_live_stream_runtime(monkeypat
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
def _build_stamp_env(tmp_path, monkeypatch):
"""Wire a self-contained source-stamp environment and return its key."""
state_db = tmp_path / "state.db"
state_db.write_text("db", encoding="utf-8")
state_db_wal = tmp_path / "state.db-wal"
state_db_wal.write_text("wal-1", encoding="utf-8")
gateway = tmp_path / "gateway-sessions.json"
gateway.write_text("{}", encoding="utf-8")
session_dir = tmp_path / "sessions"
session_dir.mkdir()
(session_dir / "_index.json").write_text("{}", encoding="utf-8")
settings_file = tmp_path / "settings.json"
settings_file.write_text("{}", encoding="utf-8")
monkeypatch.setattr(routes, "_active_state_db_path", lambda: str(state_db))
monkeypatch.setattr(routes, "_gateway_session_metadata_path", lambda: gateway)
monkeypatch.setattr(routes, "SESSION_DIR", session_dir)
monkeypatch.setattr(routes, "SETTINGS_FILE", settings_file)
# Make the content fingerprint deterministic and unaffected by the dummy
# text-file state.db (a real sqlite connect would just return None here).
fingerprint = {"value": (1, 1)}
monkeypatch.setattr(
routes,
"_session_list_cache_state_db_fingerprint",
lambda _p: fingerprint["value"],
)
key = routes._session_list_cache_key(
active_profile="default",
all_profiles=False,
show_cli_sessions=True,
show_previous_messaging_sessions=False,
show_cron_sessions=False,
)
return key, state_db_wal, settings_file, fingerprint
def test_source_stamp_freezes_during_streaming_message_writes(tmp_path, monkeypatch):
"""#4672: per-token state.db churn must not bust the cache mid-stream.
With an active stream the volatile state.db-derived components are collapsed
to a stream-set marker, so advancing the WAL stat AND the content
fingerprint (the per-message-write signals) leaves the stamp unchanged.
Before the fix each of these advanced the stamp and forced a rebuild on
every poll.
"""
key, state_db_wal, _settings_file, fingerprint = _build_stamp_env(
tmp_path, monkeypatch
)
monkeypatch.setattr(routes, "_active_stream_ids", lambda: {"turn-1"})
before = routes._session_list_cache_source_stamp(key)
# Simulate the writes an active chat turn makes to state.db: WAL grows and
# the messages-table fingerprint advances.
state_db_wal.write_text("wal-2-grew-a-lot", encoding="utf-8")
fingerprint["value"] = (2, 99)
after = routes._session_list_cache_source_stamp(key)
assert after == before
def test_source_stamp_changes_when_stream_set_transitions(tmp_path, monkeypatch):
"""The hold-down marker tracks the active-stream SET, so a turn starting or
finishing re-validates the cache and the final title/count is picked up."""
key, _wal, _settings, _fp = _build_stamp_env(tmp_path, monkeypatch)
streams = {"value": set()}
monkeypatch.setattr(routes, "_active_stream_ids", lambda: set(streams["value"]))
idle = routes._session_list_cache_source_stamp(key)
streams["value"] = {"turn-1"}
streaming = routes._session_list_cache_source_stamp(key)
streams["value"] = {"turn-1", "turn-2"}
two_streams = routes._session_list_cache_source_stamp(key)
streams["value"] = set()
idle_again = routes._session_list_cache_source_stamp(key)
assert idle != streaming
assert streaming != two_streams
# Returning to idle re-engages the live state.db stamp path.
assert idle_again != streaming
def test_source_stamp_tracks_settings_even_while_streaming(tmp_path, monkeypatch):
"""Settings-file / sidebar-toggle changes must still invalidate the cache
during streaming so user-initiated changes are never held stale."""
key, _wal, settings_file, _fp = _build_stamp_env(tmp_path, monkeypatch)
monkeypatch.setattr(routes, "_active_stream_ids", lambda: {"turn-1"})
before = routes._session_list_cache_source_stamp(key)
settings_file.write_text('{"show_cli_sessions": true}', encoding="utf-8")
after = routes._session_list_cache_source_stamp(key)
assert after != before
def test_source_stamp_still_tracks_wal_when_idle(tmp_path, monkeypatch):
"""Regression guard: with NO active stream the stamp must still advance on a
state.db/WAL write (the original commit-reliable behavior is preserved)."""
key, state_db_wal, _settings, _fp = _build_stamp_env(tmp_path, monkeypatch)
monkeypatch.setattr(routes, "_active_stream_ids", lambda: set())
before = routes._session_list_cache_source_stamp(key)
state_db_wal.write_text("wal-2-more", encoding="utf-8")
after = routes._session_list_cache_source_stamp(key)
assert after != before