fix(desktop): settle transform_llm_output final onto the streamed bubble (#96465)

Renderer half of #96467 (7eeb0d4b03), hand-applied onto main's
use-message-stream: completeAssistantMessage takes response_transformed
(appended after persistedTurn) and settles a transformed final in place even
with no prefix relationship, gated on sawAssistantPayload. Contributor vitest
kept (trimmed to the core invariant).

apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/message-stream.ts

@@ -371,7 +371,8 @@ export function handleMessageStreamEvent(ctx: GatewayEventContext): boolean {
       payload?.response_previewed,
       failure,
       occurredAt,
-      payload?.persisted_turn
+      payload?.persisted_turn,
+      payload?.response_transformed
     )
 
     // Onboarding's first build: between turns is the only moment Setup may

apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/types.ts

@@ -21,7 +21,8 @@ export interface GatewayEventDeps {
     responsePreviewed?: boolean,
     failure?: { error: string; partial: boolean },
     occurredAt?: number,
-    persistedTurn?: PersistedTurn | null
+    persistedTurn?: PersistedTurn | null,
+    responseTransformed?: boolean
   ) => void
   failAssistantMessage: (
     sessionId: string,

apps/desktop/src/app/session/hooks/use-message-stream/index.ts

@@ -623,7 +623,8 @@ export function useMessageStream({
       responsePreviewed?: boolean,
       failure?: { error: string; partial: boolean; surface?: ErrorSurface | null },
       occurredAt = Date.now() / 1000,
-      persistedTurn?: PersistedTurn | null
+      persistedTurn?: PersistedTurn | null,
+      responseTransformed?: boolean
     ) => {
       let shouldHydrate = false
 
@@ -800,7 +801,17 @@ export function useMessageStream({
               (finalText === existingText || finalText.startsWith(existingText) || existingText.startsWith(finalText))
             )
 
-            if (existing.pending || (!interimBoundaryPending && finalText && existingText === finalText)) {
+            // A transform_llm_output hook rewrites the final after streaming (e.g. pseudonym
+            // restore), so it may share no prefix with the streamed text yet is still this
+            // turn's authoritative reply: settle it in place rather than append a duplicate.
+            // Gated on sawAssistantPayload so a stale completion can't overwrite an older bubble.
+            const transformedCurrentReply = Boolean(responseTransformed && state.sawAssistantPayload && finalText)
+
+            if (
+              existing.pending ||
+              transformedCurrentReply ||
+              (!interimBoundaryPending && finalText && existingText === finalText)
+            ) {
               nextMessages = settleAt(index)
             } else if ((interimBoundaryPending && responsePreviewed) || finalContinuesInterim) {
               // Settle the interim in place instead of creating a duplicate —

apps/desktop/src/app/session/hooks/use-message-stream/transformed-final.test.tsx

@@ -0,0 +1,71 @@
+import { act, cleanup } from '@testing-library/react'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+
+import type { ClientSessionState } from '@/app/types'
+import { chatMessageText } from '@/lib/chat-messages'
+import { clearSessionTodos } from '@/store/todos'
+
+import { type MessageStreamHarness, renderMessageStream } from './test-harness'
+
+const SID = 'session-1'
+
+let stream: MessageStreamHarness
+
+function mountStream() {
+  stream = renderMessageStream(SID)
+}
+
+const start = () => act(() => stream.handleEvent({ payload: {}, session_id: SID, type: 'message.start' }))
+
+const delta = (text: string) =>
+  act(() => stream.handleEvent({ payload: { text }, session_id: SID, type: 'message.delta' }))
+
+const interim = (text: string) =>
+  act(() => stream.handleEvent({ payload: { text, already_streamed: true }, session_id: SID, type: 'message.interim' }))
+
+const completeTransformed = (text: string) =>
+  act(() =>
+    stream.handleEvent({ payload: { text, response_transformed: true }, session_id: SID, type: 'message.complete' })
+  )
+
+function getState(): ClientSessionState {
+  return stream.state()
+}
+
+function assistantTexts(): string[] {
+  const state = getState()
+  return state.messages
+    .filter(m => m.role === 'assistant' && !m.hidden)
+    .map(m => chatMessageText(m))
+    .filter(Boolean)
+}
+
+describe('useMessageStream response_transformed settlement', () => {
+  beforeEach(() => {
+    clearSessionTodos(SID)
+  })
+
+  afterEach(() => {
+    cleanup()
+    clearSessionTodos(SID)
+    vi.restoreAllMocks()
+  })
+
+  it('replaces the streamed bubble with transformed final text that shares no prefix', async () => {
+    mountStream()
+    await start()
+
+    // A transform_llm_output plugin hook (e.g. pseudonym restore) rewrites the
+    // final text after streaming finishes. The rewritten text shares NO prefix
+    // relationship with what was streamed: the prefix-continuity heuristic must
+    // not reject it — it is this turn's authoritative reply.
+    await delta('TOKEN_1')
+    await interim('TOKEN_1')
+    await completeTransformed('example-service.internal')
+
+    const texts = assistantTexts()
+    expect(texts).toHaveLength(1)
+    expect(texts[0]).toBe('example-service.internal')
+    expect(texts).not.toContain('TOKEN_1')
+  })
+})

apps/desktop/src/lib/chat-messages/types.ts

@@ -204,6 +204,9 @@ export type GatewayEventPayload = {
   // message.complete — signals the final text was already previewed via
   // interim_assistant_callback, so the UI can settle instead of duplicating.
   response_previewed?: boolean
+  // message.complete — a transform_llm_output hook rewrote the final text after streaming;
+  // it authoritatively replaces the current turn's streamed text even without a prefix match.
+  response_transformed?: boolean
   persisted_turn?: PersistedTurn | null
   // message.complete — history-commit note the gateway surfaced instead of dropping.
   warning?: string