fix(ui): keep streaming deltas through pipeline

packages/ui/src/sync/__tests__/event-pipeline.test.js

@@ -275,7 +275,7 @@ describe('createEventPipeline', () => {
     expect(received[0].payload.type).toBe('server.connected');
   });
 
-  it('skips stale message.part.delta events after a newer message.part.updated for the same field', async () => {
+  it('keeps message.part.delta events when a newer message.part.updated is queued for the same field', async () => {
     installDomStubs();
 
     let releaseStream;
@@ -285,8 +285,8 @@ describe('createEventPipeline', () => {
 
     const received = [];
 
-    // Simulate: part.updated arrives, then delta, then a newer part.updated for the
-    // same part. The older queued delta becomes stale and must be skipped.
+    // The pipeline only routes/coalesces events. Whether this delta is already
+    // represented by the newer snapshot is reducer state, not queue state.
     const directory = '/test/dir';
     const sdk = createSdkWithEvents([
       // T0: message.part.updated for part-A
@@ -299,7 +299,7 @@ describe('createEventPipeline', () => {
           },
         },
       },
-      // T1: message.part.delta for part-A (should be dropped as stale)
+      // T1: message.part.delta for part-A
       {
         payload: {
           type: 'message.part.delta',
@@ -329,7 +329,7 @@ describe('createEventPipeline', () => {
         sdk,
         onEvent: (dir, payload) => {
           received.push({ directory: dir, payload });
-          if (received.length === 1) {
+          if (received.length === 2) {
             cleanup();
             releaseStream();
             resolve();
@@ -340,8 +340,10 @@ describe('createEventPipeline', () => {
 
     await delivered;
 
-    expect(received.length).toBe(1);
+    expect(received.length).toBe(2);
     expect(received[0].payload.type).toBe('message.part.updated');
+    expect(received[1].payload.type).toBe('message.part.delta');
+    expect(received[1].payload.properties.delta).toBe(' world');
   });
 
   it('keeps delta events for other fields on the same part', async () => {

packages/ui/src/sync/event-pipeline.ts

@@ -1,6 +1,10 @@
 /**
  * Event Pipeline — transport connection, event coalescing, and batched flush.
  *
+ * This module must not make state-dependent decisions about event validity.
+ * For example, deciding whether a delta is already represented by a full part
+ * snapshot belongs in the reducer, which has access to the current state.
+ *
  * Plain closure API:
  *   const { cleanup } = createEventPipeline({ sdk, onEvent })
  *
@@ -145,7 +149,6 @@ type DirectoryQueue = {
   queue: Event[]
   buffer: Event[]
   coalesced: Map<string, number>
-  staleDeltas: Set<string>
   timer: ReturnType<typeof setTimeout> | undefined
   last: number
 }
@@ -185,7 +188,6 @@ export function createEventPipeline(input: EventPipelineInput) {
       queue: [],
       buffer: [],
       coalesced: new Map(),
-      staleDeltas: new Set(),
       timer: undefined,
       last: 0,
     }
@@ -212,8 +214,6 @@ export function createEventPipeline(input: EventPipelineInput) {
     return undefined
   }
 
-  const deltaKey = (messageID: string, partID: string, field: string) => `${messageID}:${partID}:${field}`
-
   const flushDir = (directory: string) => {
     const d = directories.get(directory)
     if (!d) return
@@ -224,22 +224,14 @@ export function createEventPipeline(input: EventPipelineInput) {
     if (d.queue.length === 0) return
 
     const events = d.queue
-    const staleDeltas = d.staleDeltas.size > 0 ? new Set(d.staleDeltas) : undefined
     d.queue = d.buffer
     d.buffer = events
     d.queue.length = 0
     d.coalesced.clear()
-    d.staleDeltas.clear()
 
     d.last = Date.now()
     syncDebug.pipeline.flush(events.length)
     for (const payload of events) {
-      if (staleDeltas && payload.type === "message.part.delta") {
-        const props = payload.properties as { messageID: string; partID: string; field: string }
-        if (staleDeltas.has(deltaKey(props.messageID, props.partID, props.field))) {
-          continue
-        }
-      }
       onEvent(directory, payload)
     }
 
@@ -313,15 +305,6 @@ export function createEventPipeline(input: EventPipelineInput) {
           } as unknown as Event
         } else {
           d.queue[i] = normalizedPayload
-          if (normalizedPayload.type === "message.part.updated") {
-            const part = (normalizedPayload.properties as { part: Record<string, unknown> & { messageID: string; id: string } }).part
-            for (const field of ["text", "output"] as const) {
-              const value = part[field]
-              if (typeof value === "string" && value.length > 0) {
-                d.staleDeltas.add(deltaKey(part.messageID, part.id, field))
-              }
-            }
-          }
         }
         syncDebug.pipeline.coalesced(normalizedPayload.type, k)
         return

packages/ui/src/sync/reconnect-recovery.test.ts

@@ -54,6 +54,17 @@ describe("getReconnectCandidateSessionIds", () => {
     }).sort()).toContain("active")
   })
 
+  test("includes completed assistant sessions when the latest assistant parts are missing", () => {
+    expect(getReconnectCandidateSessionIds({
+      session: [createSession("blank")],
+      session_status: { blank: { type: "idle" } as SessionStatus },
+      message: {
+        blank: [createAssistantMessage("m-1", "blank", 1)],
+      },
+      part: {},
+    })).toEqual(["blank"])
+  })
+
   test("does not include a viewed session from another directory", () => {
     expect(getReconnectCandidateSessionIds({
       session: [createSession("active")],

packages/ui/src/sync/reconnect-recovery.ts

@@ -1,10 +1,11 @@
-import type { SessionStatus, Message } from "@opencode-ai/sdk/v2/client"
+import type { SessionStatus, Message, Part } from "@opencode-ai/sdk/v2/client"
 import type { Session } from "@opencode-ai/sdk/v2"
 
 type ReconnectRecoveryState = {
   session: Session[]
   session_status?: Record<string, SessionStatus>
   message?: Record<string, Message[]>
+  part?: Record<string, Part[]>
 }
 
 export type ViewedSessionRecoveryTarget = {
@@ -26,12 +27,17 @@ export function getReconnectCandidateSessionIds(state: ReconnectRecoveryState, o
 
   for (const [sessionId, messages] of Object.entries(state.message ?? {})) {
     const lastMessage = messages[messages.length - 1]
+    const lastAssistantComplete = lastMessage
+      && lastMessage.role === "assistant"
+      && typeof (lastMessage as { time?: { completed?: number } }).time?.completed === "number"
     if (
       lastMessage
       && lastMessage.role === "assistant"
       && typeof (lastMessage as { time?: { completed?: number } }).time?.completed !== "number"
     ) {
       ids.add(sessionId)
+    } else if (lastAssistantComplete && state.part && (state.part[lastMessage.id]?.length ?? 0) === 0) {
+      ids.add(sessionId)
     }
   }