trade-message-center/apps/server/test/onetalk-websocket.test.ts

// 验证 OneTalk WebSocket 业务编排

import assert from "node:assert/strict";
import test from "node:test";

import type { WebSocket } from "@fastify/websocket";
import {
    ONETALK_CLIENT_FRAME_TYPES,
    ONETALK_ERROR_CODES,
    ONETALK_PROTOCOL_VERSION,
    createOneTalkBuyerFactFingerprint,
    createOneTalkRenderedCardContentFingerprint,
    createMockAuthorizationReader,
    decodeOneTalkFrame,
    type MockAuthorizationRecord,
    type CenterConversation,
    type OneTalkCenterMessage,
    type OneTalkMindScope,
    type OneTalkMessage,
    type OneTalkObservedMessage,
    type OneTalkAuthorizationReader,
    type OneTalkAuthorizationRequest,
    type OneTalkSendConfirmationFrame,
    type OneTalkSendRequestFrame,
} from "@trade-message-center/onetalk-contract";

import { createApp, type AppDependencies } from "../src/app.ts";
import {
    OneTalkDatabaseError,
    type OneTalkObservationResult,
    type OneTalkProfileIngestionResult,
    type OneTalkReadService,
    type OneTalkService,
} from "../src/onetalk/index.ts";
import {
    createOneTalkConnectionRegistry,
    type OneTalkRegisteredConnection,
} from "../src/websocket/registry.ts";
import type { DatabaseConnection } from "../src/database/index.ts";
import { createOneTalkCutoverPolicy } from "../src/cutover-policy.ts";

type OneTalkClientFrameType = (typeof ONETALK_CLIENT_FRAME_TYPES)[number];

const CLIENT_FRAME_COVERAGE_BASELINES = {
    "ws.hello": ["ws.accepted", "anchor.snapshot"],
    heartbeat: ["heartbeat.ack"],
    "conversation.discovered": ["conversation.ack"],
    "conversations.discovered": ["conversations.ack"],
    "sync.complete": ["sync.status", "conversation.updated"],
    "contact.profile.observed": ["contact.profile.ack"],
    "buyer.facts.observed": ["buyer.facts.ack"],
    "message.observed": ["message.ack", "message.created", "conversation.updated"],
    "messages.observed": ["messages.ack"],
    "rendered.card.observed": ["rendered.card.ack", "message.updated"],
    "storage.delete.ack": [],
    "history.sync.ack": [],
    "send.request": ["send.command"],
    "send.confirmation": ["message.created", "conversation.updated", "send.result"],
} as const satisfies Record<OneTalkClientFrameType, readonly string[]>;

const testConfig = {
    host: "127.0.0.1",
    port: 3000,
    databaseUrl: "postgres://test:test@localhost:5432/test",
    environment: "non_development" as const,
    mindAuthorization: {
        baseUrl: "https://mind.example.com",
        mindPageOrigin: "http://mind.localhost",
        pluginOrigins: ["http://plugin.localhost"],
        timeoutMs: 100,
    },
};

const pluginScope = {
    channelAccountId: "onetalk-account-1",
    deviceId: "device-1",
} as const;

const mindScope: OneTalkMindScope = {
    mindUserId: "mind-user-1",
    workspaceId: "workspace-1",
    channelAccountId: pluginScope.channelAccountId,
};

const authorizationRecord: MockAuthorizationRecord = {
    scope: pluginScope,
    mindScope,
    binding: "binding-1",
    permissions: ["read", "send"],
    authorizationVersion: "version-1",
    active: true,
};

const createDatabaseStub = (): DatabaseConnection => {
    return { db: {} as DatabaseConnection["db"], close: async () => {} };
};

const message = (messageId = "message-1"): OneTalkMessage => {
    return {
        messageId,
        conversationId: "conversation-1",
        senderId: "sender-1",
        direction: "received",
        sentAtMs: 1_700_000_000_000,
        content: { version: 1, kind: "text", text: "hello" },
        participantIds: ["sender-1", "login-user-1"],
        readStatus: 1,
        messageStatus: 2,
        unreadCount: 0,
    };
};

const orderMessage = (messageId = "message-1"): OneTalkMessage => ({
    ...message(messageId),
    content: {
        version: 1,
        kind: "order",
        orderId: "order-1",
        bizCode: null,
        contractId: null,
        id: null,
        tenant: null,
        orderAmount: 1,
        orderAmountCurrency: "USD",
        paymentAmount: 1,
        paymentAmountCurrency: "USD",
        statusMessageKey: "paid",
        actions: [],
    },
});

const renderedOrderSupplement = {
    version: 1 as const,
    kind: "rendered_order" as const,
    title: "Confirmed order",
    products: [],
    productCount: 0,
    status: { code: null, text: "Paid" },
    payment: { totalDisplay: "$1", discountDisplay: null },
    delivery: { shippingAddress: "Hangzhou", methodLabel: null, dateLabel: null },
    action: { label: null, status: null },
};

const centerMessage = (value: OneTalkMessage): OneTalkCenterMessage => ({
    messageId: value.messageId,
    conversationId: value.conversationId,
    senderId: value.senderId,
    participantIds: [...value.participantIds],
    direction: value.direction,
    sentAtMs: value.sentAtMs,
    content: { ...value.content },
});

const createService = (observeMessage: OneTalkService["observeMessage"]): OneTalkService => {
    return {
        discoverConversation: async (_context, conversationId, _guard, conversationKind) => ({
            channelAccountId: pluginScope.channelAccountId,
            conversationId,
            conversationKind,
            syncPhase: "initial",
            syncResult: "incomplete",
            latestMessageId: null,
            historyComplete: false,
            messageCount: 0,
            historyGeneration: "initial",
        }),
        discoverConversations: async (_context, entries, _guard) =>
            entries.map((entry) => ({
                channelAccountId: pluginScope.channelAccountId,
                conversationId: entry.conversationId,
                conversationKind: "direct" as const,
                syncPhase: "initial" as const,
                syncResult: "incomplete" as const,
                latestMessageId: null,
                historyComplete: false,
                messageCount: 0,
                historyGeneration: "initial",
            })),
        listAnchors: async () => [],
        listConversations: async () => [],
        readConversation: async () => null,
        readHistory: async () => null,
        observeMessage,
        observeMessages: async (context, observations, guard) =>
            Promise.all(
                observations.map((observation) =>
                    observeMessage(
                        context,
                        observation.observationSource,
                        observation.message,
                        guard,
                    ),
                ),
            ),
        completeSync: async () => ({
            status: "accepted",
            conversation: {
                channelAccountId: pluginScope.channelAccountId,
                conversationId: "conversation-1",
                conversationKind: null,
                syncPhase: "initial",
                syncResult: "incomplete",
                latestMessageId: null,
                historyComplete: false,
                messageCount: 0,
                historyGeneration: "initial",
            },
            anchorAdvanced: false,
        }),
    };
};

const createReadService = (): OneTalkReadService => ({
    listConversations: async () => ({
        status: "accepted",
        conversations: [],
        page: { hasMore: false, nextCursor: null },
    }),
    readConversation: async ({ conversationId }) => ({
        status: "accepted",
        conversation: {
            conversationId,
            conversationType: "direct",
            name: null,
            avatarUrl: null,
            lastContactTimeLong: null,
            messagePreview: null,
        },
    }),
    readHistory: async () => ({ status: "not_found" }),
});

const createTestApp = (dependencies: AppDependencies): ReturnType<typeof createApp> =>
    createApp(testConfig, {
        ...dependencies,
        readService: dependencies.readService ?? createReadService(),
    });

type TestFrame = Record<string, unknown>;
type FramePredicate = (frame: TestFrame) => boolean;
type FrameWaiter = {
    predicate: FramePredicate;
    resolve: (frame: TestFrame) => void;
    reject: (error: Error) => void;
};
type FrameReader = {
    frames: TestFrame[];
    waiters: FrameWaiter[];
    failure?: Error;
};

const frameReaders = new WeakMap<WebSocket, FrameReader>();

const drainFrameReader = (reader: FrameReader): void => {
    while (reader.frames.length > 0 && reader.waiters.length > 0) {
        const waiter = reader.waiters[0];
        const frame = reader.frames[0];
        let matches = false;
        try {
            matches = waiter.predicate(frame);
        } catch (error: unknown) {
            reader.waiters.shift();
            waiter.reject(error instanceof Error ? error : new Error(String(error)));
            continue;
        }
        if (!matches) {
            reader.waiters.shift();
            waiter.reject(new Error(`Unexpected WebSocket frame type: ${String(frame.type)}`));
            continue;
        }
        reader.waiters.shift();
        reader.frames.shift();
        waiter.resolve(frame);
    }
};

const frameReaderFor = (socket: WebSocket): FrameReader => {
    const existing = frameReaders.get(socket);
    if (existing) return existing;

    const reader: FrameReader = { frames: [], waiters: [] };
    const onMessage = (data: Buffer): void => {
        try {
            reader.frames.push(JSON.parse(data.toString()) as TestFrame);
            drainFrameReader(reader);
        } catch (error: unknown) {
            const failure = error instanceof Error ? error : new Error(String(error));
            reader.failure = failure;
            while (reader.waiters.length > 0) reader.waiters.shift()?.reject(failure);
        }
    };
    // Keep one listener for the socket lifetime. Frames arriving between awaits
    // stay in the queue and are checked by the next predicate in wire order.
    socket.on("message", onMessage);
    const onError = (error: unknown): void => {
        reader.failure = error instanceof Error ? error : new Error(String(error));
        while (reader.waiters.length > 0) reader.waiters.shift()?.reject(reader.failure);
    };
    socket.on("error", onError);
    socket.once("close", () => {
        if (reader.failure) return;
        reader.failure = new Error("WebSocket closed before the expected frame arrived");
        while (reader.waiters.length > 0) reader.waiters.shift()?.reject(reader.failure);
    });
    frameReaders.set(socket, reader);
    return reader;
};

