fix: Paginate conversation fetches to exhaustion

package.json

@@ -51,7 +51,7 @@
     ],
     "dependencies": {
         "@doist/cli-core": "0.26.0",
-        "@doist/comms-sdk": "0.7.1",
+        "@doist/comms-sdk": "0.11.0",
         "@pnpm/tabtab": "0.5.4",
         "chalk": "5.6.2",
         "commander": "14.0.3",

src/commands/conversation/conversation.test.ts

@@ -105,7 +105,11 @@ function createClient({
     return {
         conversations: {
             getConversations: vi.fn(async ({ archived }: { archived?: boolean }) =>
-                archived ? archivedConversations : activeConversations,
+                archived === undefined
+                    ? [...activeConversations, ...archivedConversations]
+                    : archived
+                      ? archivedConversations
+                      : activeConversations,
             ),
             getUnread: vi.fn().mockResolvedValue({ data: [], version: 1 }),
             getConversation: vi.fn(async (id: string) => conversationsById.get(id)),
@@ -263,7 +267,11 @@ describe('conversation with', () => {
         expect(refsMocks.resolveUserRefs).toHaveBeenCalledWith('Alice', 1)
         expect(refsMocks.resolveConversationId).not.toHaveBeenCalled()
         expect(consoleSpy).toHaveBeenCalledWith('Conversation with Me, Alice Example')
-        expect(client.conversations.getConversations).toHaveBeenCalledWith({ workspaceId: 1 })
+        expect(client.conversations.getConversations).toHaveBeenCalledWith({
+            workspaceId: 1,
+            limit: 500,
+            archived: false,
+        })
     })
 
     it('searches archived conversations when no active 1:1 is found', async () => {
@@ -285,9 +293,14 @@ describe('conversation with', () => {
 
         await program.parseAsync(['node', 'tdc', 'conversation', 'with', 'Alice'])
 
-        expect(client.conversations.getConversations).toHaveBeenCalledWith({ workspaceId: 1 })
         expect(client.conversations.getConversations).toHaveBeenCalledWith({
             workspaceId: 1,
+            limit: 500,
+            archived: false,
+        })
+        expect(client.conversations.getConversations).toHaveBeenCalledWith({
+            workspaceId: 1,
+            limit: 500,
             archived: true,
         })
     })
@@ -423,11 +436,14 @@ describe('conversation list', () => {
         expect(output).toContain('id:43')
         expect(output).toContain('id:42')
         // Active-only default never touches the archived fetch.
-        expect(client.conversations.getConversations).toHaveBeenCalledWith({ workspaceId: 1 })
-        expect(client.conversations.getConversations).not.toHaveBeenCalledWith({
+        expect(client.conversations.getConversations).toHaveBeenCalledWith({
             workspaceId: 1,
-            archived: true,
+            limit: 500,
+            archived: false,
         })
+        expect(client.conversations.getConversations).not.toHaveBeenCalledWith(
+            expect.objectContaining({ archived: true }),
+        )
     })
 
     it('filters to conversations that include a given participant', async () => {
@@ -611,9 +627,12 @@ describe('conversation list', () => {
 
         expect(client.conversations.getConversations).toHaveBeenCalledWith({
             workspaceId: 1,
+            limit: 500,
             archived: true,
         })
-        expect(client.conversations.getConversations).not.toHaveBeenCalledWith({ workspaceId: 1 })
+        expect(client.conversations.getConversations).not.toHaveBeenCalledWith(
+            expect.objectContaining({ archived: false }),
+        )
         expect(JSON.parse(consoleSpy.mock.calls[0][0]).map((c: { id: string }) => c.id)).toEqual([
             '43',
         ])
@@ -1252,3 +1271,149 @@ describe('conversation reply --file', () => {
         expect(consoleSpy).toHaveBeenCalledWith(expect.stringContaining(files.png))
     })
 })
+
+describe('conversation pagination', () => {
+    beforeEach(() => {
+        vi.clearAllMocks()
+        refsMocks.resolveUserRefs.mockResolvedValue([2])
+    })
+
+    const paginationUsers = {
+        1: { id: 1, fullName: 'Me' },
+        2: { id: 2, fullName: 'Alice Example' },
+        3: { id: 3, fullName: 'Bob Example' },
+    }
+
+    function descendingConversations(count: number, userIds: number[] = [1, 2, 3]) {
+        // Distinct, strictly descending lastActive so page boundaries are unambiguous.
+        return Array.from({ length: count }, (_, index) =>
+            createConversation(
+                1000 + index,
+                userIds,
+                new Date(Date.UTC(2026, 0, 20, 0, 0, 0) - index * 60_000).toISOString(),
+            ),
+        )
+    }
+
+    /** Client whose getConversations honors limit + the strict compound cursor. */
+    function paginatedClient(conversations: TestConversation[]) {
+        const client = createClient({ users: paginationUsers })
+        const sorted = [...conversations].sort(
+            (a, b) => b.lastActive.getTime() - a.lastActive.getTime(),
+        )
+        client.conversations.getConversations = vi.fn(
+            async ({
+                archived,
+                limit = 20,
+                olderThan,
+                beforeId,
+            }: {
+                workspaceId: number
+                archived?: boolean
+                limit?: number
+                olderThan?: Date
+                beforeId?: string
+            }) => {
+                let rows = sorted.filter(
+                    (conversation) => archived === undefined || conversation.archived === archived,
+                )
+                if (olderThan && beforeId) {
+                    rows = rows.filter(
+                        (conversation) =>
+                            conversation.lastActive.getTime() < olderThan.getTime() ||
+                            (conversation.lastActive.getTime() === olderThan.getTime() &&
+                                conversation.id < beforeId),
+                    )
+                }
+                return rows.slice(0, limit)
+            },
+        ) as never
+        return client
+    }
+
+    it('pages past the 500-row server limit to find quiet conversations', async () => {
+        const conversations = descendingConversations(501)
+        const needle = conversations[conversations.length - 1] as TestConversation
+        needle.title = 'Backend Leads'
+        const client = paginatedClient(conversations)
+        apiMocks.getCommsClient.mockResolvedValue(client)
+
+        const program = createProgram()
+        const consoleSpy = captureConsole('log')
+
+        await program.parseAsync(['node', 'tdc', 'conversation', 'list', '--name', 'backend'])
+
+        const output = consoleSpy.mock.calls.map((call) => call[0]).join('\n')
+        expect(output).toContain('Backend Leads')
+
+        expect(client.conversations.getConversations).toHaveBeenCalledTimes(2)
+        const boundary = conversations[499] as TestConversation
+        expect(client.conversations.getConversations).toHaveBeenLastCalledWith({
+            workspaceId: 1,
+            limit: 500,
+            archived: false,
+            olderThan: boundary.lastActive,
+            beforeId: boundary.id,
+        })
+    })
+
+    it('dedupes boundary rows repeated by a legacy inclusive server', async () => {
+        const conversations = descendingConversations(501)
+        const client = createClient({ users: paginationUsers })
+        const sorted = [...conversations].sort(
+            (a, b) => b.lastActive.getTime() - a.lastActive.getTime(),
+        )
+        // Legacy server: ignores beforeId, repeats the boundary row (<=).
+        client.conversations.getConversations = vi.fn(
+            async ({ limit = 20, olderThan }: { limit?: number; olderThan?: Date }) => {
+                const rows = olderThan
+                    ? sorted.filter(
+                          (conversation) =>
+                              conversation.lastActive.getTime() <= olderThan.getTime(),
+                      )
+                    : sorted
+                return rows.slice(0, limit)
+            },
+        ) as never
+        apiMocks.getCommsClient.mockResolvedValue(client)
+
+        const program = createProgram()
+        const consoleSpy = captureConsole('log')
+
+        await program.parseAsync(['node', 'tdc', 'conversation', 'list', '--json', '--full'])
+
+        const ids = JSON.parse(consoleSpy.mock.calls[0][0]).map(
+            (conversation: { id: string }) => conversation.id,
+        )
+        expect(ids).toHaveLength(501)
+        expect(new Set(ids).size).toBe(501)
+    })
+
+    it('fails loudly when the server never advances the page', async () => {
+        const stuckPage = descendingConversations(500)
+        const client = createClient({ users: paginationUsers })
+        client.conversations.getConversations = vi.fn(async () => stuckPage) as never
+        apiMocks.getCommsClient.mockResolvedValue(client)
+
+        const program = createProgram()
+
+        await expect(
+            program.parseAsync(['node', 'tdc', 'conversation', 'list']),
+        ).rejects.toHaveProperty('code', 'PAGINATION_STALLED')
+    })
+
+    it('finds a 1:1 conversation beyond the first page with `conversation with`', async () => {
+        const groups = descendingConversations(500)
+        const direct = createConversation(9999, [1, 2], '2020-01-01T00:00:00.000Z')
+        const client = paginatedClient([...groups, direct])
+        apiMocks.getCommsClient.mockResolvedValue(client)
+
+        const program = createProgram()
+        const consoleSpy = captureConsole('log')
+
+        await program.parseAsync(['node', 'tdc', 'conversation', 'with', 'Alice'])
+
+        expect(consoleSpy).toHaveBeenCalledWith('Conversation with Me, Alice Example')
+        expect(client.conversations.getConversations).toHaveBeenCalledTimes(2)
+    })
+})

src/commands/conversation/helpers.ts

@@ -69,42 +69,72 @@ export function sortByLastActiveDescending(a: Conversation, b: Conversation): nu
     return new Date(b.lastActive).getTime() - new Date(a.lastActive).getTime()
 }
 
+const CONVERSATION_PAGE_LIMIT = 500
+
 /**
- * Fetch a workspace's conversations for the requested {@link ConversationState},
- * sorted by last activity (newest first). `active` and `archived` are single
- * calls; `all` fetches both and dedupes by id (the SDK's `getConversations` is
- * not paginated, so each state returns the full set in one call).
+ * Stream every conversations/get page for one archived state
+ * (undefined = the server's unfiltered active+archived stream). Pages
+ * advance by the strict compound (lastActive, id) cursor taken from the
+ * last raw row, so quiet conversations beyond the first page are reached.
  */
-export async function getConversationsByState(
+async function* iterateConversationPages(
     workspaceId: number,
-    state: ConversationState = 'all',
-): Promise<Conversation[]> {
+    archived: boolean | undefined,
+): AsyncGenerator<Conversation[]> {
     const client = await getCommsClient()
+    const seenIds = new Set<string>()
+    let cursor: { olderThan: Date; beforeId: string } | undefined
 
-    if (state === 'active') {
-        const active = await client.conversations.getConversations({ workspaceId })
-        return [...active].sort(sortByLastActiveDescending)
-    }
-
-    if (state === 'archived') {
-        const archived = await client.conversations.getConversations({
+    for (;;) {
+        const page = await client.conversations.getConversations({
             workspaceId,
-            archived: true,
+            limit: CONVERSATION_PAGE_LIMIT,
+            ...(archived === undefined ? {} : { archived }),
+            ...cursor,
         })
-        return [...archived].sort(sortByLastActiveDescending)
-    }
 
-    const [active, archived] = await Promise.all([
-        client.conversations.getConversations({ workspaceId }),
-        client.conversations.getConversations({ workspaceId, archived: true }),
-    ])
+        const sizeBefore = seenIds.size
+        for (const conversation of page) seenIds.add(conversation.id)
+        yield page
+
+        if (page.length < CONVERSATION_PAGE_LIMIT) return
+        if (seenIds.size === sizeBefore) {
+            // A full page of only known rows means the cursor is not
+            // advancing; truncating silently is how conversations "disappear".
+            throw new CliError(
+                'PAGINATION_STALLED',
+                `conversations/get returned a full page with no new conversations (workspace ${workspaceId}); results would be incomplete`,
+            )
+        }
+        const boundary = page[page.length - 1] as Conversation
+        cursor = { olderThan: new Date(boundary.lastActive), beforeId: boundary.id }
+    }
+}
 
+async function fetchAllConversations(
+    workspaceId: number,
+    archived: boolean | undefined,
+): Promise<Conversation[]> {
     const byId = new Map<string, Conversation>()
-    for (const conversation of [...active, ...archived]) {
-        byId.set(conversation.id, conversation)
+    for await (const page of iterateConversationPages(workspaceId, archived)) {
+        for (const conversation of page) byId.set(conversation.id, conversation)
     }
+    return [...byId.values()]
+}
 
-    return [...byId.values()].sort(sortByLastActiveDescending)
+/**
+ * Fetch ALL of a workspace's conversations for the requested
+ * {@link ConversationState}, sorted by last activity (newest first).
+ * `active`/`archived` page one filtered stream each; `all` pages the
+ * server's unfiltered stream once.
+ */
+export async function getConversationsByState(
+    workspaceId: number,
+    state: ConversationState = 'all',
+): Promise<Conversation[]> {
+    const archived = state === 'all' ? undefined : state === 'archived'
+    const conversations = await fetchAllConversations(workspaceId, archived)
+    return conversations.sort(sortByLastActiveDescending)
 }
 
 function scanForDirectConversation(
@@ -135,23 +165,18 @@ export async function findDirectConversation(
     sessionUserId: number,
     targetUserId: number,
 ): Promise<ConversationLookupResult> {
-    const client = await getCommsClient()
-    // Active first — only scan archived on miss.
-    const active = await client.conversations.getConversations({ workspaceId })
-    const activeScan = scanForDirectConversation(active, sessionUserId, targetUserId)
-    if (activeScan.match) {
-        return {
-            directConversation: activeScan.match,
-            groupConversationCount: activeScan.extraGroupCount,
+    let groupConversationCount = 0
+    // Active first — only scan archived on miss; stop at the first match.
+    for (const archived of [false, true]) {
+        for await (const page of iterateConversationPages(workspaceId, archived)) {
+            const scan = scanForDirectConversation(page, sessionUserId, targetUserId)
+            groupConversationCount += scan.extraGroupCount
+            if (scan.match) {
+                return { directConversation: scan.match, groupConversationCount }
+            }
         }
     }
-
-    const archived = await client.conversations.getConversations({ workspaceId, archived: true })
-    const archivedScan = scanForDirectConversation(archived, sessionUserId, targetUserId)
-    return {
-        directConversation: archivedScan.match,
-        groupConversationCount: activeScan.extraGroupCount + archivedScan.extraGroupCount,
-    }
+    return { groupConversationCount }
 }
 
 export async function renderConversationList(

src/lib/errors.ts

@@ -53,6 +53,7 @@ export type ErrorCode =
     // State errors
     | 'ALREADY_INSTALLED'
     | 'BATCH_FAILED'
+    | 'PAGINATION_STALLED'
     | 'FILE_NOT_FOUND'
     | 'FILE_READ_ERROR'
     | 'NOT_CREATOR'