fix(core): preserve private history paging failures

Split larger private history reads into bounded pages and honor the end of history. Surface rejected or stalled pages instead of returning misleading empty or partial results.

Fixes #605

packages/core/src/bridge/apis/message.ts

@@ -254,6 +254,7 @@ type FriendMessage = Extract<QQEventVariant, { kind: 'friend_message' }>;
 // ── Group-history fetch guards (the server frequency-limits / kicks abusive
 //    pulls, so keep each request small, bound the loop, and space the sends). ──
 const HISTORY_MAX_COUNT = 200; // hard cap on messages returned per call
+const PRIVATE_HISTORY_PAGE_SIZE = 100;
 const HISTORY_CHUNK_SEQ = 30; // sequence span per server request (small packet)
 const HISTORY_MAX_REQUESTS = 12; // bound the walk-back loop
 const HISTORY_MIN_GAP_MS = 300; // minimum spacing between SsoGetGroupMsg sends
@@ -1543,7 +1544,7 @@ export class MessageApi {
                     selfUin,
                     friendUid,
                     cursor.time,
-                    want - collected.size,
+                    Math.min(want - collected.size, PRIVATE_HISTORY_PAGE_SIZE),
                     cursor.random,
                 ),
             );
@@ -1564,7 +1565,7 @@ export class MessageApi {
                 break;
             }
 