const nextFrame = async (
    socket: WebSocket,
    predicate: FramePredicate = () => true,
): Promise<TestFrame> => {
    const reader = frameReaderFor(socket);
    return new Promise((resolve, reject) => {
        reader.waiters.push({ predicate, resolve, reject });
        if (reader.failure && reader.frames.length === 0) {
            reader.waiters.shift();
            reject(reader.failure);
            return;
        }
        drainFrameReader(reader);
    });
};

const nextMessages = async (socket: WebSocket, count: number): Promise<TestFrame[]> => {
    const messages: TestFrame[] = [];
    for (let index = 0; index < count; index += 1) {
        messages.push(await nextFrame(socket));
    }
    return messages;
};

const nextMessage = (socket: WebSocket): Promise<TestFrame> => nextFrame(socket);

const nextPluginStatus = async (
    socket: WebSocket,
    status: "online" | "offline",
): Promise<TestFrame> => {
    const frame = await nextFrame(socket, (candidate) => candidate.type === "plugin.status");
    assert.deepEqual(frame.payload, { status });
    return frame;
};

const nextCloseCode = async (socket: {
    once: (event: string, listener: (code: number) => void) => void;
}): Promise<number> => {
    return new Promise((resolve, reject) => {
        socket.once("close", resolve);
        socket.once("error", reject);
    });
};

const openSocket = async (app: ReturnType<typeof createApp>, path = "/ws/plugin") => {
    await app.ready();
    const url = path === "/ws/mind" ? `/ws/mind?workspaceId=${mindScope.workspaceId}` : path;
    const origin = url.startsWith("/ws/mind") ? "http://mind.localhost" : "http://plugin.localhost";
    const socket = await app.injectWS(url, { headers: { origin } });
    frameReaderFor(socket);
    return socket;
};

const connectPlugin = async (socket: Awaited<ReturnType<typeof openSocket>>): Promise<void> => {
    const handshake = nextMessages(socket, 2);
    socket.send(
        JSON.stringify({
            protocolVersion: ONETALK_PROTOCOL_VERSION,
            connectionType: "plugin",
            type: "ws.hello",
            requestId: "plugin-hello",
            scope: pluginScope,
            payload: { binding: "binding-1", requestedPermissions: ["read", "send"] },
        }),
    );
    const [accepted, snapshot] = await handshake;
    assert.equal(accepted.type, "ws.accepted");
    assert.deepEqual(snapshot.payload, { anchors: [] });
};

const connectMindPage = async (socket: Awaited<ReturnType<typeof openSocket>>): Promise<void> => {
    const accepted = nextFrame(socket, (frame) => frame.type === "ws.accepted");
    const status = nextPluginStatus(socket, "offline");
    socket.send(
        JSON.stringify({
            protocolVersion: ONETALK_PROTOCOL_VERSION,
            connectionType: "mind_page",
            type: "ws.hello",
            requestId: "mind-hello",
            scope: mindScope,
            payload: { requestedPermissions: ["read"] },
        }),
    );
    assert.equal((await accepted).type, "ws.accepted");
    await status;
};

const connectMindPageForSend = async (
    socket: Awaited<ReturnType<typeof openSocket>>,
): Promise<void> => {
    const accepted = nextFrame(socket, (frame) => frame.type === "ws.accepted");
    const status = nextPluginStatus(socket, "offline");
    socket.send(
        JSON.stringify({
            protocolVersion: ONETALK_PROTOCOL_VERSION,
            connectionType: "mind_page",
            type: "ws.hello",
            requestId: "mind-send-hello",
            scope: mindScope,
            payload: { requestedPermissions: ["read", "send"] },
        }),
    );
    assert.equal((await accepted).type, "ws.accepted");
    await status;
};

const observedFrame = (
    requestId: string,
    observedMessage: OneTalkObservedMessage = message(),
): Record<string, unknown> => {
    return {
        protocolVersion: ONETALK_PROTOCOL_VERSION,
        connectionType: "plugin",
        type: "message.observed",
        requestId,
        scope: pluginScope,
        payload: {
            historyGeneration: "initial",
            observationSource: "live",
            message: observedMessage,
        },
    };
};

const observedBatchFrame = (
    requestId: string,
    observationSource: "history" | "incremental",
    messages: OneTalkObservedMessage[],
): Record<string, unknown> => {
    return {
        protocolVersion: ONETALK_PROTOCOL_VERSION,
        connectionType: "plugin",
        type: "messages.observed",
        requestId,
        scope: pluginScope,
        payload: { historyGeneration: "initial", observationSource, messages },
    };
};

const closeApp = async (
    app: ReturnType<typeof createApp>,
    sockets: Array<{ terminate: () => void }>,
): Promise<void> => {
    for (const socket of sockets) socket.terminate();
    for (const client of app.websocketServer.clients) client.terminate();
    await app.close();
};

