fix(state): the kept-prefix match compares rows through the same lens the loader applies

Gate review: _rows_to_messages sanitizes and strips user/assistant text on load, so a rewind
issued from a reloaded session compared stripped content against raw rows, missed at the first
whitespace-trailing turn, and fell back to archive-all + re-insert once per reload. Both sides
now go through _loaded_view_content (shared with the loader). Test: rewind after a reload keeps
the prefix ids (red on the previous head).

hermes_state_messages.py

@@ -583,14 +583,24 @@ class SessionMessagesMixin:
         kept = 0
         for row, msg in zip(live, messages):
             tool_calls = _parse_tool_calls(msg.get("tool_calls"))
-            identity = (msg.get("role", "unknown"), self._encode_content(msg.get("content")),
+            role = msg.get("role", "unknown")
+            identity = (role, self._encode_content(self._loaded_view_content(role, msg.get("content"))),
                         msg.get("tool_call_id"), json.dumps(tool_calls) if tool_calls else None)
-            if identity != (row[1], row[2], row[3], row[4]):
+            row_content = self._encode_content(self._loaded_view_content(row[1], self._decode_content(row[2])))
+            if identity != (row[1], row_content, row[3], row[4]):
                 break
             msg["_row_id"] = row[0]
             kept += 1
         return kept
 
+    @staticmethod
+    def _loaded_view_content(role: str, content: Any) -> Any:
+        """Content as ``_rows_to_messages`` hands it to callers: user/assistant strings are sanitized and
+        stripped on load, so a reloaded session's messages must be compared to rows through the same lens."""
+        if role in {"user", "assistant"} and isinstance(content, str):
+            return sanitize_context(content).strip()
+        return content
+
     def has_archived_messages(self, session_id: str) -> bool:
         """True if the session has any soft-archived (``active = 0``) rows (tests/diagnostics).
 
@@ -1081,9 +1091,7 @@ class SessionMessagesMixin:
         messages = []
         exact_user_clones: Dict[Tuple[Any, str], Dict[str, Any]] = {}
         for row in rows:
-            content = self._decode_content(row["content"])
-            if row["role"] in {"user", "assistant"} and isinstance(content, str):
-                content = sanitize_context(content).strip()
+            content = self._loaded_view_content(row["role"], self._decode_content(row["content"]))
             # Underscore-prefixed like ``_row_id``: transports strip it before the wire; compression's
             # assembly copies strip it so rotated child handoffs still flush (_fresh_compaction_message_copy).
             msg = {"role": row["role"], "content": content, _DB_PERSISTED_MARKER_KEY: True}

tests/hermes_state/test_replace_messages_archives_suffix_only.py

@@ -40,3 +40,21 @@ def test_archive_mode_rewind_archives_only_the_dropped_suffix(tmp_path):
     assert len(rows) == 80
     assert len([m for m in rows if m["active"]]) == 50
     assert db.get_session(sid)["message_count"] == 50
+
+
+def test_reloaded_session_prefix_still_matches_its_rows(tmp_path):
+    """Loading strips/sanitizes user+assistant text; a rewind issued from that loaded view must still
+    recognise the stored rows as the same prefix, or every post-reload rewind falls back to archive-all."""
+    db = SessionDB(tmp_path / "state.db")
+    sid = "rewind-after-reload"
+    db.create_session(sid, "test")
+    for i in range(6):
+        db.append_message(sid, "user" if i % 2 == 0 else "assistant", f"turn {i}   \n")
+
+    loaded = db.get_messages_as_conversation(sid)
+    prefix_ids = [m["id"] for m in db.get_messages(sid)][:4]
+
+    db.replace_messages(sid, [dict(m) for m in loaded[:4]], active_only=True, archive_dropped=True)
+
+    assert [m["id"] for m in db.get_messages(sid)] == prefix_ids
+    assert len([m for m in db.get_messages(sid, include_inactive=True) if not m["active"]]) == 2