fix(v2): render native response deltas progressively

Apply OpenCode V2 text and reasoning delta events directly to the normalized message store so assistant responses appear as they are generated instead of waiting for periodic HTTP snapshots.

Deduplicate replayed events, preserve ordinal part ordering, reconcile stale or superseded snapshots, settle terminal messages, retry failed terminal refreshes, and clear streaming state when sessions or workspaces are removed. This keeps snapshot refreshes as authoritative recovery without erasing visible streamed content or recreating removed stores.

Add focused streaming regression coverage and extend native event tests for immediate text, reasoning, periodic refresh behavior, replay handling, deletion cleanup, and terminal reconciliation. Validated with 223 standard UI tests, 64 browser/integration tests, full workspace typechecks, a production Tauri build, and a live CDP smoke covering progressive rendering and reload restoration.

.github/workflows/pr-build.yml

@@ -148,6 +148,7 @@ jobs:
           packages/ui/src/lib/hooks/use-active-session-message-load.test.ts
           packages/ui/src/stores/forms.test.ts
           packages/ui/src/stores/instances-restore-ownership.test.ts
+          packages/ui/src/stores/native-session-streaming.test.ts
           packages/ui/src/stores/permission-lifecycle.test.ts
           packages/ui/src/stores/pty-store-reactivity.test.ts
           packages/ui/src/stores/session-actions.test.ts

packages/ui/src/stores/instances.ts

@@ -41,6 +41,7 @@ import {
 } from "./session-state"
 import { setHasInstances } from "./ui"
 import { messageStoreBus } from "./message-v2/bus"