test("executes a wire assertion for every declared OneTalk client frame", async () => {
    const service = createService(async (_context, source, observedMessage) => ({
        status: "accepted" as const,
        message: observedMessage,
        ...(source === "send_confirmation" ? { renderedCardContent: renderedOrderSupplement } : {}),
    }));
    const renderedCardInputs: unknown[] = [];
    const app = createApp(testConfig, {
        database: createDatabaseStub(),
        authorization: createMockAuthorizationReader([
            { ...authorizationRecord, permissions: ["read", "send", "rebuild"] },
        ]),
        oneTalkService: service,
        profileService: {
            ingestProfiles: async ({ profiles }) => ({
                status: "accepted" as const,
                inputProfileCount: profiles.length,
                profileCount: profiles.length,
                writtenProfileCount: profiles.length,
                staleProfileCount: 0,
            }),
        },
        buyerFactService: {
            ingestFacts: async ({ facts }) => ({
                status: "accepted" as const,
                inputFactCount: facts.length,
                factCount: facts.length,
            }),
        },
        renderedCardService: {
            ingest: async (input) => {
                renderedCardInputs.push(input);
                return {
                    status: "accepted" as const,
                    renderedCardContent: input.observation.content,
                    message: orderMessage(input.observation.messageId),
                };
            },
        },
        readService: {
            listConversations: async () => ({
                status: "accepted" as const,
                conversations: [],
                page: { hasMore: false, nextCursor: null },
            }),
            readConversation: async (input) => ({
                status: "accepted" as const,
                conversation: {
                    conversationId: input.conversationId,
                    conversationType: "direct" as const,
                    name: null,
                    avatarUrl: null,
                    lastContactTimeLong: null,
                    messagePreview: null,
                },
            }),
            readHistory: async () => ({ status: "not_found" as const }),
        },
    });
    const mind = await openSocket(app, "/ws/mind");
    const plugin = await openSocket(app);
    const coverage = new Map<OneTalkClientFrameType, readonly string[]>();
    const recordCoverage = (
        clientType: OneTalkClientFrameType,
        frames: readonly Record<string, unknown>[],
    ): void => {
        const frameTypes: string[] = [];
        for (const frame of frames) {
            if (typeof frame.type !== "string") assert.fail("wire frame must have a string type");
            frameTypes.push(frame.type);
        }
        coverage.set(clientType, frameTypes);
    };

    try {
        const mindHandshake = nextMessages(mind, 2);
        mind.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "mind_page",
                type: "ws.hello",
                requestId: "matrix-mind-hello",
                scope: mindScope,
                payload: { requestedPermissions: ["read", "send", "rebuild"] },
            }),
        );
        const [mindAccepted, initialPluginStatus] = await mindHandshake;
        assert.equal(mindAccepted.type, "ws.accepted");
        assert.deepEqual(initialPluginStatus.payload, { status: "offline" });

        const pluginHandshake = nextMessages(plugin, 2);
        const pluginOnline = nextPluginStatus(mind, "online");
        plugin.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "plugin",
                type: "ws.hello",
                requestId: "matrix-plugin-hello",
                scope: pluginScope,
                payload: {
                    binding: "binding-1",
                    requestedPermissions: ["read", "send", "rebuild"],
                },
            }),
        );
        const handshakeFrames = await pluginHandshake;
        assert.deepEqual(
            handshakeFrames.map((frame) => frame.type),
            CLIENT_FRAME_COVERAGE_BASELINES["ws.hello"],
        );
        assert.equal((await pluginOnline).type, "plugin.status");
        recordCoverage("ws.hello", handshakeFrames);

        const heartbeat = nextMessage(plugin);
        plugin.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "plugin",
                type: "heartbeat",
                requestId: "matrix-heartbeat",
                scope: pluginScope,
                payload: { sentAtMs: 1_700_000_000_000 },
            }),
        );
        const heartbeatAck = await heartbeat;
        assert.deepEqual([heartbeatAck.type], CLIENT_FRAME_COVERAGE_BASELINES.heartbeat);
        recordCoverage("heartbeat", [heartbeatAck]);

        plugin.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "plugin",
                type: "storage.delete.ack",
                requestId: "matrix-storage-delete",
                scope: pluginScope,
                payload: {
                    rebuildId: "unknown-rebuild",
                    conversationId: "conversation-1",
                    status: "cleared",
                },
            }),
        );
        await new Promise((resolve) => setTimeout(resolve, 0));
        recordCoverage("storage.delete.ack", []);

        plugin.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "plugin",
                type: "history.sync.ack",
                requestId: "matrix-history-sync",
                scope: pluginScope,
                payload: {
                    rebuildId: "unknown-rebuild",
                    conversationId: "conversation-1",
                    status: "started",
                },
            }),
        );
        await new Promise((resolve) => setTimeout(resolve, 0));
        recordCoverage("history.sync.ack", []);

        const discovery = nextMessage(plugin);
        const discoveryStatus = nextFrame(mind, (frame) => frame.type === "sync.status");
        plugin.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "plugin",
                type: "conversation.discovered",
                requestId: "matrix-discovery",
                scope: pluginScope,
                payload: {
                    conversationId: "conversation-1",
                    conversationType: "direct",
                    lastContactTimeLong: null,
                    messagePreview: null,
                },
            }),
        );
        const discoveryAck = await discovery;
        await discoveryStatus;
        assert.deepEqual(
            [discoveryAck.type],
            CLIENT_FRAME_COVERAGE_BASELINES["conversation.discovered"],
        );
        recordCoverage("conversation.discovered", [discoveryAck]);

        const discoveries = nextMessage(plugin);
        plugin.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "plugin",
                type: "conversations.discovered",
                requestId: "matrix-discoveries",
                scope: pluginScope,
                payload: {
                    batchId: "matrix-discoveries",
                    entries: [
                        {
                            conversationId: "conversation-2",
                            lastContactTimeLong: null,
                            messagePreview: null,
                        },
                    ],
                    fragment: { sequence: 0, isFinal: false },
                },
            }),
        );
        plugin.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "plugin",
                type: "conversations.discovered",
                requestId: "matrix-discoveries",
                scope: pluginScope,
                payload: {
                    batchId: "matrix-discoveries",
                    entries: [
                        {
                            conversationId: "conversation-3",
                            lastContactTimeLong: null,
                            messagePreview: null,
                        },
                    ],
                    fragment: { sequence: 1, isFinal: true },
                },
            }),
        );
        const discoveriesAck = await discoveries;
        assert.deepEqual(
            [discoveriesAck.type],
            CLIENT_FRAME_COVERAGE_BASELINES["conversations.discovered"],
        );
        recordCoverage("conversations.discovered", [discoveriesAck]);

        const syncStatus = nextMessages(mind, 2);
        plugin.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "plugin",
                type: "sync.complete",
                requestId: "matrix-sync-complete",
                scope: pluginScope,
                payload: {
                    conversationId: "conversation-1",
                    historyGeneration: "initial",
                    mode: "full",
                    historyComplete: true,
                    result: "succeeded",
                    latestMessageId: null,
                },
            }),
        );
        const syncStatusFrames = await syncStatus;
        assert.deepEqual(
            syncStatusFrames.map((frame) => frame.type),
            CLIENT_FRAME_COVERAGE_BASELINES["sync.complete"],
        );
        recordCoverage("sync.complete", syncStatusFrames);

        const profileAck = nextMessage(plugin);
        plugin.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "plugin",
                type: "contact.profile.observed",
                requestId: "matrix-profile",
                scope: pluginScope,
                payload: {
                    profiles: [
                        {
                            conversationId: "conversation-1",
                            aliId: "2208314000798",
                            accountId: "243340382",
                            loginId: "hzhago",
                            name: "Heena Liu",
                            companyName: "Hago",
                            countryCode: "CN",
                            currentTimeZone: -9,
                            serviceType: "cgs",
                            avatarUrl: "https://cdn.example.com/avatar/profile-1.jpg",
                            observedAtMs: 1_700_000_000_000,
                            profileFingerprint: "matrix-profile",
                            observationStatus: "confirmed",
                        },
                    ],
                },
            }),
        );
        const profileAckFrame = await profileAck;
        assert.deepEqual(
            [profileAckFrame.type],
            CLIENT_FRAME_COVERAGE_BASELINES["contact.profile.observed"],
        );
        recordCoverage("contact.profile.observed", [profileAckFrame]);

        const buyerAck = nextMessage(plugin);
        plugin.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "plugin",
                type: "buyer.facts.observed",
                requestId: "matrix-buyer",
                scope: pluginScope,
                payload: {
                    facts: [
                        {
                            conversationId: "conversation-1",
                            buyerTags: ["tag"],
                            buyerFeatures: ["feature"],
                            tags: {
                                state: "confirmed",
                                errorCode: null,
                                attemptedAtMs: 1,
                                confirmedAtMs: 1,
                            },
                            features: {
                                state: "confirmed",
                                errorCode: null,
                                attemptedAtMs: 1,
                                confirmedAtMs: 1,
                            },
                            observedAtMs: 1,
                            factFingerprint: createOneTalkBuyerFactFingerprint(
                                ["tag"],
                                ["feature"],
                            ),
                        },
                    ],
                },
            }),
        );
        const buyerAckFrame = await buyerAck;
        assert.deepEqual(
            [buyerAckFrame.type],
            CLIENT_FRAME_COVERAGE_BASELINES["buyer.facts.observed"],
        );
        recordCoverage("buyer.facts.observed", [buyerAckFrame]);

        const observedAck = nextMessage(plugin);
        const observedEvents = nextMessages(mind, 2);
        plugin.send(JSON.stringify(observedFrame("matrix-observed")));
        const observedFrames = [await observedAck, ...(await observedEvents)];
        assert.deepEqual(
            observedFrames.map((frame) => frame.type),
            CLIENT_FRAME_COVERAGE_BASELINES["message.observed"],
        );
        recordCoverage("message.observed", observedFrames);

        const observedBatchAck = nextMessage(plugin);
        plugin.send(
            JSON.stringify(
                observedBatchFrame("matrix-observed-batch", "history", [message("batch-1")]),
            ),
        );
        const observedBatchFrames = [await observedBatchAck];
        assert.deepEqual(
            observedBatchFrames.map((frame) => frame.type),
            CLIENT_FRAME_COVERAGE_BASELINES["messages.observed"],
        );
        recordCoverage("messages.observed", observedBatchFrames);

        const command = nextMessage(plugin);
        mind.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "mind_page",
                type: "send.request",
                requestId: "matrix-send-request",
                sendRequestId: "matrix-send",
                scope: mindScope,
                payload: {
                    conversationId: "conversation-1",
                    content: { kind: "text", text: "hello" },
                },
            }),
        );
        const commandFrame = await command;
        assert.deepEqual([commandFrame.type], CLIENT_FRAME_COVERAGE_BASELINES["send.request"]);
        recordCoverage("send.request", [commandFrame]);

        const renderedCardContent = {
            version: 1 as const,
            kind: "rendered_order" as const,
            title: "Matrix order",
            products: [],
            productCount: 0,
            status: { code: null, text: "Paid" },
            payment: { totalDisplay: "$1", discountDisplay: null },
            delivery: { shippingAddress: "Matrix address", methodLabel: null, dateLabel: null },
            action: { label: null, status: null },
        };
        const renderedCardAck = nextMessage(plugin);
        const renderedCardUpdate = nextFrame(mind, (frame) => frame.type === "message.updated");
        plugin.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "plugin",
                type: "rendered.card.observed",
                requestId: "matrix-rendered-card",
                scope: pluginScope,
                payload: {
                    conversationId: "conversation-1",
                    messageId: "matrix-rendered-card-message",
                    content: renderedCardContent,
                    contentFingerprint:
                        createOneTalkRenderedCardContentFingerprint(renderedCardContent),
                    observedAtMs: 1_700_000_000_100,
                },
            }),
        );
        const renderedCardFrames = [await renderedCardAck, await renderedCardUpdate];
        assert.deepEqual(
            renderedCardFrames.map((frame) => frame.type),
            CLIENT_FRAME_COVERAGE_BASELINES["rendered.card.observed"],
        );
        assert.deepEqual(renderedCardFrames[0]?.payload, {
            conversationId: "conversation-1",
            messageId: "matrix-rendered-card-message",
            contentFingerprint: createOneTalkRenderedCardContentFingerprint(renderedCardContent),
            observedAtMs: 1_700_000_000_100,
            status: "accepted",
        });
        assert.deepEqual(renderedCardFrames[1]?.payload, {
            message: {
                ...centerMessage(orderMessage("matrix-rendered-card-message")),
                content: {
                    version: 1,
                    kind: "order",
                    orderId: "order-1",
                    bizCode: null,
                    contractId: null,
                    id: null,
                    tenant: null,
                    orderAmount: 1,
                    orderAmountCurrency: "USD",
                    paymentAmount: 1,
                    paymentAmountCurrency: "USD",
                    statusMessageKey: "paid",
                    actions: [],
                    title: "Matrix order",
                    products: [],
                    productCount: 0,
                    status: { code: null, text: "Paid" },
                    payment: { totalDisplay: "$1", discountDisplay: null },
                    delivery: {
                        shippingAddress: "Matrix address",
                        methodLabel: null,
                        dateLabel: null,
                    },
                    action: { label: null, status: null },
                },
            },
        });
        assert.equal(renderedCardInputs.length, 1);
        recordCoverage("rendered.card.observed", renderedCardFrames);

        const sentMessage = { ...orderMessage("matrix-sent"), direction: "sent" as const };
        const confirmationFrames = nextMessages(mind, 3);
        plugin.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "plugin",
                type: "send.confirmation",
                requestId: "matrix-send-confirmation",
                sendRequestId: "matrix-send",
                scope: pluginScope,
                payload: {
                    status: "confirmed_sent",
                    message: sentMessage,
                },
            }),
        );
        const sentConfirmationFrames = await confirmationFrames;
        assert.deepEqual(
            sentConfirmationFrames.map((frame) => frame.type),
            CLIENT_FRAME_COVERAGE_BASELINES["send.confirmation"],
        );
        const sendResult = sentConfirmationFrames.find((frame) => frame.type === "send.result");
        const messageCreated = sentConfirmationFrames.find(
            (frame) => frame.type === "message.created",
        );
        if (!sendResult) assert.fail("send confirmation must produce a send.result");
        if (!messageCreated) assert.fail("send confirmation must produce a message.created frame");
        assert.deepEqual(sendResult.payload, {
            status: "confirmed_sent",
            message: centerMessage(sentMessage),
        });
        assert.deepEqual(messageCreated.payload, {
            message: {
                ...centerMessage(sentMessage),
                content: {
                    ...sentMessage.content,
                    title: renderedOrderSupplement.title,
                    products: renderedOrderSupplement.products,
                    productCount: renderedOrderSupplement.productCount,
                    status: renderedOrderSupplement.status,
                    payment: renderedOrderSupplement.payment,
                    delivery: renderedOrderSupplement.delivery,
                    action: renderedOrderSupplement.action,
                },
            },
        });
        const decodedSendResult = decodeOneTalkFrame(sendResult);
        assert.equal(decodedSendResult.ok, true);
        recordCoverage("send.confirmation", sentConfirmationFrames);

        assert.deepEqual([...coverage.keys()].sort(), [...ONETALK_CLIENT_FRAME_TYPES].sort());
        assert.deepEqual(Object.fromEntries(coverage), CLIENT_FRAME_COVERAGE_BASELINES);
    } finally {
        await closeApp(app, [mind, plugin]);
    }
});

