From dbe470aa6fb3b136fd0e99c8fe9f4adcb7c17102 Mon Sep 17 00:00:00 2001 From: allenliang2022 Date: Sat, 27 Jun 2026 22:56:15 +0800 Subject: [PATCH] test(webui): add registration-window race regression + grace guard --- api/models.py | 26 ++++++++ .../test_session_lost_response_regression.py | 64 +++++++++++++++++++ 2 files changed, 90 insertions(+) diff --git a/api/models.py b/api/models.py index 54517a81..5a032ebd 100644 --- a/api/models.py +++ b/api/models.py @@ -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 diff --git a/tests/test_session_lost_response_regression.py b/tests/test_session_lost_response_regression.py index 6ff50515..750e62ba 100644 --- a/tests/test_session_lost_response_regression.py +++ b/tests/test_session_lost_response_regression.py @@ -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" + + +