diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py
index bb0cac5083168..263cf1563a27e 100644
--- a/agent/agent_runtime_helpers.py
+++ b/agent/agent_runtime_helpers.py
@@ -530,6 +530,12 @@ def repair_message_sequence(agent, messages: List[Dict]) -> int:
             or m.get("finish_reason") == "incomplete"
         )
 
+    def _is_verification_candidate(m: Dict) -> bool:
+        return m.get("finish_reason") in {
+            "verification_required",
+            "verify_hook_continue",
+        }
+
     collapsed: List[Dict] = []
     for msg in messages:
         if (
@@ -542,6 +548,16 @@ def repair_message_sequence(agent, messages: List[Dict]) -> int:
             and not _is_codex_interim(collapsed[-1])
         ):
             prev = collapsed[-1]
+            # Verification candidate collapsing: when the earlier assistant
+            # message is a provisional candidate (finish_reason =
+            # verification_required / verify_hook_continue), the later
+            # response supersedes it for model replay — replace rather than
+            # union. Both remain durable in state.db; this only affects the
+            # in-memory sequence sent to the model. (#65919 §7)
+            if _is_verification_candidate(prev):
+                collapsed[-1] = msg
+                repairs += 1
+                continue
             # Union tool_calls (preserve order, both may carry them).
             prev_calls = list(prev.get("tool_calls") or [])
             new_calls = list(msg.get("tool_calls") or [])
diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py
index 75ce40b304704..501e49b54d558 100644
--- a/agent/conversation_loop.py
+++ b/agent/conversation_loop.py
@@ -690,6 +690,12 @@ def run_conversation(
     # user-facing result available; it must not be confused with error or
     # recovery text produced by unrelated exit paths.
     _pending_verification_response = None
+    # Tracks whether the pending verification candidate was already streamed
+    # to the user as interim content. The finalizer uses this to set
+    # ``_response_was_previewed`` ONLY when the pending candidate is actually
+    # reused as the final response — not merely because any interim was
+    # streamed. (#65919 review: response-loss blocker)
+    _pending_verification_response_previewed = False
 
     # Per-turn tally of consecutive successful credential-pool token refreshes,
     # keyed by (provider, pool-entry-id). A persistent upstream 401 lets
@@ -5522,17 +5528,17 @@ def run_conversation(
                         getattr(agent, "_verification_stop_nudges", 0) + 1
                     )
                     final_msg["finish_reason"] = "verification_required"
-                    final_msg["_verification_stop_synthetic"] = True
+                    # The assistant response is real content — persist it and
+                    # emit to the UI as an interim message so the user sees the
+                    # attempted final answer before the verification loop runs.
+                    # Only the nudge is flagged synthetic so it gets stripped
+                    # from the durable transcript (#65919 §7).
+                    agent._emit_interim_assistant_message(final_msg)
                     messages.append(final_msg)
-                    # Keep the attempted final answer in model history so the
-                    # synthetic user nudge preserves role alternation, but do
-                    # not surface it to the user as an interim answer. The
-                    # whole point of this guard is to prevent premature
-                    # "done" claims before checks run. Both the attempted
-                    # answer and the nudge are flagged synthetic so neither
-                    # persists — otherwise the resumed transcript keeps a
-                    # premature "done" with the nudge stripped, producing an
-                    # assistant→assistant adjacency. (#55733)
+                    try:
+                        agent._flush_messages_to_session_db(messages, conversation_history)
+                    except Exception:
+                        logger.debug("verify-on-stop interim flush failed", exc_info=True)
                     messages.append({
                         "role": "user",
                         "content": _verify_nudge,
@@ -5548,7 +5554,13 @@ def run_conversation(
                     # continuation-budget exhaustion.  ``final_response`` itself
                     # must be cleared so the finalizer can distinguish this gate
                     # from unrelated error/recovery exits. (#61631)
+                    # Track whether this candidate was already streamed so the
+                    # finalizer can mark the turn previewed only if the
+                    # candidate is actually reused as the final response.
                     _pending_verification_response = final_response
+                    _pending_verification_response_previewed = (
+                        agent._interim_content_was_streamed(final_response or "")
+                    )
                     final_response = None
                     continue
 
@@ -5587,12 +5599,17 @@ def run_conversation(
                 if _verify_nudge2:
                     agent._pre_verify_nudges = _attempt + 1
                     final_msg["finish_reason"] = "verify_hook_continue"
-                    final_msg["_pre_verify_synthetic"] = True
-                    # Same alternation contract as verify-on-stop: keep the
-                    # attempted answer in history, follow it with a synthetic
-                    # user nudge, and don't surface the premature answer. Both
-                    # are flagged synthetic so neither persists. (#55733)
+                    # The assistant response is real content — persist it and
+                    # emit to the UI as an interim message so the user sees the
+                    # attempted final answer before the pre_verify loop runs.
+                    # Only the nudge is flagged synthetic so it gets stripped
+                    # from the durable transcript (#65919 §7).
+                    agent._emit_interim_assistant_message(final_msg)
                     messages.append(final_msg)
+                    try:
+                        agent._flush_messages_to_session_db(messages, conversation_history)
+                    except Exception:
+                        logger.debug("pre_verify interim flush failed", exc_info=True)
                     messages.append({
                         "role": "user",
                         "content": _verify_nudge2,
@@ -5602,6 +5619,9 @@ def run_conversation(
                     logger.debug("pre_verify nudge issued (attempt %d)",
                                  agent._pre_verify_nudges)
                     _pending_verification_response = final_response
+                    _pending_verification_response_previewed = (
+                        agent._interim_content_was_streamed(final_response or "")
+                    )
                     final_response = None
                     continue
 
@@ -5649,6 +5669,9 @@ def run_conversation(
                     # exhaustion path does not treat the narrated stop as
                     # a completed answer.
                     _pending_verification_response = final_response
+                    _pending_verification_response_previewed = (
+                        agent._interim_content_was_streamed(final_response or "")
+                    )
                     final_response = None
                     continue
 
@@ -5773,6 +5796,7 @@ def run_conversation(
         _should_review_memory=_should_review_memory,
         _turn_exit_reason=_turn_exit_reason,
         _pending_verification_response=_pending_verification_response,
+        _pending_verification_response_previewed=_pending_verification_response_previewed,
     )
 
 
diff --git a/agent/turn_finalizer.py b/agent/turn_finalizer.py
index ccbaac9f6190f..1a7b52ff516cb 100644
--- a/agent/turn_finalizer.py
+++ b/agent/turn_finalizer.py
@@ -42,6 +42,30 @@ def _is_pure_tool_call_tail(msg: dict) -> bool:
     return not flatten_message_text(msg.get("content")).strip()
 
 
+# Verification continuation scaffolding flags: verify-on-stop / pre_verify
+# inject a synthetic user nudge to keep the agent going one more turn.
+# These nudges must be stripped from returned/live history to avoid
+# role-alternation breaks and poisoning the resumed transcript. The
+# assistant response is real content and is not flagged. (#65919 §7)
+_VERIFICATION_CONTINUATION_FLAGS = (
+    "_verification_stop_synthetic",
+    "_pre_verify_synthetic",
+)
+
+
+def _drop_verification_continuation_scaffolding(messages) -> None:
+    """Remove verification-continuation nudge messages from *messages* in place.
+
+    Only the synthetic nudges carry these flags, so this strips just the
+    nudges while preserving the real attempted-final-answer that was
+    persisted to state.db.
+    """
+    messages[:] = [
+        m for m in messages
+        if not (isinstance(m, dict) and any(m.get(f) for f in _VERIFICATION_CONTINUATION_FLAGS))
+    ]
+
+
 def finalize_turn(
     agent,
     *,
@@ -58,6 +82,7 @@ def finalize_turn(
     _should_review_memory,
     _turn_exit_reason,
     _pending_verification_response=None,
+    _pending_verification_response_previewed=False,
 ):
     """Run the post-loop finalization and return the turn ``result`` dict.
 
@@ -91,6 +116,11 @@ def finalize_turn(
         # fallible model call. The explicit pending value is the provenance
         # guard: unrelated error/recovery exits can never enter this branch.
         final_response = _pending_verification_response
+        # Mark the turn as previewed only when the reused candidate was
+        # actually streamed to the user as interim content. (#65919 review:
+        # response-loss blocker)
+        if _pending_verification_response_previewed:
+            agent._response_was_previewed = True
         _turn_exit_reason = f"max_iterations_reached({api_call_count}/{agent.max_iterations})"
         iteration_limit_fallback = True
         preserved_verification_fallback = True
@@ -206,6 +236,12 @@ def finalize_turn(
     try:
         agent._drop_trailing_empty_response_scaffolding(messages)
 
+        # Drop verification-continuation nudges (synthetic user messages)
+        # from the live history before the tail-assistant check — only the
+        # nudges need stripping; the assistant candidate persists in
+        # state.db. (#65919 §7)
+        _drop_verification_continuation_scaffolding(messages)
+
         # When the turn was interrupted and the last message is a tool
         # result, append a synthetic assistant message to close the
         # tool-call sequence. Without this, the session persists a
@@ -235,6 +271,10 @@ def finalize_turn(
         # single chokepoint every recovery ``break`` flows through, so the
         # invariant "delivered final_response ⇒ assistant row in transcript"
         # holds regardless of which path produced it. (#43849 / #44100)
+        #
+        # Compare content (not just role) so a verification candidate that
+        # matches the final response is not duplicated at budget
+        # exhaustion. (#65919 §7)
         if final_response and not interrupted:
             try:
                 _tail = messages[-1] if messages else None
@@ -242,8 +282,10 @@ def finalize_turn(
                 _tail = None
             _tail_role = _tail.get("role") if isinstance(_tail, dict) else None
             if _tail_role != "assistant":
+                # Tail is not an assistant row — append the final response
+                # so the durable turn closes with the answer (#43849/#44100).
                 messages.append({"role": "assistant", "content": final_response})
-            elif isinstance(_tail, dict) and _is_pure_tool_call_tail(_tail):
+            elif isinstance(_tail, dict) and _tail.get("content") != final_response and _is_pure_tool_call_tail(_tail):
                 # The tail IS an assistant row, but a *pure tool-call turn*:
                 # tool_calls with no text of its own. The role check alone
                 # leaves the #43849/#44100 invariant unmet — the user saw a
@@ -253,6 +295,11 @@ def finalize_turn(
                 # instead of appending, so the durable turn ends with the answer
                 # without disturbing the tool-call structure or creating an
                 # assistant→assistant pair.
+                #
+                # The ``content != final_response`` guard prevents filling when
+                # the tail already carries the final response text (verification
+                # candidate collapse — the provisional answer was persisted and
+                # reused as the terminal response, #65919 §7).
                 _tail["content"] = final_response
                 # The row may have already been flushed to SQLite by the
                 # incremental tool-call persist (conversation_loop.py:4990),
diff --git a/agent/verification_evidence.py b/agent/verification_evidence.py
index 9849cdd73a98a..d66a1534045ce 100644
--- a/agent/verification_evidence.py
+++ b/agent/verification_evidence.py
@@ -122,13 +122,13 @@ def _ensure_schema(conn: sqlite3.Connection) -> None:
     conn.commit()
 
 
-def _split_segment_tokens(command: str) -> list[list[str]]:
+def _split_segment_tokens(command: str, *, posix: bool = True) -> list[list[str]]:
     segments: list[list[str]] = []
     for segment in _SHELL_SPLIT_RE.split(command.strip()):
         if not segment:
             continue
         try:
-            tokens = shlex.split(segment)
+            tokens = shlex.split(segment, posix=posix)
         except ValueError:
             continue
         if tokens:
@@ -298,10 +298,13 @@ def _ad_hoc_script_args(tokens: list[str], root: str | Path | None) -> Optional[
 
 
 def _find_ad_hoc_match(command: str, root: str | Path | None) -> Optional[list[str]]:
-    for tokens in _split_segment_tokens(command):
-        trailing_args = _ad_hoc_script_args(tokens, root)
-        if trailing_args is not None:
-            return trailing_args
+    # Try both posix=True (default) and posix=False (Windows backslash paths)
+    # so ad-hoc verification scripts with backslash paths are matched on Windows.
+    for posix in (True, False):
+        for tokens in _split_segment_tokens(command, posix=posix):
+            trailing_args = _ad_hoc_script_args(tokens, root)
+            if trailing_args is not None:
+                return trailing_args
     return None
 
 
diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts
index 35dc67a8e79b0..1533181af11c6 100644
--- a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts
+++ b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts
@@ -53,6 +53,7 @@ import { hasSessionInfoStatePatch, sessionInfoStatePatch, SUBAGENT_EVENT_TYPES,
 
 const COMPACTION_RESUME_EVENT_TYPES = new Set([
   'message.delta',
+  'message.interim',
   'thinking.delta',
   'reasoning.delta',
   'reasoning.available',
@@ -71,9 +72,10 @@ interface GatewayEventDeps {
   nativeSubagentSessionsRef: MutableRefObject<Set<string>>
   appendAssistantDelta: (sessionId: string, delta: string) => void
   appendReasoningDelta: (sessionId: string, delta: string, replace?: boolean) => void
-  completeAssistantMessage: (sessionId: string, text: string) => void
+  completeAssistantMessage: (sessionId: string, text: string, responsePreviewed?: boolean) => void
   failAssistantMessage: (sessionId: string, errorMessage: string) => void
   flushQueuedDeltas: (sessionId?: string) => void
+  finalizeInterimAssistantMessage: (sessionId: string, text: string) => void
   queryClient: QueryClient
   refreshHermesConfig: () => Promise<void>
   sessionInterrupted: (sessionId: string) => boolean
@@ -103,6 +105,7 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) {
     completeAssistantMessage,
     failAssistantMessage,
     flushQueuedDeltas,
+    finalizeInterimAssistantMessage,
     queryClient,
     refreshHermesConfig,
     sessionInterrupted,
@@ -386,6 +389,7 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) {
             awaitingResponse: true,
             sawAssistantPayload: false,
             interrupted: false,
+            interimBoundaryPending: false,
             turnStartedAt: Date.now()
           }
         })
@@ -397,6 +401,18 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) {
         if (sessionId) {
           appendAssistantDelta(sessionId, coerceGatewayText(payload?.text))
         }
+      } else if (event.type === 'message.interim') {
+        // The agent emitted interim assistant commentary (text alongside tool
+        // calls, or the attempted final answer before a verify-on-stop nudge).
+        // Finalize it as its own sealed bubble so message.complete doesn't wipe
+        // it — the text was already streamed via message.delta and is visible.
+        if (sessionId) {
+          flushQueuedDeltas(sessionId)
+          const text = coerceGatewayText(payload?.text)
+          if (text) {
+            finalizeInterimAssistantMessage(sessionId, text)
+          }
+        }
       } else if (event.type === 'thinking.delta') {
         // thinking.delta carries the kawaii spinner status (face + verb from
         // KawaiiSpinner), not real reasoning. The bottom-of-thread loading
@@ -469,7 +485,7 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) {
         playCompletionSound()
 
         const finalText = coerceGatewayText(payload?.text) || coerceGatewayText(payload?.rendered)
-        completeAssistantMessage(sessionId, finalText)
+        completeAssistantMessage(sessionId, finalText, payload?.response_previewed)
 
         if (isActiveEvent) {
           setTurnStartedAt(null)
@@ -803,6 +819,7 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) {
       compactedTurnRef,
       completeAssistantMessage,
       failAssistantMessage,
+      finalizeInterimAssistantMessage,
       flushQueuedDeltas,
       lastCwdInfoSessionRef,
       nativeSubagentSessionsRef,
diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/index.ts b/apps/desktop/src/app/session/hooks/use-message-stream/index.ts
index 12aa158a860d8..7331ca9462671 100644
--- a/apps/desktop/src/app/session/hooks/use-message-stream/index.ts
+++ b/apps/desktop/src/app/session/hooks/use-message-stream/index.ts
@@ -10,6 +10,7 @@ import {
   type ChatMessagePart,
   chatMessageText,
   type GatewayEventPayload,
+  mergeFinalAssistantText,
   reasoningPart,
   renderMediaTags,
   upsertToolPart
@@ -370,8 +371,60 @@ export function useMessageStream({
     [flushQueuedDeltas, mutateStream, sessionInterrupted]
   )
 
-  const completeAssistantMessage = useCallback(
+  const finalizeInterimAssistantMessage = useCallback(
     (sessionId: string, text: string) => {
+      updateSessionState(sessionId, state => {
+        if (state.interrupted) {
+          return state
+        }
+
+        const authoritativeText = renderMediaTags(text).trim()
+        if (!authoritativeText) {
+          return state
+        }
+
+        const streamId = state.streamId
+
+        const replaceTextPart = (parts: ChatMessagePart[]) => {
+          const visibleText = stripGeneratedImageEchoes(
+            authoritativeText, generatedImageEchoSources(parts)
+          ).trim()
+          return mergeFinalAssistantText(parts, visibleText)
+        }
+
+        let nextMessages = state.messages
+        if (streamId && nextMessages.some(m => m.id === streamId)) {
+          // Finalize the existing streaming bubble in place
+          nextMessages = nextMessages.map(m =>
+            m.id === streamId
+              ? { ...m, parts: replaceTextPart(m.parts), pending: false }
+              : m
+          )
+        } else {
+          // No streaming bubble — create a standalone interim message
+          nextMessages = [...nextMessages, {
+            id: `assistant-interim-${Date.now()}`,
+            role: 'assistant' as const,
+            parts: [assistantTextPart(authoritativeText)],
+            pending: false,
+            branchGroupId: state.pendingBranchGroup ?? undefined
+          }]
+        }
+
+        return {
+          ...state,
+          messages: nextMessages,
+          streamId: null,
+          interimBoundaryPending: true,
+          sawAssistantPayload: state.sawAssistantPayload || Boolean(authoritativeText)
+        }
+      })
+    },
+    [updateSessionState]
+  )
+
+  const completeAssistantMessage = useCallback(
+    (sessionId: string, text: string, responsePreviewed?: boolean) => {
       let shouldHydrate = false
 
       const completedState = updateSessionState(sessionId, state => {
@@ -394,27 +447,11 @@ export function useMessageStream({
         const streamId = state.streamId
         const finalText = renderMediaTags(text).trim()
         const completionError = completionErrorText(finalText)
-        const normalize = (value: string) => value.replace(/\s+/g, ' ').trim()
+        const interimBoundaryPending = state.interimBoundaryPending
 
         const replaceTextPart = (parts: ChatMessagePart[]) => {
           const visibleFinalText = stripGeneratedImageEchoes(finalText, generatedImageEchoSources(parts)).trim()
-          const dedupeReference = normalize(visibleFinalText)
-
-          const kept = parts.filter(part => {
-            if (part.type === 'text') {
-              return false
-            }
-
-            if (part.type !== 'reasoning' || !dedupeReference) {
-              return true
-            }
-
-            const r = normalize(part.text)
-
-            return !(r && (dedupeReference.startsWith(r) || r.startsWith(dedupeReference)))
-          })
-
-          return visibleFinalText ? [...kept, assistantTextPart(visibleFinalText)] : kept
+          return mergeFinalAssistantText(parts, visibleFinalText)
         }
 
         const completeMessage = (message: ChatMessage): ChatMessage =>
@@ -454,7 +491,27 @@ export function useMessageStream({
             const existing = prev[index]
             const existingText = chatMessageText(existing).trim()
 
-            if (existing.pending || (finalText && existingText === finalText)) {
+            if (existing.pending || (!interimBoundaryPending && finalText && existingText === finalText)) {
+              nextMessages = prev.map((message, messageIndex) =>
+                messageIndex === index ? completeMessage(message) : message
+              )
+            } else if (
+              interimBoundaryPending &&
+              responsePreviewed &&
+              finalText &&
+              existingText &&
+              finalText.startsWith(existingText)
+            ) {
+              // The verification candidate was published provisionally as an
+              // interim message and then reused as the terminal response
+              // (continuation-budget fallback). Settle the interim in place
+              // instead of creating a duplicate — the DB has one row, so the
+              // live UI must agree. (#65919 review: duplicate-message blocker)
+              //
+              // Prefix match (not exact equality): the final response may be
+              // the streamed text plus a trailing delta.  mergeFinalAssistantText
+              // (called via completeMessage) handles the actual merge — it
+              // strips the old text parts and appends the full final text.
               nextMessages = prev.map((message, messageIndex) =>
                 messageIndex === index ? completeMessage(message) : message
               )
@@ -480,6 +537,7 @@ export function useMessageStream({
           awaitingResponse: false,
           busy: false,
           needsInput: false,
+          interimBoundaryPending: false,
           turnStartedAt: null
         }
       })
@@ -543,6 +601,7 @@ export function useMessageStream({
           awaitingResponse: false,
           busy: false,
           needsInput: false,
+          interimBoundaryPending: false,
           turnStartedAt: null
         }
       })
@@ -560,6 +619,7 @@ export function useMessageStream({
     completeAssistantMessage,
     failAssistantMessage,
     flushQueuedDeltas,
+    finalizeInterimAssistantMessage,
     queryClient,
     refreshHermesConfig,
     sessionInterrupted,
@@ -573,6 +633,7 @@ export function useMessageStream({
     appendReasoningDelta,
     completeAssistantMessage,
     handleGatewayEvent,
+    finalizeInterimAssistantMessage,
     upsertToolCall
   }
 }
diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/interim-sealing.test.tsx b/apps/desktop/src/app/session/hooks/use-message-stream/interim-sealing.test.tsx
new file mode 100644
index 0000000000000..a31d20a9b4fe0
--- /dev/null
+++ b/apps/desktop/src/app/session/hooks/use-message-stream/interim-sealing.test.tsx
@@ -0,0 +1,244 @@
+import { QueryClient } from '@tanstack/react-query'
+import { act, cleanup, render, waitFor } from '@testing-library/react'
+import { useEffect, useRef } from 'react'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+
+import type { ClientSessionState } from '@/app/types'
+import { createClientSessionState } from '@/lib/chat-runtime'
+import { chatMessageText } from '@/lib/chat-messages'
+import { $todosBySession, clearSessionTodos, setSessionTodos } from '@/store/todos'
+import type { RpcEvent } from '@/types/hermes'
+
+import { useMessageStream } from './index'
+
+const SID = 'session-1'
+
+let handleEvent: ((event: RpcEvent) => void) | null = null
+let sessionStates: Map<string, ClientSessionState>
+let mockCompleteSound: ReturnType<typeof vi.fn>
+let mockHaptic: ReturnType<typeof vi.fn>
+
+function Harness() {
+  const activeSessionIdRef = useRef<string | null>(SID)
+  const sessionStateByRuntimeIdRef = useRef(new Map<string, ClientSessionState>())
+  const queryClientRef = useRef(new QueryClient())
+
+  const stream = useMessageStream({
+    activeSessionIdRef,
+    hydrateFromStoredSession: vi.fn(async () => undefined),
+    queryClient: queryClientRef.current,
+    refreshHermesConfig: vi.fn(async () => undefined),
+    refreshSessions: vi.fn(async () => undefined),
+    sessionStateByRuntimeIdRef,
+    updateSessionState: (sessionId, updater) => {
+      const current = sessionStateByRuntimeIdRef.current.get(sessionId) ?? createClientSessionState()
+      const next = updater(current)
+      sessionStateByRuntimeIdRef.current.set(sessionId, next)
+      sessionStates.set(sessionId, next)
+
+      return next
+    }
+  })
+
+  useEffect(() => {
+    handleEvent = stream.handleGatewayEvent
+  }, [stream.handleGatewayEvent])
+
+  return null
+}
+
+async function mountStream() {
+  sessionStates = new Map()
+  render(<Harness />)
+  await waitFor(() => expect(handleEvent).not.toBeNull())
+}
+
+const start = () => act(() => handleEvent!({ payload: {}, session_id: SID, type: 'message.start' }))
+const delta = (text: string) => act(() => handleEvent!({ payload: { text }, session_id: SID, type: 'message.delta' }))
+const interim = (text: string) =>
+  act(() => handleEvent!({ payload: { text, already_streamed: true }, session_id: SID, type: 'message.interim' }))
+const complete = (text: string) =>
+  act(() => handleEvent!({ payload: { text }, session_id: SID, type: 'message.complete' }))
+const completePreviewed = (text: string) =>
+  act(() => handleEvent!({ payload: { text, response_previewed: true }, session_id: SID, type: 'message.complete' }))
+
+function getState(): ClientSessionState {
+  return sessionStates.get(SID) ?? createClientSessionState()
+}
+
+function assistantText(): string {
+  const state = getState()
+  const last = [...state.messages].reverse().find(m => m.role === 'assistant' && !m.hidden)
+  return last ? chatMessageText(last) : ''
+}
+
+function assistantMessages(): string[] {
+  const state = getState()
+  return state.messages
+    .filter(m => m.role === 'assistant' && !m.hidden)
+    .map(m => chatMessageText(m))
+    .filter(Boolean)
+}
+
+describe('useMessageStream interim text sealing', () => {
+  beforeEach(() => {
+    handleEvent = null
+    clearSessionTodos(SID)
+  })
+
+  afterEach(() => {
+    cleanup()
+    clearSessionTodos(SID)
+    vi.restoreAllMocks()
+  })
+
+  it('preserves interim text that the final response does not include', async () => {
+    await mountStream()
+    await start()
+
+    await delta('awaaaaa clean!! tsc zero errors')
+    await interim('awaaaaa clean!! tsc zero errors')
+
+    await complete('All checks passed.')
+
+    const texts = assistantMessages()
+    expect(texts).toContain('awaaaaa clean!! tsc zero errors')
+    expect(texts).toContain('All checks passed.')
+  })
+
+  it('dedupes interim text when the final response includes it', async () => {
+    await mountStream()
+    await start()
+
+    await delta('Let me check the files.')
+    await interim('Let me check the files.')
+
+    await complete('Let me check the files. Everything looks good.')
+
+    const texts = assistantMessages()
+    expect(texts).not.toContain('Let me check the files.Let me check the files.')
+    expect(texts.some(t => t.includes('Let me check the files. Everything looks good.'))).toBe(true)
+  })
+
+  it('clears interimBoundaryPending at turn end so the next turn starts clean', async () => {
+    await mountStream()
+    await start()
+
+    await delta('interim text')
+    await interim('interim text')
+    await complete('final text')
+
+    expect(getState().interimBoundaryPending).toBe(false)
+
+    await start()
+    expect(getState().interimBoundaryPending).toBe(false)
+
+    await complete('new turn final')
+
+    const texts = assistantMessages()
+    expect(texts[texts.length - 1]).toBe('new turn final')
+  })
+
+  it('finalizes an interim segment without settling the turn', async () => {
+    await mountStream()
+    await start()
+
+    await delta('streaming text')
+    await interim('streaming text')
+
+    // Turn is still active — busy stays true
+    expect(getState().busy).toBe(true)
+    expect(getState().interimBoundaryPending).toBe(true)
+  })
+
+  it('keeps an identical final completion distinct from an interim reply without response_previewed', async () => {
+    await mountStream()
+    await start()
+
+    await interim('same reply')
+    await complete('same reply')
+
+    // Without response_previewed, the interim and terminal replies are
+    // distinct messages — the gateway didn't signal that the final reuses
+    // the provisional candidate.
+    const texts = assistantMessages()
+    expect(texts.filter(t => t === 'same reply')).toHaveLength(2)
+  })
+
+  it('settles an identical final completion onto the interim when response_previewed', async () => {
+    await mountStream()
+    await start()
+
+    await interim('same reply')
+    await completePreviewed('same reply')
+
+    // With response_previewed, the final text is the same model response
+    // that was published provisionally as an interim — settle onto the
+    // existing interim instead of creating a duplicate. (#65919 review)
+    const texts = assistantMessages()
+    expect(texts.filter(t => t === 'same reply')).toHaveLength(1)
+  })
+
+  it('settles a prefix-matched final onto the interim when response_previewed', async () => {
+    await mountStream()
+    await start()
+
+    // Interim text is a PREFIX of the final — the model streamed part of
+    // its answer before the verify nudge fired, then the final includes
+    // the same text plus a trailing delta.
+    await interim('partial answer')
+    await completePreviewed('partial answer with more detail')
+
+    // Prefix match: the final starts with the interim text, so settle
+    // onto the interim instead of creating a duplicate bubble.
+    const texts = assistantMessages()
+    expect(texts.filter(t => t.includes('partial answer'))).toHaveLength(1)
+    expect(texts[0]).toBe('partial answer with more detail')
+  })
+
+  it('dedupes partial-stream-then-nudge: streamed prefix + interim + previewed final settles to one bubble', async () => {
+    await mountStream()
+    await start()
+
+    // The model streamed part of its answer via message.delta, then the
+    // verify nudge fired. The interim seals the streamed text, then the
+    // final response is the same text plus a trailing delta.
+    await delta('partial streamed')
+    await interim('partial streamed')
+    await completePreviewed('partial streamed answer continued')
+
+    // One bubble, containing the full final text — not two.
+    const texts = assistantMessages()
+    expect(texts.filter(t => t.includes('partial streamed'))).toHaveLength(1)
+    expect(texts[0]).toBe('partial streamed answer continued')
+  })
+
+  it('ignores malformed message.interim payload', async () => {
+    await mountStream()
+    await start()
+
+    // No payload at all
+    await act(() => handleEvent!({ type: 'message.interim' } as RpcEvent))
+    // Empty text
+    await act(() => handleEvent!({ payload: { text: '' }, session_id: SID, type: 'message.interim' } as RpcEvent))
+    // Undefined text
+    await act(() => handleEvent!({ payload: { text: undefined }, session_id: SID, type: 'message.interim' } as RpcEvent))
+
+    // Turn continues without finalizing or throwing
+    expect(getState().busy).toBe(true)
+    expect(getState().interimBoundaryPending).toBe(false)
+  })
+
+  it('clears interimBoundaryPending on message.start', async () => {
+    await mountStream()
+    await start()
+
+    await delta('interim text')
+    await interim('interim text')
+    expect(getState().interimBoundaryPending).toBe(true)
+
+    // New turn starts
+    await start()
+    expect(getState().interimBoundaryPending).toBe(false)
+  })
+})
diff --git a/apps/desktop/src/app/session/hooks/use-session-actions.test.tsx b/apps/desktop/src/app/session/hooks/use-session-actions.test.tsx
index d13717ad48e9a..a06e642d6e0cf 100644
--- a/apps/desktop/src/app/session/hooks/use-session-actions.test.tsx
+++ b/apps/desktop/src/app/session/hooks/use-session-actions.test.tsx
@@ -820,6 +820,7 @@ describe('resumeSession failure recovery', () => {
             busy: false,
             cwd: '',
             fast: false,
+            interimBoundaryPending: false,
             interrupted: false,
             messages: [],
             model: '',
diff --git a/apps/desktop/src/app/types.ts b/apps/desktop/src/app/types.ts
index 6eac1a3dd7548..3f8da4144334e 100644
--- a/apps/desktop/src/app/types.ts
+++ b/apps/desktop/src/app/types.ts
@@ -154,6 +154,8 @@ export interface ClientSessionState {
   sawAssistantPayload: boolean
   pendingBranchGroup: string | null
   interrupted: boolean
+  /** True after message.interim finalized a bubble in the still-running turn. */
+  interimBoundaryPending: boolean
   /** A blocking clarify prompt is waiting on the user for this session. Drives
    *  the sidebar "needs input" indicator; cleared when the turn resumes/ends. */
   needsInput: boolean
diff --git a/apps/desktop/src/lib/chat-messages.test.ts b/apps/desktop/src/lib/chat-messages.test.ts
index 44a6915fed7e9..64dc414642be0 100644
--- a/apps/desktop/src/lib/chat-messages.test.ts
+++ b/apps/desktop/src/lib/chat-messages.test.ts
@@ -5,7 +5,9 @@ import {
   appendAssistantTextPart,
   appendReasoningPart,
   chatMessageText,
+  mergeFinalAssistantText,
   preserveLocalAssistantErrors,
+  reasoningPart,
   renderMediaTags,
   toChatMessages,
   upsertToolPart
@@ -785,3 +787,69 @@ describe('upsertToolPart', () => {
     })
   })
 })
+
+describe('mergeFinalAssistantText', () => {
+  it('removes all text parts and appends the final text', () => {
+    const parts = [
+      { type: 'text' as const, text: 'streamed delta 1' },
+      { type: 'text' as const, text: 'streamed delta 2' },
+      { type: 'tool-call' as const, toolCallId: 'tc1', toolName: 'terminal', args: {} as never, argsText: '{}' }
+    ]
+
+    const result = mergeFinalAssistantText(parts, 'final answer')
+
+    expect(result.filter(p => p.type === 'text')).toHaveLength(1)
+    expect(result.filter(p => p.type === 'text')[0]).toMatchObject({ text: 'final answer' })
+    expect(result.some(p => p.type === 'tool-call')).toBe(true)
+  })
+
+  it('drops reasoning that the final text fully covers (reasoning ⊆ final)', () => {
+    const parts = [
+      reasoningPart('Let me check the files.'),
+      { type: 'text' as const, text: 'streamed' }
+    ]
+
+    const result = mergeFinalAssistantText(parts, 'Let me check the files. Everything looks good.')
+
+    expect(result.filter(p => p.type === 'reasoning')).toHaveLength(0)
+    expect(result.filter(p => p.type === 'text')).toHaveLength(1)
+  })
+
+  it('keeps a longer reasoning block when the final text is only a short prefix', () => {
+    // #61447: a short final ("Done.") must NOT swallow a longer reasoning block
+    // that merely starts with it.
+    const parts = [
+      reasoningPart('Done. The root cause was a bare catch block swallowing Stripe errors. The fix adds proper error logging.'),
+      { type: 'text' as const, text: 'streamed' }
+    ]
+
+    const result = mergeFinalAssistantText(parts, 'Done.')
+
+    expect(result.filter(p => p.type === 'reasoning')).toHaveLength(1)
+    expect(result.filter(p => p.type === 'text')[0]).toMatchObject({ text: 'Done.' })
+  })
+
+  it('keeps non-restating reasoning', () => {
+    const parts = [
+      reasoningPart('I analyzed the issue and found a race condition in the event loop.'),
+      { type: 'text' as const, text: 'streamed' }
+    ]
+
+    const result = mergeFinalAssistantText(parts, 'Fixed the race condition.')
+
+    expect(result.filter(p => p.type === 'reasoning')).toHaveLength(1)
+    expect(result.filter(p => p.type === 'text')).toHaveLength(1)
+  })
+
+  it('handles empty final text', () => {
+    const parts = [
+      { type: 'text' as const, text: 'streamed' },
+      reasoningPart('some reasoning')
+    ]
+
+    const result = mergeFinalAssistantText(parts, '')
+
+    expect(result.filter(p => p.type === 'text')).toHaveLength(0)
+    expect(result.filter(p => p.type === 'reasoning')).toHaveLength(1)
+  })
+})
diff --git a/apps/desktop/src/lib/chat-messages.ts b/apps/desktop/src/lib/chat-messages.ts
index 9488bd4cb4f08..221f6a16a064c 100644
--- a/apps/desktop/src/lib/chat-messages.ts
+++ b/apps/desktop/src/lib/chat-messages.ts
@@ -87,6 +87,9 @@ export type GatewayEventPayload = {
   label?: string
   index?: number
   aggregator?: string
+  // message.complete — signals the final text was already previewed via
+  // interim_assistant_callback, so the UI can settle instead of duplicating.
+  response_previewed?: boolean
 }
 
 export function textPart(text: string): ChatMessagePart {
@@ -136,6 +139,49 @@ export function chatMessageText(message: ChatMessage): string {
     .join('')
 }
 
+const normalizeWs = (value: string) => value.replace(/\s+/g, ' ').trim()
+
+/**
+ * Merge the final assistant text into a message's parts.
+ *
+ * - Removes all existing `text` parts (they were streamed deltas, now superseded
+ *   by the authoritative final response).
+ * - Keeps `reasoning` parts, but drops one that the final text fully covers
+ *   (reasoning ⊆ final) — the final restates it. A short final ("Done.") must
+ *   NOT swallow a longer reasoning block that merely starts with it (#61447).
+ * - Keeps all other part types (tool-call, image, etc.).
+ * - Appends the final text as a new text part.
+ */
+export function mergeFinalAssistantText(
+  parts: ChatMessagePart[],
+  finalText: string
+): ChatMessagePart[] {
+  const dedupeReference = normalizeWs(finalText)
+
+  const kept = parts.filter(part => {
+    if (part.type === 'text') {
+      // Sealed text parts were already finalized into their own bubbles —
+      // this filter only runs on the LAST streaming bubble, so there are no
+      // sealed parts here. All text parts are streamed deltas that get
+      // replaced by the authoritative final text.
+      return false
+    }
+
+    if (part.type !== 'reasoning' || !dedupeReference) {
+      return true
+    }
+
+    // Reasoning is a restatement only when the final FULLY covers it.
+    // The reverse direction is not considered — a short final must not
+    // swallow a longer reasoning block (#61447).
+    const r = normalizeWs(part.text)
+
+    return !(r && dedupeReference.startsWith(r))
+  })
+
+  return finalText ? [...kept, assistantTextPart(finalText)] : kept
+}
+
 const ATTACHED_CONTEXT_MARKER_RE = /(?:^|\n)--- Attached Context ---\s*\n/
 const CONTEXT_WARNINGS_MARKER_RE = /(?:^|\n)--- Context Warnings ---[\s\S]*$/
 const CONTEXT_REF_RE = /@(file|folder|url|image|tool|terminal):(?:"[^"\n]+"|'[^'\n]+'|`[^`\n]+`|\S+)/g
diff --git a/apps/desktop/src/lib/chat-runtime.ts b/apps/desktop/src/lib/chat-runtime.ts
index 0e200a50da8ee..775961d9546b7 100644
--- a/apps/desktop/src/lib/chat-runtime.ts
+++ b/apps/desktop/src/lib/chat-runtime.ts
@@ -54,6 +54,7 @@ export function createClientSessionState(
     sawAssistantPayload: false,
     pendingBranchGroup: null,
     interrupted: false,
+    interimBoundaryPending: false,
     needsInput: false,
     turnStartedAt: null,
     usage: null
diff --git a/apps/desktop/src/lib/gateway-events.test.ts b/apps/desktop/src/lib/gateway-events.test.ts
index 7435d22d6ee49..02c3f643ca6a6 100644
--- a/apps/desktop/src/lib/gateway-events.test.ts
+++ b/apps/desktop/src/lib/gateway-events.test.ts
@@ -13,6 +13,7 @@ describe('gateway event routing', () => {
     // output, and dropping them loses the live response until a refetch (#42178).
     expect(gatewayEventRequiresSessionId('message.delta')).toBe(false)
     expect(gatewayEventRequiresSessionId('message.complete')).toBe(false)
+    expect(gatewayEventRequiresSessionId('message.interim')).toBe(false)
     expect(gatewayEventRequiresSessionId('reasoning.delta')).toBe(false)
     expect(gatewayEventRequiresSessionId('tool.start')).toBe(false)
     expect(gatewayEventRequiresSessionId('approval.request')).toBe(false)
diff --git a/apps/desktop/src/lib/gateway-events.ts b/apps/desktop/src/lib/gateway-events.ts
index 67753c5fcd922..005e79705c15b 100644
--- a/apps/desktop/src/lib/gateway-events.ts
+++ b/apps/desktop/src/lib/gateway-events.ts
@@ -24,6 +24,7 @@ const UNSCOPED_STREAM_EVENT_TYPES = new Set([
   'error',
   'message.complete',
   'message.delta',
+  'message.interim',
   'message.start',
   'reasoning.available',
   'reasoning.delta',
diff --git a/apps/desktop/src/types/hermes.ts b/apps/desktop/src/types/hermes.ts
index 1d07515fcfba0..fcfae5cf7bde6 100644
--- a/apps/desktop/src/types/hermes.ts
+++ b/apps/desktop/src/types/hermes.ts
@@ -256,6 +256,7 @@ export interface HermesConfig {
   display?: {
     personality?: string
     skin?: string
+    interim_assistant_messages?: boolean
   }
   terminal?: {
     cwd?: string
diff --git a/apps/shared/src/json-rpc-gateway.ts b/apps/shared/src/json-rpc-gateway.ts
index b083d8e0e1a08..cff91305f27fd 100644
--- a/apps/shared/src/json-rpc-gateway.ts
+++ b/apps/shared/src/json-rpc-gateway.ts
@@ -3,6 +3,7 @@ export type GatewayEventName =
   | 'session.info'
   | 'message.start'
   | 'message.delta'
+  | 'message.interim'
   | 'message.complete'
   | 'thinking.delta'
   | 'reasoning.delta'
diff --git a/cli-config.yaml.example b/cli-config.yaml.example
index d74b5f3c6ba0d..5e121904ca4a5 100644
--- a/cli-config.yaml.example
+++ b/cli-config.yaml.example
@@ -1181,11 +1181,14 @@ display:
   #         cleanup_progress: true
   cleanup_progress: false
 
-  # Gateway-only natural mid-turn assistant updates.
-  # When true, completed assistant status messages are sent as separate chat
-  # messages. This is independent of tool_progress and gateway streaming.
-  #   true:  Send mid-turn assistant updates as separate messages (default)
-  #   false: Only send the final response
+  # Natural mid-turn assistant updates.
+  # On gateway platforms, when true, completed assistant status messages are
+  # sent as separate chat messages. On the Desktop app, when true, mid-turn
+  # assistant narration streamed between tool calls is kept in the transcript
+  # instead of the bubble collapsing to only the final message on completion.
+  # Independent of tool_progress and gateway streaming.
+  #   true:  Keep/send mid-turn assistant updates (default)
+  #   false: Only keep/send the final response
   interim_assistant_messages: true
 
   # Gateway-only long-running status heartbeats.
diff --git a/gateway/stream_consumer.py b/gateway/stream_consumer.py
index f1aaa393829c6..0e9a623a7174f 100644
--- a/gateway/stream_consumer.py
+++ b/gateway/stream_consumer.py
@@ -189,6 +189,13 @@ class GatewayStreamConsumer:
         # subsequently failed.
         self._final_content_delivered = False
         self._delivered_commentary_texts: list[str] = []
+        # Retains the finalized visible text of each streaming segment so
+        # ``has_delivered_text`` can still match after ``_reset_segment_state``
+        # clears ``_last_sent_text``. Without this, a segment break (triggered
+        # by ``on_segment_break`` or ``on_commentary``) erases the only record
+        # of what was delivered, and the gateway's final-send suppression
+        # can't recognize an already-delivered response. (#65919 review)
+        self._delivered_segment_texts: list[str] = []
         # Cache adapter lifecycle capability: only platforms that need an
         # explicit finalize call (e.g. DingTalk AI Cards) force us to make
         # a redundant final edit.  Everyone else keeps the fast path.
@@ -320,7 +327,10 @@ class GatewayStreamConsumer:
         visible_prefix = self._visible_prefix().strip()
         if visible_prefix == target:
             return True
-        return any(sent.strip() == target for sent in self._delivered_commentary_texts)
+        return any(
+            sent.strip() == target
+            for sent in (*self._delivered_commentary_texts, *self._delivered_segment_texts)
+        )
 
     def on_segment_break(self) -> None:
         """Finalize the current stream segment and start a fresh message."""
@@ -344,6 +354,13 @@ class GatewayStreamConsumer:
     def _reset_segment_state(self, *, preserve_no_edit: bool = False) -> None:
         if preserve_no_edit and self._message_id == "__no_edit__":
             return
+        # Retain the finalized visible text of the current segment before
+        # clearing ``_last_sent_text``, so ``has_delivered_text`` can still
+        # match it after a segment break. (#65919 review)
+        if self._last_sent_text:
+            finalized = self._clean_for_display(self._last_sent_text).strip()
+            if finalized:
+                self._delivered_segment_texts.append(finalized)
         self._message_id = None
         self._message_created_ts = None
         self._accumulated = ""
diff --git a/hermes_cli/config.py b/hermes_cli/config.py
index 762dfd9a0191c..947fddabf4157 100644
--- a/hermes_cli/config.py
+++ b/hermes_cli/config.py
@@ -1948,7 +1948,7 @@ DEFAULT_CONFIG = {
             "first_lines": 2,
             "last_lines": 2,
         },
-        "interim_assistant_messages": True,  # Gateway: show natural mid-turn assistant status messages
+        "interim_assistant_messages": True,  # Gateway: send natural mid-turn assistant status messages. Desktop: keep mid-turn narration between tool calls instead of collapsing to the final message.
         # Codex Responses models narrate progress in a dedicated commentary
         # channel. When true (default), completed commentary messages are
         # delivered as visible mid-turn updates via the interim message path.
diff --git a/run_agent.py b/run_agent.py
index 7280e29422605..6c13f737c8619 100644
--- a/run_agent.py
+++ b/run_agent.py
@@ -229,11 +229,12 @@ _EPHEMERAL_SCAFFOLDING_FLAGS = (
     "_empty_recovery_synthetic",
     "_empty_terminal_sentinel",
     "_thinking_prefill",
-    # verify-on-stop and pre_verify nudges append a synthetic assistant
-    # "done" plus a synthetic user nudge to keep the agent going one more
-    # turn before it can claim completion. Those messages exist only to
-    # drive the verification loop; persisting them poisons the resumed
-    # transcript and breaks prompt-prefix cache reuse on later turns. (#55733)
+    # verify-on-stop and pre_verify nudges append a synthetic user nudge to
+    # keep the agent going one more turn before it can claim completion.
+    # The nudge exists only to drive the verification loop; persisting it
+    # poisons the resumed transcript and breaks prompt-prefix cache reuse
+    # on later turns. The assistant candidate is NOT synthetic — it is
+    # persisted and emitted as an interim message (#65919).
     "_verification_stop_synthetic",
     "_pre_verify_synthetic",
     # kanban worker stop-guard: narrated exit without kanban_complete/block
@@ -4914,7 +4915,16 @@ class AIAgent:
         streamed = self._normalize_interim_visible_text(
             self._strip_think_blocks(getattr(self, "_current_streamed_assistant_text", "") or "")
         )
-        return bool(streamed) and streamed == visible_content
+        # Prefix match (not exact equality): the final response may be the
+        # streamed text plus a trailing delta, or the stream may have been
+        # partial when the verify nudge fired.  In both cases the streamed
+        # content is a prefix of the final — that's enough to mark it
+        # previewed (fails safe to a benign duplicate, never loses text).
+        # The reverse direction (streamed longer than final) is NOT matched:
+        # that could suppress a needed resend in the gateway path where
+        # already_streamed=True calls on_segment_break() instead of
+        # on_commentary() (#65919 review).
+        return bool(streamed) and visible_content.startswith(streamed)
 
     def _extract_codex_interim_visible_parts(
         self,
@@ -5019,8 +5029,19 @@ class AIAgent:
         except Exception:
             logger.debug("interim_assistant_callback error", exc_info=True)
 
-    def _emit_interim_assistant_message(self, assistant_msg: Dict[str, Any]) -> None:
-        """Surface a real mid-turn assistant commentary message to the UI layer."""
+    def _emit_interim_assistant_message(
+        self, assistant_msg: Dict[str, Any]
+    ) -> None:
+        """Surface a real mid-turn assistant commentary message to the UI layer.
+
+        Does NOT set ``_response_was_previewed`` — that flag means "the final
+        response was already shown to the user," but this helper is called for
+        ordinary tool-call narration, intermediate acknowledgements, and
+        verification candidates alike. Setting it here would cause the CLI to
+        suppress a *different* final summary (e.g. from ``_handle_max_iterations``)
+        when the only streamed text was unrelated mid-turn commentary. (#65919
+        review: response-loss blocker)
+        """
         cb = getattr(self, "interim_assistant_callback", None)
         if cb is None or not isinstance(assistant_msg, dict):
             return
diff --git a/tests/agent/test_turn_finalizer_iteration_limit_exit.py b/tests/agent/test_turn_finalizer_iteration_limit_exit.py
index 3d1d342f5f401..f1920634b779b 100644
--- a/tests/agent/test_turn_finalizer_iteration_limit_exit.py
+++ b/tests/agent/test_turn_finalizer_iteration_limit_exit.py
@@ -296,3 +296,84 @@ def test_pending_response_records_kanban_timeout(monkeypatch):
         end_run=True,
         event_payload_extra={"budget_used": 60, "budget_max": 60},
     )
+
+
+def test_published_pending_candidate_is_not_duplicated_by_finalizer(monkeypatch):
+    """When budget exhaustion preserves a verification candidate that is
+    already the tail assistant message, the finalizer must NOT append a
+    duplicate. The content-comparison guard prevents this. (#65919 §7)
+    """
+    monkeypatch.setattr("hermes_cli.plugins.invoke_hook", lambda *_a, **_kw: [])
+    agent = _LimitAgent()
+    report = "the composed report"
+
+    result = finalize_turn(
+        agent,
+        final_response=report,
+        api_call_count=60,
+        interrupted=False,
+        failed=False,
+        # The candidate is already in messages as the tail assistant.
+        messages=[
+            {"role": "user", "content": "task"},
+            {"role": "assistant", "content": report},
+        ],
+        conversation_history=[],
+        effective_task_id="task",
+        turn_id="turn",
+        user_message="task",
+        original_user_message="task",
+        _should_review_memory=False,
+        _turn_exit_reason="unknown",
+        _pending_verification_response=report,
+    )
+
+    # The tail assistant already matches final_response — no duplicate appended.
+    roles = [m["role"] for m in result["messages"]]
+    assert roles == ["user", "assistant"]
+    # Persisted messages should also have no duplicate.
+    assert agent.persisted_messages is not None
+    persisted_roles = [m["role"] for m in agent.persisted_messages]
+    assert persisted_roles == ["user", "assistant"]
+
+
+def test_terminal_verification_failure_is_persisted_as_one_correction(monkeypatch):
+    """When verification fails terminally (nudge present but budget exhausted),
+    the finalizer drops the synthetic nudge and the assistant candidate
+    persists as a single correction. No duplicate assistant appended. (#65919 §7)
+    """
+    monkeypatch.setattr("hermes_cli.plugins.invoke_hook", lambda *_a, **_kw: [])
+    agent = _LimitAgent()
+    report = "terminal failure correction"
+
+    result = finalize_turn(
+        agent,
+        final_response=report,
+        api_call_count=60,
+        interrupted=False,
+        failed=False,
+        messages=[
+            {"role": "user", "content": "task"},
+            {"role": "assistant", "content": report},
+            # Synthetic nudge — should be dropped by _drop_verification_continuation_scaffolding.
+            {"role": "user", "content": "[System: run tests]", "_verification_stop_synthetic": True},
+        ],
+        conversation_history=[],
+        effective_task_id="task",
+        turn_id="turn",
+        user_message="task",
+        original_user_message="task",
+        _should_review_memory=False,
+        _turn_exit_reason="unknown",
+        _pending_verification_response=report,
+    )
+
+    # The nudge is dropped; the assistant candidate is the tail and matches
+    # final_response, so no duplicate is appended.
+    roles = [m["role"] for m in result["messages"]]
+    assert roles == ["user", "assistant"]
+    # The nudge is gone from persisted messages too.
+    assert agent.persisted_messages is not None
+    persisted_contents = [m.get("content") for m in agent.persisted_messages]
+    assert "[System: run tests]" not in persisted_contents
+    assert report in persisted_contents
diff --git a/tests/agent/test_verification_evidence.py b/tests/agent/test_verification_evidence.py
index 5f957f54efbe8..c176f37b9fc2a 100644
--- a/tests/agent/test_verification_evidence.py
+++ b/tests/agent/test_verification_evidence.py
@@ -391,3 +391,36 @@ def test_recording_expires_old_edit_only_state(tmp_path, monkeypatch):
     status = verification_status(session_id="old-session", cwd=tmp_path)
     assert status["status"] == "unverified"
     assert status["changed_paths"] == []
+
+
+def test_windows_backslash_ad_hoc_script_path_is_matched(tmp_path, monkeypatch):
+    """Ad-hoc verification scripts with Windows backslash paths must be
+    matched by ``_find_ad_hoc_match`` trying ``posix=False`` in addition to
+    the default ``posix=True``. (#53553 / #65919)
+
+    On Linux, ``Path`` doesn't parse Windows backslash paths, so we mock
+    ``_is_temp_script_path`` to simulate the Windows environment where the
+    path resolves correctly. The test verifies the posix=False splitting
+    fallback — the actual fix from #53553.
+    """
+    from agent.verification_evidence import _find_ad_hoc_match
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
+    (tmp_path / "package.json").write_text("{}", encoding="utf-8")
+
+    # On Windows, shlex.split(posix=True) eats backslashes as escape chars;
+    # posix=False preserves them. Mock _is_temp_script_path so the test
+    # focuses on the splitting fallback without needing a real Windows FS.
+    def mock_is_temp_script(token, root):
+        return "hermes-ad-hoc" in token and ".py" in token
+
+    monkeypatch.setattr(
+        "agent.verification_evidence._is_temp_script_path",
+        mock_is_temp_script,
+    )
+
+    win_script = r"C:\Users\test\AppData\Local\Temp\hermes-ad-hoc-check.py"
+    result = _find_ad_hoc_match(f"python {win_script}", tmp_path)
+    assert result is not None, (
+        "Windows backslash path should be matched via posix=False fallback"
+    )
diff --git a/tests/agent/test_verification_stop_caching.py b/tests/agent/test_verification_stop_caching.py
index 41fee3b337479..7620d9eb51706 100644
--- a/tests/agent/test_verification_stop_caching.py
+++ b/tests/agent/test_verification_stop_caching.py
@@ -1,15 +1,15 @@
 """Verification-loop synthetic scaffolding must never reach durable session state.
 
-verify_on_stop / pre_verify append a synthetic assistant "done" plus a synthetic
-user nudge to keep the agent going one more turn before it can claim completion.
-These messages exist only to drive the loop; persisting them poisons the resumed
-transcript and breaks prompt-prefix cache reuse on later turns (#55733).
+verify_on_stop / pre_verify inject a synthetic user nudge to keep the agent
+going one more turn before it can claim completion. The assistant response is
+real content that persists and is emitted to the UI as an interim message.
+Only the nudge (the synthetic user message) is flagged, so only the nudge
+gets stripped from the durable transcript. This test file verifies:
 
-Both persistence sinks (SQLite flush + JSON snapshot) route through the single
-``_is_ephemeral_scaffolding`` chokepoint, which is driven by
-``_EPHEMERAL_SCAFFOLDING_FLAGS``. These tests assert that the verification-loop
-flags are registered there and that both sinks drop the flagged messages while
-keeping the real conversation.
+  - The verification-loop flags remain registered in
+    ``_EPHEMERAL_SCAFFOLDING_FLAGS`` (so nudges are stripped).
+  - The DB flush drops only the nudge, keeping the assistant candidate.
+  - The JSON log drops only the nudge, keeping the assistant candidate.
 """
 
 import json
@@ -34,15 +34,16 @@ def test_verification_flags_registered_as_ephemeral(tmp_path, monkeypatch):
     assert "_verification_stop_synthetic" in ra._EPHEMERAL_SCAFFOLDING_FLAGS
     assert "_pre_verify_synthetic" in ra._EPHEMERAL_SCAFFOLDING_FLAGS
 
-    # The central classifier drives both persistence sinks.
-    assert ra._is_ephemeral_scaffolding(
-        {"role": "assistant", "content": "done", "_verification_stop_synthetic": True}
-    )
+    # The nudge messages ARE scaffolding (they carry the synthetic flag).
     assert ra._is_ephemeral_scaffolding(
         {"role": "user", "content": "[System: run tests]", "_pre_verify_synthetic": True}
     )
-    # Real messages are not scaffolding.
+    assert ra._is_ephemeral_scaffolding(
+        {"role": "user", "content": "[System: run tests]", "_verification_stop_synthetic": True}
+    )
+    # Real messages (including the assistant candidate) are not.
     assert not ra._is_ephemeral_scaffolding({"role": "user", "content": "hi"})
+    assert not ra._is_ephemeral_scaffolding({"role": "assistant", "content": "premature done"})
 
 
 def _make_agent(ra, session_id, tmp_path):
@@ -64,14 +65,18 @@ def _make_agent(ra, session_id, tmp_path):
     return agent
 
 
-def test_db_flush_drops_verification_scaffolding(tmp_path, monkeypatch):
+def test_db_flush_drops_only_nudge_keeps_candidate(tmp_path, monkeypatch):
+    """The assistant candidate is NOT flagged synthetic, so it persists.
+    Only the nudge (flagged synthetic) is dropped from the DB flush."""
     monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
     ra = _fresh_run_agent(tmp_path)
     agent = _make_agent(ra, "sess_db", tmp_path)
 
     messages = [
         {"role": "user", "content": "hi"},
-        {"role": "assistant", "content": "premature done", "_verification_stop_synthetic": True},
+        # Assistant candidate — NOT flagged synthetic, persists.
+        {"role": "assistant", "content": "premature done"},
+        # Nudge — flagged synthetic, gets dropped.
         {"role": "user", "content": "[System: run tests]", "_verification_stop_synthetic": True},
         {"role": "assistant", "content": "verified and clean"},
     ]
@@ -84,18 +89,24 @@ def test_db_flush_drops_verification_scaffolding(tmp_path, monkeypatch):
     ]
     assert "hi" in persisted
     assert "verified and clean" in persisted
-    assert "premature done" not in persisted
+    # The assistant candidate persists — it is real content.
+    assert "premature done" in persisted
+    # Only the nudge is dropped.
     assert "[System: run tests]" not in persisted
 
 
-def test_json_log_drops_verification_scaffolding(tmp_path, monkeypatch):
+def test_json_log_drops_only_nudge_keeps_candidate(tmp_path, monkeypatch):
+    """The assistant candidate is NOT flagged synthetic, so it persists in the
+    JSON log. Only the nudge (flagged synthetic) is dropped."""
     monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
     ra = _fresh_run_agent(tmp_path)
     agent = _make_agent(ra, "sess_json", tmp_path)
 
     messages = [
         {"role": "user", "content": "hi"},
-        {"role": "assistant", "content": "premature done", "_pre_verify_synthetic": True},
+        # Assistant candidate — NOT flagged synthetic, persists.
+        {"role": "assistant", "content": "premature done"},
+        # Nudge — flagged synthetic, gets dropped.
         {"role": "user", "content": "[System: run tests]", "_pre_verify_synthetic": True},
         {"role": "assistant", "content": "verified and clean"},
     ]
@@ -106,5 +117,10 @@ def test_json_log_drops_verification_scaffolding(tmp_path, monkeypatch):
     assert log_file.exists()
     data = json.loads(log_file.read_text(encoding="utf-8"))
     contents = [m.get("content") for m in data["messages"]]
-    assert contents == ["hi", "verified and clean"]
+    # The assistant candidate persists — it is real content.
+    assert "premature done" in contents
+    assert "verified and clean" in contents
+    assert "hi" in contents
+    # Only the nudge is dropped.
+    assert "[System: run tests]" not in contents
     assert all(not m.get("_pre_verify_synthetic") for m in data["messages"])
diff --git a/tests/gateway/test_stream_consumer.py b/tests/gateway/test_stream_consumer.py
index b43bacf046a55..430f44da02cb0 100644
--- a/tests/gateway/test_stream_consumer.py
+++ b/tests/gateway/test_stream_consumer.py
@@ -2379,3 +2379,40 @@ class TestStripOrphanCloseTags:
             assert tag not in consumer._accumulated
         assert "trailing prose" in consumer._accumulated
         assert "more" in consumer._accumulated
+
+
+class TestHasDeliveredTextAfterSegmentBreak:
+    """has_delivered_text must find a delivered segment after a segment break,
+    but must not claim text from a failed delivery. (#65919 review)"""
+
+    def test_finds_delivered_segment_after_segment_break(self):
+        """A successfully delivered segment must still be found by
+        has_delivered_text after _reset_segment_state runs."""
+        c = _make_consumer()
+        # Simulate a successfully delivered segment
+        c._last_sent_text = "Here is the first segment"
+        c._reset_segment_state()
+        # After the reset, has_delivered_text must still find it
+        assert c.has_delivered_text("Here is the first segment") is True
+
+    def test_does_not_find_undelivered_text(self):
+        """Text that was never delivered must not be claimed."""
+        c = _make_consumer()
+        c._last_sent_text = "delivered text"
+        c._reset_segment_state()
+        assert c.has_delivered_text("never sent text") is False
+
+    def test_finds_commentary_text(self):
+        """has_delivered_text must find commentary text delivered via
+        on_commentary."""
+        c = _make_consumer()
+        c._delivered_commentary_texts.append("interim commentary")
+        assert c.has_delivered_text("interim commentary") is True
+
+    def test_does_not_match_empty(self):
+        """Empty/whitespace text must not match."""
+        c = _make_consumer()
+        c._last_sent_text = "some text"
+        c._reset_segment_state()
+        assert c.has_delivered_text("") is False
+        assert c.has_delivered_text("   ") is False
diff --git a/tests/run_agent/test_run_agent_codex_responses.py b/tests/run_agent/test_run_agent_codex_responses.py
index c57485bb0d5e2..071596e204654 100644
--- a/tests/run_agent/test_run_agent_codex_responses.py
+++ b/tests/run_agent/test_run_agent_codex_responses.py
@@ -2371,6 +2371,37 @@ def test_interim_commentary_is_not_marked_already_streamed_when_stream_callback_
     }
 
 
+def test_interim_content_was_streamed_matches_prefix_not_exact(monkeypatch):
+    """_interim_content_was_streamed should return True when the streamed text
+    is a PREFIX of the final content (trailing delta added after stream, or
+    partial stream before verify nudge).  Exact equality is too strict — it
+    fails safe to a benign duplicate bubble instead of settling the interim.
+    (#65919 review: prefix-based match like the TUI's finalTail dedup.)"""
+    agent = _build_agent(monkeypatch)
+
+    # Exact match still works
+    agent._current_streamed_assistant_text = "hello world"
+    assert agent._interim_content_was_streamed("hello world") is True
+
+    # Streamed is a prefix of the final (trailing delta) — should match
+    agent._current_streamed_assistant_text = "hello"
+    assert agent._interim_content_was_streamed("hello world") is True
+
+    # Streamed is empty — should not match
+    agent._current_streamed_assistant_text = ""
+    assert agent._interim_content_was_streamed("hello world") is False
+
+    # Final is empty — should not match
+    agent._current_streamed_assistant_text = "hello"
+    assert agent._interim_content_was_streamed("") is False
+
+    # Streamed is LONGER than final (reverse direction) — should NOT match.
+    # This is the unsafe direction: it could suppress a needed resend in the
+    # gateway path where already_streamed=True calls on_segment_break().
+    agent._current_streamed_assistant_text = "hello world extra"
+    assert agent._interim_content_was_streamed("hello") is False
+
+
 def test_interim_commentary_preserves_assistant_content(monkeypatch):
     """Interim commentary must not silently mutate assistant text containing
     literal <memory-context> markers — that's legitimate model output (docs,
diff --git a/tests/run_agent/test_verification_continuation_budget.py b/tests/run_agent/test_verification_continuation_budget.py
index d6f65407e7abc..a3fc5d6031fea 100644
--- a/tests/run_agent/test_verification_continuation_budget.py
+++ b/tests/run_agent/test_verification_continuation_budget.py
@@ -1,4 +1,4 @@
-"""End-to-end regression coverage for verification budget exhaustion (#61631)."""
+"""End-to-end regression coverage for verification budget exhaustion (#61631, #65919 §7)."""
 
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
@@ -51,11 +51,12 @@ def _assert_pending_response_survives(agent, result):
     assert result["turn_exit_reason"] == "max_iterations_reached(1/1)"
     assert result["completed"] is False
     assert agent._handle_max_iterations.call_count == 0
+    # The nudge is stripped by _drop_verification_continuation_scaffolding,
+    # so the role sequence is [user, assistant] — the candidate is the
+    # tail and matches final_response so it is not duplicated. (#65919 §7)
     assert [message["role"] for message in result["messages"]] == [
         "user",
         "assistant",
-        "user",
-        "assistant",
     ]
 
 
@@ -75,8 +76,8 @@ def test_verify_on_stop_preserves_composed_report_at_budget_limit(agent, monkeyp
         result = agent.run_conversation("edit changed.py")
 
     _assert_pending_response_survives(agent, result)
-    assert result["messages"][1]["_verification_stop_synthetic"] is True
-    assert result["messages"][2]["_verification_stop_synthetic"] is True
+    # The assistant response persists (it is real, unflagged content).
+    assert not result["messages"][1].get("_verification_stop_synthetic")
 
 
 def test_pre_verify_preserves_composed_report_at_budget_limit(agent, monkeypatch):
@@ -100,8 +101,8 @@ def test_pre_verify_preserves_composed_report_at_budget_limit(agent, monkeypatch
         result = agent.run_conversation("edit changed.py")
 
     _assert_pending_response_survives(agent, result)
-    assert result["messages"][1]["_pre_verify_synthetic"] is True
-    assert result["messages"][2]["_pre_verify_synthetic"] is True
+    # The assistant response persists (it is real, unflagged content).
+    assert not result["messages"][1].get("_pre_verify_synthetic")
 
 
 def test_intermediate_ack_uses_summary_instead_of_premature_text(agent, monkeypatch):
@@ -144,3 +145,177 @@ def test_later_verified_response_supersedes_pending_report(agent, monkeypatch):
     assert result["turn_exit_reason"] == "text_response(finish_reason=stop)"
     assert result["completed"] is True
     agent._handle_max_iterations.assert_not_called()
+
+
+def test_multiple_verification_retries_publish_each_candidate_once(agent, monkeypatch):
+    """Multiple verification retries should publish each candidate once, in order."""
+    agent.max_iterations = 3
+    agent.iteration_budget.max_total = 3
+    answers = iter([
+        _response("candidate one"),
+        _response("candidate two"),
+        _response("candidate three"),
+    ])
+    agent._interruptible_api_call = lambda _kwargs: next(answers)
+    agent._handle_max_iterations = MagicMock(return_value="replacement summary")
+    monkeypatch.setenv("HERMES_VERIFY_ON_STOP", "1")
+
+    # Three nudges, then None (so the third candidate is the final response).
+    nudge_side_effects = ["verify it", "verify it", None]
+
+    emitted = []
+    agent.interim_assistant_callback = lambda text, **kw: emitted.append(text)
+
+    with (
+        patch(
+            "agent.verification_stop.build_verify_on_stop_nudge",
+            side_effect=nudge_side_effects,
+        ),
+        patch("hermes_cli.plugins.invoke_hook", return_value=[]),
+    ):
+        result = agent.run_conversation("edit changed.py")
+
+    # Each candidate was emitted as an interim message, in order.
+    assert emitted == ["candidate one", "candidate two"]
+    # The final response is the last candidate.
+    assert result["final_response"] == "candidate three"
+    assert result["turn_exit_reason"] == "text_response(finish_reason=stop)"
+    assert result["completed"] is True
+    agent._handle_max_iterations.assert_not_called()
+
+
+def test_verification_false_finalizes_candidate_once(agent, monkeypatch):
+    """When verification returns false/exception, the candidate is finalized once."""
+    agent._interruptible_api_call = lambda _kwargs: _response("the answer")
+    agent._handle_max_iterations = MagicMock(return_value="replacement summary")
+    monkeypatch.setenv("HERMES_VERIFY_ON_STOP", "1")
+
+    emitted = []
+    agent.interim_assistant_callback = lambda text, **kw: emitted.append(text)
+
+    with (
+        # build_verify_on_stop_nudge raises — simulates verification check failure
+        patch(
+            "agent.verification_stop.build_verify_on_stop_nudge",
+            side_effect=RuntimeError("verify check crashed"),
+        ),
+        patch("hermes_cli.plugins.invoke_hook", return_value=[]),
+    ):
+        result = agent.run_conversation("edit changed.py")
+
+    # No interim emission because verification did not run (exception path
+    # sets _verify_nudge = None, so the candidate becomes the final response
+    # without an interim emission).
+    assert result["final_response"] == "the answer"
+    assert result["completed"] is True
+    agent._handle_max_iterations.assert_not_called()
+
+
+def test_verify_on_stop_emits_interim_response_to_ui(agent, monkeypatch):
+    """The verify-on-stop path must emit the full response to the UI callback.
+
+    With no streaming set up in this test, _interim_content_was_streamed
+    returns False, so already_streamed is False — the callback reports
+    content the UI has not seen yet.
+    """
+    agent._interruptible_api_call = lambda _kwargs: _response("composed report")
+    agent._handle_max_iterations = MagicMock(return_value="replacement summary")
+    monkeypatch.setenv("HERMES_VERIFY_ON_STOP", "1")
+
+    callback_calls = []
+
+    def capture_callback(text, *, already_streamed=None):
+        callback_calls.append({"text": text, "already_streamed": already_streamed})
+
+    agent.interim_assistant_callback = capture_callback
+
+    with (
+        patch("agent.verification_stop.build_verify_on_stop_nudge", return_value="verify it"),
+        patch("hermes_cli.plugins.invoke_hook", return_value=[]),
+    ):
+        result = agent.run_conversation("edit changed.py")
+
+    # The callback was called with the full response text and already_streamed=False
+    assert len(callback_calls) == 1
+    assert callback_calls[0]["text"] == "composed report"
+    assert callback_calls[0]["already_streamed"] is False
+
+    # The candidate persists as the final response.
+    assert result["final_response"] == "composed report"
+
+
+def test_streamed_interim_then_different_summary_not_marked_previewed(agent, monkeypatch):
+    """Ordinary interim narration followed by a different non-streamed summary.
+
+    The model streams "I'll inspect the files now" as an intermediate ack.
+    _emit_interim_assistant_message is called for this ordinary narration,
+    which must NOT set _response_was_previewed. Then _handle_max_iterations
+    produces a different summary through the non-streaming Chat Completions
+    path. The final result must NOT be marked as previewed — the interim was
+    unrelated mid-turn commentary, not the final response — so the CLI renders
+    the summary instead of suppressing it. (#65919 review: response-loss blocker)
+    """
+    agent.valid_tool_names = ["web_search"]
+    agent._intent_ack_continuation = True
+    agent._looks_like_codex_intermediate_ack = MagicMock(return_value=True)
+    agent._interruptible_api_call = lambda _kwargs: _response("I'll inspect the files now")
+    agent._handle_max_iterations = MagicMock(return_value="Here is the summary of what I found.")
+    monkeypatch.setenv("HERMES_VERIFY_ON_STOP", "0")
+
+    emitted = []
+    agent.interim_assistant_callback = lambda text, **kw: emitted.append(text)
+
+    with (
+        patch("hermes_cli.plugins.has_hook", return_value=False),
+        patch("hermes_cli.plugins.invoke_hook", return_value=[]),
+    ):
+        result = agent.run_conversation("inspect /tmp/project")
+
+    # The final response is the different summary from _handle_max_iterations.
+    assert result["final_response"] == "Here is the summary of what I found."
+    # CRITICAL: response_previewed must be False — the interim narration was
+    # NOT the final response, so the CLI must render the summary.
+    assert result["response_previewed"] is False
+
+
+def test_streamed_verification_candidate_reused_marked_previewed(agent, monkeypatch):
+    """Verification candidate reused at budget exhaustion is marked previewed.
+
+    The model streams a verification candidate that is already streamed as
+    interim content. The continuation budget is exhausted, so the finalizer
+    reuses the pending verification candidate as the final response. The result
+    must be marked as previewed so the CLI/desktop settle it once instead of
+    duplicating. (#65919 review)
+    """
+    agent._interruptible_api_call = lambda _kwargs: _response("composed report")
+    agent._handle_max_iterations = MagicMock(return_value="replacement summary")
+    monkeypatch.setenv("HERMES_VERIFY_ON_STOP", "1")
+
+    agent._turn_file_mutation_paths = {"changed.py"}
+
+    callback_calls = []
+
+    def capture_callback(text, *, already_streamed=None):
+        callback_calls.append({"text": text, "already_streamed": already_streamed})
+
+    agent.interim_assistant_callback = capture_callback
+
+    # Simulate that the candidate text was already streamed. The streaming
+    # buffer is cleared after the response is processed, so mock the check
+    # directly — this is the condition the test validates: when the candidate
+    # was streamed, the previewed flag propagates to the finalizer.
+    with (
+        patch.object(agent, "_interim_content_was_streamed", return_value=True),
+        patch("agent.verification_stop.build_verify_on_stop_nudge", return_value="verify it"),
+        patch("hermes_cli.plugins.invoke_hook", return_value=[]),
+    ):
+        result = agent.run_conversation("edit changed.py")
+
+    # The candidate was already streamed, so the callback reports already_streamed=True.
+    assert len(callback_calls) == 1
+    assert callback_calls[0]["already_streamed"] is True
+    # The candidate is reused as the final response.
+    assert result["final_response"] == "composed report"
+    # CRITICAL: response_previewed must be True — the reused candidate was
+    # streamed as interim content, so the CLI/desktop settle it once.
+    assert result["response_previewed"] is True
diff --git a/tests/tui_gateway/test_interim_assistant_callback.py b/tests/tui_gateway/test_interim_assistant_callback.py
new file mode 100644
index 0000000000000..833c914136753
--- /dev/null
+++ b/tests/tui_gateway/test_interim_assistant_callback.py
@@ -0,0 +1,105 @@
+"""Tests for the interim_assistant_callback config gating in tui_gateway.
+
+These tests exercise the real _agent_cbs() wiring rather than a local
+imitation, so a break in the production callback registration is caught.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+
+def test_load_interim_assistant_messages_defaults_true():
+    from tui_gateway.server import _load_interim_assistant_messages
+
+    with patch("tui_gateway.server._load_cfg", return_value={}):
+        assert _load_interim_assistant_messages() is True
+
+
+def test_load_interim_assistant_messages_explicit_true():
+    from tui_gateway.server import _load_interim_assistant_messages
+
+    with patch("tui_gateway.server._load_cfg", return_value={"display": {"interim_assistant_messages": True}}):
+        assert _load_interim_assistant_messages() is True
+
+
+def test_load_interim_assistant_messages_explicit_false():
+    from tui_gateway.server import _load_interim_assistant_messages
+
+    with patch("tui_gateway.server._load_cfg", return_value={"display": {"interim_assistant_messages": False}}):
+        assert _load_interim_assistant_messages() is False
+
+
+def test_load_interim_assistant_messages_string_off():
+    from tui_gateway.server import _load_interim_assistant_messages
+
+    with patch("tui_gateway.server._load_cfg", return_value={"display": {"interim_assistant_messages": "off"}}):
+        assert _load_interim_assistant_messages() is False
+
+
+def test_agent_cbs_includes_interim_callback_when_enabled():
+    """_agent_cbs() includes interim_assistant_callback when the config is on.
+
+    Exercises the real _agent_cbs() wiring: the callback must be present in
+    the returned dict and, when invoked, must emit a message.interim event
+    with the text and already_streamed flag passed through.
+    """
+    from tui_gateway.server import _agent_cbs
+
+    emitted: list[tuple] = []
+
+    def fake_emit(event_type, sid, payload=None):
+        emitted.append((event_type, sid, payload))
+
+    with patch("tui_gateway.server._load_cfg", return_value={}), \
+         patch("tui_gateway.server._emit", side_effect=fake_emit):
+        cbs = _agent_cbs("test-session")
+
+        assert "interim_assistant_callback" in cbs
+        cb = cbs["interim_assistant_callback"]
+        assert callable(cb)
+
+        # Invoke the real callback inside the patch context — the lambda
+        # resolves _emit by name at call time, so it must be called while
+        # the patch is active.
+        cb("hello world", already_streamed=True)
+
+    assert len(emitted) == 1
+    assert emitted[0][0] == "message.interim"
+    assert emitted[0][1] == "test-session"
+    assert emitted[0][2]["text"] == "hello world"
+    assert emitted[0][2]["already_streamed"] is True
+
+
+def test_agent_cbs_omits_interim_callback_when_disabled():
+    """_agent_cbs() omits interim_assistant_callback when the config is off.
+
+    Exercises the real _agent_cbs() wiring: the callback must NOT be present
+    in the returned dict when display.interim_assistant_messages is false.
+    """
+    from tui_gateway.server import _agent_cbs
+
+    with patch("tui_gateway.server._load_cfg", return_value={"display": {"interim_assistant_messages": False}}):
+        cbs = _agent_cbs("test-session")
+
+    assert "interim_assistant_callback" not in cbs
+
+
+def test_agent_cbs_interim_callback_passes_already_streamed_false():
+    """The real callback passes already_streamed=False by default."""
+    from tui_gateway.server import _agent_cbs
+
+    emitted: list[tuple] = []
+
+    def fake_emit(event_type, sid, payload=None):
+        emitted.append((event_type, sid, payload))
+
+    with patch("tui_gateway.server._load_cfg", return_value={}), \
+         patch("tui_gateway.server._emit", side_effect=fake_emit):
+        cbs = _agent_cbs("test-session")
+
+        cb = cbs["interim_assistant_callback"]
+        cb("interim text")
+
+    assert emitted[0][2]["already_streamed"] is False
+    assert emitted[0][2]["text"] == "interim text"
diff --git a/tui_gateway/server.py b/tui_gateway/server.py
index 798abaedb7df5..bdf1f4d672c54 100644
--- a/tui_gateway/server.py
+++ b/tui_gateway/server.py
@@ -430,6 +430,20 @@ def _load_busy_input_mode() -> str:
     return raw if raw in {"queue", "steer", "interrupt"} else "interrupt"
 
 
+def _load_interim_assistant_messages() -> bool:
+    """Return whether interim assistant commentary should be surfaced to UIs.
+
+    Honors ``display.interim_assistant_messages`` (default true). When false,
+    the tui_gateway does not install ``interim_assistant_callback``, so
+    interim text from tool-call turns and verify-on-stop candidates is never
+    emitted as ``message.interim`` — mirroring the messaging gateway's gating.
+    """
+    display = _load_cfg().get("display")
+    if not isinstance(display, dict):
+        return True
+    return is_truthy_value(display.get("interim_assistant_messages", True))
+
+
 def _notify_session_boundary(
     event_type: str, session_id: str | None, platform: str | None = None
 ) -> None:
@@ -4307,7 +4321,7 @@ def _mirror_subagent_to_child(event_type: str, payload: dict) -> None:
 
 
 def _agent_cbs(sid: str) -> dict:
-    return {
+    callbacks = {
         "tool_start_callback": lambda tc_id, name, args: _on_tool_start(
             sid, tc_id, name, args
         ),
@@ -4362,6 +4376,22 @@ def _agent_cbs(sid: str) -> dict:
         ),
     }
 
+    # Interim assistant commentary (text alongside tool calls, or the attempted
+    # final answer before a verify-on-stop nudge). Gated on
+    # display.interim_assistant_messages (default true). Also set per-turn in
+    # _run_prompt_submit as defense-in-depth — the per-turn set overwrites
+    # this, and the finally block clears it so a stale closure can't fire.
+    if _load_interim_assistant_messages():
+        callbacks["interim_assistant_callback"] = (
+            lambda text, *, already_streamed=False: _emit(
+                "message.interim",
+                sid,
+                {"text": str(text), "already_streamed": bool(already_streamed)},
+            )
+        )
+
+    return callbacks
+
 
 def _apply_project_workspace(task_id: str, path: str, _name: str = "") -> None:
     """Intentional workspace move from the project_* tools: re-anchor the live
@@ -10034,6 +10064,22 @@ def _run_prompt_submit(rid, sid: str, session: dict, text: Any) -> None:
                     payload["rendered"] = r
                 _emit("message.delta", sid, payload)
 
+            # Surface interim assistant text (commentary emitted alongside
+            # tool calls, or the attempted final answer before a verify-on-stop
+            # nudge) so the desktop can seal it as its own segment instead of
+            # losing it when message.complete replaces the streaming buffer.
+            # Gated on display.interim_assistant_messages (default true).
+            if _load_interim_assistant_messages():
+                def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None:
+                    _emit("message.interim", sid, {
+                        "text": text,
+                        "already_streamed": already_streamed,
+                    })
+
+                agent.interim_assistant_callback = _interim_assistant_cb
+            else:
+                agent.interim_assistant_callback = None
+
             run_kwargs = {
                 "conversation_history": list(history),
                 "stream_callback": _stream,
@@ -10155,6 +10201,8 @@ def _run_prompt_submit(rid, sid: str, session: dict, text: Any) -> None:
                 payload["reasoning"] = last_reasoning
             if status_note:
                 payload["warning"] = status_note
+            if result.get("response_previewed"):
+                payload["response_previewed"] = True
             rendered = render_message(raw, cols)
             if rendered:
                 payload["rendered"] = rendered
@@ -10338,6 +10386,9 @@ def _run_prompt_submit(rid, sid: str, session: dict, text: Any) -> None:
             if home_token is not None:
                 reset_hermes_home_override(home_token)
             _clear_session_context(session_tokens)
+            # Clear the per-turn interim callback so a stale closure from
+            # this turn can't fire during a later turn on the same agent.
+            agent.interim_assistant_callback = None
             with session["history_lock"]:
                 session["running"] = False
                 session["last_active"] = time.time()
diff --git a/ui-tui/src/__tests__/createGatewayEventHandler.test.ts b/ui-tui/src/__tests__/createGatewayEventHandler.test.ts
index 9103877bacc18..daf8617c30913 100644
--- a/ui-tui/src/__tests__/createGatewayEventHandler.test.ts
+++ b/ui-tui/src/__tests__/createGatewayEventHandler.test.ts
@@ -1683,4 +1683,81 @@ describe('createGatewayEventHandler', () => {
       expect(openExternalUrlMock).not.toHaveBeenCalled()
     })
   })
+
+  describe('message.interim', () => {
+    it('finalizes an interim segment without settling the turn', () => {
+      const appended: Msg[] = []
+      const onEvent = createGatewayEventHandler(buildCtx(appended))
+
+      onEvent({ payload: {}, type: 'message.start' } as any)
+      onEvent({ payload: { text: 'streaming text' }, type: 'message.delta' } as any)
+      onEvent({ payload: { already_streamed: true, text: 'streaming text' }, type: 'message.interim' } as any)
+
+      // Turn is still active — busy stays true, no completion messages appended
+      expect(getUiState().busy).toBe(true)
+      expect(appended).toHaveLength(0)
+    })
+
+    it('keeps identical interim and terminal replies as separate messages without response_previewed', () => {
+      const appended: Msg[] = []
+      const onEvent = createGatewayEventHandler(buildCtx(appended))
+
+      onEvent({ payload: {}, type: 'message.start' } as any)
+      onEvent({ payload: { already_streamed: true, text: 'same reply' }, type: 'message.interim' } as any)
+      onEvent({ payload: { text: 'same reply' }, type: 'message.complete' } as any)
+
+      const assistantMsgs = appended.filter(m => m.role === 'assistant' && m.text)
+      expect(assistantMsgs).toHaveLength(2)
+    })
+
+    it('settles identical terminal reply onto interim when response_previewed', () => {
+      const appended: Msg[] = []
+      const onEvent = createGatewayEventHandler(buildCtx(appended))
+
+      onEvent({ payload: {}, type: 'message.start' } as any)
+      onEvent({ payload: { already_streamed: true, text: 'same reply' }, type: 'message.interim' } as any)
+      onEvent({ payload: { response_previewed: true, text: 'same reply' }, type: 'message.complete' } as any)
+
+      // With response_previewed, the terminal reply is the same model
+      // response that was published provisionally — settle onto the
+      // interim instead of duplicating. (#65919 review)
+      const assistantMsgs = appended.filter(m => m.role === 'assistant' && m.text)
+      expect(assistantMsgs).toHaveLength(1)
+      expect(assistantMsgs[0]?.text).toBe('same reply')
+    })
+
+    it('deduplicates flushed chunks within the terminal message after an interim boundary', () => {
+      const appended: Msg[] = []
+      const onEvent = createGatewayEventHandler(buildCtx(appended))
+
+      onEvent({ payload: {}, type: 'message.start' } as any)
+      // Interim seals the first segment
+      onEvent({ payload: { already_streamed: true, text: 'interim answer' }, type: 'message.interim' } as any)
+      // Post-interim deltas that match the final text — these get deduped
+      onEvent({ payload: { text: 'final answer' }, type: 'message.delta' } as any)
+      onEvent({ payload: { text: 'final answer' }, type: 'message.complete' } as any)
+
+      const texts = appended.filter(m => m.role === 'assistant' && m.text).map(m => m.text)
+      // interim + final, no duplication of the final
+      expect(texts).toContain('interim answer')
+      expect(texts.filter(t => t === 'final answer')).toHaveLength(1)
+    })
+
+    it('ignores malformed message.interim payload', () => {
+      const appended: Msg[] = []
+      const onEvent = createGatewayEventHandler(buildCtx(appended))
+
+      onEvent({ payload: {}, type: 'message.start' } as any)
+      // No payload at all
+      onEvent({ type: 'message.interim' } as any)
+      // Empty text
+      onEvent({ payload: { text: '' }, type: 'message.interim' } as any)
+      // Undefined text
+      onEvent({ payload: { text: undefined }, type: 'message.interim' } as any)
+
+      // Turn continues without finalizing or throwing
+      expect(getUiState().busy).toBe(true)
+      expect(appended).toHaveLength(0)
+    })
+  })
 })
diff --git a/ui-tui/src/app/createGatewayEventHandler.ts b/ui-tui/src/app/createGatewayEventHandler.ts
index 051f09347777e..d708cc8ccfbbf 100644
--- a/ui-tui/src/app/createGatewayEventHandler.ts
+++ b/ui-tui/src/app/createGatewayEventHandler.ts
@@ -946,6 +946,14 @@ export function createGatewayEventHandler(ctx: GatewayEventHandlerContext): (ev:
         turnController.recordMessageDelta(ev.payload ?? {})
 
         return
+      case 'message.interim': {
+        const text = ev.payload?.text
+        if (typeof text === 'string' && text.trim()) {
+          turnController.recordInterimMessage(text)
+        }
+
+        return
+      }
       case 'message.complete': {
         const { finalMessages, finalText, wasInterrupted } = turnController.recordMessageComplete(ev.payload ?? {})
 
diff --git a/ui-tui/src/app/turnController.ts b/ui-tui/src/app/turnController.ts
index a15810d0dbd65..63acc70276b9b 100644
--- a/ui-tui/src/app/turnController.ts
+++ b/ui-tui/src/app/turnController.ts
@@ -126,6 +126,7 @@ class TurnController {
   private activeTools: ActiveTool[] = []
   private activeReasoningText = ''
   private reasoningSegmentIndex: null | number = null
+  private interimBoundaryIndex: null | number = null
   private activityId = 0
   private reasoningStreamingTimer: Timer = null
   private reasoningTimer: Timer = null
@@ -554,7 +555,7 @@ class TurnController {
     this.flushPendingNotice()
   }
 
-  recordMessageComplete(payload: { rendered?: string; reasoning?: string; text?: string }) {
+  recordMessageComplete(payload: { rendered?: string; reasoning?: string; response_previewed?: boolean; text?: string }) {
     this.closeReasoningSegment()
 
     // Ink renders markdown via <Md>; the gateway's Rich-rendered ANSI
@@ -565,7 +566,15 @@ class TurnController {
     // only when the gateway elected not to send any (#16391).
     const rawText = (payload.text ?? payload.rendered ?? this.bufRef).trimStart()
     const split = splitReasoning(rawText)
-    const finalText = finalTail(split.text, this.segmentMessages)
+    // Only dedupe segments AFTER the interim boundary — interim-sealed
+    // segments are preserved even if the final text includes them.
+    // Exception: when response_previewed is true, the final text is the
+    // same model response that was published provisionally as an interim
+    // message. Dedupe against ALL segments (including sealed interims) so
+    // the identical text doesn't render as a duplicate message. (#65919
+    // review: duplicate-message blocker)
+    const dedupeStart = payload.response_previewed ? 0 : (this.interimBoundaryIndex ?? 0)
+    const finalText = finalTail(split.text, this.segmentMessages.slice(dedupeStart))
     const existingReasoning = this.reasoningText.trim() || String(payload.reasoning ?? '').trim()
     const savedReasoning = [existingReasoning, existingReasoning ? '' : split.reasoning].filter(Boolean).join('\n\n')
     const savedToolTokens = this.toolTokenAcc
@@ -672,6 +681,31 @@ class TurnController {
     }
   }
 
+  recordInterimMessage(text: string) {
+    if (this.interrupted) {
+      return
+    }
+
+    const authoritativeText = text.trimStart()
+    if (!authoritativeText) {
+      return
+    }
+
+    // If the streaming buffer hasn't caught up to the authoritative interim
+    // text (e.g. the backend didn't stream every token), sync it so the
+    // sealed segment matches what the user should see.
+    if (this.bufRef.trimStart() !== authoritativeText) {
+      this.bufRef = authoritativeText
+    }
+
+    // Flush the current streaming buffer into a sealed segment — this is the
+    // TUI equivalent of the desktop's finalizeInterimAssistantMessage. The
+    // segment survives message.complete's finalTail dedupe because
+    // interimBoundaryIndex marks it as interim-sealed.
+    this.flushStreamingSegment()
+    this.interimBoundaryIndex = this.segmentMessages.length
+  }
+
   recordReasoningAvailable(text: string, force = false) {
     if (this.interrupted || (!force && !getUiState().showReasoning)) {
       return
@@ -885,6 +919,7 @@ class TurnController {
     this.pendingSegmentTools = []
     this.protocolWarned = false
     this.reasoningSegmentIndex = null
+    this.interimBoundaryIndex = null
     this.segmentMessages = []
     this.turnTools = []
     this.toolTokenAcc = 0
@@ -941,6 +976,7 @@ class TurnController {
     this.activeTools = []
     this.activeReasoningText = ''
     this.reasoningSegmentIndex = null
+    this.interimBoundaryIndex = null
     this.turnTools = []
     this.toolTokenAcc = 0
     this.interrupted = false
diff --git a/ui-tui/src/gatewayTypes.ts b/ui-tui/src/gatewayTypes.ts
index 953b8a812579c..7fa2c34b37aef 100644
--- a/ui-tui/src/gatewayTypes.ts
+++ b/ui-tui/src/gatewayTypes.ts
@@ -641,7 +641,12 @@ export type GatewayEvent =
   | { payload: SubagentEventPayload; session_id?: string; type: 'subagent.complete' }
   | { payload: { rendered?: string; text?: string }; session_id?: string; type: 'message.delta' }
   | {
-      payload?: { reasoning?: string; rendered?: string; text?: string; usage?: Usage }
+      payload: { already_streamed?: boolean; text: string }
+      session_id?: string
+      type: 'message.interim'
+    }
+  | {
+      payload?: { reasoning?: string; rendered?: string; response_previewed?: boolean; text?: string; usage?: Usage }
       session_id?: string
       type: 'message.complete'
     }