test("publishes exact false profile updates only for newer committed profiles", async () => {
    const conversation = {
        conversationId: "conversation-1",
        conversationType: "direct" as const,
        name: "New profile name",
        avatarUrl: "https://cdn.example.com/avatar/new.jpg",
        lastContactTimeLong: 1_780_000_000_000,
        messagePreview: "latest preview",
    };
    const profileResults: OneTalkProfileIngestionResult[] = [
        {
            status: "accepted",
            inputProfileCount: 1,
            profileCount: 1,
            writtenProfileCount: 1,
            staleProfileCount: 0,
            writtenConversationIds: [conversation.conversationId],
        },
        {
            status: "accepted",
            inputProfileCount: 1,
            profileCount: 1,
            writtenProfileCount: 0,
            staleProfileCount: 1,
            writtenConversationIds: [],
        },
    ];
    const app = createApp(testConfig, {
        database: createDatabaseStub(),
        authorization: createMockAuthorizationReader([authorizationRecord]),
        oneTalkService: createService(async () => ({
            status: "duplicate",
            message: message(),
        })),
        profileService: {
            ingestProfiles: async () => {
                const result = profileResults.shift();
                if (!result) throw new Error("profile result queue exhausted");
                return result;
            },
        },
        readService: {
            listConversations: async () => ({
                status: "accepted" as const,
                conversations: [conversation],
                page: { hasMore: false, nextCursor: null },
            }),
            readConversation: async () => ({
                status: "accepted" as const,
                conversation,
            }),
            readHistory: async () => ({ status: "not_found" as const }),
        },
    });
    const mind = await openSocket(app, "/ws/mind");
    const plugin = await openSocket(app);

    try {
        await connectMindPage(mind);
        const online = nextPluginStatus(mind, "online");
        await connectPlugin(plugin);
        await online;

        const newerUpdate = nextFrame(mind, (frame) => frame.type === "conversation.updated");
        const newerAck = nextFrame(plugin, (frame) => frame.type === "contact.profile.ack");
        plugin.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "plugin",
                type: "contact.profile.observed",
                requestId: "profile-newer",
                scope: pluginScope,
                payload: {
                    profiles: [
                        {
                            conversationId: conversation.conversationId,
                            aliId: "ali-1",
                            accountId: "account-1",
                            loginId: "login-1",
                            name: conversation.name,
                            companyName: null,
                            countryCode: "CN",
                            currentTimeZone: 8,
                            serviceType: "standard",
                            avatarUrl: conversation.avatarUrl,
                            observedAtMs: 1_780_000_000_000,
                            profileFingerprint: "profile-newer",
                            observationStatus: "confirmed",
                        },
                    ],
                },
            }),
        );
        const updateFrame = await newerUpdate;
        const updatePayload = updateFrame.payload as {
            conversation: Record<string, unknown>;
            moveToTop: boolean;
        };
        assert.equal(updatePayload.moveToTop, false);
        assert.deepEqual(updatePayload.conversation, conversation);
        assert.deepEqual(Object.keys(updatePayload.conversation).sort(), [
            "avatarUrl",
            "conversationId",
            "conversationType",
            "lastContactTimeLong",
            "messagePreview",
            "name",
        ]);
        assert.deepEqual((await newerAck).payload, {
            status: "delivered",
            profileCount: 1,
        });

        const staleAck = nextFrame(plugin, (frame) => frame.type === "contact.profile.ack");
        plugin.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "plugin",
                type: "contact.profile.observed",
                requestId: "profile-stale",
                scope: pluginScope,
                payload: {
                    profiles: [
                        {
                            conversationId: conversation.conversationId,
                            aliId: "ali-1",
                            accountId: "account-1",
                            loginId: "login-1",
                            name: "Stale profile name",
                            companyName: null,
                            countryCode: "CN",
                            currentTimeZone: 8,
                            serviceType: "standard",
                            avatarUrl: "https://cdn.example.com/avatar/stale.jpg",
                            observedAtMs: 1_779_999_999_999,
                            profileFingerprint: "profile-stale",
                            observationStatus: "confirmed",
                        },
                    ],
                },
            }),
        );
        assert.deepEqual((await staleAck).payload, {
            status: "delivered",
            profileCount: 1,
        });
        await new Promise<void>((resolve) => setImmediate(resolve));
        assert.equal(frameReaderFor(mind).frames.length, 0);
    } finally {
        await closeApp(app, [mind, plugin]);
    }
});

test("emits complete frame payloads through the injected diagnostics sink", async () => {
    const diagnostics: Record<string, unknown>[] = [];
    const app = createApp(testConfig, {
        database: createDatabaseStub(),
        authorization: createMockAuthorizationReader([authorizationRecord]),
        oneTalkService: createService(async () => ({
            status: "duplicate",
            message: message(),
        })),
        onOneTalkDiagnostic: (event) => diagnostics.push(event),
    });
    const socket = await openSocket(app);
    try {
        await connectPlugin(socket);
        assert.deepEqual(
            diagnostics.slice(0, 5).map((event) => [event.event, event.direction, event.frameType]),
            [
                ["ws_frame", "inbound", "ws.hello"],
                ["ws_hello", undefined, "ws.hello"],
                ["ws_decision", undefined, "ws.hello"],
                ["ws_frame", "outbound", "ws.accepted"],
                ["ws_frame", "outbound", "anchor.snapshot"],
            ],
        );
        assert.ok(diagnostics.some((event) => event.event === "ws_hello"));
        assert.ok(
            diagnostics.some((event) => event.event === "ws_decision" && event.code === "accepted"),
        );
        assert.ok(
            diagnostics.some(
                (event) => event.event === "ws_frame" && event.frameType === "ws.hello",
            ),
        );
        assert.deepEqual(diagnostics[0].payload, {
            binding: "binding-1",
            requestedPermissions: ["read", "send"],
        });
        assert.equal(JSON.stringify(diagnostics).includes("binding-1"), true);
        assert.equal(JSON.stringify(diagnostics).includes("device-1"), false);
    } finally {
        await closeApp(app, [socket]);
    }
});

test("isolates diagnostics sink failures from handshake, business frames, and close", async () => {
    const diagnosticAttempts: string[] = [];
    const app = createApp(testConfig, {
        database: createDatabaseStub(),
        authorization: createMockAuthorizationReader([authorizationRecord]),
        oneTalkService: createService(async () => ({
            status: "duplicate",
            message: message(),
        })),
        onOneTalkDiagnostic: (event) => {
            diagnosticAttempts.push(event.event);
            throw new Error("diagnostics unavailable");
        },
    });
    const socket = await openSocket(app);
    try {
        await connectPlugin(socket);
        const ack = nextMessage(socket);
        socket.send(JSON.stringify(observedFrame("diagnostics-business-frame")));
        assert.deepEqual((await ack).payload, {
            status: "duplicate",
            historyGeneration: "initial",
            conversationId: "conversation-1",
            messageId: "message-1",
        });
        const closed = new Promise<void>((resolve) => socket.once("close", () => resolve()));
        socket.terminate();
        await closed;
        assert.deepEqual(diagnosticAttempts.slice(0, 3), ["ws_frame", "ws_hello", "ws_decision"]);
        assert.ok(diagnosticAttempts.includes("ws_close"));
    } finally {
        await closeApp(app, [socket]);
    }
});