-            if (page.cursor.time === 0 && page.cursor.random === 0) {
+            if (page.isComplete || (page.cursor.time === 0 && page.cursor.random === 0)) {
                 reachedEnd = true;
                 break;
             }
@@ -1572,11 +1573,6 @@ export class MessageApi {
             const nextCursorKey = `${page.cursor.time}:${page.cursor.random}`;
 
             if (seenCursors.has(nextCursorKey)) {
-                if (page.messages.length === 0) {
-                    reachedEnd = true;
-                    break;
-                }
-
                 throw new Error(
                     `private roam history cursor did not advance from ${cursor.time}/${cursor.random}`,
                 );

packages/core/tests/group-history-walkback.test.ts

@@ -208,6 +208,56 @@ describe('MessageApi history pagination', () => {
         expect(result.map((message) => message.ntMsgSeq)).toEqual([63213, 63214, 63215]);
     });
 
+    it('splits a 200-message private history request into bounded pages', async () => {
+        mockC2cRoamFetch.mockImplementation(
+            async (_sender, _identity, _self, _friend, time, count) => ({
+                messages: Array.from({ length: count }, (_, index) => ({
+                    ...gm(time - index),
+                    kind: 'friend_message',
+                    ntMsgSeq: time - index,
+                })),
+                cursor: { time: time - count, random: 1 },
+                isComplete: false,
+            }),
+        );
+        const result = await runHistory(
+            new MessageApi(ctx).getC2cLatestHistory('u_friend', 200, 10001, 1000),
+        );
+        expect(result).toHaveLength(200);
+        expect(mockC2cRoamFetch.mock.calls.map((call) => call[5])).toEqual([100, 100]);
+        expect(result.map((message) => message.ntMsgSeq)).toEqual(
+            Array.from({ length: 200 }, (_, index) => 801 + index),
+        );
+    });
+
+    it('honors completion even when the last page still includes a cursor', async () => {
+        mockC2cRoamFetch.mockResolvedValue({
+            messages: [{ ...gm(1), kind: 'friend_message', ntMsgSeq: 1 }],
+            cursor: { time: 100, random: 2 },
+            isComplete: true,
+        });
+        const result = await runHistory(
+            new MessageApi(ctx).getC2cLatestHistory('u_friend', 20, 10001, 1000),
+        );
+        expect(result).toHaveLength(1);
+        expect(mockC2cRoamFetch).toHaveBeenCalledOnce();
+    });
+
+    it('does not return partial history after a later request fails', async () => {
+        mockC2cRoamFetch
+            .mockResolvedValueOnce({
+                messages: [{ ...gm(1), kind: 'friend_message', ntMsgSeq: 1 }],
+                cursor: { time: 100, random: 2 },
+                isComplete: false,
+            })
+            .mockRejectedValueOnce(new Error('SsoGetRoamMsg failed: result=123'));
+        const assertion = expect(
+            new MessageApi(ctx).getC2cLatestHistory('u_friend', 20, 10001, 1000),
+        ).rejects.toThrow('result=123');
+        await vi.runAllTimersAsync();
+        await assertion;
+    });
+
     it('rejects a stalled private roam cursor instead of returning a partial page', async () => {
         mockC2cRoamFetch.mockResolvedValue({
             cursor: { time: 1700000100, random: 0 },

packages/proto-defs/src/get-c2c-msg.ts

@@ -33,7 +33,10 @@ export interface SsoGetRoamMsg {
 }
 
 export interface SsoGetRoamMsgResponse {
+    result?: pb<1, uint_32>;
+    errorMessage?: pb<2, string>;
     friendUid?: pb<3, string>;
+    isComplete?: pb<4, bool>;
     timestamp?: pb<5, uint_32>;
     random?: pb<6, uint_32>;
     messages?: pb_repeated<7, PushMsgBody>;

packages/protocol/src/msg-push/fetch-c2c-history.ts

@@ -33,6 +33,7 @@ interface RawSender {
 export interface C2cRoamPage {
     messages: FriendMessage[];
     cursor: { time: number; random: number };
+    isComplete: boolean;
 }
 
 function decodeC2cMessages(
@@ -166,6 +167,17 @@ export async function fetchC2cRoamMessagePage(
         requirePacketResponse(result, 'SsoGetRoamMsg'),
     );
 
+    if (response.result === 100000301) {
+        log.debug('SsoGetRoamMsg reached the end of private history');
+        return { messages: [], cursor: { time: 0, random: 0 }, isComplete: true };
+    }
+
+    if (response.result) {
+        throw new Error(
+            `SsoGetRoamMsg failed: result=${response.result} ${response.errorMessage ?? ''}`.trim(),
+        );
+    }
+
     if (response.friendUid && response.friendUid !== friendUid) {
         throw new Error(`SsoGetRoamMsg response friend uid mismatch: ${response.friendUid}`);
     }
@@ -175,7 +187,7 @@ export async function fetchC2cRoamMessagePage(
     messages.sort((a, b) => (a.ntMsgSeq ?? 0) - (b.ntMsgSeq ?? 0) || a.time - b.time);
     const cursor = { time: response.timestamp ?? 0, random: response.random ?? 0 };
     log.debug(
-        'SsoGetRoamMsg decoded friend=%s before=%d/%d count=%d wire=%d usable=%d next=%d/%d',
+        'SsoGetRoamMsg decoded friend=%s before=%d/%d count=%d wire=%d usable=%d next=%d/%d complete=%s',
         friendUid,
         beforeTime,
         beforeRandom,
@@ -184,6 +196,7 @@ export async function fetchC2cRoamMessagePage(
         messages.length,
         cursor.time,
         cursor.random,
+        response.isComplete === true,
     );
-    return { messages, cursor };
+    return { messages, cursor, isComplete: response.isComplete === true };
 }

packages/protocol/tests/msg-push/fetch-c2c-history.test.ts

@@ -23,6 +23,7 @@ import {
     fetchC2cMessageRange,
     fetchC2cRoamMessagePage,
 } from '../../src/msg-push';
+import { s, v } from '../oidb-services/_pb-oracle';
 
 const identity = { findFriend: () => undefined } as unknown as IdentityService;
 const previousLogLevel = getLogLevel();
@@ -341,6 +342,38 @@ describe('fetchC2cMessageRange / SsoGetC2cMsg', () => {
 });
 
 describe('fetchC2cRoamMessagePage / SsoGetRoamMsg', () => {
+    it('rejects business errors before interpreting an empty history page', async () => {
+        const response = Buffer.from([...v(1, 123), ...s(2, 'request rejected')]);
+        await expect(
+            fetchC2cRoamMessagePage(
+                { sendRawPacket: async () => okResult(response) },
+                identity,
+                10001,
+                'u_friend',
+                1700000100,
+                200,
+            ),
+        ).rejects.toThrow('result=123 request rejected');
+    });
+
+    it.each([
+        { bytes: [...v(1, 100000301)], cursor: { time: 0, random: 0 } },
+        {
+            bytes: [...s(3, 'u_friend'), ...v(4, 1), ...v(5, 1700000000), ...v(6, 77)],
+            cursor: { time: 1700000000, random: 77 },
+        },
+    ])('preserves a server-confirmed end of history', async ({ bytes, cursor }) => {
+        const result = await fetchC2cRoamMessagePage(
+            { sendRawPacket: async () => okResult(Buffer.from(bytes)) },
+            identity,
+            10001,
+            'u_friend',
+            1700000100,
+            20,
+        );
+        expect(result).toEqual({ messages: [], cursor, isComplete: true });
+    });
+
     it('encodes the timestamp cursor and orders both sides by canonical NT sequence', async () => {
         let captured: { cmd: string; body: Uint8Array } | null = null;
         const response = protobuf_encode<SsoGetRoamMsgResponse>({