fix(desktop, ink): don't wipe messages before final message

agent/agent_runtime_helpers.py

@@ -530,6 +530,12 @@ def _is_codex_interim(m: Dict) -> bool:
             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 _is_codex_interim(m: Dict) -> bool:
             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 [])

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 _perform_api_call(next_api_kwargs):
                         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 _perform_api_call(next_api_kwargs):
                     # 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 _perform_api_call(next_api_kwargs):
                 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 _perform_api_call(next_api_kwargs):
                     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 _perform_api_call(next_api_kwargs):
                     # 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 _perform_api_call(next_api_kwargs):
         _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,
     )
 
 

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,15 +271,21 @@ 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
             except Exception:
                 _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),

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
 
 

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,