+import { clearNativeContentDeltaState } from "./native-session-streaming"
 import { upsertPermissionV2, removePermissionV2, upsertQuestionV2, removeQuestionV2 } from "./message-v2/bridge"
 import {
   clearRepliedPermissions,
@@ -1118,6 +1119,7 @@ function removeInstance(id: string, options: { authoritative?: boolean } = {}) {
 
   // Clean up session indexes and drafts for removed instance
   clearCacheForInstance(id)
+  clearNativeContentDeltaState(id)
   messageStoreBus.unregisterInstance(id)
   clearInstanceDraftPrompts(id)
   clearSessionListRequestState(id)

packages/ui/src/stores/native-session-streaming.test.ts

@@ -0,0 +1,148 @@
+import assert from "node:assert/strict"
+import { describe, it } from "node:test"
+import { messageStoreBus } from "./message-v2/bus.ts"
+import {
+  applyNativeContentDelta,
+  clearNativeContentDeltaState,
+  reconcileNativeContentAfterSnapshot,
+  reapplyNativeContentDeltas,
+  settleNativeContentDeltas,
+} from "./native-session-streaming.ts"
+
+describe("native session streaming", () => {
+  it("creates assistant content and applies text and reasoning deltas immediately", () => {
+    const instanceId = "native-streaming"
+    const base = { id: "event", created: 10, data: { sessionID: "session", assistantMessageID: "assistant" } }
+
+    try {
+      assert.equal(applyNativeContentDelta(instanceId, {
+        ...base,
+        type: "session.text.delta",
+        data: { ...base.data, ordinal: 0, delta: "hello" },
+      }), true)
+      applyNativeContentDelta(instanceId, {
+        ...base,
+        id: "event-2",
+        type: "session.text.delta",
+        data: { ...base.data, ordinal: 0, delta: " world" },
+      })
+      applyNativeContentDelta(instanceId, {
+        ...base,
+        id: "event-3",
+        type: "session.reasoning.delta",
+        data: { ...base.data, ordinal: 1, delta: "thinking" },
+      })
+
+      const message = messageStoreBus.getOrCreate(instanceId).getMessage("assistant")
+      assert.equal(message?.status, "streaming")
+      assert.equal((message?.parts["assistant-text-0"]?.data as any)?.text, "hello world")
+      assert.equal((message?.parts["assistant-reasoning-1"]?.data as any)?.text, "thinking")
+      assert.equal(messageStoreBus.getOrCreate(instanceId).getMessageInfo("assistant")?.sessionID, "session")
+    } finally {
+      clearNativeContentDeltaState(instanceId)
+      messageStoreBus.unregisterInstance(instanceId)
+    }
+  })
+
+  it("rejects malformed deltas without creating state", () => {
+    const instanceId = "invalid-native-streaming"
+    try {
+      assert.equal(applyNativeContentDelta(instanceId, {
+        type: "session.text.delta",
+        data: { sessionID: "session", assistantMessageID: "", ordinal: 0, delta: "ignored" },
+      } as any), false)
+      assert.equal(messageStoreBus.getOrCreate(instanceId).getSessionMessageIds("session").length, 0)
+    } finally {
+      clearNativeContentDeltaState(instanceId)
+      messageStoreBus.unregisterInstance(instanceId)
+    }
+  })
+
+  it("deduplicates replayed events and restores direct content after a stale snapshot", () => {
+    const instanceId = "replayed-native-streaming"
+    const event = {
+      id: "event-1",
+      created: 10,
+      type: "session.text.delta" as const,
+      data: { sessionID: "session", assistantMessageID: "assistant", ordinal: 0, delta: "hello" },
+    }
+    try {
+      applyNativeContentDelta(instanceId, event)
+      applyNativeContentDelta(instanceId, event)
+      const store = messageStoreBus.getOrCreate(instanceId)
+      assert.equal((store.getMessage("assistant")?.parts["assistant-text-0"]?.data as any)?.text, "hello")
+
+      store.upsertMessage({
+        id: "assistant", sessionId: "session", role: "assistant", status: "streaming",
+        parts: [{ id: "assistant-text-0", type: "text", text: "", sessionID: "session", messageID: "assistant" }],
+      })
+      reapplyNativeContentDeltas(instanceId, "session")
+      assert.equal((store.getMessage("assistant")?.parts["assistant-text-0"]?.data as any)?.text, "hello")
+
+      settleNativeContentDeltas(instanceId, "session")
+      applyNativeContentDelta(instanceId, { ...event, id: "late-event", data: { ...event.data, delta: " late" } })
+      assert.equal((store.getMessage("assistant")?.parts["assistant-text-0"]?.data as any)?.text, "hello")
+      reconcileNativeContentAfterSnapshot(instanceId, "session")
+      assert.equal((store.getMessage("assistant")?.parts["assistant-text-0"]?.data as any)?.text, "hello")
+      assert.equal(store.getMessage("assistant")?.status, "complete")
+      store.upsertMessage({ id: "assistant", sessionId: "session", role: "assistant", status: "streaming" })
+      reconcileNativeContentAfterSnapshot(instanceId, "session")
+      assert.equal(store.getMessage("assistant")?.status, "complete")
+    } finally {
+      clearNativeContentDeltaState(instanceId)
+      messageStoreBus.unregisterInstance(instanceId)
+    }
+  })
+
+  it("renders parts in ordinal order even when events arrive out of order", () => {
+    const instanceId = "ordered-native-streaming"
+    const data = { sessionID: "session", assistantMessageID: "assistant" }
+    try {
+      applyNativeContentDelta(instanceId, {
+        id: "second", created: 2, type: "session.reasoning.delta",
+        data: { ...data, ordinal: 1, delta: "second" },
+      })
+      applyNativeContentDelta(instanceId, {
+        id: "first", created: 1, type: "session.text.delta",
+        data: { ...data, ordinal: 0, delta: "first" },
+      })
+      assert.deepEqual(messageStoreBus.getOrCreate(instanceId).getMessage("assistant")?.partIds, [
+        "assistant-text-0",
+        "assistant-reasoning-1",
+      ])
+    } finally {
+      clearNativeContentDeltaState(instanceId)
+      messageStoreBus.unregisterInstance(instanceId)
+    }
+  })
+
+  it("rejects negative ordinals", () => {
+    const instanceId = "negative-native-streaming"
+    try {
+      assert.equal(applyNativeContentDelta(instanceId, {
+        id: "event", created: 1, type: "session.text.delta",
+        data: { sessionID: "session", assistantMessageID: "assistant", ordinal: -1, delta: "ignored" },
+      }), false)
+    } finally {
+      clearNativeContentDeltaState(instanceId)
+      messageStoreBus.unregisterInstance(instanceId)
+    }
+  })
+
+  it("does not recreate cleared session state during late reconciliation", () => {
+    const instanceId = "cleared-native-streaming"
+    try {
+      applyNativeContentDelta(instanceId, {
+        id: "event", created: 1, type: "session.text.delta",
+        data: { sessionID: "session", assistantMessageID: "assistant", ordinal: 0, delta: "text" },
+      })
+      messageStoreBus.unregisterInstance(instanceId)
+      clearNativeContentDeltaState(instanceId, "session")
+      reapplyNativeContentDeltas(instanceId, "session")
+      assert.equal(messageStoreBus.getInstance(instanceId), undefined)
+    } finally {
+      clearNativeContentDeltaState(instanceId)
+      if (messageStoreBus.getInstance(instanceId)) messageStoreBus.unregisterInstance(instanceId)
+    }
+  })
+})

packages/ui/src/stores/native-session-streaming.ts

@@ -0,0 +1,185 @@
+import type { SessionReasoningDelta, SessionTextDelta } from "@opencode-ai/client"
+import type { ClientPart, MessageInfo } from "../types/message"
+import { messageStoreBus } from "./message-v2/bus"
+
+type NativeContentDelta = SessionTextDelta | SessionReasoningDelta
+type TrackedPart = {
+  messageId: string
+  partId: string
+  type: "text" | "reasoning"
+  ordinal: number
+  text: string
+  createdAt: number
+}
+type SessionStreamingState = {
+  seenEventIds: Set<string>
+  eventOrder: string[]
+  parts: Map<string, TrackedPart>
+  settledMessageIds: Set<string>
+  settledMessageOrder: string[]
+}
+
+const MAX_TRACKED_EVENT_IDS = 20_000
+const MAX_SETTLED_MESSAGE_IDS = 100
+const streamingState = new Map<string, Map<string, SessionStreamingState>>()
+
+function getSessionState(instanceId: string, sessionId: string): SessionStreamingState {
+  let instance = streamingState.get(instanceId)
+  if (!instance) {
+    instance = new Map()
+    streamingState.set(instanceId, instance)
+  }
+  let session = instance.get(sessionId)
+  if (!session) {
+    session = {
+      seenEventIds: new Set(), eventOrder: [], parts: new Map(),
+      settledMessageIds: new Set(), settledMessageOrder: [],
+    }
+    instance.set(sessionId, session)
+  }
+  return session
+}
+
+function markEventSeen(state: SessionStreamingState, eventId: string): boolean {
+  if (state.seenEventIds.has(eventId)) return false
+  state.seenEventIds.add(eventId)
+  state.eventOrder.push(eventId)
+  while (state.eventOrder.length > MAX_TRACKED_EVENT_IDS) {
+    const expired = state.eventOrder.shift()
+    if (expired) state.seenEventIds.delete(expired)
+  }
+  return true
+}
+
+function renderTrackedPart(instanceId: string, sessionId: string, tracked: TrackedPart): void {
+  const store = messageStoreBus.getOrCreate(instanceId)
+  const current = store.getMessage(tracked.messageId)
+  const existingPart = current?.parts[tracked.partId]?.data as ClientPart | undefined
+  const existingText = existingPart && "text" in existingPart ? existingPart.text : undefined
+  if (typeof existingText === "string") {
+    const terminalSnapshot = current?.status === "complete" || current?.status === "error"
+    if (terminalSnapshot) {
+      if (!tracked.text.startsWith(existingText) || existingText.length >= tracked.text.length) tracked.text = existingText
+    } else if (existingText.length > tracked.text.length && existingText.startsWith(tracked.text)) {
+      tracked.text = existingText
+    }
+  }
+
+  const part: ClientPart = {
+    ...(existingPart ?? {}),
+    id: tracked.partId,
+    type: tracked.type,
+    text: tracked.text,
+    sessionID: sessionId,
+    messageID: tracked.messageId,
+  } as ClientPart
+  const parts = (current?.partIds ?? [])
+    .filter((partId) => partId !== tracked.partId)
+    .map((partId) => current?.parts[partId]?.data)
+    .filter((candidate): candidate is ClientPart => Boolean(candidate))
+  parts.splice(Math.min(tracked.ordinal, parts.length), 0, part)
+
+  store.addOrUpdateSession({ id: sessionId })
+  store.upsertMessage({
+    id: tracked.messageId,
+    sessionId,
+    role: "assistant",
+    status: "streaming",
+    createdAt: current?.createdAt ?? tracked.createdAt,
+    updatedAt: Date.now(),
+    parts,
+  })
+  if (!store.getMessageInfo(tracked.messageId)) {
+    const info: MessageInfo = {
+      id: tracked.messageId,
+      sessionID: sessionId,
+      role: "assistant",
+      time: { created: tracked.createdAt },
+    }
+    store.setMessageInfo(tracked.messageId, info)
+  }
+}
+
+export function applyNativeContentDelta(instanceId: string, event: NativeContentDelta): boolean {
+  const { sessionID, assistantMessageID, ordinal, delta } = event.data
+  if (!instanceId || !event.id || !sessionID || !assistantMessageID || !Number.isInteger(ordinal) || ordinal < 0 || typeof delta !== "string") {
+    return false
+  }
+
+  const state = getSessionState(instanceId, sessionID)
+  if (!markEventSeen(state, event.id) || state.settledMessageIds.has(assistantMessageID)) return true
+  const createdAt = typeof event.created === "number" ? event.created : Date.now()
+  const partType = event.type === "session.text.delta" ? "text" : "reasoning"
+  const partId = `${assistantMessageID}-${partType}-${ordinal}`
+  const tracked = state.parts.get(partId) ?? {
+    messageId: assistantMessageID,
+    partId,
+    type: partType,
+    ordinal,
+    text: "",
+    createdAt,
+  }
+  tracked.text += delta
+  state.parts.set(partId, tracked)
+  renderTrackedPart(instanceId, sessionID, tracked)
+  return true
+}
+
+export function reapplyNativeContentDeltas(instanceId: string, sessionId: string): void {
+  const state = streamingState.get(instanceId)?.get(sessionId)
+  if (!state) return
+  for (const tracked of [...state.parts.values()].sort((a, b) => a.ordinal - b.ordinal)) {
+    renderTrackedPart(instanceId, sessionId, tracked)
+  }
+}
+
+function markMessageComplete(instanceId: string, sessionId: string, messageId: string): void {
+  const store = messageStoreBus.getOrCreate(instanceId)
+  const message = store.getMessage(messageId)
+  if (!message || message.status === "complete" || message.status === "error") return
+  store.upsertMessage({
+    id: message.id,
+    sessionId,
+    role: "assistant",
+    status: "complete",
+    createdAt: message.createdAt,
+    updatedAt: Date.now(),
+  })
+}
+
+export function settleNativeContentDeltas(instanceId: string, sessionId: string): void {
+  const state = streamingState.get(instanceId)?.get(sessionId)
+  if (!state) return
+  for (const tracked of state.parts.values()) {
+    if (!state.settledMessageIds.has(tracked.messageId)) {
+      state.settledMessageIds.add(tracked.messageId)
+      state.settledMessageOrder.push(tracked.messageId)
+    }
+    markMessageComplete(instanceId, sessionId, tracked.messageId)
+  }
+  while (state.settledMessageOrder.length > MAX_SETTLED_MESSAGE_IDS) {
+    const expired = state.settledMessageOrder.shift()
+    if (!expired) continue
+    state.settledMessageIds.delete(expired)
+    for (const [partId, tracked] of state.parts) {
+      if (tracked.messageId === expired) state.parts.delete(partId)
+    }
+  }
+}
+
+export function reconcileNativeContentAfterSnapshot(instanceId: string, sessionId: string): void {
+  const state = streamingState.get(instanceId)?.get(sessionId)
+  if (!state) return
+  reapplyNativeContentDeltas(instanceId, sessionId)
+  for (const messageId of state.settledMessageIds) markMessageComplete(instanceId, sessionId, messageId)
+}
+
+export function clearNativeContentDeltaState(instanceId: string, sessionId?: string): void {
+  if (!sessionId) {
+    streamingState.delete(instanceId)
+    return
+  }
+  const instance = streamingState.get(instanceId)
+  instance?.delete(sessionId)
+  if (instance?.size === 0) streamingState.delete(instanceId)
+}

packages/ui/src/stores/session-api.ts

@@ -57,6 +57,7 @@ import { normalizeSessionMessage } from "./message-v2/normalizers"
 import { updateSessionInfo } from "./message-v2/session-info"
 import { seedSessionMessagesV2, reconcilePendingPermissionsV2, reconcilePendingQuestionsV2 } from "./message-v2/bridge"
 import { messageStoreBus } from "./message-v2/bus"
+import { clearNativeContentDeltaState, reconcileNativeContentAfterSnapshot } from "./native-session-streaming"
 import { clearCacheForSession } from "../lib/global-cache"
 import { getLogger } from "../lib/logger"
 import { getOpencodeErrorMessage } from "../lib/opencode-api"
@@ -733,6 +734,7 @@ async function deleteSession(instanceId: string, sessionId: string): Promise<voi
 }
 
 function removeSessionRuntimeState(instanceId: string, sessionId: string, authoritative = true): void {
+  clearNativeContentDeltaState(instanceId, sessionId)
   cancelSessionGenerationAdmissions(instanceId, sessionId)
   if (authoritative) markSessionDeletedAuthoritative(instanceId, sessionId)
   deleteSessionAttachments(instanceId, sessionId)
@@ -968,7 +970,9 @@ async function loadMessages(
       if (cursor) seenCursors.add(cursor)
     } while (cursor)
 
-    if (!isCurrentMessageLoad(instanceId, sessionId, loadEpoch) || !sessions().get(instanceId)?.has(sessionId)) return
+    if (!instances().has(instanceId)
+      || !isCurrentMessageLoad(instanceId, sessionId, loadEpoch)
+      || !sessions().get(instanceId)?.has(sessionId)) return
 
     if (!Array.isArray(apiMessages)) {
       return
@@ -1034,14 +1038,16 @@ async function loadMessages(
 
       if (!agentName && !providerID && !modelID) {
         const defaultModel = await getDefaultModel(instanceId, session.agent)
-        if (!isCurrentMessageLoad(instanceId, sessionId, loadEpoch) || !sessions().get(instanceId)?.has(sessionId)) return
+        if (!instances().has(instanceId)
+          || !isCurrentMessageLoad(instanceId, sessionId, loadEpoch)
+          || !sessions().get(instanceId)?.has(sessionId)) return
         agentName = session.agent
         providerID = defaultModel.providerId
         modelID = defaultModel.modelId
       }
 
       setSessions((prev) => {
-        if (!isCurrentMessageLoad(instanceId, sessionId, loadEpoch)) return prev
+        if (!instances().has(instanceId) || !isCurrentMessageLoad(instanceId, sessionId, loadEpoch)) return prev
         const next = new Map(prev)
         const nextInstanceSessions = next.get(instanceId)
         if (!nextInstanceSessions) return next
@@ -1059,7 +1065,7 @@ async function loadMessages(
       const sessionForV2 = sessions().get(instanceId)?.get(sessionId) ?? {
         id: sessionId, title: session?.title, parentId: session?.parentId ?? null, revert: session?.revert,
       }
-      if (!isCurrentMessageLoad(instanceId, sessionId, loadEpoch)) return
+      if (!instances().has(instanceId) || !isCurrentMessageLoad(instanceId, sessionId, loadEpoch)) return
       if (!seedSessionMessagesV2(instanceId, sessionForV2, messages, messagesInfo, messageRevision)) {
         retryAfterRevisionConflict = true
       } else {
@@ -1092,6 +1098,9 @@ async function loadMessages(
         return next
       })
     }
+    if (instances().has(instanceId) && sessions().get(instanceId)?.has(sessionId)) {
+      reconcileNativeContentAfterSnapshot(instanceId, sessionId)
+    }
   }
 
   if (retryAfterRevisionConflict && sessions().get(instanceId)?.has(sessionId)) {
@@ -1104,7 +1113,9 @@ async function loadMessages(
     })
   }
 
-  if (!isCurrentMessageLoad(instanceId, sessionId, loadEpoch) || !sessions().get(instanceId)?.has(sessionId)) return
+  if (!instances().has(instanceId)
+    || !isCurrentMessageLoad(instanceId, sessionId, loadEpoch)
+    || !sessions().get(instanceId)?.has(sessionId)) return
   updateSessionInfo(instanceId, sessionId)
 
   if (!skipChildren && session.parentId === null) {

packages/ui/src/stores/session-events.ts

@@ -68,6 +68,11 @@ import {
 } from "./message-v2/bridge"
 import { messageStoreBus } from "./message-v2/bus"
 import { handleConversationAssistantPartUpdated } from "./conversation-speech"
+import {
+  applyNativeContentDelta,
+  clearNativeContentDeltaState,
+  settleNativeContentDeltas,
+} from "./native-session-streaming"
 
 const log = getLogger("sse")
 const pendingSessionFetches = new Map<string, {
@@ -75,6 +80,8 @@ const pendingSessionFetches = new Map<string, {
   retry?: SessionRetryState | null
 }>()
 const NATIVE_REFRESH_DELAY_MS = 75
+const NATIVE_TERMINAL_RETRY_DELAY_MS = 500
+const MAX_NATIVE_TERMINAL_REFRESH_ATTEMPTS = 3
 const nativeRefreshes = new Map<string, {
   instanceId: string
   sessionId: string
@@ -82,6 +89,7 @@ const nativeRefreshes = new Map<string, {
   speakAfter: boolean
   timer?: ReturnType<typeof setTimeout>
   running?: Promise<void>
+  terminalAttempts?: number
 }>()
 let activeRetryToast: ToastHandle | null = null
 
@@ -115,21 +123,47 @@ function requestNativeSessionRefresh(instanceId: string, sessionId: string, fina
     if (refresh.running) return refresh.running
     refresh.pending = false
     refresh.running = (async () => {
+      const terminal = refresh.speakAfter
+      if (!instances().has(refresh.instanceId)
+        || !sessions().get(refresh.instanceId)?.has(refresh.sessionId)
+        || getAuthoritativelyDeletedSessionIdsForInstance(refresh.instanceId).has(refresh.sessionId)) {
+        clearNativeContentDeltaState(refresh.instanceId, refresh.sessionId)
+        return
+      }
       try {
         await loadMessages(refresh.instanceId, refresh.sessionId, { force: true, skipChildren: true })
+        if (!instances().has(refresh.instanceId)
+          || !sessions().get(refresh.instanceId)?.has(refresh.sessionId)
+          || getAuthoritativelyDeletedSessionIdsForInstance(refresh.instanceId).has(refresh.sessionId)) {
+          clearNativeContentDeltaState(refresh.instanceId, refresh.sessionId)
+          return
+        }
+        if (terminal) {
+          settleNativeContentDeltas(refresh.instanceId, refresh.sessionId)
+          refresh.terminalAttempts = 0
+        }
       } catch (error) {
         log.error("Failed to refresh native session messages", { instanceId, sessionId, error })
+        if (terminal
+          && (refresh.terminalAttempts ?? 0) + 1 < MAX_NATIVE_TERMINAL_REFRESH_ATTEMPTS
+          && instances().has(refresh.instanceId)
+          && sessions().get(refresh.instanceId)?.has(refresh.sessionId)) {
+          refresh.terminalAttempts = (refresh.terminalAttempts ?? 0) + 1
+          refresh.pending = true
+        }
       }
     })().finally(() => {
       refresh.running = undefined
       if (refresh.pending) {
-        if (refresh.speakAfter) void run()
+        if (refresh.speakAfter) schedule(NATIVE_TERMINAL_RETRY_DELAY_MS)
         else schedule(NATIVE_REFRESH_DELAY_MS)
         return
       }
       if (refresh.speakAfter) {
         refresh.speakAfter = false
-        speakCompletedAssistantText(refresh.instanceId, refresh.sessionId)
+        if (instances().has(refresh.instanceId) && sessions().get(refresh.instanceId)?.has(refresh.sessionId)) {
+          speakCompletedAssistantText(refresh.instanceId, refresh.sessionId)
+        }
       }
       nativeRefreshes.delete(key)
     })
@@ -158,6 +192,13 @@ function clearNativeSessionRefresh(instanceId: string, sessionId: string): void
 }
 
 function handleNativeSessionEvent(instanceId: string, event: NativeSessionEvent): void {
+  if (event.type === "session.text.delta" || event.type === "session.reasoning.delta") {
+    const sessionId = event.data.sessionID
+    if (!instances().has(instanceId) || getAuthoritativelyDeletedSessionIdsForInstance(instanceId).has(sessionId)) return
+    ensureSessionStatus(instanceId, sessionId, "working", event.location?.directory)
+    applyNativeContentDelta(instanceId, event)
+    return
+  }
   switch (event.type) {
     case "form.created":
       addPendingForm(instanceId, event.data.form)
@@ -225,6 +266,7 @@ function setTerminalNativeSessionStatus(instanceId: string, sessionId: string, f
   if (existing) setSessionStatus(instanceId, sessionId, "idle", { force: true })
   else ensureSessionStatus(instanceId, sessionId, "idle", directory)
   if (failed) messageStoreBus.getOrCreate(instanceId).failPendingSends(sessionId)
+  settleNativeContentDeltas(instanceId, sessionId)
   requestNativeSessionRefresh(instanceId, sessionId, true)
 }
 
@@ -535,6 +577,7 @@ function handleSessionDeleted(instanceId: string, event: EventSessionDeleted): v
 
   log.info(`[SSE] Session deleted: ${sessionId}`)
   clearNativeSessionRefresh(instanceId, sessionId)
+  clearNativeContentDeltaState(instanceId, sessionId)
   removeSessionRuntimeState(instanceId, sessionId)
 }
 
@@ -550,6 +593,7 @@ function handleSessionIdle(instanceId: string, event: SessionIdle): void {
   }
 
   ensureSessionStatus(instanceId, sessionId, "idle", event.location?.directory)
+  settleNativeContentDeltas(instanceId, sessionId)
   requestNativeSessionRefresh(instanceId, sessionId, true)
   log.info(`[SSE] Session idle: ${sessionId}`)
 }

packages/ui/src/stores/session-native-events.test.ts

@@ -5,6 +5,7 @@ import { sdkManager } from "../lib/sdk-manager.ts"
 import type { Session } from "../types/session.ts"
 import { addInstance, removeInstance } from "./instances.ts"
 import { messageStoreBus } from "./message-v2/bus.ts"
+import { clearNativeContentDeltaState } from "./native-session-streaming.ts"
 import { handleNativeSessionEvent, handleSessionIdle, handleSessionStatus } from "./session-events.ts"
 import { clearInstanceDeletedSessionAuthority, sessions, setSessions } from "./session-state.ts"
 
@@ -100,13 +101,14 @@ describe("native session event reducer", () => {
         id: "text", created: 1, type: "session.text.delta",
         data: { sessionID: sessionId, assistantMessageID: "assistant", ordinal: 0, delta: "streaming text" },
       })
+      const store = messageStoreBus.getOrCreate(instanceId)
+      assert.equal((store.getMessage("assistant")?.parts["assistant-text-0"]?.data as any)?.text, "streaming text")
       handleNativeSessionEvent(instanceId, {
         id: "tool", created: 2, type: "session.tool.progress",
         data: { sessionID: sessionId, assistantMessageID: "assistant", id: "tool", metadata: {} },
       })
       await delay(120)
 
-      const store = messageStoreBus.getOrCreate(instanceId)
       assert.equal(calls, 1)
       assert.equal(sessions().get(instanceId)?.get(sessionId)?.status, "working")
       assert.equal((store.getMessage("assistant")?.parts["assistant-text-0"]?.data as any)?.text, "streaming text")
@@ -142,9 +144,17 @@ describe("native session event reducer", () => {
     const eventTypes = ["session.text.delta", "session.reasoning.delta", "session.tool.progress"] as const
     let eventIndex = 0
     const stream = setInterval(() => {
+      const type = eventTypes[eventIndex % eventTypes.length]
+      const id = `event-${eventIndex++}`
       handleNativeSessionEvent(instanceId, {
-        type: eventTypes[eventIndex++ % eventTypes.length],
-        data: { sessionID: sessionId },
+        id,
+        created: eventIndex,
+        type,
+        data: type === "session.text.delta"
+          ? { sessionID: sessionId, assistantMessageID: "assistant", ordinal: 0, delta: "text " }
+          : type === "session.reasoning.delta"
+            ? { sessionID: sessionId, assistantMessageID: "assistant", ordinal: 1, delta: "reason " }
+            : { sessionID: sessionId, assistantMessageID: "assistant", id: "tool", metadata: {} },
       } as any)
     }, 10)
 
@@ -155,8 +165,12 @@ describe("native session event reducer", () => {
 
       assert.ok(calls >= 2, `expected periodic refreshes, received ${calls}`)
       assert.ok(calls <= 5, `expected refreshes to stay bounded, received ${calls}`)
+      const streamed = messageStoreBus.getOrCreate(instanceId).getMessage("assistant")
+      assert.match((streamed?.parts["assistant-text-0"]?.data as any)?.text ?? "", /text/)
+      assert.match((streamed?.parts["assistant-reasoning-1"]?.data as any)?.text ?? "", /reason/)
     } finally {
       clearInterval(stream)
+      clearNativeContentDeltaState(instanceId)
       messageStoreBus.unregisterInstance(instanceId)
       setSessions((prev) => { const next = new Map(prev); next.delete(instanceId); return next })
       clearInstanceDeletedSessionAuthority(instanceId)