fix(webui): render a server-initiated turn that finished during an SSE gap on a visible tab

A visible tab can permanently miss a server-initiated turn (self-wake /
cron / restart hook) when its per-session SSE is momentarily down at the
emit instant. server_turn_started is a fire-and-forget broadcast with no
replay buffer; if the turn also finishes before the tab reconnects, the
on-subscribe self-heal finds no live run in ACTIVE_RUNS and replays
nothing, so the transcript stays stale until a hard refresh.

Extend the self-heal: the (re)subscribing tab reports its last-known
message_count (?known_count); when there is no live run to replay, the
handler compares the persisted count and, if the server is ahead, emits a
lightweight session-updated frame. The frontend syncs incrementally via
the #5189 keepStaleUntilLoaded swap-in-place loadSession path (no
clear+refetch, so the #5177/#5189 blank-gap jump is not reintroduced).
Idle-only and pane-scoped; emits only when the server is strictly ahead.
This commit is contained in:
allenliang2022 authored and nesquena-hermes committed 2026-06-30 17:13:45 +00:00
1 parent a09eb83f52
commit 4ef9d7c5cb
9 files changed
+400 -8

No files matched your search

+65
View File
@@ -288,6 +288,71 @@ def active_stream_id_for_session(session_id: str) -> Optional[str]:
return None
def persisted_message_count_for_session(session_id: str) -> Optional[int]:
"""Cheap, metadata-only persisted ``message_count`` for *session_id*, or None.
Companion to ``active_stream_id_for_session`` for the per-session SSE
on-subscribe self-heal. ``active_stream_id_for_session`` recovers a turn
that is live RIGHT NOW (replay ``server_turn_started``). But a
SERVER-initiated turn (self-wake / cron / restart hook) can start AND
finish entirely inside an SSE gap: the fire-and-forget
``server_turn_started`` reached no subscriber AND the run already cleared
from ``ACTIVE_RUNS`` by the time the tab reconnects, so the replay finds
nothing (returns None) and the tab's transcript stays stale until a hard
refresh — the reported visible-tab defect. To detect that case the handler
compares the freshly-(re)subscribed tab's last-known count against this
persisted count; a server that is AHEAD means a turn landed during the gap.
Reads via ``metadata_only=True`` so it never parses the full transcript
(this runs on every per-session SSE (re)connect). The persisted count is
written by ``Session.save`` as ``meta['message_count'] = len(messages)`` —
the SAME basis the frontend's ``S.session.message_count`` is built from —
so the comparison is apples-to-apples. Returns None when the count is
unknown (legacy sidecars without a persisted count); the caller treats
None as "cannot tell, do nothing", never as a trigger.
"""
try:
from api.models import get_session
s = get_session(session_id, metadata_only=True)
count = getattr(s, "_metadata_message_count", None)
if count is None:
msgs = getattr(s, "messages", None)
count = len(msgs) if isinstance(msgs, list) and msgs else None
return int(count) if count is not None else None
except Exception:
logger.debug(
"persisted_message_count_for_session lookup failed for %s",
session_id,
exc_info=True,
)
return None
def should_emit_session_updated(
subscriber_known_count: Optional[int],
persisted_count: Optional[int],
) -> bool:
"""Gate for the per-session SSE "finished during the gap" self-heal emit.
Single source of truth shared by the SSE handler and its tests so the two
cannot drift (the handler MUST call this, not inline the comparison).
Emit a ``session-updated`` frame ONLY when:
* the (re)subscribing tab reported a last-known count (``?known_count``),
AND
* the persisted server-side count is known, AND
* the server is STRICTLY ahead (a turn landed during the gap).
A missing known count (tab didn't report), an unknown persisted count
(legacy sidecar), or an equal/behind count all return False — never a
spurious reload.
"""
if subscriber_known_count is None:
return False
if persisted_count is None:
return False
return persisted_count > subscriber_known_count
def _reaper_loop() -> None:
logger.info("SessionChannel reaper thread started")
while not _REAPER_STOP.is_set():
+42
View File
@@ -16187,9 +16187,22 @@ def _handle_session_sse_stream(handler, parsed):
if not sid:
return bad(handler, "session_id is required")
# The (re)subscribing tab reports its last-known message_count via
# ?known_count=N so the on-subscribe self-heal can detect a server-initiated
# turn that started AND finished entirely inside this tab's SSE gap (see the
# "server-initiated turn finished during the gap" self-heal block below).
# Absent/blank/non-numeric => None ("tab didn't report", never triggers).
_known_count_raw = parse_qs(parsed.query).get("known_count", [""])[0]
try:
subscriber_known_count = int(_known_count_raw) if _known_count_raw != "" else None
except (TypeError, ValueError):
subscriber_known_count = None
from api.background_process import (
subscribe_to_session_channel,
active_stream_id_for_session,
persisted_message_count_for_session,
should_emit_session_updated,
)
# Atomic get-or-create + subscribe under SESSION_CHANNELS_LOCK. Doing these
@@ -16274,6 +16287,35 @@ def _handle_session_sse_stream(handler, parsed):
"source": "subscribe_recovery",
"recovered": True,
})
else:
# ── Server-initiated turn that FINISHED during the SSE gap ──
# The block above only heals a turn that is live RIGHT NOW. But
# a server-initiated turn (self-wake / cron / restart hook) can
# start AND finish entirely inside the gap: the fire-and-forget
# `server_turn_started` reached no subscriber, and by the time
# this tab reconnects the run has already cleared from
# ACTIVE_RUNS — so active_stream_id_for_session returns None and
# nothing above replays. The turn IS persisted, but this tab's
# transcript stays stale until a hard refresh (the reported
# visible-tab defect). Detect it by comparing the persisted
# message_count against what this (re)subscribing tab last knew
# (?known_count). If the server is AHEAD, emit a lightweight
# `session-updated` frame so the tab does an INCREMENTAL,
# swap-in-place message sync (frontend reuses #5189's
# keepStaleUntilLoaded loadSession path — NO clear+refetch, so
# the #5177/#5189 blank-gap jump is not reintroduced). Carries
# only counts (no transcript) to stay cheap. Skipped
# entirely when the tab didn't report a count or the persisted
# count is unknown (legacy sidecar) → never a spurious reload.
if subscriber_known_count is not None:
persisted_count = persisted_message_count_for_session(sid)
if should_emit_session_updated(subscriber_known_count, persisted_count):
_sse(handler, 'session-updated', {
"session_id": sid,
"message_count": persisted_count,
"known_count": subscriber_known_count,
"source": "subscribe_recovery",
})
except _CLIENT_DISCONNECT_ERRORS:
# Client vanished mid-recovery — re-raise so the outer handler
# treats it as a normal disconnect and the finally still cleans up.
+54 -1
View File
@@ -6663,7 +6663,25 @@ function startSessionStream(sid) {
// Tab is visible — the real SSE owns live-view; ensure no stale hidden poll.
_stopHiddenActiveStreamPoll();
try {
const es = new EventSource(_apiUrl('api/session/stream?session_id=' + encodeURIComponent(sid)));
// Report our last-known message_count so the server's on-subscribe
// self-heal can detect a server-initiated turn (self-wake / cron / restart)
// that started AND finished entirely inside an SSE gap — the fire-and-forget
// `server_turn_started` was missed AND the run already cleared from
// ACTIVE_RUNS by the time we reconnect, so the `server_turn_started` replay
// finds nothing. When the server is ahead of this count it emits a
// `session-updated` frame (handled below) and we incrementally sync. Use the
// session's full message_count (same basis the server persists), NOT
// S.messages.length, which is only the rendered tail window for long
// sessions and would false-trigger a reload on every reconnect.
let _knownCount = '';
try {
if (S.session && S.session.session_id === sid && Number.isFinite(Number(S.session.message_count))) {
_knownCount = String(Number(S.session.message_count));
}
} catch (_) {}
const _streamUrl = 'api/session/stream?session_id=' + encodeURIComponent(sid)
+ (_knownCount !== '' ? '&known_count=' + encodeURIComponent(_knownCount) : '');
const es = new EventSource(_apiUrl(_streamUrl));
_sessionEventSource = es;
es.addEventListener('initial', () => { /* connection confirmed */ });
es.addEventListener('bg_task_complete', e => {
@@ -6672,6 +6690,41 @@ function startSessionStream(sid) {
_handleBgTaskCompleteEvent(e, sid, {source: 'session'});
}
});
// ── Visible-tab self-heal: a server-initiated turn finished during an SSE
// gap ─────────────────────────────────────────────────────────────────
// Distinct from `server_turn_started` (which attaches a LIVE stream): this
// frame fires when the server detected, at our (re)subscribe, that a turn
// started AND finished while our EventSource was momentarily down (so we
// missed the live broadcast and the run already cleared). The turn is
// persisted but our transcript is stale and there is NO live stream left to
// attach. Incrementally sync via the SAME swap-in-place path #5189 uses for
// the hidden-tab return (loadSession force + keepStaleUntilLoaded): the new
// transcript replaces the old in a single render frame — NO clear+refetch,
// so the #5177/#5189 blank-gap "jump" is not reintroduced.
es.addEventListener('session-updated', e => {
try {
const d = JSON.parse(e.data || '{}');
const evSid = d.session_id || sid;
if (evSid !== sid) return;
// Only act when this session is the one on screen and we're idle (no
// live turn rendering — that path owns its own message updates).
const isCurrent = (typeof _isSessionCurrentPane === 'function')
? _isSessionCurrentPane(sid)
: (S.session && S.session.session_id === sid);
if (!isCurrent) return;
if (S.activeStreamId) return;
// Re-check against our CURRENT known count — a concurrent load may have
// already caught us up between the server's emit and now.
const localCount = (S.session && S.session.session_id === sid && Number.isFinite(Number(S.session.message_count)))
? Number(S.session.message_count)
: (Array.isArray(S.messages) ? S.messages.length : 0);
const serverCount = Number(d.message_count);
if (!Number.isFinite(serverCount) || serverCount <= localCount) return;
if (typeof loadSession === 'function') {
void loadSession(sid, {force: true, externalRefreshReason: 'session-updated', keepStaleUntilLoaded: true});
}
} catch (_) {}
});
// ── Defect B: live-view of server-initiated (Option Z) turns ──────────
// The drain thread starts the wakeup turn server-side and the server
// fans a `server_turn_started` {stream_id} frame onto this per-session
+16 -1
View File
@@ -1418,7 +1418,10 @@ async function loadSession(sid){
_setActiveSessionUrl(S.session.session_id);
if(typeof startSessionStream==='function') startSessionStream(S.session.session_id);
const activeStreamId=S.session.active_stream_id||null;
// `let` (not const): re-read below, after the awaited _ensureMessagesLoaded,
// so a server_turn_started that attaches a live stream MID-RELOAD is honored
// by the attach/idle decision instead of being clobbered by the stale snapshot.
let activeStreamId=S.session.active_stream_id||null;
// If the server says the session is idle, reset browser-side streaming flags
// NOW — before the async _ensureMessagesLoaded gap below. Without this,
// S.busy can remain true from a still-running stream in the PREVIOUS session
@@ -1654,6 +1657,18 @@ async function loadSession(sid){
// Attach pending user message if one is queued.
_mergePendingSessionMessage(S.session,S.messages);
// Self-heal-vs-live-render race guard (maintainer/Codex-reproduced; verified
// in an isolated instance). `activeStreamId` was snapshotted BEFORE the
// awaited _ensureMessagesLoaded above. During a force reload (the
// `session-updated` self-heal or any keepStaleUntilLoaded recovery), a
// server-initiated turn can fire `server_turn_started` mid-await and set
// S.activeStreamId for THIS sid. Without re-reading, the idle branch below
// would clear S.activeStreamId/S.busy off the stale (null) snapshot and
// silently kill the live turn's render. Fold a concurrently-attached
// same-session stream into activeStreamId so the existing attach branch
// (and all its `attachLiveStream(sid, activeStreamId, ...)` calls) keeps it.
activeStreamId = activeStreamId || ((S.activeStreamId && S.session && S.session.session_id===sid) ? S.activeStreamId : null);
if(activeStreamId){
S.busy=true;
S.activeStreamId=activeStreamId;
+1 -1
View File
@@ -129,7 +129,7 @@ if(!_isDuplicateGatewaySessionSnapshot([null, {session_id:'web-2', session_sourc
def test_load_session_persists_only_after_metadata_loads():
"""Do not overwrite the last good localStorage sid before /api/session succeeds."""
src = _read(SESSIONS_JS)
load = _block(src, "async function loadSession(sid)", "\n const activeStreamId=")
load = _block(src, "async function loadSession(sid)", "activeStreamId=S.session.active_stream_id")
api_pos = load.index("data = await api(`/api/session")
persist_pos = load.index("localStorage.setItem('hermes-webui-session',S.session.session_id)")
+5 -1
View File
@@ -138,7 +138,11 @@ def test_attach_live_stream_different_stream_still_reopens_transport():
def test_load_session_reattach_path_uses_attach_live_stream_for_running_sessions():
"""The session switch-back path should still route through attachLiveStream()."""
body = _function_body(SESSIONS_JS, "loadSession")
active_pos = body.find("const activeStreamId=S.session.active_stream_id||null")
# Anchor without the `const`/`let` keyword: the declaration was widened to
# `let` so the race-guard can re-read activeStreamId after the awaited
# message load (#5248). The invariant this test pins — the snapshot is taken
# before, and used by, the attachLiveStream reattach — is unchanged.
active_pos = body.find("activeStreamId=S.session.active_stream_id||null")
reattach_pos = body.find("attachLiveStream(sid, activeStreamId")
assert active_pos != -1
assert reattach_pos != -1
@@ -42,7 +42,7 @@ def test_model_selection_records_pending_state_before_async_session_update():
def test_load_session_applies_pending_model_before_first_topbar_sync():
"""Reload should project the pending selection before server old metadata wins."""
body = _body_between(SESSIONS_JS, "async function loadSession", "const activeStreamId=")
body = _body_between(SESSIONS_JS, "async function loadSession", "activeStreamId=S.session.active_stream_id")
apply_idx = body.index("_applyPendingSessionModelForSession")
sync_idx = body.index("syncTopbar()")
+211 -1
View File
@@ -368,7 +368,7 @@ def test_session_sse_handler_wires_on_subscribe_recovery():
# The recovery must be inside the session SSE handler and use the
# recovered marker so the frontend uses the replay attach path.
handler_ix = src.index("def _handle_session_sse_stream")
handler_src = src[handler_ix:handler_ix + 6000]
handler_src = src[handler_ix:handler_ix + 9000]
assert "active_stream_id_for_session" in handler_src
assert '"recovered": True' in handler_src
assert "server_turn_started" in handler_src
@@ -551,3 +551,213 @@ def test_frontend_handler_requires_event_id_to_surface():
assert "_bgTaskCompleteRingBufferAdd(sid, evt_id)" in fn_src
# Ack body carries event_id so server can correlate.
assert "event_id: evt_id" in fn_src
# ---------------------------------------------------------------------------
# Visible-tab self-heal: server-initiated turn FINISHED during the SSE gap
# ---------------------------------------------------------------------------
#
# Root cause (distinct from the live-run replay above): a server-initiated
# turn (self-wake / cron / restart hook) can start AND finish entirely while a
# VISIBLE tab's per-session EventSource is momentarily down (transient SSE
# drop / reverse-proxy idle-timeout / connection-pool starvation). The
# fire-and-forget `server_turn_started` reaches no subscriber, and by the time
# the tab reconnects the run has already cleared from ACTIVE_RUNS — so
# active_stream_id_for_session returns None and the live-run replay finds
# nothing. The turn IS persisted; only the tab's live-view is stale until a
# hard refresh.
#
# Fix: the (re)subscribing tab reports its last-known message_count
# (?known_count); when the live-run replay is a no-op, the handler compares the
# persisted count against it and, if the server is AHEAD, emits a lightweight
# `session-updated` frame so the tab incrementally syncs via the #5189
# keepStaleUntilLoaded swap-in-place path (NO clear+refetch → no blank-gap
# jump). Reproduced deterministically with the controlled Playwright harness in
# references/visible-tab-misses-wake-during-sse-gap.md (SSE force-closed at the
# wakeup-emit instant; pre-fix gotServerTurnStarted:false / msgCount:0 while the
# sidecar already had the turn).
# ---------------------------------------------------------------------------
def _make_meta_session(sid, count, updated_at=123.0):
"""Build a fake metadata-only session stub mirroring load_metadata_only()."""
class _MetaSession:
session_id = sid
_metadata_message_count = count
_loaded_metadata_only = True
messages = []
updated_at = None
s = _MetaSession()
s.updated_at = updated_at
return s
def test_persisted_message_count_uses_metadata_only(monkeypatch):
"""The companion lookup must return the persisted count via a metadata-only
load (never parsing the full transcript) and None when unknown."""
from api import background_process as bp
import api.models as models
sid = "sess-persisted-count"
# Normal: metadata stub carries _metadata_message_count.
monkeypatch.setattr(models, "get_session", lambda _sid, metadata_only=False: _make_meta_session(_sid, 7), raising=True)
assert bp.persisted_message_count_for_session(sid) == 7
# Unknown count (legacy sidecar, no persisted count, empty messages) → None
# so the caller treats it as "cannot tell", never a spurious trigger.
monkeypatch.setattr(models, "get_session", lambda _sid, metadata_only=False: _make_meta_session(_sid, None), raising=True)
assert bp.persisted_message_count_for_session(sid) is None
# Lookup failure (e.g. corrupt sidecar) is swallowed → None, never raises.
def _boom(_sid, metadata_only=False):
raise RuntimeError("decode error")
monkeypatch.setattr(models, "get_session", _boom, raising=True)
assert bp.persisted_message_count_for_session(sid) is None
def test_persisted_message_count_requests_metadata_only(monkeypatch):
"""Guard the perf contract: the lookup MUST pass metadata_only=True so it
never parses a 400KB+ transcript on every per-session SSE (re)connect."""
from api import background_process as bp
import api.models as models
seen = {}
def _spy(_sid, metadata_only=False):
seen["metadata_only"] = metadata_only
return _make_meta_session(_sid, 3)
monkeypatch.setattr(models, "get_session", _spy, raising=True)
assert bp.persisted_message_count_for_session("sess-spy") == 3
assert seen.get("metadata_only") is True
def test_session_sse_handler_wires_finished_during_gap_self_heal():
"""Source-grep: the per-session SSE handler must (a) parse ?known_count,
(b) in the live-run-absent branch compare the persisted count against it,
and (c) emit a `session-updated` frame when the server is ahead — all so a
turn that finished during the SSE gap still self-heals on a visible tab."""
src = (REPO_ROOT / "api" / "routes.py").read_text()
handler_ix = src.index("def _handle_session_sse_stream")
handler_src = src[handler_ix:handler_ix + 9000]
# (a) the subscriber reports its last-known count.
assert "known_count" in handler_src
assert "subscriber_known_count" in handler_src
# (b) the no-live-run branch consults the persisted count companion.
assert "persisted_message_count_for_session" in handler_src
# (c) and emits the incremental-sync frame only when ahead.
assert "session-updated" in handler_src
# The persisted-count comparison must live in the ELSE of the live-run
# replay (a finished-during-gap turn has NO active stream to replay), and
# the count comparison gates the emit. Anchor on the EMIT call itself
# (`_sse(handler, 'session-updated'`), not the first `session-updated`
# occurrence — that one is in the explanatory comment, before the gate.
su_ix = handler_src.index("_sse(handler, 'session-updated'")
gate_src = handler_src[max(0, su_ix - 1200):su_ix]
assert "should_emit_session_updated(" in gate_src
def test_session_updated_emit_gate_is_the_real_handler_function():
"""Behavioural contract for the count gate — exercises the ACTUAL function
the SSE handler calls (`should_emit_session_updated`), NOT a hand-rolled
copy. If the gate drifts (e.g. `>` becomes `>=`, or a short-circuit is
added), this test moves with it because it imports the real implementation.
Emit ONLY when a known subscriber count exists AND the persisted count
strictly exceeds it.
"""
from api.background_process import should_emit_session_updated as gate
# Server ahead → emit.
assert gate(5, 7) is True
# Equal (tab already current) → no emit.
assert gate(7, 7) is False
# Behind (shouldn't happen, but never reload backwards) → no emit.
assert gate(7, 5) is False
# Tab didn't report a count → no emit.
assert gate(None, 7) is False
# Persisted count unknown (legacy sidecar) → no emit.
assert gate(5, None) is False
# Both unknown → no emit.
assert gate(None, None) is False
def test_sse_handler_uses_shared_emit_gate_not_inline_comparison():
"""The handler MUST call `should_emit_session_updated` rather than inline
the `persisted_count > subscriber_known_count` comparison — that shared gate
is the single source of truth the behavioural test above locks down. An
inline comparison would let the handler's real logic drift away from the
tested function (greptile P2 r…: the gate test must exercise the handler,
not a copy)."""
src = (REPO_ROOT / "api" / "routes.py").read_text()
handler_ix = src.index("def _handle_session_sse_stream")
handler_src = src[handler_ix:handler_ix + 9000]
# The shared gate is imported and called in the emit branch.
assert "should_emit_session_updated" in handler_src
# And the inline form is GONE (the comparison lives only in the gate fn).
su_ix = handler_src.index("_sse(handler, 'session-updated'")
gate_src = handler_src[max(0, su_ix - 1200):su_ix]
assert "should_emit_session_updated(" in gate_src
assert "> subscriber_known_count" not in gate_src
def test_frontend_subscribes_with_known_count_and_handles_session_updated():
"""Source-grep: the frontend must (a) append &known_count from the
session's FULL message_count (not the rendered tail S.messages.length), and
(b) handle the `session-updated` frame via the keepStaleUntilLoaded
swap-in-place loadSession path (no clear+refetch → no blank-gap jump)."""
js = (REPO_ROOT / "static" / "messages.js").read_text()
# (a) known_count is sent on subscribe, sourced from S.session.message_count.
assert "known_count" in js
ss_ix = js.index("function startSessionStream")
ss_src = js[ss_ix:ss_ix + 4000]
assert "known_count=" in ss_src
assert "S.session.message_count" in ss_src
# Must NOT key the reported count on S.messages.length (tail window) — that
# would false-trigger a reload on every reconnect for long sessions.
knc_ix = ss_src.index("_knownCount")
knc_src = ss_src[knc_ix:knc_ix + 400]
assert "S.messages.length" not in knc_src
# (b) the session-updated listener routes through the swap-in-place path.
su_ix = js.index("addEventListener('session-updated'")
su_src = js[su_ix:su_ix + 1400]
assert "keepStaleUntilLoaded: true" in su_src or "keepStaleUntilLoaded:true" in su_src
assert "loadSession(" in su_src
# Idle-only: must bail when a live turn is rendering (that path owns its own
# message updates) and when the session isn't the one on screen.
assert "S.activeStreamId" in su_src
assert "isCurrent" in su_src
def test_loadsession_idle_cleanup_does_not_clobber_concurrent_live_stream():
"""Self-heal-vs-live-render race guard (maintainer/Codex-reproduced; verified
in an isolated instance).
The `session-updated` self-heal calls `loadSession(force, keepStaleUntilLoaded)`,
which awaits `_ensureMessagesLoaded`. A server-initiated turn can fire
`server_turn_started` DURING that await and set `S.activeStreamId` for the
same sid. The post-await branch must NOT clear `S.activeStreamId`/`S.busy`
based only on the stale `activeStreamId` snapshot taken before the await —
that silently kills the live turn's render. It must re-read the LIVE state and
prefer a concurrently-attached same-session stream over the stale snapshot.
"""
js = (REPO_ROOT / "static" / "sessions.js").read_text()
compact = js.replace(" ", "").replace("\n", "")
# `activeStreamId` is declared `let` (not const) so it can be re-read.
assert "letactiveStreamId=S.session.active_stream_id||null" in compact, (
"activeStreamId must be `let` so the race-guard can re-read it post-await"
)
# Before the attach/idle branch, fold a concurrently-attached same-session
# live stream into activeStreamId so a mid-reload server_turn_started is
# preserved (attached), not clobbered by the idle cleanup.
assert "activeStreamId=activeStreamId||((S.activeStreamId&&S.session&&S.session.session_id===sid)?S.activeStreamId:null)" in compact, (
"missing the post-await race-guard re-read that folds a concurrent "
"same-session live stream into activeStreamId before the idle branch"
)
# The re-read must sit BEFORE the attach/idle decision (`if(activeStreamId){`).
reread_ix = compact.find("activeStreamId=activeStreamId||((S.activeStreamId")
branch_ix = compact.find("if(activeStreamId){")
assert reread_ix != -1 and branch_ix != -1 and reread_ix < branch_ix, (
"race-guard re-read must precede the attach/idle branch"
)
+5 -2
View File
@@ -642,8 +642,11 @@ def test_reload_path_restores_pending_message_and_reattaches_live_stream(cleanup
assert 'pending_user_message' in ui_src
assert 'function attachLiveStream' in messages_src
assert 'const pendingMsg=typeof getPendingSessionMessage' in sessions_src
assert ('const activeStreamId=data.session.active_stream_id||null;' in sessions_src or
'const activeStreamId=S.session.active_stream_id||null;' in sessions_src)
# `activeStreamId` declaration was widened const → let (#5248 race-guard
# re-reads it after the awaited message load). Accept either keyword via a
# prefix anchor that omits const/let.
assert ('activeStreamId=data.session.active_stream_id||null;' in sessions_src or
'activeStreamId=S.session.active_stream_id||null;' in sessions_src)
assert 'attachLiveStream(sid, activeStreamId' in sessions_src
assert 'if (S.activeStreamId && S.activeStreamId === streamId) return;' in ui_src
active_branch_start = sessions_src.index("if(activeStreamId){\n S.busy=true;")