fix(sync): treat part snapshot as a delta coalescing barrier (#1693)

A `message.part.updated` snapshot did not invalidate the pending delta
coalescing key for its message/part. A delta arriving after an intervening
snapshot merged into a delta queued before it, and the snapshot then
overwrote that slot, dropping the later delta's text (e.g. `abc` rendered
as `ab`). Enqueueing a part snapshot now drops that part's pending delta
coalescing keys, while leaving already-queued delta events in place, so
post-snapshot deltas start a fresh entry.

Closes #1647.

Co-authored-by: Ibrahim Khan <ibrakhxn@amazon.com>

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

@@ -97,6 +97,51 @@ describe("createEventPipeline", () => {
     })).toEqual(["updated:a", "delta:b", "updated:ab"])
   })
 
+  test("does not merge deltas across an intervening part snapshot", async () => {
+    let resolveStreamFinished!: () => void
+    const streamFinished = new Promise<void>((resolve) => {
+      resolveStreamFinished = resolve
+    })
+    let resolveDelivered!: () => void
+    const deliveredAll = new Promise<void>((resolve) => {
+      resolveDelivered = resolve
+    })
+    const delivered: Event[] = []
+    const pipeline = createEventPipeline({
+      sdk: createSdk([
+        partUpdatedEvent("a"),
+        deltaEvent("b"),
+        partUpdatedEvent("ab"),
+        deltaEvent("c"),
+      ], resolveStreamFinished),
+      onEvent: (_directory, payload) => {
+        delivered.push(payload)
+        if (delivered.length === 4) {
+          resolveDelivered()
+        }
+      },
+      transport: "sse",
+      heartbeatTimeoutMs: 1_000,
+    })
+
+    try {
+      await streamFinished
+      await Promise.race([deliveredAll, new Promise<void>((resolve) => setTimeout(resolve, 300))])
+    } finally {
+      pipeline.cleanup()
+    }
+
+    // The "ab" snapshot is a coalescing barrier: the trailing "c" delta must
+    // stay a separate event after it, not merge into the "b" delta queued
+    // before the snapshot (which the snapshot would then overwrite).
+    expect(delivered.map((event) => {
+      if (event.type === "message.part.delta") {
+        return `delta:${(event.properties as { delta: string }).delta}`
+      }
+      return `updated:${((event.properties as { part: { text: string } }).part).text}`
+    })).toEqual(["updated:a", "delta:b", "updated:ab", "delta:c"])
+  })
+
   test("normalizes openchamber session status events", async () => {
     let resolveStreamFinished!: () => void
     const streamFinished = new Promise<void>((resolve) => {

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

@@ -447,6 +447,26 @@ export function createEventPipeline(input: EventPipelineInput): EventPipeline {
     const normalizedPayload = normalizeEventType(payload)
     const routedDirectory = routeDirectory?.(directory, normalizedPayload) || directory
     const d = getOrCreateDir(routedDirectory)
+
+    // A full part snapshot is a coalescing barrier for that part's deltas:
+    // drop its pending delta coalescing keys so a delta arriving after the
+    // snapshot starts a fresh queue entry instead of merging into a delta
+    // queued before the snapshot, which the snapshot would then overwrite and
+    // drop the later delta's text. The already-queued delta event stays.
+    if (normalizedPayload.type === "message.part.updated") {
+      const part = (normalizedPayload.properties as { part?: { id?: unknown; messageID?: unknown } }).part
+      const messageID = typeof part?.messageID === "string" ? part.messageID : undefined
+      const partID = typeof part?.id === "string" ? part.id : undefined
+      if (messageID && partID) {
+        const deltaPrefix = `message.part.delta:${messageID}:${partID}:`
+        for (const coalesceKey of d.coalesced.keys()) {
+          if (coalesceKey.startsWith(deltaPrefix)) {
+            d.coalesced.delete(coalesceKey)
+          }
+        }
+      }
+    }
+
     const k = key(normalizedPayload)
     if (k) {
       const i = d.coalesced.get(k)