test("rejects a Mind send before dispatch when no plugin is connected", async () => {
    let observed = false;
    const service = createService(async () => {
        observed = true;
        return { status: "accepted", message: message() };
    });
    const authorization = createMockAuthorizationReader([authorizationRecord]);
    const app = createApp(testConfig, {
        database: createDatabaseStub(),
        authorization,
        oneTalkService: service,
    });
    const mind = await openSocket(app, "/ws/mind");

    try {
        const accepted = nextFrame(mind, (frame) => frame.type === "ws.accepted");
        const initialStatus = nextPluginStatus(mind, "offline");
        mind.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "mind_page",
                type: "ws.hello",
                requestId: "mind-send-hello",
                scope: mindScope,
                payload: { requestedPermissions: ["read", "send"] },
            }),
        );
        assert.equal((await accepted).type, "ws.accepted");
        await initialStatus;

        const result = nextMessage(mind);
        mind.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "mind_page",
                type: "send.request",
                requestId: "mind-send",
                sendRequestId: "send-request-1",
                scope: mindScope,
                payload: {
                    conversationId: "conversation-1",
                    content: { kind: "text", text: "hello" },
                },
            }),
        );
        assert.deepEqual((await result).payload, {
            status: "rejected_before_send",
            reason: "waiting_for_page",
        });
        assert.equal(observed, false);
    } finally {
        await closeApp(app, [mind]);
    }
});

test("resolves a valid non-confirmation immediately after plugin confirmation", async () => {
    let observed = false;
    const diagnostics: Array<Record<string, unknown>> = [];
    const service = createService(async () => {
        observed = true;
        return { status: "accepted", message: message("should-not-be-ingested") };
    });
    const app = createApp(testConfig, {
        database: createDatabaseStub(),
        authorization: createMockAuthorizationReader([authorizationRecord]),
        oneTalkService: service,
        onOneTalkDiagnostic: (event) => diagnostics.push(event),
    });
    const mind = await openSocket(app, "/ws/mind");
    const plugin = await openSocket(app);

    try {
        await connectMindPageForSend(mind);
        const online = nextPluginStatus(mind, "online");
        await connectPlugin(plugin);
        assert.equal((await online).type, "plugin.status");

        const command = nextMessage(plugin);
        const result = nextMessage(mind);
        mind.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "mind_page",
                type: "send.request",
                requestId: "mind-send-request",
                sendRequestId: "send-request-1",
                scope: mindScope,
                payload: {
                    conversationId: "conversation-1",
                    content: { kind: "text", text: "hello" },
                },
            }),
        );
        assert.deepEqual((await command).payload, {
            conversationId: "conversation-1",
            content: { kind: "text", text: "hello" },
        });

        plugin.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "plugin",
                type: "send.confirmation",
                requestId: "plugin-send-confirmation",
                sendRequestId: "send-request-1",
                scope: pluginScope,
                payload: {
                    status: "delivery_unknown",
                    reason: "send_state_lost",
                },
            }),
        );

        assert.deepEqual((await result).payload, {
            status: "delivery_unknown",
            reason: "send_state_lost",
        });
        assert.equal(observed, false);
        const inboundConfirmation = diagnostics.findIndex(
            (event) =>
                event.event === "ws_frame" &&
                event.direction === "inbound" &&
                event.frameType === "send.confirmation",
        );
        const outboundResult = diagnostics.findIndex(
            (event) =>
                event.event === "ws_frame" &&
                event.direction === "outbound" &&
                event.frameType === "send.result",
        );
        assert.ok(inboundConfirmation >= 0);
        assert.ok(outboundResult > inboundConfirmation);
        assert.equal(
            diagnostics.some((event) => event.code === "send_timeout"),
            false,
        );
    } finally {
        await closeApp(app, [mind, plugin]);
    }
});

test("returns a structurally matching duplicate confirmation without publishing message.created", async () => {
    const events: string[] = [];
    const confirmationMessage: OneTalkMessage = {
        ...message("duplicate-message-1"),
        direction: "sent",
        content: { kind: "text", text: "hello", version: 1 },
    };
    const storedMessage: OneTalkMessage = {
        ...confirmationMessage,
        content: { version: 1, kind: "text", text: "hello" },
    };
    assert.notEqual(JSON.stringify(confirmationMessage), JSON.stringify(storedMessage));
    const service = createService(async () => {
        events.push("observe");
        return { status: "duplicate", message: storedMessage };
    });
    const app = createApp(testConfig, {
        database: createDatabaseStub(),
        authorization: createMockAuthorizationReader([authorizationRecord]),
        oneTalkService: service,
    });
    const mind = await openSocket(app, "/ws/mind");
    const plugin = await openSocket(app);

    try {
        await connectMindPageForSend(mind);
        const online = nextPluginStatus(mind, "online");
        await connectPlugin(plugin);
        assert.equal((await online).type, "plugin.status");

        const command = nextMessage(plugin);
        mind.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "mind_page",
                type: "send.request",
                requestId: "mind-send-duplicate",
                sendRequestId: "send-request-duplicate",
                scope: mindScope,
                payload: {
                    conversationId: "conversation-1",
                    content: { kind: "text", text: "hello" },
                },
            }),
        );
        assert.equal((await command).type, "send.command");

        const result = nextMessage(mind);
        plugin.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "plugin",
                type: "send.confirmation",
                requestId: "plugin-send-duplicate",
                sendRequestId: "send-request-duplicate",
                scope: pluginScope,
                payload: { status: "confirmed_sent", message: confirmationMessage },
            }),
        );

        const serverResult = await result;
        assert.deepEqual(serverResult.payload, {
            status: "confirmed_sent",
            message: centerMessage(storedMessage),
        });
        const decoded = decodeOneTalkFrame(serverResult);
        assert.equal(decoded.ok, true);
        if (decoded.ok) assert.equal(decoded.frame.type, "send.result");
        assert.deepEqual(events, ["observe"]);
    } finally {
        await closeApp(app, [mind, plugin]);
    }
});

test("does not confirm a duplicate when its message content differs", async () => {
    const confirmationMessage: OneTalkMessage = {
        ...message("duplicate-message-mismatch"),
        direction: "sent",
    };
    const storedMessage: OneTalkMessage = {
        ...confirmationMessage,
        content: { version: 1, kind: "text", text: "different" },
    };
    const service = createService(async () => ({
        status: "duplicate" as const,
        message: storedMessage,
    }));
    const app = createApp(testConfig, {
        database: createDatabaseStub(),
        authorization: createMockAuthorizationReader([authorizationRecord]),
        oneTalkService: service,
    });
    const mind = await openSocket(app, "/ws/mind");
    const plugin = await openSocket(app);

    try {
        await connectMindPageForSend(mind);
        const online = nextPluginStatus(mind, "online");
        await connectPlugin(plugin);
        assert.equal((await online).type, "plugin.status");

        const command = nextMessage(plugin);
        mind.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "mind_page",
                type: "send.request",
                requestId: "mind-send-duplicate-mismatch",
                sendRequestId: "send-request-duplicate-mismatch",
                scope: mindScope,
                payload: {
                    conversationId: "conversation-1",
                    content: { kind: "text", text: "hello" },
                },
            }),
        );
        assert.equal((await command).type, "send.command");

        const result = nextMessage(mind);
        plugin.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "plugin",
                type: "send.confirmation",
                requestId: "plugin-send-duplicate-mismatch",
                sendRequestId: "send-request-duplicate-mismatch",
                scope: pluginScope,
                payload: { status: "confirmed_sent", message: confirmationMessage },
            }),
        );

        assert.deepEqual((await result).payload, {
            status: "delivery_unknown",
            reason: "database_unavailable",
        });
    } finally {
        await closeApp(app, [mind, plugin]);
    }
});

test("continues after anomaly and rejection with independent ACKs", async () => {
    const results: OneTalkObservationResult[] = [
        { status: "anomaly", anomalyCode: "invalid_message_observation" },
        { status: "rejected", reason: "conversation_not_discovered" },
        { status: "accepted", message: message("message-3") },
    ];
    const service = createService(async () => {
        const result = results.shift();
        if (!result) throw new Error("test observation queue exhausted");
        return result;
    });
    const authorization = createMockAuthorizationReader([authorizationRecord]);
    const app = createApp(testConfig, {
        database: createDatabaseStub(),
        authorization,
        oneTalkService: service,
    });
    const plugin = await openSocket(app);

    try {
        await connectPlugin(plugin);

        const anomalyAck = nextMessage(plugin);
        plugin.send(JSON.stringify(observedFrame("observe-anomaly")));
        assert.deepEqual((await anomalyAck).payload, {
            status: "anomaly",
            historyGeneration: "initial",
            conversationId: "conversation-1",
            messageId: "message-1",
            anomalyCode: "invalid_message_observation",
        });

        const rejectedAck = nextMessage(plugin);
        plugin.send(JSON.stringify(observedFrame("observe-rejected")));
        assert.deepEqual((await rejectedAck).payload, {
            status: "rejected",
            historyGeneration: "initial",
            conversationId: "conversation-1",
            messageId: "message-1",
        });

        const acceptedAck = nextMessage(plugin);
        plugin.send(JSON.stringify(observedFrame("observe-accepted", message("message-3"))));
        assert.deepEqual((await acceptedAck).payload, {
            status: "accepted",
            historyGeneration: "initial",
            conversationId: "conversation-1",
            messageId: "message-3",
        });
    } finally {
        await closeApp(app, [plugin]);
    }
});

