feat: support client private-message reporting

Handle confirmed private messages sent from the client and preserve account isolation when sessions change. Surface unavailable messages without reporting incomplete content. Refs #528.
2026-10-21 16:59:26 +08:00
parent 617e7581dd
commit 491b531841
12 changed files with 963 additions and 4 deletions
@@ -957,6 +957,17 @@ export class HookSession extends EventEmitter {
            return;
        }

+        if (
+            packet.source === 'client-message' &&
+            (!isRealUin(packet.uin) || packet.uin !== this._uin)
+        ) {
+            this.log.warn(
+                'Client message receipt belongs to an inactive account: PID=%d',
+                this.pid,
+            );
+            return;
+        }
+
        const uin = packet.uin || this._uin;
        // Routing requires a real UIN, but even a frame carrying UIN=0 proves the
        // receive hook is alive. Attribute that evidence to the current epoch
@@ -986,6 +997,7 @@ export class HookSession extends EventEmitter {
            seqId: packet.seq,
            retCode: packet.error,
            fromClient: false,
+            ...(packet.source ? { source: packet.source } : {}),
            body: Buffer.from(packet.body),
        });
    }
@@ -7,6 +7,7 @@ import {
 } from '@snowluma/common/logger';
 import { renderParamsVerbose } from '@snowluma/common/log-summary';
 import { isRealUin } from '@snowluma/common/uin';
+import { CLIENT_MESSAGE_RECEIPT_CMD } from '@snowluma/common/protocol-types';
 import { readdirSync, promises as fs } from 'fs';
 import net from 'net';
 import path from 'path';
@@ -38,6 +39,8 @@ export enum PipeOp {
    recvPacket = 6,
    loginState = 7,
    loginIdentityHint = 17,
+    clientMessageReceipt = 21,
+    clientMessageState = 22,
 }

 const PipeFlagWantReply = 1 << 0;
