fix(sync): remove stale-delta skip and add parts-gap recovery (#889)

The pipeline's stale-delta mechanism incorrectly marked all
message.part.delta events as stale when a message.part.updated
coalesced, regardless of queue position. This caused valid streaming
deltas to be silently dropped, resulting in blank or incomplete
assistant messages.

Additionally, when part events were dropped by the reducer (missing
parts array or partID not found), there was no recovery path — the
state stayed permanently out of sync until the next SSE reconnect
or manual refresh.

Also discovered: message.updated that successfully writes an assistant
message but has empty parts would render a blank bubble, with no
repair triggered since repair only ran on reducer return false.

Changes:
- Remove staleDeltas Set and deltaKey from event-pipeline.ts
- Coalesce still replaces same-key events, but deltas are never skipped
- Add enqueuePartsRepair + repairSessionParts to sync-context.tsx
  (5s cooldown, deduped, async SDK re-fetch)
- Trigger repair on reducer return false for part events
- Trigger repair on message.updated return true with empty parts
- Add sync debug.ts with gated diagnostic logging
- Add pipeline coalescing tests

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

@@ -35,6 +35,22 @@ function createSdkWithSingleEvent(event, hold) {
   };
 }
 
+// Helper to create an SDK that yields multiple events in sequence, then holds.
+function createSdkWithEvents(events, hold) {
+  return {
+    global: {
+      event: async () => ({
+        stream: (async function* () {
+          for (const event of events) {
+            yield event;
+          }
+          await hold;
+        })(),
+      }),
+    },
+  };
+}
+
 describe('createEventPipeline', () => {
   it('falls back to payload.properties.directory when the SDK event omits top-level directory', async () => {
     installDomStubs();
@@ -192,4 +208,136 @@ describe('createEventPipeline', () => {
     expect(received[0].directory).toBe('global');
     expect(received[0].payload.type).toBe('server.connected');
   });
+
+  it('delivers message.part.delta events after a coalesced message.part.updated (no stale-delta skip)', async () => {
+    installDomStubs();
+
+    let releaseStream;
+    const hold = new Promise((resolve) => {
+      releaseStream = resolve;
+    });
+
+    const received = [];
+
+    // Simulate: part.updated arrives first, then delta, then part.updated again (coalesces with first).
+    // After coalescing, the delta should still be delivered — NOT skipped.
+    const directory = '/test/dir';
+    const sdk = createSdkWithEvents([
+      // T0: message.part.updated for part-A
+      {
+        payload: {
+          type: 'message.part.updated',
+          properties: {
+            directory,
+            part: { id: 'part-A', type: 'text', messageID: 'msg-1' },
+          },
+        },
+      },
+      // T1: message.part.delta for part-A (should flow through even after coalesce)
+      {
+        payload: {
+          type: 'message.part.delta',
+          properties: {
+            directory,
+            messageID: 'msg-1',
+            partID: 'part-A',
+            field: 'text',
+            delta: ' world',
+          },
+        },
+      },
+      // T2: message.part.updated for part-A — coalesces with T0
+      {
+        payload: {
+          type: 'message.part.updated',
+          properties: {
+            directory,
+            part: { id: 'part-A', type: 'text', messageID: 'msg-1' },
+          },
+        },
+      },
+    ], hold);
+
+    const delivered = new Promise((resolve) => {
+      const { cleanup } = createEventPipeline({
+        sdk,
+        onEvent: (dir, payload) => {
+          received.push({ directory: dir, payload });
+          if (received.length === 2) {
+            cleanup();
+            releaseStream();
+            resolve();
+          }
+        },
+      });
+    });
+
+    await delivered;
+
+    // Coalescing means T0 and T2 merge into one event at T0's queue position.
+    // The delta is a different event type with no coalesce key, so it gets
+    // its own queue slot. After coalesce:
+    //   - queue[0] = coalesced part.updated (from T2, replacing T0)
+    //   - queue[1] = part.delta (from T1)
+    // Total: 2 events delivered
+    expect(received.length).toBe(2);
+
+    // The first event should be the coalesced message.part.updated
+    expect(received[0].payload.type).toBe('message.part.updated');
+
+    // The delta MUST be delivered — it should NOT be skipped
+    expect(received[1].payload.type).toBe('message.part.delta');
+    expect(received[1].payload.properties.delta).toBe(' world');
+  });
+
+  it('coalesces message.part.updated events for the same part', async () => {
+    installDomStubs();
+
+    let releaseStream;
+    const hold = new Promise((resolve) => {
+      releaseStream = resolve;
+    });
+
+    const received = [];
+    const directory = '/test/dir';
+
+    const sdk = createSdkWithEvents([
+      {
+        payload: {
+          type: 'message.part.updated',
+          properties: {
+            directory,
+            part: { id: 'part-A', type: 'text', messageID: 'msg-1' },
+          },
+        },
+      },
+      {
+        payload: {
+          type: 'message.part.updated',
+          properties: {
+            directory,
+            part: { id: 'part-A', type: 'text', messageID: 'msg-1' },
+          },
+        },
+      },
+    ], hold);
+
+    const delivered = new Promise((resolve) => {
+      const { cleanup } = createEventPipeline({
+        sdk,
+        onEvent: (dir, payload) => {
+          received.push({ directory: dir, payload });
+          cleanup();
+          releaseStream();
+          resolve();
+        },
+      });
+    });
+
+    await delivered;
+
+    // Only 1 event should be delivered (coalesced)
+    expect(received.length).toBe(1);
+    expect(received[0].payload.type).toBe('message.part.updated');
+  });
 });

packages/ui/src/sync/debug.ts

@@ -0,0 +1,83 @@
+/**
+ * Sync debug logging — gated behind localStorage flag.
+ *
+ * Enable in browser console:
+ *   localStorage.setItem("openchamber:sync:debug", "1")
+ *
+ * Disable:
+ *   localStorage.removeItem("openchamber:sync:debug")
+ *
+ * All checks are early-returns on the hot path — zero cost when disabled.
+ */
+
+const FLAG_KEY = "openchamber:sync:debug"
+
+let _enabled: boolean | undefined
+
+export function isSyncDebugEnabled(): boolean {
+  if (_enabled !== undefined) return _enabled
+  try {
+    _enabled = typeof localStorage !== "undefined" && localStorage.getItem(FLAG_KEY) === "1"
+  } catch {
+    _enabled = false
+  }
+  return _enabled
+}
+
+/** Force-refresh the flag (call after user toggles localStorage). */
+export function refreshSyncDebugFlag(): void {
+  _enabled = undefined
+}
+
+type SyncDebugCategory = "pipeline" | "reducer" | "dispatch"
+
+function log(cat: SyncDebugCategory, ...args: unknown[]): void {
+  if (!isSyncDebugEnabled()) return
+  const tag = `%c[sync:${cat}]`
+  const style = "color: #888"
+  console.log(tag, style, ...args)
+}
+
+export const syncDebug = {
+  pipeline: {
+    /** Event coalesced (replaced an earlier event in the queue). */
+    coalesced: (eventType: string, coalesceKey: string) =>
+      log("pipeline", "coalesced", eventType, coalesceKey),
+
+    /** Flush batch dispatched. */
+    flush: (count: number) =>
+      log("pipeline", "flush", `${count} events`),
+  },
+
+  reducer: {
+    /** message.updated skipped because role/finish/completed matched existing. */
+    messageUpdatedUnchanged: (sessionID: string, messageID: string, role: string, finish: unknown, completed: unknown) =>
+      log("reducer", "message.updated UNCHANGED (skipped)", { sessionID, messageID, role, finish, completed }),
+
+    /** message.part.updated arrived but no parts array exists for this messageID. */
+    partUpdatedNoExistingParts: (messageID: string, partID: string, partType: string) =>
+      log("reducer", "message.part.updated NO EXISTING PARTS", { messageID, partID, partType }),
+
+    /** message.part.delta arrived but parts array missing — silently dropped. */
+    partDeltaNoParts: (messageID: string, partID: string) =>
+      log("reducer", "message.part.delta DROPPED (no parts array)", { messageID, partID }),
+
+    /** message.part.delta arrived but partID not found in parts array. */
+    partDeltaNotFound: (messageID: string, partID: string) =>
+      log("reducer", "message.part.delta DROPPED (partID not found)", { messageID, partID }),
+
+    /** SKIP_PARTS filtered out a part. */
+    partSkipped: (messageID: string, partID: string, partType: string) =>
+      log("reducer", "message.part.updated SKIPPED (type filtered)", { messageID, partID, partType }),
+  },
+
+  dispatch: {
+    /** Event dispatched to store but reducer returned false (no state change). */
+    eventNoChange: (eventType: string, sessionID?: string, messageID?: string) =>
+      log("dispatch", "event → no change", { eventType, sessionID, messageID }),
+
+    /** Event applied to store successfully. */
+    eventApplied: (eventType: string, sessionID?: string, messageID?: string) =>
+      log("dispatch", "event → applied", { eventType, sessionID, messageID }),
+  },
+} as const

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

@@ -9,6 +9,7 @@
  */
 
 import type { Event, OpencodeClient } from "@opencode-ai/sdk/v2/client"
+import { syncDebug } from "./debug"
 
 // ---------------------------------------------------------------------------
 // Types
@@ -86,13 +87,9 @@ export function createEventPipeline(input: EventPipelineInput) {
   let queue: QueuedEvent[] = []
   let buffer: QueuedEvent[] = []
   const coalesced = new Map<string, number>()
-  const staleDeltas = new Set<string>()
   let timer: ReturnType<typeof setTimeout> | undefined
   let last = 0
 
-  const deltaKey = (directory: string, messageID: string, partID: string) =>
-    `${directory}:${messageID}:${partID}`
-
   // Coalesce key — same-type events for the same entity replace earlier ones
   const key = (directory: string, payload: Event): string | undefined => {
     if (payload.type === "session.status") {
@@ -109,7 +106,7 @@ export function createEventPipeline(input: EventPipelineInput) {
     return undefined
   }
 
-  // Flush — swap queue, dispatch events, skip stale deltas
+  // Flush — swap queue, dispatch events
   const flush = () => {
     if (timer) clearTimeout(timer)
     timer = undefined
@@ -117,21 +114,16 @@ export function createEventPipeline(input: EventPipelineInput) {
     if (queue.length === 0) return
 
     const events = queue
-    const skip = staleDeltas.size > 0 ? new Set(staleDeltas) : undefined
     queue = buffer
     buffer = events
     queue.length = 0
     coalesced.clear()
-    staleDeltas.clear()
 
     last = Date.now()
+    syncDebug.pipeline.flush(events.length)
     // React 18 batches synchronous setState calls automatically,
     // equivalent to SolidJS batch()
     for (const event of events) {
-      if (skip && event.payload.type === "message.part.delta") {
-        const props = event.payload.properties as { messageID: string; partID: string }
-        if (skip.has(deltaKey(event.directory, props.messageID, props.partID))) continue
-      }
       onEvent(event.directory, event.payload)
     }
 
@@ -214,10 +206,7 @@ export function createEventPipeline(input: EventPipelineInput) {
             const i = coalesced.get(k)
             if (i !== undefined) {
               queue[i] = { directory, payload: normalizedPayload }
-              if (normalizedPayload.type === "message.part.updated") {
-                const part = (normalizedPayload.properties as { part: { messageID: string; id: string } }).part
-                staleDeltas.add(deltaKey(directory, part.messageID, part.id))
-              }
+              syncDebug.pipeline.coalesced(normalizedPayload.type, k)
               continue
             }
             coalesced.set(k, queue.length)

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

@@ -14,6 +14,7 @@ import { Binary } from "./binary"
 import type { GlobalState, State } from "./types"
 import { dropSessionCaches } from "./session-cache"
 import { stripSessionDiffSnapshots } from "./sanitize"
+import { syncDebug } from "./debug"
 
 const SKIP_PARTS = new Set(["patch", "step-start", "step-finish"])
 
@@ -160,6 +161,7 @@ export function applyDirectoryEvent(
           && (existing as { finish?: unknown }).finish === (info as { finish?: unknown }).finish
           && (existing.time as { completed?: number })?.completed === (info.time as { completed?: number })?.completed
         if (unchanged) {
+          syncDebug.reducer.messageUpdatedUnchanged(info.sessionID, info.id, info.role, (info as { finish?: unknown }).finish, (info.time as { completed?: number })?.completed)
           return false
         }
         const next = [...messages]
@@ -190,10 +192,14 @@ export function applyDirectoryEvent(
 
     case "message.part.updated": {
       const part = (event.properties as { part: Part }).part
-      if (SKIP_PARTS.has(part.type)) return false
+      if (SKIP_PARTS.has(part.type)) {
+        syncDebug.reducer.partSkipped((part as { messageID: string }).messageID, part.id, part.type)
+        return false
+      }
       const messageID = (part as { messageID: string }).messageID
       const parts = draft.part[messageID]
       if (!parts) {
+        syncDebug.reducer.partUpdatedNoExistingParts(messageID, part.id, part.type)
         draft.part[messageID] = [part]
         return true
       }
@@ -246,9 +252,15 @@ export function applyDirectoryEvent(
         delta: string
       }
       const parts = draft.part[props.messageID]
-      if (!parts) return false
+      if (!parts) {
+        syncDebug.reducer.partDeltaNoParts(props.messageID, props.partID)
+        return false
+      }
       const result = Binary.search(parts, props.partID, (p) => p.id)
-      if (!result.found) return false
+      if (!result.found) {
+        syncDebug.reducer.partDeltaNotFound(props.messageID, props.partID)
+        return false
+      }
       const existing = parts[result.index] as Record<string, unknown>
       const existingValue = existing[props.field] as string | undefined
       // Create new Part object + new array so React detects the change

packages/ui/src/sync/sync-context.tsx

@@ -15,6 +15,7 @@ import { updateStreamingState } from "./streaming"
 import { setActionRefs } from "./session-actions"
 import { setSyncRefs } from "./sync-refs"
 import { stripMessageDiffSnapshots, stripSessionDiffSnapshots } from "./sanitize"
+import { syncDebug } from "./debug"
 import { opencodeClient } from "@/lib/opencode/client"
 import { usePermissionStore } from "@/stores/permissionStore"
 import { autoRespondsPermission, normalizeDirectory } from "@/stores/utils/permissionAutoAccept"
@@ -100,6 +101,79 @@ const RECONNECT_SKIP_PARTS = new Set(["patch", "step-start", "step-finish"])
 
 const cmp = (a: string, b: string) => (a < b ? -1 : a > b ? 1 : 0)
 
+// ---------------------------------------------------------------------------
+// Parts-gap recovery — when SSE events arrive but parts are missing,
+// trigger a targeted re-fetch for the affected sessions.
+// Tracked per-directory, deduplicated, and auto-expiring.
+// ---------------------------------------------------------------------------
+
+type PendingRepair = {
+  sessionID: string
+  directory: string
+  enqueuedAt: number
+}
+
+const REPAIR_COOLDOWN_MS = 5_000
+const pendingRepairs = new Map<string, PendingRepair>() // key: directory:sessionID
+
+const repairKey = (directory: string, sessionID: string) => `${directory}:${sessionID}`
+
+function enqueuePartsRepair(directory: string, sessionID: string, childStores: ChildStoreManager) {
+  if (!directory || directory === "global" || !sessionID) return
+  const k = repairKey(directory, sessionID)
+  const existing = pendingRepairs.get(k)
+  if (existing && Date.now() - existing.enqueuedAt < REPAIR_COOLDOWN_MS) return
+
+  pendingRepairs.set(k, { sessionID, directory, enqueuedAt: Date.now() })
+
+  // Defer to next microtask so we don't hold up the current event batch
+  void Promise.resolve().then(async () => {
+    const store = childStores.getChild(directory)
+    if (!store) {
+      pendingRepairs.delete(k)
+      return
+    }
+    try {
+      await repairSessionParts(directory, sessionID, store)
+    } catch {
+      // Transient failure — next SSE event or reconnect will catch up.
+    } finally {
+      pendingRepairs.delete(k)
+    }
+  })
+}
+
+async function repairSessionParts(
+  directory: string,
+  sessionID: string,
+  store: StoreApi<DirectoryStore>,
+) {
+  const scopedClient = opencodeClient.getScopedSdkClient(directory)
+  const result = await retry(() =>
+    scopedClient.session.messages({ sessionID, limit: RECONNECT_MESSAGE_LIMIT }),
+  )
+  const records = (result.data ?? []).filter((record: { info?: { id?: string } }) => !!record?.info?.id)
+  if (records.length === 0) return
+
+  store.setState((state: DirectoryStore) => {
+    const nextPartState = { ...state.part }
+    for (const record of records) {
+      const messageId = record?.info?.id
+      if (!messageId) continue
+      const newParts = (record.parts ?? [])
+        .filter((part: Part) => !!part?.id && !RECONNECT_SKIP_PARTS.has(part.type))
+        .sort((a: Part, b: Part) => cmp(a.id, b.id))
+
+      const existing = nextPartState[messageId]
+      // Only patch if parts were missing or fewer than server has
+      if (!existing || existing.length < newParts.length) {
+        nextPartState[messageId] = newParts
+      }
+    }
+    return { part: nextPartState }
+  })
+}
+
 // Module-level refs for notification viewed check.
 // Used to determine if user is currently viewing the session when a notification arrives.
 let _activeDirectory = ""
@@ -804,6 +878,33 @@ function handleEvent(
 
   if (applyDirectoryEvent(draft, payload)) {
     store.setState(draft)
+    const sessionID = getSessionIdFromPayload(payload) ?? undefined
+    const messageID = getMessageIdFromPayload(payload) ?? undefined
+    syncDebug.dispatch.eventApplied(payload.type, sessionID, messageID)
+
+    // Parts-gap recovery on message.updated: if the message was inserted or
+    // replaced but draft.part[messageID] is empty, the parts were lost or
+    // never arrived. Trigger repair so the UI doesn't render a blank bubble.
+    if (sessionID && messageID && payload.type === "message.updated") {
+      const after = store.getState()
+      const info = (payload.properties as { info: Message }).info
+      if (info.role === "assistant" && (!after.part[messageID] || after.part[messageID].length === 0)) {
+        enqueuePartsRepair(resolvedDirectory, sessionID, childStores)
+      }
+    }
+  } else {
+    const sessionID = getSessionIdFromPayload(payload) ?? undefined
+    const messageID = getMessageIdFromPayload(payload) ?? undefined
+    syncDebug.dispatch.eventNoChange(payload.type, sessionID, messageID)
+
+    // Parts-gap recovery: if a part event was dropped because the parts array
+    // was missing (message not yet inserted or parts lost), trigger a repair
+    // fetch for the session.
+    if (sessionID && messageID && (
+      payload.type === "message.part.delta" || payload.type === "message.part.updated"
+    )) {
+      enqueuePartsRepair(resolvedDirectory, sessionID, childStores)
+    }
   }
 
   updateRoutingIndexFromEvent(routingIndex, resolvedDirectory, payload)