test("prevents Mind pages from modifying technical conversations", async () => {
    let discovered = false;
    const service = createService(async () => ({
        status: "accepted",
        message: message(),
    }));
    service.discoverConversation = async () => {
        discovered = true;
        return {
            channelAccountId: pluginScope.channelAccountId,
            conversationId: "conversation-1",
            conversationKind: null,
            syncPhase: "initial",
            syncResult: "incomplete",
            latestMessageId: null,
            historyComplete: false,
            messageCount: 0,
            historyGeneration: "initial",
        };
    };
    const authorization = createMockAuthorizationReader([authorizationRecord]);
    const app = createApp(testConfig, {
        database: createDatabaseStub(),
        authorization,
        oneTalkService: service,
    });
    const mind = await openSocket(app, "/ws/mind");

    try {
        await connectMindPage(mind);
        const errorMessage = nextMessage(mind);
        mind.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "mind_page",
                type: "conversation.discovered",
                requestId: "mind-discover",
                scope: mindScope,
                payload: {
                    conversationId: "conversation-1",
                    conversationType: "direct",
                    latestMessageAtMs: null,
                },
            }),
        );
        assert.deepEqual((await errorMessage).payload, {
            code: ONETALK_ERROR_CODES.invalidMessage,
        });
        assert.equal(discovered, false);
    } finally {
        await closeApp(app, [mind]);
    }
});

test("rejects duplicate discovery IDs across fragments before service invocation", async () => {
    let calls = 0;
    const service = createService(async () => ({ status: "anomaly", anomalyCode: "unused" }));
    service.discoverConversations = async () => {
        calls += 1;
        return [];
    };
    const app = createApp(testConfig, {
        database: createDatabaseStub(),
        authorization: createMockAuthorizationReader([authorizationRecord]),
        oneTalkService: service,
    });
    const plugin = await openSocket(app);

    try {
        await connectPlugin(plugin);
        plugin.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "plugin",
                type: "conversations.discovered",
                requestId: "batch-duplicate",
                scope: pluginScope,
                payload: {
                    batchId: "batch-duplicate",
                    entries: [{ conversationId: "conversation-1", latestMessageAtMs: null }],
                    fragment: { sequence: 0, isFinal: false },
                },
            }),
        );
        const error = nextMessage(plugin);
        const close = nextCloseCode(plugin);
        plugin.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "plugin",
                type: "conversations.discovered",
                requestId: "batch-duplicate",
                scope: pluginScope,
                payload: {
                    batchId: "batch-duplicate",
                    entries: [{ conversationId: "conversation-1", latestMessageAtMs: null }],
                    fragment: { sequence: 1, isFinal: true },
                },
            }),
        );
        assert.deepEqual((await error).payload, { code: ONETALK_ERROR_CODES.invalidMessage });
        assert.equal(await close, 1003);
        assert.equal(calls, 0);
    } finally {
        await closeApp(app, [plugin]);
    }
});

test("returns database_unavailable without ACK or publish on storage failure", async () => {
    let published = false;
    const service = createService(async () => {
        throw new OneTalkDatabaseError(new Error("database unavailable in test"));
    });
    const authorization = createMockAuthorizationReader([authorizationRecord]);
    const app = createApp(testConfig, {
        database: createDatabaseStub(),
        authorization,
        oneTalkService: service,
        onOneTalkPublishFailure: () => {
            published = true;
        },
    });
    const plugin = await openSocket(app);

    try {
        await connectPlugin(plugin);
        const errorMessage = nextMessage(plugin);
        const closeCode = nextCloseCode(plugin);
        plugin.send(JSON.stringify(observedFrame("observe-database-error")));
        assert.deepEqual((await errorMessage).payload, {
            code: ONETALK_ERROR_CODES.databaseUnavailable,
        });
        assert.equal(await closeCode, 1011);
        assert.equal(published, false);
    } finally {
        await closeApp(app, [plugin]);
    }
});

test("flushes observations before completing sync and publishes accepted live facts after ACK", async () => {
    let markObservationStarted!: () => void;
    const observationStarted = new Promise<void>((resolve) => {
        markObservationStarted = resolve;
    });
    let releaseObservation!: () => void;
    const observationRelease = new Promise<void>((resolve) => {
        releaseObservation = resolve;
    });
    let completeCalls = 0;
    const service = createService(async (_context, _source, observedMessage) => ({
        status: "accepted" as const,
        message: observedMessage,
    }));
    service.observeMessages = async (_context, observations) => {
        markObservationStarted();
        await observationRelease;
        return observations.map((observation) => ({
            status: "accepted" as const,
            message: observation.message,
        }));
    };
    service.completeSync = async () => {
        completeCalls += 1;
        return {
            status: "accepted" as const,
            conversation: {
                channelAccountId: pluginScope.channelAccountId,
                conversationId: "conversation-1",
                conversationKind: "direct" as const,
                syncPhase: "initial" as const,
                syncResult: "incomplete" as const,
                latestMessageId: null,
                historyComplete: false,
                messageCount: 0,
                historyGeneration: "initial",
            },
            anchorAdvanced: false,
        };
    };
    const app = createTestApp({
        database: createDatabaseStub(),
        authorization: createMockAuthorizationReader([authorizationRecord]),
        oneTalkService: service,
    });
    const mind = await openSocket(app, "/ws/mind");
    const plugin = await openSocket(app);

    try {
        await connectMindPage(mind);
        const online = nextPluginStatus(mind, "online");
        await connectPlugin(plugin);
        await online;

        const acknowledgement = nextMessage(plugin);
        const mindEvents = nextMessages(mind, 4);
        plugin.send(JSON.stringify(observedFrame("deferred-observation")));
        plugin.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "plugin",
                type: "sync.complete",
                requestId: "deferred-completion",
                scope: pluginScope,
                payload: {
                    conversationId: "conversation-1",
                    historyGeneration: "initial",
                    mode: "full",
                    historyComplete: true,
                    result: "succeeded",
                    latestMessageId: null,
                },
            }),
        );

        await observationStarted;
        assert.equal(completeCalls, 0);
        releaseObservation();

        assert.equal((await acknowledgement).type, "message.ack");
        assert.deepEqual(
            (await mindEvents).map((frame) => frame.type),
            ["message.created", "conversation.updated", "sync.status", "conversation.updated"],
        );
        assert.equal(completeCalls, 1);
    } finally {
        await closeApp(app, [mind, plugin]);
    }
});

test("publishes an existing rendered-card supplement when its base live message arrives", async () => {
    const renderedCardContent = {
        version: 1 as const,
        kind: "rendered_order" as const,
        title: "Late order",
        products: [],
        productCount: 0,
        status: { code: null, text: "Paid" },
        payment: { totalDisplay: "$1", discountDisplay: null },
        delivery: { shippingAddress: "Hangzhou", methodLabel: null, dateLabel: null },
        action: { label: null, status: null },
    };
    const service = createService(async (_context, _source, observedMessage) => ({
        status: "accepted" as const,
        message: observedMessage,
        renderedCardContent,
    }));
    const app = createTestApp({
        database: createDatabaseStub(),
        authorization: createMockAuthorizationReader([authorizationRecord]),
        oneTalkService: service,
    });
    const mind = await openSocket(app, "/ws/mind");
    const plugin = await openSocket(app);

    try {
        await connectMindPage(mind);
        const online = nextPluginStatus(mind, "online");
        await connectPlugin(plugin);
        await online;

        const acknowledgement = nextMessage(plugin);
        const created = nextFrame(mind, (value) => value.type === "message.created");
        plugin.send(
            JSON.stringify(observedFrame("late-base-message", orderMessage("late-message"))),
        );

        assert.equal((await acknowledgement).type, "message.ack");
        assert.deepEqual((await created).payload, {
            message: {
                ...centerMessage(orderMessage("late-message")),
                content: {
                    version: 1,
                    kind: "order",
                    orderId: "order-1",
                    bizCode: null,
                    contractId: null,
                    id: null,
                    tenant: null,
                    orderAmount: 1,
                    orderAmountCurrency: "USD",
                    paymentAmount: 1,
                    paymentAmountCurrency: "USD",
                    statusMessageKey: "paid",
                    actions: [],
                    title: "Late order",
                    products: [],
                    productCount: 0,
                    status: { code: null, text: "Paid" },
                    payment: { totalDisplay: "$1", discountDisplay: null },
                    delivery: { shippingAddress: "Hangzhou", methodLabel: null, dateLabel: null },
                    action: { label: null, status: null },
                },
            },
        });
    } finally {
        await closeApp(app, [mind, plugin]);
    }
});

