fix(state): make _row_id opt-in per consumer instead of universal

CI caught ACP session restore seeing an unexpected _row_id in restored
history — get_messages_as_conversation feeds more than the desktop, and
changing the default shape broke the strictest consumer. Row ids are now
include_row_ids=True, requested only by the gateway's resume/display
projections; ACP restore, export, and inspection get the transcript in its
historical shape.

hermes_state.py

@@ -6335,6 +6335,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         include_ancestors: bool = False,
         include_inactive: bool = False,
         repair_alternation: bool = False,
+        include_row_ids: bool = False,
     ) -> List[Dict[str, Any]]:
         """
         Load messages in the OpenAI conversation format (role + content dicts).
@@ -6381,6 +6382,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
             session_id=session_id,
             include_ancestors=include_ancestors,
             repair_alternation=repair_alternation,
+            include_row_ids=include_row_ids,
         )
 
     # Columns every conversation projection decodes. Shared by
@@ -6400,6 +6402,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         session_id: str,
         include_ancestors: bool,
         repair_alternation: bool,
+        include_row_ids: bool = False,
     ) -> List[Dict[str, Any]]:
         """Decode fetched message rows into the OpenAI conversation format.
 
@@ -6415,10 +6418,12 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                 content = sanitize_context(content).strip()
             msg = {"role": row["role"], "content": content}
             # Durable per-message identity for surfaces that need to address a
-            # specific row later (desktop reactions). Underscore-prefixed so
-            # every transport's convert_messages() strips it before the wire —
-            # the established escape hatch for agent-internal bookkeeping.
-            if row["id"] is not None:
+            # specific row later (desktop reactions). OPT-IN: only the gateway
+            # asks for it — every other consumer (ACP restore, export,
+            # inspection) gets the transcript in its historical shape.
+            # Underscore-prefixed so every transport's convert_messages()
+            # strips it before the wire.
+            if include_row_ids and row["id"] is not None:
                 msg["_row_id"] = row["id"]
             # api_content is the byte-fidelity sidecar: the exact string sent
             # to the API when it differed from the clean content. Returned
@@ -6556,12 +6561,14 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
             session_id=session_id,
             include_ancestors=False,
             repair_alternation=True,
+            include_row_ids=True,
         )
         display_history = self._rows_to_conversation(
             rows,
             session_id=session_id,
             include_ancestors=True,
             repair_alternation=False,
+            include_row_ids=True,
         )
         return model_history, display_history
 

tests/test_message_reactions.py

@@ -21,7 +21,7 @@ def session(db):
     key = db.create_session("react-test", "test")
     db.append_message(key, "user", "how do i center a div")
     db.append_message(key, "assistant", "use flexbox")
-    rows = [m["_row_id"] for m in db.get_messages_as_conversation(key)]
+    rows = [m["_row_id"] for m in db.get_messages_as_conversation(key, include_row_ids=True)]
 
     return key, rows
 
@@ -155,15 +155,21 @@ def test_latest_user_message_is_the_agents_default_target(session, db):
     assert db.latest_user_message_row_id(key) == rows[0]
 
     db.append_message(key, "user", "thanks!")
-    newest = db.get_messages_as_conversation(key)[-1]["_row_id"]
+    newest = db.get_messages_as_conversation(key, include_row_ids=True)[-1]["_row_id"]
 
     assert db.latest_user_message_row_id(key) == newest
 
 
-def test_row_id_never_reaches_the_provider(session, db):
-    """_row_id is underscore-prefixed so transports strip it before the wire."""
+def test_row_id_is_opt_in_and_never_reaches_the_provider(session, db):
+    """Only include_row_ids=True consumers see _row_id — and it's underscore-
+    prefixed so transports strip it before the wire even for them. Default
+    consumers (ACP restore, export) get the transcript in its historical shape.
+    """
     key, _rows = session
 
     for message in db.get_messages_as_conversation(key):
+        assert "_row_id" not in message
+
+    for message in db.get_messages_as_conversation(key, include_row_ids=True):
         assert "_row_id" in message
         assert all(not k.startswith("_") or k == "_row_id" for k in message)

tui_gateway/server.py

@@ -7506,7 +7506,7 @@ def _live_visible_history(session: dict, db, in_memory_fallback: list[dict]) ->
     key = session.get("session_key")
     if db is not None and key:
         try:
-            display = db.get_messages_as_conversation(key, include_ancestors=True)
+            display = db.get_messages_as_conversation(key, include_ancestors=True, include_row_ids=True)
             return _reconcile_display_with_live(display, in_memory_fallback)
         except Exception:
             logger.debug("live display projection read failed", exc_info=True)
@@ -11689,7 +11689,7 @@ def _format_live_history_output(session: dict) -> str:
     if db is not None and session.get("session_key"):
         try:
             history = db.get_messages_as_conversation(
-                session["session_key"], include_ancestors=True
+                session["session_key"], include_ancestors=True, include_row_ids=True
             )
         except Exception:
             pass
@@ -11729,7 +11729,9 @@ def _format_live_context_output(session: dict) -> str:
     if db is not None and session.get("session_key"):
         try:
             messages = _history_to_messages(
-                db.get_messages_as_conversation(session["session_key"], include_ancestors=True)
+                db.get_messages_as_conversation(
+                    session["session_key"], include_ancestors=True, include_row_ids=True
+                )
             )
         except Exception:
             messages = []