diff --git a/hermes_state_messages.py b/hermes_state_messages.py
index ade9b0d8cd..61fef4b6f9 100644
--- a/hermes_state_messages.py
+++ b/hermes_state_messages.py
@@ -1057,9 +1057,10 @@ class SessionMessagesMixin:
             raise ValueError("after_id is incompatible with include_compacted (deduped display reads use offset paging)")
         active_clause = self._active_clause(include_inactive, include_compacted)
         if include_compacted and not include_inactive and self._ensure_display_order(session_id):
-            with self._read_ctx() as conn:
-                rows = self._display_rows_from_conn(
-                    conn, session_id, limit=limit, offset=offset, latest=latest)
+            # Route through the IOERR-retrying reader (#100871), never a bare _read_ctx.
+            rows = self._read_retrying_ioerr(
+                lambda conn: self._display_rows_from_conn(
+                    conn, session_id, limit=limit, offset=offset, latest=latest))
         elif include_compacted:
             # Read-only legacy stores cannot persist display identities; keep only fixed-width
             # identities and representative ids while scanning, then fetch the selected payloads.
diff --git a/tests/hermes_state/test_read_path_transient_ioerr.py b/tests/hermes_state/test_read_path_transient_ioerr.py
index ca2c80678e..1e6be6cf54 100644
--- a/tests/hermes_state/test_read_path_transient_ioerr.py
+++ b/tests/hermes_state/test_read_path_transient_ioerr.py
@@ -8,14 +8,14 @@ import hermes_state
 from hermes_state import SessionDB
 
 
-_STATE = {"failures_left": 0, "attempts": 0}  # module-level: the tracking factory subclasses _FlakyReads
+_STATE = {"failures_left": 0, "attempts": 0, "fail_prefix": "SELECT"}  # module-level: the tracking factory subclasses _FlakyReads
 
 
 class _FlakyReads(sqlite3.Connection):
     """Real SQLite connection whose first N SELECTs fail the way a mid-checkpoint mode=ro reader does."""
 
     def execute(self, sql, *args, **kwargs):  # type: ignore[override]
-        if str(sql).lstrip().upper().startswith("SELECT"):
+        if str(sql).lstrip().upper().startswith(_STATE["fail_prefix"]):
             _STATE["attempts"] += 1
             if _STATE["failures_left"] > 0:
                 _STATE["failures_left"] -= 1
@@ -41,7 +41,7 @@ def db(tmp_path, monkeypatch):
     monkeypatch.setattr(hermes_state, "_connect_tracked_db", flaky_connect)
     while db._evict_one_idle_read_conn():  # the next read opens through the flaky factory
         pass
-    _STATE.update(failures_left=0, attempts=0)
+    _STATE.update(failures_left=0, attempts=0, fail_prefix="SELECT")
     yield db
     db.close()
 
@@ -60,3 +60,14 @@ def test_persistent_ioerr_propagates_after_the_budget(db):
         db.get_session("s")
     assert _STATE["attempts"] == hermes_state._READ_ONLY_IOERR_RETRY_ATTEMPTS + 1
     assert db._db_corrupt is False  # busy/EIO is not corruption: no quarantine
+
+
+@pytest.mark.parametrize("include_compacted, fail_prefix", [(False, "SELECT"), (True, "WITH")])
+def test_transient_ioerr_on_get_messages_is_retried(db, include_compacted, fail_prefix):
+    """Both message read paths -- live rows and the deduped display-history CTE -- replay a transient IOERR."""
+    db.append_message("s", "user", "hi")
+    _STATE.update(failures_left=1, attempts=0, fail_prefix=fail_prefix)  # WITH: only the display CTE can fail
+    rows = db.get_messages("s", include_compacted=include_compacted)
+    assert [r["content"] for r in rows] == ["hi"]
+    assert _STATE["failures_left"] == 0 and _STATE["attempts"] == 2  # one failure, one replay
+    assert db._db_corrupt is False and db._db_wal_generation_lost is False