fix(tui_gateway): serve candidate-inclusive display on warm/live resume

#65919 persists verification candidates (finish_reason=verification_required
/ verify_hook_continue) to state.db but collapses them out of the in-memory
model history via repair_message_sequence. The eager session.resume + REST
paths read the verbatim display lineage (candidate present), but the
warm/live-reuse payload (_live_session_payload) built its user-visible
messages from the collapsed in-memory model history — so switching to a
still-live session dropped the substantive verification answer that a cold
resume of the SAME session showed. That divergence is the cross-session
"substantive text vanishes on switch" class, and the direct sibling of the
resume-duplication regression fixed in #68149.

Reconcile the persisted display lineage (candidate-inclusive, the same
get_messages_as_conversation(..., include_ancestors=True) read the eager
resume + REST paths use) with the fresh in-memory tail in
_live_visible_history, so all three surfaces agree by construction while a
not-yet-flushed live turn is still shown. Extracted
_reconcile_display_with_live as a pure, DI-testable function (anchors on the
last persisted row's (role, text); appends only the uncovered in-memory tail;
trusts the DB display when the tail can't be anchored).

Tests: unit coverage for candidate-inclusion, freshness, empty/raising-DB
fallback, and the combined candidate+fresh-tail case. The existing freshness
guard (test_session_resume_live_payload_uses_current_history_with_ancestors)
stays green.

tests/test_tui_gateway_server.py

@@ -1464,6 +1464,124 @@ def test_session_resume_uses_parent_lineage_for_display(monkeypatch):
     assert captured["history_calls"] == [("tip", False), ("tip", True)]
 
 
+def test_live_visible_history_prefers_db_display_with_candidate():
+    """A warm/live session must serve the persisted DISPLAY lineage, not the
+    collapsed in-memory model history.
+
+    Regression for #65919's cross-session fallout: verification candidates
+    (finish_reason=verification_required) are persisted but collapsed out of the
+    model working history by repair_message_sequence. Building the live-reuse
+    payload from ``display_history_prefix + history`` therefore dropped the
+    substantive answer, while the eager session.resume path still showed it —
+    the two payloads for the same session disagreed. This asserts the live path
+    now matches the eager/REST display projection by construction.
+    """
+    # In-memory model history: the candidate has been collapsed away.
+    in_memory = [
+        {"role": "user", "content": "do the thing"},
+        {"role": "assistant", "content": "terse verified reply"},
+    ]
+    # Persisted display lineage: the candidate (substantive answer) survives.
+    display_with_candidate = [
+        {"role": "user", "content": "do the thing"},
+        {"role": "assistant", "content": "long substantive answer",
+         "finish_reason": "verification_required"},
+        {"role": "assistant", "content": "terse verified reply"},
+    ]
+
+    class DB:
+        def get_messages_as_conversation(
+            self, key, include_ancestors=False, repair_alternation=False
+        ):
+            assert key == "s1"
+            assert include_ancestors is True
+            return list(display_with_candidate)
+
+    result = server._live_visible_history({"session_key": "s1"}, DB(), in_memory)
+    assert result == display_with_candidate
+
+
+def test_live_visible_history_falls_back_without_db_or_key():
+    in_memory = [{"role": "user", "content": "hi"}]
+    # No DB handle available.
+    assert server._live_visible_history({"session_key": "s"}, None, in_memory) == in_memory
+
+    # DB available but the session has no persist key yet.
+    class DB:
+        def get_messages_as_conversation(self, *a, **k):  # pragma: no cover - not reached
+            raise AssertionError("must not query without a session_key")
+
+    assert server._live_visible_history({}, DB(), in_memory) == in_memory
+
+
+def test_live_visible_history_falls_back_when_db_empty():
+    """A brand-new live session whose first turn hasn't been flushed keeps its
+    in-memory history rather than rendering empty."""
+    in_memory = [{"role": "user", "content": "fresh turn not flushed yet"}]
+
+    class EmptyDB:
+        def get_messages_as_conversation(self, *a, **k):
+            return []
+
+    assert server._live_visible_history({"session_key": "s"}, EmptyDB(), in_memory) == in_memory
+
+
+def test_live_visible_history_falls_back_when_db_raises():
+    in_memory = [{"role": "user", "content": "hi"}, {"role": "assistant", "content": "yo"}]
+
+    class BrokenDB:
+        def get_messages_as_conversation(self, *a, **k):
+            raise RuntimeError("db exploded")
+
+    assert server._live_visible_history({"session_key": "s"}, BrokenDB(), in_memory) == in_memory
+
+
+def test_live_visible_history_keeps_candidate_and_fresh_tail():
+    """The hard case: the persisted candidate (missing from in-memory) AND a
+    not-yet-flushed live turn (missing from the DB) must BOTH survive."""
+    # Persisted display: has the verification candidate, lags the newest turn.
+    db_display = [
+        {"role": "user", "content": "turn 1"},
+        {"role": "assistant", "content": "long substantive answer",
+         "finish_reason": "verification_required"},
+        {"role": "assistant", "content": "terse verified reply"},
+    ]
+    # In-memory model history: candidate collapsed out, but has a fresh turn 2.
+    in_memory = [
+        {"role": "user", "content": "turn 1"},
+        {"role": "assistant", "content": "terse verified reply"},
+        {"role": "user", "content": "turn 2 not flushed"},
+        {"role": "assistant", "content": "turn 2 reply not flushed"},
+    ]
+
+    class DB:
+        def get_messages_as_conversation(self, key, include_ancestors=False, repair_alternation=False):
+            return list(db_display)
+
+    result = server._live_visible_history({"session_key": "s1"}, DB(), in_memory)
+    assert result == [
+        {"role": "user", "content": "turn 1"},
+        {"role": "assistant", "content": "long substantive answer",
+         "finish_reason": "verification_required"},
+        {"role": "assistant", "content": "terse verified reply"},
+        {"role": "user", "content": "turn 2 not flushed"},
+        {"role": "assistant", "content": "turn 2 reply not flushed"},
+    ]
+
+
+def test_reconcile_display_with_live_trusts_db_when_tail_absent():
+    """If the DB tail isn't in memory (DB ahead / diverged), don't duplicate —
+    serve the persisted display."""
+    db_display = [
+        {"role": "user", "content": "a"},
+        {"role": "assistant", "content": "b"},
+    ]
+    in_memory = [{"role": "user", "content": "unrelated"}]
+    assert server._reconcile_display_with_live(db_display, in_memory) == db_display
+    assert server._reconcile_display_with_live([], in_memory) == in_memory
+    assert server._reconcile_display_with_live(db_display, []) == db_display
+
+
 def test_session_resume_follows_compression_tip(monkeypatch, tmp_path):
     """Resuming a rotated-out parent id must load the continuation's messages.
 

tui_gateway/server.py

@@ -6656,6 +6656,77 @@ def _fallback_session_info(session: dict) -> dict:
     }
 
 
+def _reconcile_display_with_live(
+    db_display: list[dict], in_memory: list[dict]
+) -> list[dict]:
+    """Merge the persisted DISPLAY lineage with the in-memory live history.
+
+    Two projections of the same session that each hold something the other
+    lacks:
+
+    - ``db_display`` — the verbatim persisted lineage. It includes
+      *model-invisible* rows (verification candidates, finish_reason
+      ``verification_required`` / ``verify_hook_continue``) that the in-memory
+      model history collapses out via ``repair_message_sequence`` (#65919), but
+      it can lag the newest turn by a flush.
+    - ``in_memory`` — ``display_history_prefix + session["history"]``. It is the
+      freshest recency authority (a just-appended turn may not be flushed yet)
+      but it is the collapsed *model* projection, so it is missing candidates.
+
+    The merge keeps the DB display (candidate-inclusive) as the base and appends
+    only the in-memory tail that the DB does not yet cover, anchored on the last
+    DB row's ``(role, text)``. This satisfies BOTH invariants at once: the
+    substantive verification answer survives a warm/live switch (matching the
+    eager resume + REST payloads), and a not-yet-flushed live turn is not
+    dropped.
+    """
+    if not db_display:
+        return in_memory
+    if not in_memory:
+        return db_display
+
+    def _key(msg: dict) -> tuple:
+        return (msg.get("role"), _coerce_message_text(msg.get("content")))
+
+    anchor = _key(db_display[-1])
+    last_shared = -1
+    for idx, msg in enumerate(in_memory):
+        if isinstance(msg, dict) and _key(msg) == anchor:
+            last_shared = idx
+    if last_shared == -1:
+        # The DB tail isn't present in memory (DB is ahead, or the histories
+        # diverged) — trust the persisted display rather than risk duplicating.
+        return db_display
+    return list(db_display) + list(in_memory[last_shared + 1 :])
+
+
+def _live_visible_history(session: dict, db, in_memory_fallback: list[dict]) -> list[dict]:
+    """Return the user-visible DISPLAY projection for a live/warm session.
+
+    Serving the raw in-memory *model* history for the user-visible payload
+    dropped model-invisible rows (verification candidates persisted by #65919)
+    whenever a warm/live session was reused, while the eager ``session.resume``
+    path (which reads the verbatim display lineage) still showed them — the two
+    payloads disagreed about the same session, which is the cross-session
+    "substantive answer vanishes on switch" class of bug.
+
+    This reconciles the persisted display lineage (candidate-inclusive, via
+    ``get_messages_as_conversation(..., include_ancestors=True)`` — the same
+    read the eager resume and REST paths use) with the fresh in-memory tail, so
+    all surfaces agree while a not-yet-flushed turn is still shown. Falls back to
+    the in-memory history when the DB/session_key is unavailable or the DB read
+    fails.
+    """
+    key = session.get("session_key")
+    if db is not None and key:
+        try:
+            display = db.get_messages_as_conversation(key, include_ancestors=True)
+            return _reconcile_display_with_live(display, in_memory_fallback)
+        except Exception:
+            logger.debug("live display projection read failed", exc_info=True)
+    return in_memory_fallback
+
+
 def _live_session_payload(
     sid: str,
     session: dict,
@@ -6671,12 +6742,16 @@ def _live_session_payload(
             session["transport"] = transport
         if touch:
             session["last_active"] = time.time()
-        history = list(session.get("display_history_prefix") or []) + list(
+        in_memory_history = list(session.get("display_history_prefix") or []) + list(
             session.get("history") or []
         )
         inflight = _inflight_snapshot(session)
         queued = _queued_prompt_snapshot(session)
         running = bool(session.get("running"))
+    # Prefer the persisted display lineage (candidate-inclusive) so this payload
+    # matches the eager session.resume + REST transcript; the DB has its own
+    # lock, so read it outside the session history lock.
+    history = _live_visible_history(session, _get_db(), in_memory_history)
     payload = {
         "info": _fallback_session_info(session),
         "message_count": len(history),