test(webui): add registration-window race regression + grace guard

This commit is contained in:
allenliang2022 authored and nesquena-hermes committed 2026-06-27 15:24:15 +00:00
1 parent f3da26dc07
commit dbe470aa6f
2 files changed
+90

No files matched your search

+26
View File
@@ -2311,6 +2311,15 @@ def _repair_stale_pending(session) -> bool:
return False
# Grace window (seconds) for the state.db->sidecar read-side self-heal. A turn's
# active_stream_id + pending_started_at are persisted to the sidecar a moment
# BEFORE the worker registers the stream in STREAMS/ACTIVE_RUNS. Within this
# window an in-flight, not-yet-registered stream must not be mistaken for a dead
# one and cleared. Streams that genuinely lost their terminal event are far
# older than this, so they still self-heal.
_STATE_DB_SYNC_REGISTRATION_GRACE_SECONDS = 30.0
def _sync_sidecar_from_state_db_if_newer(session) -> bool:
"""Read-side self-heal when WebUI sidecar lags Hermes state.db.
@@ -2346,6 +2355,22 @@ def _sync_sidecar_from_state_db_if_newer(session) -> bool:
active_stream_id = getattr(session, 'active_stream_id', None)
if active_stream_id and active_stream_id in _active_stream_ids():
return False
# Registration-window race guard. A turn is registered in STREAMS/ACTIVE_RUNS
# by the worker thread a moment AFTER the request handler persists
# active_stream_id + pending_started_at to the sidecar. In that brief window
# the stream is legitimately in flight yet not yet visible in the registries,
# so the check above would mis-classify it as a dead stream and clear it.
# Treat a *recent* pending_started_at as "still starting up" and refuse to
# self-heal it; a stream that genuinely lost its terminal event will have a
# pending_started_at well older than this window.
if active_stream_id:
pending_started_at = getattr(session, 'pending_started_at', None)
try:
pending_age = time.time() - float(pending_started_at)
except (TypeError, ValueError):
pending_age = None
if pending_age is not None and pending_age < _STATE_DB_SYNC_REGISTRATION_GRACE_SECONDS:
return False
try:
state_summary = get_state_db_session_summary(
sid,
@@ -2356,6 +2381,7 @@ def _sync_sidecar_from_state_db_if_newer(session) -> bool:
except Exception:
logger.debug("state.db summary check failed for session %s", sid, exc_info=True)
return False
if state_count <= 0:
return False
@@ -17,6 +17,7 @@ The scenario this test pins down:
"""
from concurrent.futures import ThreadPoolExecutor
import threading
import time
import pytest
@@ -978,3 +979,66 @@ def test_sync_persists_recovered_state_db_tail_when_stream_dead(monkeypatch):
assert reloaded.messages[-1]["content"] == "recovered tail"
def test_sync_skips_during_registration_window_recent_pending(monkeypatch):
"""Registration-window race: a just-submitted, not-yet-registered stream
must NOT be cleared by the self-heal.
The request handler persists active_stream_id + pending_started_at to the
sidecar a moment BEFORE the worker thread registers the stream in
STREAMS/ACTIVE_RUNS. In that window the stream is absent from the registries
but the turn is legitimately starting. A *recent* pending_started_at must
keep the self-heal off so it can't drop a turn that is about to run.
(maintainer-requested regression)
"""
sid = "state_registration_window_sid"
stream_id = "stream_just_submitted_not_registered"
s = Session(
session_id=sid,
title="Registration window",
messages=[
{"role": "user", "content": "old question", "timestamp": 100.0},
{"role": "assistant", "content": "old answer", "timestamp": 101.0},
],
active_stream_id=stream_id,
pending_user_message="new request",
# Submitted just now: inside the registration grace window.
pending_started_at=time.time(),
)
s.save()
# state.db is ahead (the new turn's worker has started writing rows), which
# would otherwise satisfy the "newer transcript" gate.
state_messages = [
{"role": "user", "content": "old question", "timestamp": 100.0},
{"role": "assistant", "content": "old answer", "timestamp": 101.0},
{"role": "user", "content": "new request", "timestamp": time.time()},
{"role": "assistant", "content": "first streamed token", "timestamp": time.time()},
]
monkeypatch.setattr(
models,
"get_state_db_session_summary",
lambda sid_arg, profile=None: {"message_count": len(state_messages), "last_message_at": time.time()},
)
monkeypatch.setattr(
models,
"get_state_db_session_messages",
lambda sid_arg, **kwargs: list(state_messages),
)
# The stream is NOT in STREAMS/ACTIVE_RUNS yet (registration not done).
config.STREAMS.pop(stream_id, None)
config.ACTIVE_RUNS.pop(stream_id, None)
result = models._sync_sidecar_from_state_db_if_newer(s)
# Self-heal must back off: the turn is still starting up.
assert result is False
assert s.active_stream_id == stream_id
assert s.pending_user_message == "new request"
assert [m["content"] for m in s.messages] == ["old question", "old answer"]
reloaded = Session.load(sid)
assert reloaded.active_stream_id == stream_id
assert reloaded.pending_user_message == "new request"