DasPoschi/hermes-webui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: recover pending turn after stale stream restart

Browse Source
This commit is contained in:
Manfred
2026-05-03 19:47:40 +02:00
parent 84e74407c9
commit afaeb03532
3 changed files with 99 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
api/models.py
View File

@@ -542,11 +542,23 @@ def _apply_core_sync_or_error_marker(
if require_stream_dead and session.active_stream_id in _active_stream_ids():
return False
# When messages is already non-empty the core-sync overwrite and recovered
# user turn are skipped (we cannot clobber in-memory mutations), but the
# stuck pending fields MUST still be cleared and an error marker appended
# so the session isn't permanently left in stale-pending state.
# When messages is already non-empty, do not overwrite history from any core
# transcript. The pending user turn may still be the only durable copy of a
# prompt submitted just before a server restart, so materialize it before
# clearing runtime stream state.
if len(session.messages) != 0:
_recovered_ts = int(time.time())
if isinstance(session.pending_started_at, (int, float)) and session.pending_started_at > 0:
_recovered_ts = int(session.pending_started_at)
recovered = {
'role': 'user',
'content': session.pending_user_message,
'timestamp': _recovered_ts,
'_recovered': True,
}
if session.pending_attachments:
recovered['attachments'] = list(session.pending_attachments)
session.messages.append(recovered)
session.active_stream_id = None
session.pending_user_message = None
session.pending_attachments = []
@@ -559,7 +571,7 @@ def _apply_core_sync_or_error_marker(
})
session.save()
logger.info(
"Session %s: pending cleared (messages non-empty), added error marker",
"Session %s: recovered pending user turn (messages non-empty), added error marker",
sid,
)
return True
@@ -636,8 +648,7 @@ def _repair_stale_pending(session) -> bool:
# _apply_core_sync_or_error_marker uses this to detect a rotated active_stream_id
# (e.g. context compression) or a stream that came back alive.
_seen_stream_id = session.active_stream_id
if (len(session.messages) != 0
or not session.pending_user_message
if (not session.pending_user_message
or not _seen_stream_id
or _seen_stream_id in _active_stream_ids()):
return False

54
tests/test_session_sidecar_repair.py
View File

@@ -457,14 +457,14 @@ class TestCancelInProgressGuard:
class TestEmptyMessagesGuard:
"""_apply_core_sync_or_error_marker bails out when session.messages is
non-empty, preventing it from clobbering in-memory mutations made by the
streaming thread or cancel path."""
"""_apply_core_sync_or_error_marker preserves existing messages when
session.messages is non-empty, while still recovering the pending user turn
before clearing stale stream runtime fields."""
def test_pending_cleared_when_messages_nonempty_direct(self, hermes_home, monkeypatch):
"""When _apply_core_sync_or_error_marker is called on a session with
non-empty messages and pending set, it clears the pending fields and
appends an error marker, returning True."""
non-empty messages and pending set, it recovers the pending user turn,
clears the pending fields, and appends an error marker."""
s = _make_session(messages=[{"role": "user", "content": "hello"}])
s.pending_user_message = "Another question"
s.active_stream_id = "stream_1"
@@ -477,11 +477,14 @@ class TestEmptyMessagesGuard:
)
assert result is True
# Original message should be untouched
assert len(s.messages) == 2 # original + error marker
# Original message should be untouched, pending turn recovered, then marker appended
assert len(s.messages) == 3 # original + recovered user turn + error marker
assert s.messages[0]["content"] == "hello"
assert s.messages[1]["role"] == "user"
assert s.messages[1]["content"] == "Another question"
assert s.messages[1].get("_recovered") is True
# Error marker appended
assert s.messages[1].get("_error") is True
assert s.messages[2].get("_error") is True
# Pending fields cleared
assert s.pending_user_message is None
assert s.active_stream_id is None
@@ -517,13 +520,13 @@ class TestEmptyMessagesGuard:
class TestNonEmptyMessagesPendingCleared:
"""When messages is non-empty and pending is stuck, _last_resort_sync_from_core
clears the pending fields and appends exactly one error marker without
clobbering existing messages or syncing from core."""
preserves existing messages, recovers the pending user turn, and appends
exactly one error marker without syncing from core."""
def test_pending_cleared_when_messages_nonempty(self, hermes_home, monkeypatch):
"""_last_resort_sync_from_core on a session with both messages and
pending_user_message clears pending fields and appends exactly one
error marker."""
pending_user_message recovers that pending turn before clearing runtime
fields and appending exactly one error marker."""
s = _make_session(messages=[{"role": "user", "content": "existing turn"}])
s.pending_user_message = "Stuck draft"
s.pending_attachments = [{"type": "image", "name": "screenshot.png"}]
@@ -543,9 +546,9 @@ class TestNonEmptyMessagesPendingCleared:
streaming._last_resort_sync_from_core(s, "stale_stream", agent_lock)
# Existing messages preserved untouched
assert len(s.messages) == 2, (
f"Expected 2 messages (original + error marker), got {len(s.messages)}"
# Existing messages preserved untouched, pending turn recovered, error marker appended
assert len(s.messages) == 3, (
f"Expected 3 messages (original + recovered turn + error marker), got {len(s.messages)}"
)
assert s.messages[0]["role"] == "user"
assert s.messages[0]["content"] == "existing turn"
@@ -553,15 +556,18 @@ class TestNonEmptyMessagesPendingCleared:
"Core transcript must NOT be synced when messages is non-empty"
)
# Exactly one recovered user turn
recovered_msgs = [m for m in s.messages if m.get("_recovered")]
assert len(recovered_msgs) == 1
assert recovered_msgs[0]["role"] == "user"
assert recovered_msgs[0]["content"] == "Stuck draft"
assert recovered_msgs[0]["attachments"] == [{"type": "image", "name": "screenshot.png"}]
# Exactly one error marker
error_msgs = [m for m in s.messages if m.get("_error")]
assert len(error_msgs) == 1
assert "Previous turn did not complete" in error_msgs[0]["content"]
# No recovered user turn (messages is non-empty, so skip that)
recovered_msgs = [m for m in s.messages if m.get("_recovered")]
assert len(recovered_msgs) == 0
# Pending fields fully cleared
assert s.pending_user_message is None
assert s.pending_attachments == []
@@ -719,14 +725,18 @@ class TestRepairStalePendingIntegration:
error_msgs = [m for m in s.messages if m.get("_error")]
assert len(error_msgs) == 1
def test_skips_when_messages_nonempty(self, hermes_home, monkeypatch):
"""Pre-check: if messages is non-empty, repair is skipped entirely."""
def test_recovers_when_messages_nonempty(self, hermes_home, monkeypatch):
"""Pre-check: if messages is non-empty, repair still preserves the
pending user turn instead of silently discarding it."""
s = _make_session(messages=[{"role": "user", "content": "hi"}])
s.pending_user_message = "more"
s.active_stream_id = "stream_1"
result = _repair_stale_pending(s)
assert result is False
assert result is True
assert [m["content"] for m in s.messages if m["role"] == "user"] == ["hi", "more"]
assert s.messages[1].get("_recovered") is True
assert any(m.get("_error") for m in s.messages)
def test_skips_when_stream_alive(self, hermes_home, monkeypatch):
"""Pre-check: if the stream is still alive in STREAMS, repair is skipped."""

49
tests/test_stale_stream_pending_recovery.py Normal file
View File

@@ -0,0 +1,49 @@
"""Regression: stale stream cleanup must not discard pending user turns.
A server restart drops the in-memory STREAMS table. Browser reload then calls
get_session(), which clears stale active_stream_id state. For long conversations
that already have messages, the pending_user_message can be the only durable copy
of the user turn that was submitted just before the restart.
"""
import api.config as config
import api.models as models
from api.models import Session, get_session
def test_stale_stream_cleanup_recovers_pending_turn_on_non_empty_session(tmp_path, monkeypatch):
session_dir = tmp_path / "sessions"
session_dir.mkdir()
monkeypatch.setattr(models, "SESSION_DIR", session_dir)
monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json")
models.SESSIONS.clear()
config.STREAMS.clear()
s = Session(
session_id="stale_stream_nonempty",
title="Existing long chat",
messages=[
{"role": "user", "content": "previous prompt", "timestamp": 100},
{"role": "assistant", "content": "previous answer", "timestamp": 101},
],
)
s.active_stream_id = "dead_stream"
s.pending_user_message = "new prompt that must survive restart"
s.pending_attachments = [{"name": "note.txt", "path": "/tmp/note.txt"}]
s.pending_started_at = 123
s.save()
recovered = get_session("stale_stream_nonempty")
assert recovered.active_stream_id is None
assert recovered.pending_user_message is None
assert any(
msg.get("role") == "user"
and msg.get("content") == "new prompt that must survive restart"
and msg.get("_recovered") is True
for msg in recovered.messages
)
assert any(
msg.get("role") == "assistant" and msg.get("_error") is True
for msg in recovered.messages
)