feat(tui): hydrate pending session work

Author: kitlangton

packages/tui/src/context/data.tsx

@@ -199,6 +199,31 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({
         const item = messages.findLast((item) => item.type === "compaction" && item.status === "running")
         return item?.type === "compaction" ? item : undefined
       },
+      fromPending(item: SessionPendingInfo): SessionMessageInfo {
+        if (item.type === "user")
+          return {
+            id: item.id,
+            type: "user",
+            ...item.data,
+            time: { created: item.timeCreated },
+          }
+        if (item.type === "synthetic")
+          return {
+            id: item.id,
+            type: "synthetic",
+            ...item.data,
+            time: { created: item.timeCreated },
+          }
+        return {
+          id: item.id,
+          type: "compaction",
+          status: "running",
+          reason: "manual",
+          summary: "",
+          recent: "",
+          time: { created: item.timeCreated },
+        }
+      },
       latestTool(assistant: SessionMessageAssistant | undefined, callID?: string) {
         return assistant?.content.findLast(
           (item): item is SessionMessageAssistantTool =>
@@ -960,16 +985,29 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({
           },
           sync(sessionID: string) {
             return sync.run(`session.message:${sessionID}`, async () => {
-              const pending = new Set(store.session.input[sessionID] ?? [])
-              const messages = (
-                await client.api.message.list({ sessionID, limit: 200, order: "desc" })
-              ).data.toReversed()
-              const index = new Map(messages.map((message, index) => [message.id, index]))
-              store.session.message[sessionID]
-                ?.filter((message) => pending.has(message.id))
-                .forEach((item) => message.append(messages, index, item))
-              messageIndex.set(sessionID, index)
-              setStore("session", "message", sessionID, reconcile(messages))
+              const inputsBeforeSync = new Set(store.session.input[sessionID] ?? [])
+              const [projected, pending] = await Promise.all([
+                client.api.message.list({ sessionID, limit: 200, order: "desc" }),
+                client.api.session.pending.list({ sessionID }),
+              ])
+              const messages = projected.data.toReversed()
+              const projectedIDs = new Set(messages.map((message) => message.id))
+              const pendingMessages = pending.filter((item) => !projectedIDs.has(item.id)).map(message.fromPending)
+              const next = [...messages, ...pendingMessages]
+              messageIndex.set(sessionID, new Map(next.map((message, index) => [message.id, index])))
+              const inputsAfterSync = new Set(store.session.input[sessionID] ?? [])
+              setStore(
+                "session",
+                "input",
+                sessionID,
+                pendingMessages.flatMap((message) =>
+                  (message.type === "user" || message.type === "synthetic") &&
+                  (!inputsBeforeSync.has(message.id) || inputsAfterSync.has(message.id))
+                    ? [message.id]
+                    : [],
+                ),
+              )
+              setStore("session", "message", sessionID, reconcile(next))
             })
           },
           invalidate(sessionID: string) {

packages/tui/test/cli/tui/data.test.tsx

@@ -2381,9 +2381,33 @@ test("preserves admitted prompts when hydration races with promotion", async ()
   const events = createEventStream()
   const sessionID = "session-1"
   const messageID = "msg_user_1"
+  const queuedID = "msg_user_2"
   const requested = Promise.withResolvers<void>()
   const response = Promise.withResolvers<Response>()
   const calls = createFetch((url) => {
+    if (url.pathname === `/api/session/${sessionID}/pending`)
+      return json({
+        data: [
+          {
+            admittedSeq: 0,
+            id: messageID,
+            sessionID,
+            timeCreated: 0,
+            type: "user",
+            data: { text: "hello" },
+            delivery: "steer",
+          },
+          {
+            admittedSeq: 1,
+            id: queuedID,
+            sessionID,
+            timeCreated: 1,
+            type: "user",
+            data: { text: "queued" },
+            delivery: "queue",
+          },
+        ],
+      })
     if (url.pathname !== `/api/session/${sessionID}/message`) return
     requested.resolve()
     return response.promise
@@ -2463,14 +2487,14 @@ test("preserves admitted prompts when hydration races with promotion", async ()
     response.resolve(json({ data: [], cursor: {} }))
     await refresh
 
-    const message = sync.session.message.list(sessionID)?.[0]
+    const message = sync.session.message.get(sessionID, messageID)
     expect(message?.type).toBe("user")
     if (message?.type !== "user") return
     expect(message).toMatchObject({ id: messageID, text: "hello" })
     expect(message.metadata).toBeUndefined()
-    expect(sync.session.pending.list(sessionID)).toEqual([])
-    expect(sync.session.input.list(sessionID)).toEqual([])
-    expect(sync.session.message.list(sessionID).map((message) => message.id)).toEqual([messageID])
+    expect(sync.session.input.list(sessionID)).toEqual([queuedID])
+    expect(sync.session.message.list(sessionID).map((message) => message.id)).toEqual([messageID, queuedID])
+    expect(sync.session.message.get(sessionID, queuedID)).toMatchObject({ id: queuedID, text: "queued" })
     expect(sync.session.message.list("missing")).toEqual([])
     expect(sync.session.message.get(sessionID, messageID)).toBe(message)
     expect(sync.session.message.get(sessionID, "missing")).toBeUndefined()

packages/tui/test/fixture/tui-client.ts

@@ -109,6 +109,7 @@ export function createFetch(override?: FetchHandler, events?: ReturnType<typeof
       })
     if (url.pathname === "/api/session") return json({ data: [], cursor: {} })
     if (url.pathname === "/api/session/active") return json({ data: {} })
+    if (/^\/api\/session\/[^/]+\/pending$/.test(url.pathname)) return json({ data: [] })
     if (url.pathname === "/api/permission/request")
       return json({ location: { directory, project: { id: "proj_test", directory: worktree } }, data: [] })
     if (url.pathname === "/api/form/request")