test("does not publish duplicate live observations", async () => {
    const service = createService(async () => ({
        status: "duplicate" as const,
        message: message(),
    }));
    const app = createTestApp({
        database: createDatabaseStub(),
        authorization: createMockAuthorizationReader([authorizationRecord]),
        oneTalkService: service,
    });
    const mind = await openSocket(app, "/ws/mind");
    const plugin = await openSocket(app);

    try {
        await connectMindPage(mind);
        const online = nextPluginStatus(mind, "online");
        await connectPlugin(plugin);
        await online;

        const acknowledgement = nextMessage(plugin);
        plugin.send(JSON.stringify(observedFrame("duplicate-observation")));
        assert.deepEqual((await acknowledgement).payload, {
            status: "duplicate",
            historyGeneration: "initial",
            conversationId: "conversation-1",
            messageId: "message-1",
        });
        await Promise.resolve();
        assert.deepEqual(frameReaderFor(mind).frames, []);
    } finally {
        await closeApp(app, [mind, plugin]);
    }
});

const fakeSocket = (send: (payload: string) => void): WebSocket => {
    return {
        readyState: 1,
        send,
        close: () => {},
    } as unknown as WebSocket;
};

test("does not publish live facts to a Mind connection that did not request read", async () => {
    const authorization = createMockAuthorizationReader([authorizationRecord]);
    const sent: string[] = [];
    const registry = createOneTalkConnectionRegistry({
        authorization,
        onPublishFailure: () => {},
    });
    const socket = fakeSocket((payload) => sent.push(payload));
    registry.register({
        socket,
        connectionType: "mind_page",
        scope: mindScope,
        mindScope,
        binding: "binding-1",
        authorizationVersion: "version-1",
        permissions: [],
    });

    await registry.publishMessageCreated({
        message: centerMessage(message()),
        requestId: "observe-unauthorized-page",
        scope: mindScope,
    });

    assert.deepEqual(
        sent.map((payload) => {
            const frame = JSON.parse(payload) as { type: string; payload: { code: string } };
            return { type: frame.type, code: frame.payload.code };
        }),
        [{ type: "ws.error", code: ONETALK_ERROR_CODES.authorizationRejected }],
    );
});

test("fences every Mind publish event when fallback authorization mismatches", async () => {
    const eventCases = [
        {
            type: "message.created",
            publish: (registry: ReturnType<typeof createOneTalkConnectionRegistry>) =>
                registry.publishMessageCreated({
                    message: centerMessage(message()),
                    requestId: "publish-message-mismatch",
                    scope: mindScope,
                }),
        },
        {
            type: "sync.status",
            publish: (registry: ReturnType<typeof createOneTalkConnectionRegistry>) =>
                registry.publishSyncStatus({
                    requestId: "publish-sync-mismatch",
                    scope: mindScope,
                    status: {
                        conversationId: "conversation-1",
                        historyGeneration: "initial",
                        mode: "full",
                        syncPhase: "initial",
                        syncResult: "succeeded",
                        latestMessageId: null,
                        historyComplete: true,
                        messageCount: 1,
                        anchorAdvanced: false,
                    },
                }),
        },
        {
            type: "conversation.updated",
            publish: (registry: ReturnType<typeof createOneTalkConnectionRegistry>) =>
                registry.publishConversationUpdated({
                    conversation: {
                        conversationId: "conversation-1",
                        conversationType: "direct",
                        name: null,
                        avatarUrl: null,
                        lastContactTimeLong: null,
                        messagePreview: null,
                    },
                    moveToTop: false,
                    requestId: "publish-conversation-mismatch",
                    scope: mindScope,
                }),
        },
    ] as const;

    for (const eventCase of eventCases) {
        const requests: OneTalkAuthorizationRequest[] = [];
        const sent: TestFrame[] = [];
        const authorization: OneTalkAuthorizationReader = {
            authorize: async (request) => {
                requests.push(request);
                return { allowed: false, code: ONETALK_ERROR_CODES.scopeMismatch };
            },
            readAuthorizationVersion: async () => "version-1",
        };
        const registry = createOneTalkConnectionRegistry({
            authorization,
            onPublishFailure: () => undefined,
        });
        const socket = fakeSocket((payload) => sent.push(JSON.parse(payload) as TestFrame));
        registry.register({
            socket,
            connectionType: "mind_page",
            scope: mindScope,
            mindScope,
            binding: "binding-1",
            authorizationVersion: "version-1",
            permissions: ["read"],
        });

        await eventCase.publish(registry);

        assert.deepEqual(requests, [
            {
                connectionType: "mind_page",
                operation: "read",
                scope: {
                    channelAccountId: mindScope.channelAccountId,
                    workspaceId: mindScope.workspaceId,
                },
            },
        ]);
        assert.deepEqual(
            sent.map((frame) => [frame.type, (frame.payload as { code?: string }).code]),
            [["ws.error", ONETALK_ERROR_CODES.scopeMismatch]],
            eventCase.type,
        );
    }
});

test("accepts rebuild ACKs from a rebuild-only plugin session", async () => {
    const authorization = createMockAuthorizationReader([
        { ...authorizationRecord, permissions: ["rebuild"] },
    ]);
    const app = createTestApp({
        database: createDatabaseStub(),
        authorization,
        oneTalkService: createService(async (_context, _source, observedMessage) => ({
            status: "accepted",
            message: observedMessage,
        })),
    });
    const plugin = await openSocket(app);

    try {
        const accepted = nextMessage(plugin);
        plugin.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "plugin",
                type: "ws.hello",
                requestId: "rebuild-only-hello",
                scope: pluginScope,
                payload: { binding: "binding-1", requestedPermissions: ["rebuild"] },
            }),
        );
        assert.equal((await accepted).type, "ws.accepted");

        plugin.send(
            JSON.stringify({
                protocolVersion: ONETALK_PROTOCOL_VERSION,
                connectionType: "plugin",
                type: "storage.delete.ack",
                requestId: "rebuild-only-clear",
                scope: pluginScope,
                payload: {
                    rebuildId: "unknown-rebuild",
                    conversationId: "conversation-1",
                    status: "cleared",
                },
            }),
        );
        await new Promise<void>((resolve) => setImmediate(resolve));
        assert.deepEqual(frameReaderFor(plugin).frames, []);
    } finally {
        await closeApp(app, [plugin]);
    }
});

test("reserves a sendRequestId before authorization and dispatches only once", async () => {
    let releaseAuthorization!: () => void;
    const authorizationGate = new Promise<void>((resolve) => {
        releaseAuthorization = resolve;
    });
    const sentCommands: string[] = [];
    const authorization = {
        authorize: async (request: Parameters<OneTalkAuthorizationReader["authorize"]>[0]) => {
            void request;
            await authorizationGate;
            return {
                allowed: true as const,
                binding: authorizationRecord.binding,
                authorizationVersion: authorizationRecord.authorizationVersion,
                permissions: authorizationRecord.permissions,
                mindScope,
            };
        },
        readAuthorizationVersion: async () => authorizationRecord.authorizationVersion,
    };
    const registry = createOneTalkConnectionRegistry({
        authorization,
        cutoverPolicy: createOneTalkCutoverPolicy(),
        onPublishFailure: () => {},
    });
    const mindSocket = fakeSocket(() => {});
    let commandSent!: () => void;
    const commandPromise = new Promise<void>((resolve) => {
        commandSent = resolve;
    });
    const pluginSocket = fakeSocket((payload) => {
        sentCommands.push(payload);
        commandSent();
    });
    const mind: OneTalkRegisteredConnection = {
        socket: mindSocket,
        connectionType: "mind_page",
        scope: mindScope,
        mindScope,
        binding: "binding-1",
        authorizationVersion: "version-1",
        permissions: ["read", "send"],
    };
    const plugin: OneTalkRegisteredConnection = {
        socket: pluginSocket,
        connectionType: "plugin",
        scope: pluginScope,
        mindScope,
        binding: "binding-1",
        authorizationVersion: "version-1",
        permissions: ["read", "send"],
    };
    registry.register(mind);
    registry.register(plugin);
    const frame: OneTalkSendRequestFrame = {
        protocolVersion: ONETALK_PROTOCOL_VERSION,
        connectionType: "mind_page",
        type: "send.request",
        requestId: "request-1",
        sendRequestId: "duplicate-id",
        scope: mindScope,
        payload: { conversationId: "conversation-1", content: { kind: "text", text: "hello" } },
    };

    const first = registry.requestSend({ mindSocket, frame });
    const imageWhileTextPending = await registry.requestSend({
        mindSocket,
        frame: {
            ...frame,
            requestId: "image-during-text",
            sendRequestId: "image-during-text",
            payload: {
                conversationId: "conversation-1",
                content: {
                    kind: "image",
                    source: {
                        downloadUrl: "https://mind.example.test/bridge/image-1",
                        fileName: "image.jpg",
                        mimeType: "image/jpeg",
                    },
                },
            },
        },
    });
    assert.deepEqual(imageWhileTextPending, {
        status: "rejected_before_send",
        reason: "send_in_progress",
    });
    const fileWhileTextPending = await registry.requestSend({
        mindSocket,
        frame: {
            ...frame,
            requestId: "file-during-text",
            sendRequestId: "file-during-text",
            payload: {
                conversationId: "conversation-1",
                content: {
                    kind: "file",
                    source: {
                        downloadUrl: "https://mind.example.test/bridge/attachment-1",
                        fileName: "quotation.zip",
                        mimeType: "application/x-mind-approved-file",
                    },
                },
            },
        },
    });
    assert.deepEqual(fileWhileTextPending, {
        status: "rejected_before_send",
        reason: "send_in_progress",
    });
    const duplicate = await registry.requestSend({ mindSocket, frame });
    assert.deepEqual(duplicate, {
        status: "rejected_before_send",
        reason: "duplicate_request",
    });
    releaseAuthorization();
    await commandPromise;
    assert.equal(JSON.parse(sentCommands[0]).type, "send.command");
    registry.unregister(pluginSocket);
    assert.deepEqual(await first, {
        status: "delivery_unknown",
        reason: "send_connection_lost",
    });
});

