diff --git a/api/models.py b/api/models.py index 50490446..8f8c491d 100644 --- a/api/models.py +++ b/api/models.py @@ -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 diff --git a/tests/test_session_sidecar_repair.py b/tests/test_session_sidecar_repair.py index 75b6b49d..e95efafb 100644 --- a/tests/test_session_sidecar_repair.py +++ b/tests/test_session_sidecar_repair.py @@ -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.""" diff --git a/tests/test_stale_stream_pending_recovery.py b/tests/test_stale_stream_pending_recovery.py new file mode 100644 index 00000000..debf4be1 --- /dev/null +++ b/tests/test_stale_stream_pending_recovery.py @@ -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 + )