@@ -55,6 +58,7 @@ export interface QqHookPacket {
    cmd: string;
    uin: string;
    body: Buffer;
+    source?: 'client-message';
 }

 export interface QqHookHello {
@@ -1369,6 +1373,44 @@ export class QqHookClient extends EventEmitter {
            return;
        }

+        if (frame.op === PipeOp.clientMessageState) {
+            if (frame.status === 1) {
+                runtimeLog.debug('Client private-message reporting is available: PID=%d', this.pid);
+            } else {
+                runtimeLog.warn(
+                    'Client private-message reporting is unavailable: PID=%d status=%d',
+                    this.pid,
+                    frame.status,
+                );
+            }
+
+            return;
+        }
+
+        if (frame.op === PipeOp.clientMessageReceipt) {
+            if (
+                frame.cmd !== '' ||
+                frame.requestId !== 0 ||
+                frame.flags !== 0 ||
+                frame.value0 <= 0n ||
+                frame.value0 > BigInt(Number.MAX_SAFE_INTEGER) ||
+                !isRealUin(frame.msg)
+            ) {
+                runtimeLog.warn('Invalid client message receipt: PID=%d', this.pid);
+                return;
+            }
+
+            this.emit('packet', {
+                seq: Number(frame.value0),
+                error: frame.status,
+                cmd: CLIENT_MESSAGE_RECEIPT_CMD,
+                uin: frame.msg,
+                body: Buffer.from(frame.body),
+                source: 'client-message',
+            } satisfies QqHookPacket);
+            return;
+        }
+
        if (frame.op !== PipeOp.recvPacket) {
            return;
        }
@@ -929,6 +929,42 @@ describe('HookSession — pipe-down must not clobber a settled error', () => {
 });

 describe('HookSession — packet forwarding', () => {
+    it('forwards client receipts only for the active account without changing direction', async () => {
+        const onPacket = vi.fn();
+        const ctx = makeSession({ pipeLive: true, onPacket });
+        await ctx.session.load();
+        ctx.session.onPipeUp();
+        await flush();
+        ctx.currentClient().fireLogin('10001');
+        const receipt: QqHookPacket = {
+            seq: 42,
+            error: 0,
+            cmd: 'ClientMessageReceipt',
+            uin: '10001',
+            source: 'client-message',
+            body: Buffer.from([1, 2, 3]),
+        };
+
+        for (const uin of ['', '0', '20002']) {
+            ctx.currentClient().firePacket({ ...receipt, uin });
+        }
+
+        expect(onPacket).not.toHaveBeenCalled();
+        ctx.currentClient().firePacket(receipt);
+        expect(onPacket).toHaveBeenCalledExactlyOnceWith(
+            expect.objectContaining({
+                pid: 1234,
+                uin: '10001',
+                source: 'client-message',
+                fromClient: false,
+            }),
+        );
+        ctx.currentClient().fireLogin('20002');
+        ctx.currentClient().firePacket(receipt);
+        expect(onPacket).toHaveBeenCalledOnce();
+        await ctx.session.dispose();
+    });
+
    // These tests cover the field-rename + filter logic that used to live
    // inside the deleted NtqqHandler.onHookPacket. Inlining it here means
    // we can exercise the transformation with a plain vi.fn() sink, no
@@ -12,6 +12,7 @@ import {
    type LogEntry,
 } from '@snowluma/common/logger';
 import { afterEach, describe, expect, it, vi } from 'vitest';
+import { CLIENT_MESSAGE_RECEIPT_CMD } from '@snowluma/common/protocol-types';
 import {
    HEADER_SIZE,
    listLiveLinuxPipePids,
@@ -147,6 +148,113 @@ describe('QqHookClient login wait lifecycle', () => {
    });
 });

+describe('QqHookClient client message receipts', () => {
+    it('preserves receipt provenance and failure status separately from normal packets', async () => {
+        const sockets = mockControlPipe(12345, () => {});
+        const client = new QqHookClient(12345);
+        const onPacket = vi.fn();
+
+        try {
+            await client.startRecv(onPacket);
+            const socket = sockets.mock.results[0]!.value;
+            const body = Buffer.from([1, 2, 3]);
+            socket.emit(
+                'data',
+                pipeFrame({ op: PipeOp.clientMessageReceipt, value0: 42n, message: '10001', body }),
+            );
+            socket.emit(
+                'data',
+                pipeFrame({
+                    op: PipeOp.clientMessageReceipt,
+                    value0: 43n,
+                    message: '10001',
+                    status: -1,
+                }),
+            );
+            socket.emit(
+                'data',
+                pipeFrame({
+                    op: PipeOp.recvPacket,
+                    value0: 44n,
+                    message: '10001',
+                    cmd: CLIENT_MESSAGE_RECEIPT_CMD,
+                    body,
+                }),
+            );
+
+            expect(onPacket.mock.calls.map(([packet]) => packet)).toEqual([
+                {
+                    seq: 42,
+                    error: 0,
+                    cmd: CLIENT_MESSAGE_RECEIPT_CMD,
+                    uin: '10001',
+                    body,
+                    source: 'client-message',
+                },
+                {
+                    seq: 43,
+                    error: -1,
+                    cmd: CLIENT_MESSAGE_RECEIPT_CMD,
+                    uin: '10001',
+                    body: Buffer.alloc(0),
+                    source: 'client-message',
+                },
+                { seq: 44, error: 0, cmd: CLIENT_MESSAGE_RECEIPT_CMD, uin: '10001', body },
+            ]);
+        } finally {
+            client.close();
+        }
+    });
+
+    it('reports unsupported capability and invalid envelopes without breaking normal receive', async () => {
+        const entries: LogEntry[] = [];
+        setLogLevel('warn');
+        const unsubscribe = subscribeLogs((entry) => entries.push(entry));
+        const sockets = mockControlPipe(12345, () => {});
+        const client = new QqHookClient(12345);
+        const onPacket = vi.fn();
+
+        try {
+            await client.startRecv(onPacket);
+            const socket = sockets.mock.results[0]!.value;
+            socket.emit('data', pipeFrame({ op: PipeOp.clientMessageState, status: -1 }));
+
+            for (const patch of [
+                { value0: 0n },
+                { value0: BigInt(Number.MAX_SAFE_INTEGER) + 1n },
+                { message: '' },
+                { cmd: 'unexpected' },
+                { requestId: 1 },
+            ]) {
+                socket.emit(
+                    'data',
+                    pipeFrame({
+                        op: PipeOp.clientMessageReceipt,
+                        value0: 42n,
+                        message: '10001',
+                        ...patch,
+                    }),
+                );
+            }
+
+            expect(onPacket).not.toHaveBeenCalled();
+            expect(
+                entries.filter(
+                    (entry) => entry.scope === 'QQHook.Runtime' && entry.level === 'warn',
+                ),
+            ).toHaveLength(6);
+            socket.emit(
+                'data',
+                pipeFrame({ op: PipeOp.recvPacket, cmd: 'normal', message: '10001', value0: 44n }),
+            );
+            expect(onPacket).toHaveBeenCalledOnce();
+        } finally {
+            client.close();
+            unsubscribe();
+        }
+    });
+});
+
 describe('listLiveLinuxPipePids', () => {
    it('ignores stale-looking control socket names when the probe rejects them', async () => {
        const runtimeDir = await makeRuntimeDir();
@@ -1,3 +1,5 @@
+export const CLIENT_MESSAGE_RECEIPT_CMD = 'ClientMessageReceipt';
+
 export interface PacketInfo {
    pid: number;
    uin: string;
@@ -6,6 +8,7 @@ export interface PacketInfo {
    seqId: number;
    retCode: number;
    fromClient: boolean;
+    source?: 'client-message';
    body: Uint8Array;
 }
 export type PacketSink = (packet: PacketInfo) => void;
@@ -4,7 +4,13 @@ import { packetTimeoutMs } from '@snowluma/bridge/packet-timeout';
 import type { PacketInfo } from '@snowluma/common/protocol-types';
 import { BridgeEventBus } from '@snowluma/protocol/event-bus';
 import { IdentityService } from '@snowluma/protocol/identity-service';
-import { MSG_PUSH_CMD, parseMsgPushOrThrow, SysMsgDedup } from '@snowluma/protocol/msg-push';
+import {
+    MSG_PUSH_CMD,
+    parseMsgPushOrThrow,
+    SysMsgDedup,
+    decodeClientMessageReceipt,
+    fetchClientMessage,
+} from '@snowluma/protocol/msg-push';
 import { KICK_NT_CMD, parseKickNT } from '@snowluma/protocol/kick-nt';
 import { IncomingPacketPipeline, type CmdParser } from '@snowluma/protocol/packet-pipeline';
 import type { OnlineDeviceInfo } from '@snowluma/protocol/events';
@@ -38,6 +44,10 @@ export class Bridge implements BridgeInterface {
    private onlineClients_: readonly Readonly<OnlineDeviceInfo>[] | null = null;
    private disposePromise_: Promise<void> | null = null;
    private readonly pendingPackets_ = new Set<Promise<void>>();
+    private readonly clientMessageTasks_ = new Map<
+        string,
+        { connection: ProcessConnection; work: Promise<void> }
+    >();
    private rosterWarmupStarted_ = false;
    private settleRosterWarmup!: (result: RosterWarmupResult) => void;
    private readonly rosterWarmup = new Promise<RosterWarmupResult>((resolve) => {
@@ -111,9 +121,13 @@ export class Bridge implements BridgeInterface {

    bindPid(pid: number, client: PacketSender): void {
        const previous = this.activeConnection;
+        const existing = this.connections.get(pid);
        this.connections.delete(pid);
        // A new login stays healthy until the watchdog confirms a failure.
-        this.connections.set(pid, { pid, client, receiveHealthy: true });
+        const connection =
+            existing?.client === client ? existing : { pid, client, receiveHealthy: true };
+        connection.receiveHealthy = true;
+        this.connections.set(pid, connection);

        if (previous?.pid !== pid || previous.client !== client) {
            runtimeLog.trace(
@@ -190,12 +204,81 @@ export class Bridge implements BridgeInterface {
            return Promise.reject(new Error('Bridge is disposed'));
        }

-        const work = this.identity.initialization.then(() => this.pipeline.process(pkt));
+        const connection = this.connections.get(pkt.pid);
+        const work = this.identity.initialization.then(() =>
+            pkt.source === 'client-message'
+                ? this.processClientMessage(pkt, connection)
+                : this.pipeline.process(pkt),
+        );
        this.pendingPackets_.add(work);
        void work.finally(() => this.pendingPackets_.delete(work)).catch(() => undefined);
        return work;
    }

+    private async processClientMessage(
+        pkt: PacketInfo,
+        connection: ProcessConnection | undefined,
+    ): Promise<void> {
+        const requireConnection = (): ProcessConnection => {
+            if (
+                this.disposePromise_ ||
+                !connection ||
+                this.connections.get(pkt.pid) !== connection
+            ) {
+                throw new Error('Client message receipt belongs to an inactive connection');
+            }
+
+            return connection;
+        };
+
+        const current = requireConnection();
+        const receipt = decodeClientMessageReceipt(pkt, this.identity.uin);
+        const key = JSON.stringify([
+            pkt.pid,
+            receipt.peerUid,
+            receipt.sequence,
+            receipt.clientSequence,
+            receipt.timestamp,
+        ]);
+        const pending = this.clientMessageTasks_.get(key);
+
+        if (pending?.connection === current) {
+            return pending.work;
+        }
+
+        if (this.clientMessageTasks_.size >= 256) {
+            throw new Error('Too many pending client message lookups');
+        }
+
+        runtimeLog.trace(
+            'client_message_lookup_start uin=%j pid=%d seqId=%d',
+            this.identity.uin,
+            pkt.pid,
+            pkt.seqId,
+        );
+        const work = (async () => {
+            const event = await fetchClientMessage(current.client, this.identity, receipt);
+            requireConnection();
+            await this.pipeline.processEvents(pkt, [event]);
+            runtimeLog.trace(
+                'client_message_lookup_complete uin=%j pid=%d seqId=%d',
+                this.identity.uin,
+                pkt.pid,
+                pkt.seqId,
+            );
+        })();
+        const task = { connection: current, work };
+        this.clientMessageTasks_.set(key, task);
+
+        try {
+            await work;
+        } finally {
+            if (this.clientMessageTasks_.get(key) === task) {
+                this.clientMessageTasks_.delete(key);
+            }
+        }
+    }
+
    rememberUploadedFile(meta: UploadedFileMeta): void {
        if (!meta.fileId) {
            return;
@@ -0,0 +1,207 @@
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import { CLIENT_MESSAGE_RECEIPT_CMD, type PacketInfo } from '@snowluma/common/protocol-types';
+import type { PacketSender, SendPacketResult } from '@snowluma/common/packet-sender';
+import type { SendMessageResponse } from '@snowluma/proto-defs/action';
+import type { SsoGetC2cMsgResponse } from '@snowluma/proto-defs/get-c2c-msg';
+import { protobuf_encode } from '@snowluma/proton';
+import { IdentityService } from '@snowluma/protocol/identity-service';
+import { Bridge } from '../src/bridge/bridge';
+
+const bridges: Bridge[] = [];
+const timestamp = 1_700_000_000;
+
+function deferred<T>() {
+    let resolve!: (value: T) => void;
+    const promise = new Promise<T>((accept) => {
+        resolve = accept;
+    });
+    return { promise, resolve };
+}
+
+afterEach(async () => {
+    await Promise.all(bridges.splice(0).map((bridge) => bridge.dispose()));
+    vi.restoreAllMocks();
+});
+
+function packet(): PacketInfo {
+    const uid = Buffer.from('u_test_peer');
+    const header = Buffer.alloc(8);
+    header[0] = 1;
+    header[1] = uid.length;
+    header.writeUInt32LE(9, 4);
+    return {
+        pid: 1234,
+        uin: '10001',
+        source: 'client-message',
+        serviceCmd: CLIENT_MESSAGE_RECEIPT_CMD,
+        seqId: 42,
+        retCode: 0,
+        fromClient: false,
+        body: Buffer.concat([
+            header,
+            uid,
+            protobuf_encode<SendMessageResponse>({ privateSequence: 87, timestamp1: timestamp }),
+        ]),
+    };
+}
+
+function response(): SendPacketResult {
+    return {
+        success: true,
+        gotResponse: true,
+        errorCode: 0,
+        errorMessage: '',
+        responseData: Buffer.from(
+            protobuf_encode<SsoGetC2cMsgResponse>({
+                friendUid: 'u_test_peer',
+                messages: [
+                    {
+                        responseHead: {
+                            fromUin: 10001,
+                            fromUid: 'u_test_self',
+                            toUin: 20002,
+                            toUid: 'u_test_peer',
+                        },
+                        contentHead: {
+                            msgType: 166,
+                            sequence: 9,
+                            ntMsgSeq: 87,
+                            timestamp,
+                            msgId: 123456,
+                        },
+                        body: { richText: { elems: [{ text: { str: 'hello' } }] } },
+                    },
+                ],
+            }),
+        ),
+    };
+}
+
+async function setup() {
+    const bridge = new Bridge(IdentityService.memory('10001'));
+    bridges.push(bridge);
+    vi.spyOn(bridge.apis.contacts, 'fetchFriendList').mockResolvedValue([
+        { uin: 10001, uid: 'u_test_self', nickname: 'self', remark: '' },
+    ]);
+    vi.spyOn(bridge.apis.contacts, 'fetchGroupList').mockResolvedValue([]);
+    const sendPacket = vi.fn<PacketSender['sendPacket']>(async () => response());
+    const sender = { sendPacket };
+    bridge.bindPid(1234, sender);
+    await bridge.whenRosterWarmupSettled();
+    const received = vi.fn();
+    bridge.events.on('friend_message', received);
+    return { bridge, sendPacket, sender, received };
+}
+
+describe('Bridge client message recovery', () => {
+    it('dispatches confirmed content through the normal event pipeline', async () => {
+        const ctx = await setup();
+        await ctx.bridge.onPacket(packet());
+        expect(ctx.sendPacket).toHaveBeenCalledOnce();
+        expect(ctx.received).toHaveBeenCalledOnce();
+        expect(ctx.received).toHaveBeenCalledWith(
+            expect.objectContaining({
+                kind: 'friend_message',
+                selfUin: 10001,
+                senderUin: 10001,
+                peerUin: 20002,
+                ntMsgSeq: 87,
+                clientSeq: 9,
+                elements: [{ type: 'text', text: 'hello' }],
+            }),
+        );
+    });
+
+    it('does not treat an ordinary packet with the same command as a receipt', async () => {
+        const ctx = await setup();
+        const ordinary = packet();
+        delete ordinary.source;
+        await ctx.bridge.onPacket(ordinary);
+        expect(ctx.sendPacket).not.toHaveBeenCalled();
+        expect(ctx.received).not.toHaveBeenCalled();
+    });
+
+    it('shares an in-flight lookup across repeated receipts and ordinary PID observations', async () => {
+        const ctx = await setup();
+        const result = deferred<SendPacketResult>();
+        const started = deferred<void>();
+        ctx.sendPacket.mockImplementation(() => {
+            started.resolve();
+            return result.promise;
+        });
+        const first = ctx.bridge.onPacket(packet());
+        await started.promise;
+        ctx.bridge.bindPid(1234, ctx.sender);
+        const duplicate = ctx.bridge.onPacket(packet());
+        await Promise.resolve();
+        result.resolve(response());
+        await Promise.all([first, duplicate]);
+        expect(ctx.sendPacket).toHaveBeenCalledOnce();
+        expect(ctx.received).toHaveBeenCalledOnce();
+    });
+
+    it.each(['detach', 'replace', 'reattach', 'dispose'] as const)(
+        'rejects a lookup completed after %s',
+        async (action) => {
+            const ctx = await setup();
+            const result = deferred<SendPacketResult>();
+            const started = deferred<void>();
+            ctx.sendPacket.mockImplementation(() => {
+                started.resolve();
+                return result.promise;
+            });
+            const work = ctx.bridge.onPacket(packet());
+            const rejected = expect(work).rejects.toThrow('inactive connection');
+            await started.promise;
+            let disposal: Promise<void> | undefined;
+
+            if (action === 'dispose') {
+                disposal = ctx.bridge.dispose();
+            } else if (action === 'replace') {
+                ctx.bridge.bindPid(1234, { sendPacket: vi.fn() });
+            } else {
+                ctx.bridge.detachPid(1234);
+
+                if (action === 'reattach') {
+                    ctx.bridge.bindPid(1234, ctx.sender);
+                }
+            }
+
+            result.resolve(response());
+            await rejected;
+            await disposal;
+            expect(ctx.received).not.toHaveBeenCalled();
+        },
+    );
+
+    it('queries the originating connection even when another process is active', async () => {
+        const ctx = await setup();
+        const other = { sendPacket: vi.fn<PacketSender['sendPacket']>() };
+        ctx.bridge.bindPid(4321, other);
+        await ctx.bridge.onPacket(packet());
+        expect(ctx.sendPacket).toHaveBeenCalledOnce();
+        expect(other.sendPacket).not.toHaveBeenCalled();
+    });
+
+    it('rejects unavailable, detached and wrong-account receipts before querying', async () => {
+        const ctx = await setup();
+        await expect(ctx.bridge.onPacket({ ...packet(), retCode: -1 })).rejects.toThrow(
+            'unavailable',
+        );
+        await expect(ctx.bridge.onPacket({ ...packet(), uin: '20002' })).rejects.toThrow('account');
+        ctx.bridge.detachPid(1234);
+        await expect(ctx.bridge.onPacket(packet())).rejects.toThrow('inactive connection');
+        expect(ctx.sendPacket).not.toHaveBeenCalled();
+        expect(ctx.received).not.toHaveBeenCalled();
+    });
+
+    it('surfaces lookup failure and allows a later receipt to retry', async () => {
+        const ctx = await setup();
+        ctx.sendPacket.mockRejectedValueOnce(new Error('lookup failed'));
+        await expect(ctx.bridge.onPacket(packet())).rejects.toThrow('lookup failed');
+        expect(ctx.received).not.toHaveBeenCalled();
+        await ctx.bridge.onPacket(packet());
+        expect(ctx.sendPacket).toHaveBeenCalledTimes(2);
+        expect(ctx.received).toHaveBeenCalledOnce();
+    });
+});
@@ -79,12 +79,20 @@ export interface SendMessageResponse {
    result?: pb<1, int_32>;
    errMsg?: pb<2, string>;
    timestamp1?: pb<3, uint_32>;
+    reviewResult?: pb<6, int_32>;
+    serverResult?: pb<7, SendMessageServerResult>;
    field10?: pb<10, uint_32>;
    groupSequence?: pb<11, uint_32>;
    timestamp2?: pb<12, uint_32>;
    privateSequence?: pb<14, uint_32>;
 }

+export interface SendMessageServerResult {
+    type?: pb<1, int_32>;
+    code?: pb<2, int_32>;
+    message?: pb<3, string>;
+}
+
 // MentionExtra (for building @mention text element)
 export interface MentionExtraSend {
    type?: pb<3, int_32>;
@@ -0,0 +1,157 @@
+import { CLIENT_MESSAGE_RECEIPT_CMD, type PacketInfo } from '@snowluma/common/protocol-types';
+import type { PacketSender } from '@snowluma/common/packet-sender';
+import { isRealUin } from '@snowluma/common/uin';
+import type { SendMessageResponse } from '@snowluma/proto-defs/action';
+import type { SsoGetC2cMsg, SsoGetC2cMsgResponse } from '@snowluma/proto-defs/get-c2c-msg';
+import { protobuf_decode, protobuf_encode } from '@snowluma/proton';
+import type { FriendMessage } from '../events';
+import type { IdentityService } from '../identity-service';
+import { classifyMessageSurvival } from './blank-filter';
+import { buildContextFromMessage } from './context';
+import { decodeFriendMessage } from './decoders/friend-message';
+import { PkgType } from './enums';
+import { SSO_GET_C2C_MSG_CMD } from './fetch-c2c-history';
+import { requirePacketResponse } from './packet-response';
+
+export interface ClientMessageReceipt {
+    readonly peerUid: string;
+    readonly clientSequence: number;
+    readonly sequence: number;
+    readonly timestamp: number;
+}
+
+export function decodeClientMessageReceipt(pkt: PacketInfo, selfUin: string): ClientMessageReceipt {
+    if (
+        pkt.source !== 'client-message' ||
+        pkt.serviceCmd !== CLIENT_MESSAGE_RECEIPT_CMD ||
+        !isRealUin(selfUin) ||
+        pkt.uin !== selfUin
+    ) {
+        throw new Error('Client message receipt has an invalid source or account');
+    }
+
+    if (pkt.retCode !== 0) {
+        throw new Error(`Client message receipt is unavailable (status=${pkt.retCode})`);
+    }
+
+    const body = Buffer.from(pkt.body);
+
+    if (body.length < 12 || body.length > 8 + 128 + 65536) {
+        throw new Error('Client message receipt has an invalid length');
+    }
+
+    const peerLength = body[1]!;
+    const ackOffset = 8 + peerLength;
+
+    if (
+        body[0] !== 1 ||
+        body.readUInt16LE(2) !== 0 ||
+        peerLength < 3 ||
+        peerLength > 128 ||
+        ackOffset >= body.length ||
+        body.length - ackOffset > 65536
+    ) {
+        throw new Error('Client message receipt has an invalid envelope');
+    }
+
+    const peerUid = body.subarray(8, ackOffset).toString('utf8');
+    const clientSequence = body.readUInt32LE(4);
+
+    if (!/^u_[A-Za-z0-9_-]+$/.test(peerUid) || clientSequence === 0) {
+        throw new Error('Client message receipt has an invalid message identity');
+    }
+
+    const ack = protobuf_decode<SendMessageResponse>(body.subarray(ackOffset));
+
+    if (
+        (ack.result ?? 0) !== 0 ||
+        (ack.reviewResult ?? 0) !== 0 ||
+        (ack.serverResult?.code ?? 0) !== 0
+    ) {
+        throw new Error('Client message was not accepted');
+    }
+
+    const sequence = ack.privateSequence;
+    const timestamp = ack.timestamp1;
+
+    if (!sequence || !timestamp || (ack.groupSequence ?? 0) !== 0) {
+        throw new Error('Client message receipt has no valid private message confirmation');
+    }
+
+    return { peerUid, clientSequence, sequence, timestamp };
+}
+
+export async function fetchClientMessage(
+    sender: PacketSender,
+    identity: IdentityService,
+    receipt: ClientMessageReceipt,
+): Promise<FriendMessage> {
+    const request = protobuf_encode<SsoGetC2cMsg>({
+        friendUid: receipt.peerUid,
+        startSequence: receipt.sequence,
+        endSequence: receipt.sequence,
+    });
+    const result = await sender.sendPacket(SSO_GET_C2C_MSG_CMD, Buffer.from(request));
+    const response = protobuf_decode<SsoGetC2cMsgResponse>(
+        requirePacketResponse(result, 'Client message lookup'),
+    );
+
+    if (response.friendUid && response.friendUid !== receipt.peerUid) {
+        throw new Error('Client message lookup returned another conversation');
+    }
+
+    if (response.messages?.length !== 1) {
+        throw new Error(
+            `Client message lookup expected one message, received ${response.messages?.length ?? 0}`,
+        );
+    }
+
+    const message = response.messages[0]!;
+    const head = message.contentHead;
+    const route = message.responseHead;
+    const selfUin = Number(identity.uin);
+
+    if (
+        !head ||
+        !route ||
+        route.fromUin !== selfUin ||
+        route.toUid !== receipt.peerUid ||
+        !route.toUin ||
+        route.grp != null ||
+        head.ntMsgSeq !== receipt.sequence ||
+        head.sequence !== receipt.clientSequence ||
+        head.timestamp !== receipt.timestamp
+    ) {
+        throw new Error('Client message lookup did not match the confirmed message');
+    }
+
+    if (
+        ![
+            PkgType.PrivateMessage,
+            PkgType.PrivateRecordMessage,
+            PkgType.PrivateFileMessage,
+            PkgType.ForwardFakePrivateMessage,
+        ].includes(head.msgType ?? 0)
+    ) {
+        throw new Error('Client message lookup returned an unsupported message type');
+    }
+
+    const context = buildContextFromMessage(message, selfUin, identity);
+
+    if (!context) {
+        throw new Error('Client message lookup has no message context');
+    }
+
+    const events = decodeFriendMessage(context);
+    const event = events[0];
+
+    if (
+        events.length !== 1 ||
+        event?.kind !== 'friend_message' ||
+        classifyMessageSurvival(context.head, event.elements, context.body) !== 'keep'
+    ) {
+        throw new Error('Client message lookup has no supported message content');
+    }
+
+    return event;
+}
@@ -26,6 +26,7 @@ import { deriveSysMsgDedupIdentity, SysMsgDedup } from './sysmsg-dedup';

 export { deriveSysMsgDedupIdentity, SysMsgDedup } from './sysmsg-dedup';
 export type { SysMsgDedupIdentity } from './sysmsg-dedup';
+export { decodeClientMessageReceipt, fetchClientMessage } from './client-message-receipt';

 export { SSO_GET_GROUP_MSG_CMD, fetchGroupMessageRange } from './fetch-group-history';
 export {
@@ -85,8 +85,18 @@ export class IncomingPacketPipeline {
    }

    process(pkt: PacketInfo): Promise<void> {
+        return this.processWithHandlers(pkt, this.cmdHandlers_.get(pkt.serviceCmd));
+    }
+
+    processEvents(pkt: PacketInfo, events: readonly QQEventVariant[]): Promise<void> {
+        return this.processWithHandlers(pkt, [() => [...events]]);
+    }
+
+    private processWithHandlers(
+        pkt: PacketInfo,
+        handlers: readonly CmdParser[] | undefined,
+    ): Promise<void> {
        const startedAt = Date.now();
-        const handlers = this.cmdHandlers_.get(pkt.serviceCmd);

        if (!handlers) {
            this.packetLog.trace(() => [
@@ -0,0 +1,292 @@
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import { CLIENT_MESSAGE_RECEIPT_CMD, type PacketInfo } from '@snowluma/common/protocol-types';
+import type { PacketSender, SendPacketResult } from '@snowluma/common/packet-sender';
+import type { SendMessageResponse } from '@snowluma/proto-defs/action';
+import type { SsoGetC2cMsg, SsoGetC2cMsgResponse } from '@snowluma/proto-defs/get-c2c-msg';
+import type { PushMsgBody } from '@snowluma/proto-defs/message';
+import { protobuf_decode, protobuf_encode } from '@snowluma/proton';
+import { IdentityService } from '../../src/identity-service';
+import {
+    decodeClientMessageReceipt,
+    fetchClientMessage,
+} from '../../src/msg-push/client-message-receipt';
+import { SSO_GET_C2C_MSG_CMD } from '../../src/msg-push/fetch-c2c-history';
+
+const self = '10001';
+const peer = 'u_test_peer';
+const time = 1_700_000_000;
+const identities: IdentityService[] = [];
+
+afterEach(async () => {
+    await Promise.all(identities.splice(0).map((identity) => identity.close()));
+});
+
+function packet(ack: SendMessageResponse = { privateSequence: 87, timestamp1: time }): PacketInfo {
+    const uid = Buffer.from(peer);
+    const header = Buffer.alloc(8);
+    header[0] = 1;
+    header[1] = uid.length;
+    header.writeUInt32LE(9, 4);
+    return {
+        pid: 1234,
+        uin: self,
+        seqId: 42,
+        retCode: 0,
+        fromClient: false,
+        source: 'client-message',
+        serviceCmd: CLIENT_MESSAGE_RECEIPT_CMD,
+        body: Buffer.concat([header, uid, protobuf_encode<SendMessageResponse>(ack)]),
+    };
+}
+
+function message(): PushMsgBody {
+    return {
+        responseHead: { fromUin: Number(self), fromUid: 'u_test_self', toUin: 20002, toUid: peer },
+        contentHead: { msgType: 166, sequence: 9, ntMsgSeq: 87, timestamp: time, msgId: 123456 },
+        body: { richText: { elems: [{ text: { str: 'hello' } }, { face: { index: 1 } }] } },
+    };
+}
+
+function setup(messages = [message()], friendUid = peer) {
+    const identity = IdentityService.memory(self);
+    identities.push(identity);
+    const sendPacket = vi.fn<PacketSender['sendPacket']>(async () => ({
+        success: true,
+        gotResponse: true,
+        errorCode: 0,
+        errorMessage: '',
+        responseData: Buffer.from(protobuf_encode<SsoGetC2cMsgResponse>({ friendUid, messages })),
+    }));
+    const receipt = decodeClientMessageReceipt(packet(), self);
+    return {
+        sendPacket,
+        identity,
+        receipt,
+        run: () => fetchClientMessage({ sendPacket }, identity, receipt),
+    };
+}
+
+describe('client message receipts', () => {
+    it('decodes a confirmation and retrieves exactly its conversation and sequence', async () => {
+        const ctx = setup();
+        expect(ctx.receipt).toEqual({
+            peerUid: peer,
+            clientSequence: 9,
+            sequence: 87,
+            timestamp: time,
+        });
+        const event = await ctx.run();
+        expect(ctx.sendPacket).toHaveBeenCalledOnce();
+        const [cmd, body] = ctx.sendPacket.mock.calls[0]!;
+        expect(cmd).toBe(SSO_GET_C2C_MSG_CMD);
+        expect(protobuf_decode<SsoGetC2cMsg>(body)).toMatchObject({
+            friendUid: peer,
+            startSequence: 87,
+            endSequence: 87,
+        });
+        expect(event).toMatchObject({
+            kind: 'friend_message',
+            selfUin: 10001,
+            senderUin: 10001,
+            peerUin: 20002,
+            clientSeq: 9,
+            ntMsgSeq: 87,
+            sequenceAuthoritative: true,
+            time,
+            elements: [
+                { type: 'text', text: 'hello' },
+                { type: 'face', faceId: 1 },
+            ],
+        });
+    });
+
+    it.each([
+        { source: undefined },
+        { serviceCmd: 'MessageSvc.PbSendMsg' },
+        { uin: '20002' },
+        { retCode: -1 },
+    ] satisfies Partial<PacketInfo>[])('rejects invalid provenance or status: %j', (patch) => {
+        expect(() => decodeClientMessageReceipt({ ...packet(), ...patch }, self)).toThrow();
+    });
+
+    it.each([
+        { result: 1 },
+        { reviewResult: 1 },
+        { serverResult: { code: -1 } },
+        { privateSequence: 0 },
+        { timestamp1: 0 },
+        { groupSequence: 5 },
+    ] satisfies SendMessageResponse[])(
+        'rejects an unsuccessful or incomplete confirmation: %j',
+        (patch) => {
+            expect(() =>
+                decodeClientMessageReceipt(
+                    packet({ privateSequence: 87, timestamp1: time, ...patch }),
+                    self,
+                ),
+            ).toThrow();
+        },
+    );
+
+    it.each([
+        [
+            'version',
+            (body: Buffer) => {
+                body[0] = 2;
+            },
+        ],
+        [
+            'reserved',
+            (body: Buffer) => {
+                body[2] = 1;
+            },
+        ],
+        [
+            'uid length',
+            (body: Buffer) => {
+                body[1] = 129;
+            },
+        ],
+        [
+            'uid',
+            (body: Buffer) => {
+                body[8] = 0xff;
+            },
+        ],
+        [
+            'client sequence',
+            (body: Buffer) => {
+                body.writeUInt32LE(0, 4);
+            },
+        ],
+    ] as const)('rejects a malformed %s', (_name, mutate) => {
+        const pkt = packet();
+        const body = Buffer.from(pkt.body);
+        mutate(body);
+        expect(() => decodeClientMessageReceipt({ ...pkt, body }, self)).toThrow();
+    });
+
+    it('rejects truncated and excessive bodies', () => {
+        for (const body of [
+            new Uint8Array(),
+            packet().body.subarray(0, 8 + peer.length),
+            new Uint8Array(8 + 128 + 65537),
+        ]) {
+            expect(() => decodeClientMessageReceipt({ ...packet(), body }, self)).toThrow();
+        }
+    });
+
+    it.each([
+        [
+            'sender',
+            (msg: PushMsgBody) => {
+                msg.responseHead!.fromUin = 20002;
+            },
+        ],
+        [
+            'recipient',
+            (msg: PushMsgBody) => {
+                msg.responseHead!.toUid = 'u_other';
+            },
+        ],
+        [
+            'missing recipient',
+            (msg: PushMsgBody) => {
+                delete msg.responseHead!.toUin;
+            },
+        ],
+        [
+            'client sequence',
+            (msg: PushMsgBody) => {
+                msg.contentHead!.sequence = 10;
+            },
+        ],
+        [
+            'server sequence',
+            (msg: PushMsgBody) => {
+                msg.contentHead!.ntMsgSeq = 88;
+            },
+        ],
+        [
+            'timestamp',
+            (msg: PushMsgBody) => {
+                msg.contentHead!.timestamp = time + 1;
+            },
+        ],
+        [
+            'group',
+            (msg: PushMsgBody) => {
+                msg.responseHead!.grp = { groupCode: 30003 };
+            },
+        ],
+        [
+            'type',
+            (msg: PushMsgBody) => {
+                msg.contentHead!.msgType = 82;
+            },
+        ],
+        [
+            'empty content',
+            (msg: PushMsgBody) => {
+                msg.body = {};
+            },
+        ],
+        [
+            'unknown content',
+            (msg: PushMsgBody) => {
+                msg.body = { richText: { elems: [{ commonElem: { serviceType: 999 } }] } };
+            },
+        ],
+        [
+            'control',
+            (msg: PushMsgBody) => {
+                msg.contentHead!.c2cCmd = 1;
+            },
+        ],
+    ] as const)('rejects an incorrect %s in the lookup', async (_name, mutate) => {
+        const msg = message();
+        mutate(msg);
+        await expect(setup([msg]).run()).rejects.toThrow();
+    });
+
+    it('rejects missing, ambiguous and cross-conversation results', async () => {
+        await expect(setup([]).run()).rejects.toThrow('expected one');
+        await expect(setup([message(), message()]).run()).rejects.toThrow('expected one');
+        await expect(setup([message()], 'u_other').run()).rejects.toThrow('conversation');
+    });
+
+    it('preserves distinct identities for equal text', async () => {
+        const first = setup();
+        const msg = message();
+        msg.contentHead = { ...msg.contentHead, sequence: 10, ntMsgSeq: 88 };
+        const second = setup([msg]);
+        const event = await fetchClientMessage(second, second.identity, {
+            ...second.receipt,
+            clientSequence: 10,
+            sequence: 88,
+        });
+        expect((await first.run()).ntMsgSeq).toBe(87);
+        expect(event.ntMsgSeq).toBe(88);
+        expect(event.elements).toEqual((await first.run()).elements);
+    });
+
+    it.each([
+        { success: false, gotResponse: false, responseData: null },
+        { success: true, gotResponse: false, responseData: null },
+        { success: true, gotResponse: true, responseData: Buffer.alloc(0) },
+    ] satisfies Partial<SendPacketResult>[])(
+        'surfaces unsuccessful lookup transport: %j',
+        async (result) => {
+            const ctx = setup();
+            ctx.sendPacket.mockResolvedValue({
+                success: true,
+                gotResponse: true,
+                errorCode: 0,
+                errorMessage: '',
+                responseData: null,
+                ...result,
+            });
+            await expect(ctx.run()).rejects.toThrow('Client message lookup');
+        },
+    );
+});