test("keeps a pre-dispatch disconnect in rejected_before_send", async () => {
    let releaseAuthorization!: () => void;
    const authorizationGate = new Promise<void>((resolve) => {
        releaseAuthorization = resolve;
    });
    const authorization = {
        authorize: async (request: Parameters<OneTalkAuthorizationReader["authorize"]>[0]) => {
            void request;
            await authorizationGate;
            return {
                allowed: true as const,
                binding: authorizationRecord.binding,
                authorizationVersion: authorizationRecord.authorizationVersion,
                permissions: authorizationRecord.permissions,
                mindScope,
            };
        },
        readAuthorizationVersion: async () => authorizationRecord.authorizationVersion,
    };
    const registry = createOneTalkConnectionRegistry({
        authorization,
        cutoverPolicy: createOneTalkCutoverPolicy(),
        onPublishFailure: () => {},
    });
    const mindSocket = fakeSocket(() => {});
    const pluginSocket = fakeSocket(() => {});
    const mind: OneTalkRegisteredConnection = {
        socket: mindSocket,
        connectionType: "mind_page",
        scope: mindScope,
        mindScope,
        binding: "binding-1",
        authorizationVersion: "version-1",
        permissions: ["read", "send"],
    };
    const plugin: OneTalkRegisteredConnection = {
        socket: pluginSocket,
        connectionType: "plugin",
        scope: pluginScope,
        mindScope,
        binding: "binding-1",
        authorizationVersion: "version-1",
        permissions: ["read", "send"],
    };
    registry.register(mind);
    registry.register(plugin);
    const frame: OneTalkSendRequestFrame = {
        protocolVersion: ONETALK_PROTOCOL_VERSION,
        connectionType: "mind_page",
        type: "send.request",
        requestId: "request-pre-dispatch-disconnect",
        sendRequestId: "pre-dispatch-disconnect",
        scope: mindScope,
        payload: { conversationId: "conversation-1", content: { kind: "text", text: "hello" } },
    };

    const request = registry.requestSend({ mindSocket, frame });
    registry.unregister(pluginSocket);
    releaseAuthorization();

    assert.deepEqual(await request, {
        status: "rejected_before_send",
        reason: "waiting_for_page",
    });
});

test("uses the fixed 45-second terminal budget for image and file sends", async () => {
    const sentCommands: string[] = [];
    let timeoutDelay = 0;
    const authorization = createMockAuthorizationReader([authorizationRecord]);
    const registry = createOneTalkConnectionRegistry({
        authorization,
        onPublishFailure: () => {},
        sendTimeoutMs: 1,
        scheduleTimeout: ((callback: () => void, delay?: number) => {
            timeoutDelay = Number(delay);
            queueMicrotask(callback);
            return undefined as unknown as ReturnType<typeof setTimeout>;
        }) as typeof setTimeout,
    });
    const mindSocket = fakeSocket(() => {});
    const pluginSocket = fakeSocket((payload) => sentCommands.push(payload));
    const mind: OneTalkRegisteredConnection = {
        socket: mindSocket,
        connectionType: "mind_page",
        scope: mindScope,
        mindScope,
        binding: "binding-1",
        authorizationVersion: "version-1",
        permissions: ["read", "send"],
    };
    const plugin: OneTalkRegisteredConnection = {
        socket: pluginSocket,
        connectionType: "plugin",
        scope: pluginScope,
        mindScope,
        binding: "binding-1",
        authorizationVersion: "version-1",
        permissions: ["read", "send"],
    };
    const frame: OneTalkSendRequestFrame = {
        protocolVersion: ONETALK_PROTOCOL_VERSION,
        connectionType: "mind_page",
        type: "send.request",
        requestId: "request-timeout",
        sendRequestId: "timeout-id",
        scope: mindScope,
        payload: {
            conversationId: "conversation-1",
            content: {
                kind: "image",
                source: {
                    downloadUrl: "https://mind.example.test/bridge/image-1",
                    fileName: "image.jpg",
                    mimeType: "image/jpeg",
                },
            },
        },
    };
    registry.register(mind);
    registry.register(plugin);

    assert.deepEqual(await registry.requestSend({ mindSocket, frame }), {
        status: "delivery_unknown",
        reason: "send_timeout",
    });
    assert.equal(timeoutDelay, 45_000);
    assert.equal(JSON.parse(sentCommands[0]).type, "send.command");
    const fileFrame: OneTalkSendRequestFrame = {
        ...frame,
        requestId: "request-file-timeout",
        sendRequestId: "file-timeout-id",
        payload: {
            conversationId: "conversation-1",
            content: {
                kind: "file",
                source: {
                    downloadUrl: "https://mind.example.test/bridge/attachment-1",
                    fileName: "quotation.zip",
                    mimeType: "application/x-mind-approved-file",
                },
            },
        },
    };
    assert.deepEqual(await registry.requestSend({ mindSocket, frame: fileFrame }), {
        status: "delivery_unknown",
        reason: "send_timeout",
    });
    assert.equal(timeoutDelay, 45_000);
    assert.equal(JSON.parse(sentCommands[1]).type, "send.command");
});

test("claims confirmation once and makes a terminal late confirmation a no-op", async () => {
    let pluginAuthorizationCalls = 0;
    const authorization = {
        authorize: async (request: Parameters<OneTalkAuthorizationReader["authorize"]>[0]) => {
            if (request.connectionType === "plugin") pluginAuthorizationCalls += 1;
            return {
                allowed: true as const,
                binding: authorizationRecord.binding,
                authorizationVersion: authorizationRecord.authorizationVersion,
                permissions: authorizationRecord.permissions,
                mindScope,
            };
        },
        readAuthorizationVersion: async () => authorizationRecord.authorizationVersion,
    };
    const registry = createOneTalkConnectionRegistry({
        authorization,
        cutoverPolicy: createOneTalkCutoverPolicy(),
        onPublishFailure: () => {},
    });
    const mindSocket = fakeSocket(() => {});
    let commandSent!: () => void;
    const commandPromise = new Promise<void>((resolve) => {
        commandSent = resolve;
    });
    const pluginSocket = fakeSocket(() => commandSent());
    const mind: OneTalkRegisteredConnection = {
        socket: mindSocket,
        connectionType: "mind_page",
        scope: mindScope,
        mindScope,
        binding: "binding-1",
        authorizationVersion: "version-1",
        permissions: ["read", "send"],
    };
    const plugin: OneTalkRegisteredConnection = {
        socket: pluginSocket,
        connectionType: "plugin",
        scope: pluginScope,
        mindScope,
        binding: "binding-1",
        authorizationVersion: "version-1",
        permissions: ["read", "send"],
    };
    registry.register(mind);
    registry.register(plugin);
    const requestFrame: OneTalkSendRequestFrame = {
        protocolVersion: ONETALK_PROTOCOL_VERSION,
        connectionType: "mind_page",
        type: "send.request",
        requestId: "request-confirm",
        sendRequestId: "confirm-id",
        scope: mindScope,
        payload: { conversationId: "conversation-1", content: { kind: "text", text: "hello" } },
    };
    const request = registry.requestSend({ mindSocket, frame: requestFrame });
    await commandPromise;
    const sentMessage = { ...message("sent-confirm"), direction: "sent" as const };
    const confirmation: OneTalkSendConfirmationFrame = {
        protocolVersion: ONETALK_PROTOCOL_VERSION,
        connectionType: "plugin",
        type: "send.confirmation",
        requestId: "confirmation-1",
        sendRequestId: "confirm-id",
        scope: pluginScope,
        payload: { status: "confirmed_sent", message: sentMessage },
    };
    let processCalls = 0;
    const process = async () => {
        processCalls += 1;
        return { status: "accepted" as const, message: sentMessage };
    };
    const firstConfirmation = registry.handleSendConfirmation(pluginSocket, confirmation, process);
    const lateConfirmation = registry.handleSendConfirmation(pluginSocket, confirmation, process);
    await Promise.all([firstConfirmation, lateConfirmation]);
    assert.equal(processCalls, 1);
    assert.equal(pluginAuthorizationCalls, 2);
    assert.deepEqual(await request, { status: "confirmed_sent", message: sentMessage });
    const afterTerminal = registry.handleSendConfirmation(pluginSocket, confirmation, process);
    await afterTerminal;
    assert.equal(processCalls, 1);
});

test("pausing Bright v6 closes existing sockets with 1013 without an error frame", async () => {
    const policy = createOneTalkCutoverPolicy();
    const app = createApp(testConfig, {
        database: createDatabaseStub(),
        authorization: createMockAuthorizationReader([authorizationRecord]),
        cutoverPolicy: policy,
        oneTalkService: createService(async () => ({ status: "anomaly", anomalyCode: "unused" })),
    });
    const plugin = await openSocket(app);
    try {
        await connectPlugin(plugin);
        const close = nextCloseCode(plugin);
        policy.pause();
        assert.equal(await close, 1013);
    } finally {
        await closeApp(app, [plugin]);
    }
});