fix(#5940): key pending-turn recovery to exact checkpoints

api/models.py

@@ -801,19 +801,26 @@ def _append_recovered_turn_to_context(session, recovered: dict) -> None:
     context_messages = getattr(session, 'context_messages', None)
     if not isinstance(context_messages, list) or not context_messages:
         return
-    role = str(recovered.get('role') or '')
-    recovered_text = " ".join(str(recovered.get('content') or '').split())
+    recovered_text = _normalize_journal_recovery_text(recovered.get('content'))
     if not recovered_text and not recovered.get('tool_call_id') and not recovered.get('tool_calls'):
         return
     if recovered_text:
-        for existing in reversed(context_messages[-8:]):
-            if not isinstance(existing, dict) or existing.get('role') != role:
-                continue
-            existing_text = " ".join(str(existing.get('content') or '').split())
-            if existing_text == recovered_text:
+        if recovered.get('role') == 'user':
+            if _message_matches_pending_checkpoint(
+                context_messages[-1] if context_messages else None,
+                recovered.get('content'),
+                recovered.get('timestamp'),
+                recovered.get('_source'),
+                recovered.get('attachments'),
+            ):
                 return
-    context_entry = {k: v for k, v in recovered.items() if k != 'timestamp'}
-    context_messages.append(context_entry)
+        else:
+            for existing in reversed(context_messages[-8:]):
+                if not isinstance(existing, dict) or existing.get('role') != recovered.get('role'):
+                    continue
+                if _normalize_journal_recovery_text(existing.get('content')) == recovered_text:
+                    return
+    context_messages.append(dict(recovered))
 
 
 def _append_recovered_pending_turn(session, *, timestamp: int | None = None) -> dict | None:
@@ -2191,6 +2198,41 @@ def _normalize_journal_recovery_text(value) -> str:
     return " ".join(str(value or "").split())
 
 
+def _message_matches_pending_checkpoint(message, pending_text, timestamp, source, attachments):
+    if not isinstance(message, dict) or message.get('role') != 'user':
+        return False
+    try:
+        message_timestamp = int(message.get('timestamp'))
+        expected_timestamp = int(timestamp)
+    except (TypeError, ValueError):
+        return False
+    return (
+        _normalize_journal_recovery_text(message.get('content'))
+        == _normalize_journal_recovery_text(pending_text)
+        and message_timestamp == expected_timestamp
+        and (message.get('_source') or 'webui') == (source or 'webui')
+        and list(message.get('attachments') or []) == list(attachments or [])
+    )
+
+
+def _message_matches_pending_text(message, pending_text):
+    if not isinstance(message, dict) or message.get('role') != 'user':
+        return False
+    return (
+        _normalize_journal_recovery_text(message.get('content'))
+        == _normalize_journal_recovery_text(pending_text)
+    )
+
+
+def _latest_user_matches_pending_text(messages, pending_text):
+    if not isinstance(messages, list) or not pending_text:
+        return False
+    for message in reversed(messages):
+        if isinstance(message, dict) and message.get('role') == 'user':
+            return _message_matches_pending_text(message, pending_text)
+    return False
+
+
 def _partial_message_signature(message: dict) -> tuple:
     """Return a stable identity for partial assistant markers recovered on load."""
     if not isinstance(message, dict):
@@ -2887,21 +2929,24 @@ def _apply_core_sync_or_error_marker(
     # prompt submitted just before a server restart, so materialize it before
     # clearing runtime stream state.
     if len(session.messages) != 0:
-        _pending_text = " ".join(str(session.pending_user_message or "").split())
-        _already_checkpointed = False
-        if _pending_text and session.messages:
-            for _last_msg in reversed(session.messages):
-                if isinstance(_last_msg, dict) and _last_msg.get('role') == 'user':
-                    _last_text = " ".join(str(_last_msg.get('content') or "").split())
-                    _already_checkpointed = _last_text == _pending_text
-                    break
         _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)
+        _already_checkpointed = _message_matches_pending_checkpoint(
+            session.messages[-1],
+            session.pending_user_message,
+            _recovered_ts,
+            session.pending_user_source,
+            session.pending_attachments,
+        )
+        _tail_user_already_checkpointed = _already_checkpointed or _message_matches_pending_text(
+            session.messages[-1],
+            session.pending_user_message,
+        )
         _stream_id = stream_id_for_recheck or session.active_stream_id
         _pending_started_at = session.pending_started_at
         if _run_journal_terminal_state(session, _stream_id) == 'completed':
-            if not _already_checkpointed:
+            if not (_already_checkpointed or _latest_user_matches_pending_text(session.messages, session.pending_user_message)):
                 _append_recovered_pending_turn(session, timestamp=_recovered_ts)
             _append_journaled_partial_output(
                 session,
@@ -2920,14 +2965,18 @@ def _apply_core_sync_or_error_marker(
                 _stream_id,
             )
             return True
-        if not _already_checkpointed:
+        if not _tail_user_already_checkpointed:
             _append_recovered_pending_turn(session, timestamp=_recovered_ts)
         else:
             recovered = {
                 'role': 'user',
                 'content': session.pending_user_message,
+                'timestamp': _recovered_ts,
                 '_recovered': True,
             }
+            pending_source = getattr(session, 'pending_user_source', None)
+            if pending_source and pending_source != 'webui':
+                recovered['_source'] = pending_source
             if session.pending_attachments:
                 recovered['attachments'] = list(session.pending_attachments)
             _append_recovered_turn_to_context(session, recovered)
@@ -2968,21 +3017,25 @@ def _apply_core_sync_or_error_marker(
                 if core.get(field) is not None:
                     setattr(session, field, core[field])
             _pending_text = _normalize_journal_recovery_text(session.pending_user_message)
-            _already_checkpointed = False
-            if _pending_text and session.messages:
-                for _last_msg in reversed(session.messages):
-                    if isinstance(_last_msg, dict) and _last_msg.get('role') == 'user':
-                        _last_text = _normalize_journal_recovery_text(_last_msg.get('content'))
-                        _already_checkpointed = _last_text == _pending_text
-                        break
+            _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)
+            _already_checkpointed = _message_matches_pending_checkpoint(
+                session.messages[-1] if session.messages else None,
+                session.pending_user_message,
+                _recovered_ts,
+                session.pending_user_source,
+                session.pending_attachments,
+            )
+            _tail_user_already_checkpointed = _already_checkpointed or _message_matches_pending_text(
+                session.messages[-1] if session.messages else None,
+                session.pending_user_message,
+            )
             if (
                 _pending_text
-                and not _already_checkpointed
+                and not _tail_user_already_checkpointed
                 and _run_journal_has_visible_output(session, _stream_id)
             ):
-                _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)
                 _append_recovered_pending_turn(session, timestamp=_recovered_ts)
             recovered_output = _append_journaled_partial_output(
                 session,

api/streaming.py

@@ -6077,30 +6077,43 @@ def _materialize_pending_user_turn_before_error(session) -> bool:
     pending_text = str(getattr(session, 'pending_user_message', None) or '')
     if not pending_text:
         return False
-    normalized_pending = " ".join(pending_text.split())
-    if normalized_pending:
-        for existing in reversed(list(getattr(session, 'messages', None) or [])[-8:]):
-            if not isinstance(existing, dict) or existing.get('role') != 'user':
-                continue
-            existing_text = " ".join(str(existing.get('content') or '').split())
-            if existing_text == normalized_pending:
-                return False
     recovered_ts = int(time.time())
     pending_started_at = getattr(session, 'pending_started_at', None)
     if isinstance(pending_started_at, (int, float)) and pending_started_at > 0:
         recovered_ts = int(pending_started_at)
+    pending_source = getattr(session, 'pending_user_source', None) or 'webui'
+    pending_attachments = list(getattr(session, 'pending_attachments', None) or [])
+
+    def is_exact_checkpoint(messages):
+        if not isinstance(messages, list) or not messages:
+            return False
+        existing = messages[-1]
+        if not isinstance(existing, dict) or existing.get('role') != 'user':
+            return False
+        existing_source = existing.get('_source') or 'webui'
+        try:
+            existing_ts = int(existing.get('timestamp'))
+        except (TypeError, ValueError):
+            return False
+        return (
+            _normalize_user_text(existing.get('content')) == _normalize_user_text(pending_text)
+            and existing_ts == recovered_ts
+            and existing_source == pending_source
+            and list(existing.get('attachments') or []) == pending_attachments
+        )
+
+    if is_exact_checkpoint(getattr(session, 'messages', None)):
+        return False
     recovered = {
         'role': 'user',
         'content': pending_text,
         'timestamp': recovered_ts,
         '_recovered': True,
     }
-    pending_source = getattr(session, 'pending_user_source', None)
-    if pending_source and pending_source != 'webui':
+    if pending_source != 'webui':
         recovered['_source'] = pending_source
-    pending_attachments = getattr(session, 'pending_attachments', None)
     if pending_attachments:
-        recovered['attachments'] = list(pending_attachments)
+        recovered['attachments'] = pending_attachments
     session.messages.append(recovered)
     # Mirror to context_messages so the _recovered flag survives the state.db
     # round-trip (#4283).  state.db has no _recovered column, so without this
@@ -6111,14 +6124,8 @@ def _materialize_pending_user_turn_before_error(session) -> bool:
     # Placing the mirror here (rather than in _persist_cancelled_turn) covers
     # all three callers: cancel, provider-error, and exception paths.
     ctx = getattr(session, 'context_messages', None)
-    if isinstance(ctx, list) and ctx:
-        rec_text = " ".join(str(recovered.get('content') or '').split())
-        if not any(
-            isinstance(e, dict) and e.get('role') == 'user'
-            and " ".join(str(e.get('content') or '').split()) == rec_text
-            for e in ctx[-8:]
-        ):
-            ctx.append({k: v for k, v in recovered.items() if k != 'timestamp'})
+    if isinstance(ctx, list) and ctx and not is_exact_checkpoint(ctx):
+        ctx.append(dict(recovered))
     # The new user turn is now committed to messages (#3831): advance a positive
     # truncation watermark left over from a prior retry/undo/edit so that
     # merge_session_messages_append_only() still filters out replaced pre-edit

tests/test_issue1361_cancel_data_loss.py

@@ -402,9 +402,16 @@ def test_stream_error_pending_materialization_does_not_duplicate_eager_checkpoin
         pending_msg="please restart the WebUI",
         messages=[
             {"role": "assistant", "content": "previous answer"},
-            {"role": "user", "content": "please restart the WebUI"},
+            {
+                "role": "user",
+                "content": "[Workspace::v1: C:\\repo]\nplease   restart the WebUI",
+                "timestamp": 1778098700,
+                "attachments": [{"name": "screen.png"}],
+            },
         ],
     )
+    s.pending_started_at = 1778098700.0
+    s.pending_attachments = [{"name": "screen.png"}]
 
     appended = _materialize_pending_user_turn_before_error(s)
 

tests/test_issue4283_recovered_context_replay.py

@@ -264,7 +264,7 @@ class _DummySession:
         self.context_messages = context_messages
         self.pending_user_message = pending_msg
         self.pending_attachments = []
-        self.pending_started_at = None
+        self.pending_started_at = 1778098700.0
         self.active_stream_id = "stream-test"
         self.truncation_watermark = None
         self.path = ""
@@ -293,6 +293,7 @@ def test_materialize_mirrors_recovered_user_to_context_messages():
     ctx_users = [m for m in s.context_messages if m.get("role") == "user"]
     assert any(m.get("_recovered") for m in ctx_users)
     assert any("restart the gateway" in m.get("content", "") for m in ctx_users)
+    assert any(m.get("timestamp") == 1778098700 for m in ctx_users)
 
 
 def test_materialize_does_not_duplicate_context_messages():

tests/test_session_sidecar_repair.py

@@ -547,6 +547,75 @@ class TestEmptyMessagesGuard:
         assert s.pending_user_message is None
         assert s.active_stream_id is None
 
+    def test_repeated_pending_prompt_uses_current_checkpoint_identity(self, hermes_home, monkeypatch):
+        """Historical identical text must not suppress the current pending turn."""
+        s = _make_session(
+            messages=[
+                {"role": "user", "content": "repeat this", "timestamp": 111, "attachments": [{"name": "old.png"}]},
+                {"role": "assistant", "content": "Earlier answer"},
+            ],
+            context_messages=[
+                {"role": "user", "content": "repeat this", "timestamp": 111, "attachments": [{"name": "old.png"}]},
+                {"role": "assistant", "content": "Earlier answer"},
+            ],
+        )
+        s.pending_user_message = "repeat this"
+        s.pending_started_at = 222.0
+        s.pending_attachments = [{"name": "current.png"}]
+        s.active_stream_id = "stream_1"
+        lock = config._get_session_agent_lock(s.session_id)
+
+        with lock:
+            core_path = hermes_home / "sessions" / f"session_{s.session_id}.json"
+            result = _apply_core_sync_or_error_marker(
+                s, core_path, stream_id_for_recheck="stream_1",
+            )
+
+        assert result is True
+        users = [m for m in s.messages if m.get("role") == "user"]
+        assert len(users) == 2
+        assert users[-1]["timestamp"] == 222
+        assert users[-1]["attachments"] == [{"name": "current.png"}]
+        context_users = [m for m in s.context_messages if m.get("role") == "user"]
+        assert len(context_users) == 2
+        assert context_users[-1]["timestamp"] == 222
+        assert context_users[-1]["attachments"] == [{"name": "current.png"}]
+
+    def test_recovered_assistant_context_row_still_deduplicates(self):
+        s = _make_session(
+            context_messages=[{"role": "assistant", "content": "Recovered answer"}],
+        )
+        recovered = {
+            "role": "assistant",
+            "content": "Recovered answer",
+            "timestamp": 222,
+        }
+
+        models._append_recovered_turn_to_context(s, recovered)
+
+        assert s.context_messages == [{"role": "assistant", "content": "Recovered answer"}]
+
+    def test_recovered_user_context_row_requires_tail_checkpoint(self):
+        s = _make_session(
+            context_messages=[
+                {"role": "user", "content": "repeat this", "timestamp": 222, "attachments": [{"name": "same.png"}]},
+                {"role": "assistant", "content": "Earlier answer"},
+            ],
+        )
+        recovered = {
+            "role": "user",
+            "content": "repeat this",
+            "timestamp": 222,
+            "attachments": [{"name": "same.png"}],
+            "_recovered": True,
+        }
+
+        models._append_recovered_turn_to_context(s, recovered)
+
+        context_users = [m for m in s.context_messages if m.get("role") == "user"]
+        assert len(context_users) == 2
+        assert context_users[-1]["timestamp"] == 222
+
     def test_bails_when_pending_user_message_none(self, hermes_home, monkeypatch):
         """If pending_user_message is None, repair bails out."""
         s = _make_session(messages=[])

tests/test_webui_gateway_chat_backend.py

@@ -417,7 +417,16 @@ def test_gateway_chat_worker_classifies_terminal_provider_error_without_text(tmp
     stream_id = "stream-gateway-terminal-provider-error-test"
     s.active_stream_id = stream_id
     s.pending_user_message = "Say hello"
-    s.pending_attachments = []
+    s.pending_started_at = 222
+    s.pending_attachments = [{"name": "current.png"}]
+    s.messages = [
+        {"role": "user", "content": "Say hello", "timestamp": 111, "attachments": [{"name": "old.png"}]},
+        {"role": "assistant", "content": "Earlier answer"},
+    ]
+    s.context_messages = [
+        {"role": "user", "content": "Say hello", "timestamp": 111, "attachments": [{"name": "old.png"}]},
+        {"role": "assistant", "content": "Earlier answer"},
+    ]
     s.save()
     STREAMS[stream_id] = channel
 
@@ -434,6 +443,16 @@ def test_gateway_chat_worker_classifies_terminal_provider_error_without_text(tmp
     assert apperrors
     assert apperrors[-1]["type"] in {"model_not_found", "auth_mismatch"}
     assert apperrors[-1]["session_id"] == s.session_id
+    saved = models.get_session(s.session_id)
+    user_messages = [m for m in saved.messages if m.get("role") == "user"]
+    assert len(user_messages) == 2
+    assert user_messages[-1]["timestamp"] == 222
+    assert user_messages[-1]["attachments"] == [{"name": "current.png"}]
+    context_users = [m for m in saved.context_messages if m.get("role") == "user"]
+    assert len(context_users) == 2
+    assert context_users[-1]["timestamp"] == 222
+    assert context_users[-1]["attachments"] == [{"name": "current.png"}]
+    assert saved.messages[-1].get("_error") is True
 
     response_error[0] = ""
     empty_stream_id = "stream-gateway-empty-response-test"