fix(client): preserve inbox events during snapshot reads (#49106)

packages/client/src/solid/data.ts

@@ -252,6 +252,7 @@ export function createData(config: CreateDataInput) {
   const messageIndex = new Map<string, Map<string, number>>()
   const sync = createSync()
   let activeUpdates: Map<string, DataSessionStatus | undefined> | undefined
+  const pendingUpdates = new Map<string, Map<string, SessionInboxInfo | SessionInbox.Delivery | undefined>>()
 
   function setSessionActive(sessionID: string, status: DataSessionStatus) {
     activeUpdates?.set(sessionID, status)
@@ -260,6 +261,7 @@ export function createData(config: CreateDataInput) {
 
   function removePending(sessionID: string, inboxID?: string) {
     if (!inboxID) return
+    pendingUpdates.get(sessionID)?.set(inboxID, undefined)
     if (store.session.pending[sessionID]?.some((item) => item.id === inboxID))
       setStore(
         "session",
@@ -309,6 +311,7 @@ export function createData(config: CreateDataInput) {
   function updatePending(sessionID: string, inboxID: string, delivery: SessionInbox.Delivery) {
     const index = store.session.pending[sessionID]?.findIndex((item) => item.id === inboxID) ?? -1
     const item = store.session.pending[sessionID]?.[index]
+    pendingUpdates.get(sessionID)?.set(inboxID, item ? { ...item, delivery } : delivery)
     if (index < 0 || !item || item.delivery === delivery) return
     setStore("session", "pending", sessionID, index, { ...item, delivery })
   }
@@ -769,12 +772,14 @@ export function createData(config: CreateDataInput) {
       }
       case "session.inbox.enqueued": {
         outbox.delete(event.data.inboxID)
-        admitLocal({
+        const item = {
           id: event.data.inboxID,
           sessionID: event.data.sessionID,
           time: { created: event.created },
           ...event.data.item,
-        })
+        }
+        pendingUpdates.get(item.sessionID)?.set(item.id, item)
+        admitLocal(item)
         if (event.data.item.type === "compaction") {
           const active = compacting.get(event.data.sessionID)
           active?.observed.add(event.data.inboxID)
@@ -1379,23 +1384,40 @@ export function createData(config: CreateDataInput) {
         },
         sync(sessionID: string) {
           return sync.run(`session.pending:${sessionID}`, async () => {
-            const pending = await api().session.inbox.list({ sessionID })
-            // A positive read acknowledges admission even when its SSE echo is delayed.
-            pending.forEach((item) => outbox.delete(item.id))
-            // Compactions also coalesce by Session, not just by the proposed ID.
-            if (pending.some((item) => item.type === "compaction"))
-              store.session.pending[sessionID]
-                ?.filter((item) => item.type === "compaction")
-                .forEach((item) => outbox.delete(item.id))
-            // Keep optimistic rows still awaiting their echo: this fetch may
-            // have raced ahead of an in-flight admission the server does not
-            // know about yet.
-            const inflight = (store.session.pending[sessionID] ?? []).filter((item) => outbox.has(item.id))
-            const merged = inflight.length === 0 ? pending : [...pending, ...inflight]
-            batch(() => {
-              setStore("session", "pending", sessionID, reconcile(merged))
-              merged.forEach(materializeInboxMessage)
-            })
+            const updates = new Map<string, SessionInboxInfo | SessionInbox.Delivery | undefined>()
+            pendingUpdates.set(sessionID, updates)
+            try {
+              const snapshot = await api().session.inbox.list({ sessionID })
+              // Events can overtake this HTTP response on a remote connection.
+              // Reconcile them before an older snapshot can resurrect delivered input.
+              const current = new Map(snapshot.map((item) => [item.id, item]))
+              updates.forEach((item, id) => {
+                if (item === undefined) current.delete(id)
+                else if (typeof item === "string") {
+                  const existing = current.get(id)
+                  if (existing) current.set(id, { ...existing, delivery: item })
+                } else current.set(id, item)
+              })
+              const pending = [...current.values()]
+              // A positive read acknowledges admission even when its SSE echo is delayed.
+              pending.forEach((item) => outbox.delete(item.id))
+              // Compactions also coalesce by Session, not just by the proposed ID.
+              if (pending.some((item) => item.type === "compaction"))
+                store.session.pending[sessionID]
+                  ?.filter((item) => item.type === "compaction")
+                  .forEach((item) => outbox.delete(item.id))
+              // Keep optimistic rows still awaiting their echo: this fetch may
+              // have raced ahead of an in-flight admission the server does not
+              // know about yet.
+              const inflight = (store.session.pending[sessionID] ?? []).filter((item) => outbox.has(item.id))
+              const merged = inflight.length === 0 ? pending : [...pending, ...inflight]
+              batch(() => {
+                setStore("session", "pending", sessionID, reconcile(merged))
+                merged.forEach(materializeInboxMessage)
+              })
+            } finally {
+              if (pendingUpdates.get(sessionID) === updates) pendingUpdates.delete(sessionID)
+            }
           })
         },
         invalidate(sessionID: string) {

packages/client/test/solid-inbox-race.test.ts

@@ -0,0 +1,162 @@
+import { expect, test } from "bun:test"
+import { createRoot } from "solid-js"
+import { createData, type CreateDataInput } from "../src/solid"
+import { OpenCode, type OpenCodeEvent, type SessionInboxItem } from "../src/promise"
+
+const item: SessionInboxItem = {
+  id: "msg_remote",
+  sessionID: "ses_remote",
+  type: "user",
+  delivery: "steer",
+  time: { created: 1 },
+  payload: { text: "Check the remote preview" },
+}
+
+function fixture(fetch: (request: Request) => Promise<Response>) {
+  const listeners = new Set<Parameters<CreateDataInput["event"]["listen"]>[0]>()
+  const api = OpenCode.make({
+    baseUrl: "http://opencode.local",
+    fetch: (input, init) => fetch(input instanceof Request ? input : new Request(input, init)),
+  })
+  return createRoot((dispose) => ({
+    data: createData({
+      api: () => api,
+      directory: "/project",
+      event: {
+        on: () => () => {},
+        listen(handler) {
+          listeners.add(handler)
+          return () => listeners.delete(handler)
+        },
+      },
+    }),
+    emit: (event: OpenCodeEvent) => listeners.forEach((listener) => listener({ name: event.type, details: event })),
+    dispose,
+  }))
+}
+
+const enqueued: OpenCodeEvent = {
+  id: "evt_enqueued",
+  type: "session.inbox.enqueued",
+  created: 1,
+  durable: { aggregateID: item.sessionID, seq: 1, version: 1 },
+  data: {
+    sessionID: item.sessionID,
+    inboxID: item.id,
+    item: { type: item.type, delivery: item.delivery, payload: item.payload },
+  },
+}
+
+test.each(["delivered", "cancelled"] as const)(
+  "a delayed inbox snapshot cannot resurrect a %s prompt",
+  async (action) => {
+    const response = Promise.withResolvers<Response>()
+    const setup = fixture(() => response.promise)
+    try {
+      setup.emit(enqueued)
+      const reading = setup.data.session.pending.sync(item.sessionID)
+      setup.emit({
+        id: "evt_consumed",
+        type: `session.inbox.${action}`,
+        created: 2,
+        durable: { aggregateID: item.sessionID, seq: 2, version: 1 },
+        data: { sessionID: item.sessionID, inboxID: item.id },
+      })
+      expect(setup.data.session.input.has(item.sessionID, item.id)).toBe(false)
+      response.resolve(Response.json({ data: [item] }))
+      await reading
+      expect(setup.data.session.input.has(item.sessionID, item.id)).toBe(false)
+      const message = setup.data.session.message.get(item.sessionID, item.id)
+      if (action === "cancelled") expect(message).toBeUndefined()
+      else expect(message?.time.created).toBe(2)
+    } finally {
+      response.resolve(Response.json({ data: [] }))
+      setup.dispose()
+    }
+  },
+)
+
+test.each(["enqueue", "delivery"] as const)("inbox hydration retains a concurrent %s event", async (action) => {
+  const response = Promise.withResolvers<Response>()
+  const setup = fixture(() => response.promise)
+  try {
+    const reading = setup.data.session.pending.sync(item.sessionID)
+    setup.emit(
+      action === "enqueue"
+        ? enqueued
+        : {
+            id: "evt_delivery",
+            type: "session.inbox.delivery.changed",
+            created: 2,
+            durable: { aggregateID: item.sessionID, seq: 2, version: 1 },
+            data: { sessionID: item.sessionID, inboxID: item.id, delivery: "steer" },
+          },
+    )
+    response.resolve(Response.json({ data: action === "enqueue" ? [] : [{ ...item, delivery: "queue" }] }))
+    await reading
+    expect(setup.data.session.pending.list(item.sessionID)).toEqual([item])
+  } finally {
+    response.resolve(Response.json({ data: [] }))
+    setup.dispose()
+  }
+})
+
+test("hydration preserves submission order across optimistic admissions", async () => {
+  const response = Promise.withResolvers<Response>()
+  const gate = Promise.withResolvers<void>()
+  const setup = fixture((request) =>
+    request.method === "GET" ? response.promise : Promise.resolve(Response.json({ data: { id: "msg_admitted" } })),
+  )
+  try {
+    const first = setup.data.session.prompt({
+      sessionID: item.sessionID,
+      id: "msg_first",
+      text: "First",
+      gate: gate.promise,
+    })
+    const reading = setup.data.session.pending.sync(item.sessionID)
+    const second = setup.data.session.prompt({
+      sessionID: item.sessionID,
+      id: "msg_second",
+      text: "Second",
+      gate: gate.promise,
+    })
+    response.resolve(Response.json({ data: [] }))
+    await reading
+    expect(setup.data.session.input.list(item.sessionID)).toEqual(["msg_first", "msg_second"])
+    gate.resolve()
+    await Promise.all([first, second])
+  } finally {
+    response.resolve(Response.json({ data: [] }))
+    gate.resolve()
+    setup.dispose()
+  }
+})
+
+test("hydration coalesces a concurrent optimistic compaction even if its POST fails", async () => {
+  const response = Promise.withResolvers<Response>()
+  const admission = Promise.withResolvers<Response>()
+  const setup = fixture((request) => (request.method === "GET" ? response.promise : admission.promise))
+  const canonical: SessionInboxItem = {
+    id: "msg_canonical",
+    sessionID: item.sessionID,
+    type: "compaction",
+    delivery: "steer",
+    time: { created: 1 },
+    payload: {},
+  }
+  try {
+    const reading = setup.data.session.pending.sync(item.sessionID)
+    const compacting = setup.data.session.compact({ sessionID: item.sessionID }).catch(() => undefined)
+    response.resolve(Response.json({ data: [canonical] }))
+    await reading
+    expect(setup.data.session.pending.list(item.sessionID)).toEqual([canonical])
+    admission.reject(new Error("Connection closed"))
+    await compacting
+    expect(setup.data.session.pending.list(item.sessionID)).toEqual([canonical])
+  } finally {
+    response.resolve(Response.json({ data: [] }))
+    admission.resolve(new Response(null, { status: 204 }))
+    setup.dispose()
+  }
+})