feat: expand message center capabilities

.trellis/spec/server/backend/index.md

@@ -39,6 +39,7 @@
 | [服务基础设施](./service-foundation.md) | Fastify、WebSocket 与 ORM 基础契约 | 已建立 |
 | [后台纪要内部网络读取](./summary-authorization.md) | 内部 7777 listener、Docker 网络边界、固定窗口与发布约束 | Center 独立契约 |
 | [Mind HTTP 授权](./mind-authorization.md) | 两个 Mind 授权 HTTP 接口、同域 Cookie、CORS/Origin 与 fail-closed 边界 | 已建立适配器与本地 mock |
+| [OneTalk 单会话历史重建](./mind-history-rebuild.md) | rebuild 独立授权、scoped clear、generation/rebuild correlation、原子 reset 与 completion anchor | 已实现；真实 PostgreSQL/浏览器联调另行验证 |
 | [OneTalk 联系人资料 Bright 持久化](./mind-contact-profile.md) | profile composite key、严格时间前进 upsert、future-skew 拒绝、transaction/ACK fence 和 read-model 内存组合 | 已实现并有 focused tests；真实 PostgreSQL 另行验证 |
 | [OneTalk 买家事实 Bright 持久化与读取](./mind-buyer-fact.md) | buyer source replace、transaction/ACK fence、无 JOIN read projection | 已实现并有 PostgreSQL integration tests |
 

.trellis/spec/server/backend/mind-history-rebuild.md

@@ -0,0 +1,76 @@
+# OneTalk 单会话历史重建契约
+
+## 1. Scope / Trigger
+
+- Trigger：Mind 需要删除并重新拉取一个 direct conversation 的历史；这是 destructive `rebuild` 操作，不是 read 的别名。
+- Scope：`POST /api/bright/onetalk/accounts/:channelAccountId/conversations/:conversationId/history/rebuild`、`storage.delete.*`、`history.sync.*`、`rebuild.status`、插件 durable ledger 与 Bright conversation generation。
+- Excluded：账户 bootstrap marker、`onetalk_contact_profile`、buyer fact、其它 conversation、发送队列及任意 raw SDK/IndexedDB key 删除。
+
+## 2. Signatures
+
+```ts
+POST /api/bright/onetalk/accounts/:channelAccountId/conversations/:conversationId/history/rebuild
+  -> { scope, conversationId, rebuildId, status: "server_reset_committed", resync }
+
+storage.delete.command { rebuildId, target: { kind: "conversation_history", conversationId } }
+storage.delete.ack     { rebuildId, conversationId, status: "cleared" | "rejected", reason? }
+history.sync.command   { rebuildId, historyGeneration, conversationId, mode: "full" }
+history.sync.ack       { rebuildId, conversationId, status: "started" | "rejected", reason? }
+sync.complete          { conversationId, historyGeneration, rebuildId?, ... }
+```
+
+`rebuildId` 是一次内存操作关联 token；`historyGeneration` 是 `onetalk_conversation` 持久化的事实边界。repository 的 reset 与 guarded sync-anomaly 写入均必须接受 `OneTalkCommitGuard`。
+
+## 3. Contracts
+
+- HTTP、Mind status 与 plugin command 都要求精确 scope、binding、authorizationVersion 和独立的 `rebuild` permission；read permission 不能替代它。
+- 顺序固定为：plugin scoped durable clear -> matching ACK -> 重新授权 -> Bright transaction delete/reset/new generation -> `history.sync.command`。HTTP 200 只表示 transaction 已提交；sync start 失败仍返回 `server_reset_committed` 加失败 `resync`。
+- 插件只删除 `(channelAccountId, conversationId)` 的 messages/checkpoints/candidates/conversation anomalies；clear 后保持 quiesced。`rebuildId + historyGeneration + page requestId` 必须关联到 history observation、progress、candidate upload 和 `sync.complete`；未关联或过期的 page callback 不得写入当前 generation。
+- `sync.complete` 只有携带匹配 rebuild token，且服务端最终持久化结果为 `historyComplete=true` 与 `succeeded|succeeded_with_anomalies` 时，才回发完整 `anchor.snapshot` 并发布 `rebuild.status: sync_completed`。anchor snapshot 是全量替换语义，不能只发送目标会话。
+- reset transaction 在每个数据库副作用边界都复核 canonical socket、connection generation、cutover epoch、rebuild permission 和 heartbeat lease；sync anomaly 写入要在锁住目标 conversation 且验证 generation 的同一 transaction 中完成。
+
+## 4. Validation & Error Matrix
+
+| 条件 | 结果 |
+| --- | --- |
+| Mind 或 plugin 只有 read | 403/拒绝 rebuild，不执行 clear/reset |
+| 无 canonical fresh rebuild plugin | `plugin_offline` 或 `plugin_heartbeat_stale`，不写库 |
+| clear ACK 不匹配、超时或 rejected | 不 reset，返回稳定 pre-commit error |
+| clear 后授权、policy、canonical connection 或 lease 失效 | 不 reset；已提交后失效只报告 resync failed，不回滚事实 |
+| 旧 generation completion/observation | reject 或丢弃；不得写 message、anomaly、anchor 或 status |
+| reset 抢在旧 completion 前提交 | 旧 anomaly 不能落库，completion 变为 rejected |
+| persisted completion 是 incomplete/failed | 不发 rebuild completion snapshot/status |
+
+## 5. Good / Base / Bad Cases
+
+- Good：同一 conversation 的旧 page history callback 在 rebuild release 后抵达，因 requestId 不匹配被丢弃。
+- Good：服务器 reset 后同步命令未启动，HTTP 仍为 200，Mind 得到 `resync.status=failed`，下次请求可安全重试。
+- Base：同一 WebSocket 上完成一次 rebuild 后，带新 rebuildId 的下一次 clear 可替换完成态 token；迟到的旧 sync command 不能释放新的 pending token。
+- Bad：用 `read` 授权 destructive route、发送任意 store/key delete 命令、把单会话 anchor 当作全量 snapshot，或在先读 generation 后无条件写 anomaly。
+
+## 6. Tests Required
+
+- Contract：rebuild permission 独立、所有 control frame 与 `sync.complete` 的 exact-key/direction 检查。
+- Extension：scoped IDB clear 保留 profile/buyer/bootstrap；quiesce、旧 command/ACK/discovery/page callback、重复 completion snapshot 和连续 rebuild 回归。
+- Server：read-only 403、single-flight account+conversation、每个 await 后 revoke/policy/lease fence、post-commit HTTP semantics、matching completion 的 full anchor/status、incomplete completion no-op。
+- Repository：可控 race 让 reset 赢过旧 completion，断言没有旧 generation anomaly；有 `TEST_DATABASE_URL` 时再验证真实 PostgreSQL 行锁交错。
+
+## 7. Wrong vs Correct
+
+### Wrong
+
+```ts
+await plugin.clear();
+await repository.deleteMessages(scope, conversationId);
+startSync(); // 未关联 rebuild token，旧 callback 可混入
+```
+
+### Correct
+
+```ts
+await matchingClearAck(rebuildId);
+await authorizeCurrentPlugin();
+await repository.resetConversationHistory(context, conversationId, historyGeneration, guard);
+sendHistorySync({ rebuildId, historyGeneration, conversationId, mode: "full" });
+// 插件仅接受同 rebuildId 绑定的 page request；服务端只完成匹配 token 的持久化结果。
+```

apps/chrome-extension/src/onetalk/main-page/current-conversation-history/page-command.ts

@@ -1,6 +1,7 @@
 // 处理 Service Worker 下发的历史同步命令
 
 import { isPlainRecord } from "@trade-message-center/onetalk-contract";
+import { createOneTalkPageObservedMessage } from "../../page-bridge/model.ts";
 import { createOneTalkPageHistoryProgressSink } from "../../page-bridge/main.ts";
 import type {
     OneTalkPageBridgeWindow,
@@ -121,9 +122,24 @@ const observeHistoryItems = (
     pageWindow: OneTalkPageWindow,
     items: unknown[],
     observedSink?: OneTalkObservedMessageSink,
+    historyRequestId?: string,
 ): void => {
     if (!observedSink) return;
-    observedSink(createHistoryMessageBatch(items, pageWindow));
+    const batch = createHistoryMessageBatch(items, pageWindow);
+    if (historyRequestId === undefined) {
+        observedSink(batch);
+        return;
+    }
+    const bridgeWindow = pageWindow as unknown as OneTalkPageBridgeWindow;
+    bridgeWindow.postMessage(
+        createOneTalkPageObservedMessage(
+            batch.messages,
+            undefined,
+            batch.diagnostics,
+            historyRequestId,
+        ),
+        bridgeWindow.location.origin,
+    );
 };
 
 /** 执行页面同步命令并只返回可序列化的进度摘要。 */
@@ -195,9 +211,10 @@ export const handleOneTalkHistoryCommand = async (
             startTimeStamps: startTimeStamps ?? undefined,
             onProgress: createOneTalkPageHistoryProgressSink(
                 pageWindow as unknown as OneTalkPageBridgeWindow,
+                message.requestId,
             ),
             onHistoryItems: (items: unknown[]) =>
-                observeHistoryItems(pageWindow, items, observedSink),
+                observeHistoryItems(pageWindow, items, observedSink, message.requestId),
         };
         if (action === "onetalk.sync.conversation") {
             const conversationId = message.command.conversationId;

apps/chrome-extension/src/onetalk/page-bridge/main.ts

@@ -196,20 +196,28 @@ export const createOneTalkPageObservedSink = (
 /** 创建把历史分页进度发布到 Service Worker 的 MAIN sink。 */
 export const createOneTalkPageHistoryProgressSink = (
     pageWindow: OneTalkPageBridgeWindow,
+    historyRequestId?: string,
 ): ((progress: HistoryPageProgress) => void) => {
     return (progress) => {
         const origin = pageOrigin(pageWindow);
         if (!origin) return;
-        const message: OneTalkPageObservedMessage = createOneTalkPageObservedMessage([], {
-            conversationId: progress.conversationId,
-            latestMessageAtMs: progress.latestMessageAtMs,
-            page: progress.page,
-            mode: progress.mode,
-            anchorMessageId: progress.anchorMessageId,
-            nextTimeStamp: progress.nextTimeStamp,
-            historyComplete: progress.historyComplete,
-            ...(progress.anchorFound === undefined ? {} : { anchorFound: progress.anchorFound }),
-        });
+        const message: OneTalkPageObservedMessage = createOneTalkPageObservedMessage(
+            [],
+            {
+                conversationId: progress.conversationId,
+                latestMessageAtMs: progress.latestMessageAtMs,
+                page: progress.page,
+                mode: progress.mode,
+                anchorMessageId: progress.anchorMessageId,
+                nextTimeStamp: progress.nextTimeStamp,
+                historyComplete: progress.historyComplete,
+                ...(progress.anchorFound === undefined
+                    ? {}
+                    : { anchorFound: progress.anchorFound }),
+            },
+            undefined,
+            historyRequestId,
+        );
         postPageMessage(pageWindow, origin, message);
     };
 };

apps/chrome-extension/src/onetalk/page-bridge/model.ts

@@ -58,6 +58,7 @@ export type OneTalkPageObservedMessage = {
     version: typeof ONE_TALK_PAGE_BRIDGE_VERSION;
     type: "onetalk.page.observed";
     batch: ObservedOneTalkMessage[];
+    historyRequestId?: string;
     historyProgress?: OneTalkPageHistoryProgress;
     diagnostics?: OneTalkObservationDiagnostics;
 };
@@ -406,7 +407,15 @@ const decodeDiagnostics = (value: unknown): OneTalkObservationDiagnostics | null
 const decodeObservedMessageEnvelope = (
     value: Record<string, unknown>,
 ): OneTalkPageObservedMessage | null => {
-    const allowedKeys = ["source", "version", "type", "batch", "historyProgress", "diagnostics"];
+    const allowedKeys = [
+        "source",
+        "version",
+        "type",
+        "batch",
+        "historyRequestId",
+        "historyProgress",
+        "diagnostics",
+    ];
     if (
         !Array.isArray(value.batch) ||
         !Object.keys(value).every((key) => allowedKeys.includes(key)) ||
@@ -417,6 +426,12 @@ const decodeObservedMessageEnvelope = (
 
     const historyProgress = decodeHistoryProgress(value.historyProgress);
     if (historyProgress === null) return null;
+    const historyRequestId = value.historyRequestId;
+    if (
+        historyRequestId !== undefined &&
+        (typeof historyRequestId !== "string" || historyRequestId.length === 0)
+    )
+        return null;
     const diagnostics = decodeDiagnostics(value.diagnostics);
     if (diagnostics === null) return null;
 
@@ -432,6 +447,7 @@ const decodeObservedMessageEnvelope = (
         version: ONE_TALK_PAGE_BRIDGE_VERSION,
         type: "onetalk.page.observed",
         batch,
+        ...(historyRequestId === undefined ? {} : { historyRequestId }),
         ...(historyProgress === undefined ? {} : { historyProgress }),
         ...(diagnostics === undefined ? {} : { diagnostics }),
     };
@@ -601,12 +617,14 @@ export const createOneTalkPageObservedMessage = (
     batch: ObservedOneTalkMessage[],
     historyProgress?: OneTalkPageHistoryProgress,
     diagnostics?: OneTalkObservationDiagnostics,
+    historyRequestId?: string,
 ): OneTalkPageObservedMessage => {
     return {
         source: ONE_TALK_PAGE_BRIDGE_SOURCE,
         version: ONE_TALK_PAGE_BRIDGE_VERSION,
         type: "onetalk.page.observed",
         batch,
+        ...(historyRequestId === undefined ? {} : { historyRequestId }),
         ...(historyProgress === undefined ? {} : { historyProgress }),
         ...(diagnostics === undefined ? {} : { diagnostics }),
     };

apps/chrome-extension/src/onetalk/service-worker/configured-sync-session.ts

@@ -14,6 +14,7 @@ import type { OneTalkServiceWorkerRuntime } from "./runtime.ts";
 import type { OneTalkPageDiagnostic } from "./runtime.ts";
 import { createOneTalkServiceWorkerFrameRouter } from "./routing/frame-router.ts";
 import { createOneTalkSendCommandFlow } from "./flows/send-command-flow.ts";
+import { createOneTalkHistoryRebuildFlow } from "./flows/history-rebuild-flow.ts";
 import {
     createOneTalkSyncEngine,
     type OneTalkSyncEngine,
@@ -206,7 +207,7 @@ export class OneTalkConfiguredSyncSession {
             url: config.brightWebSocketUrl,
             scope: pluginScope,
             binding: config.binding,
-            requestedPermissions: ["read", "send"],
+            requestedPermissions: ["read", "send", "rebuild"],
             ...(this.now === undefined ? {} : { now: this.now }),
             createRequestId,
             onStatusChange: notifyCurrentStatus,
@@ -281,11 +282,18 @@ export class OneTalkConfiguredSyncSession {
             onPageDiagnostic: this.onPageDiagnostic,
             onError: reportCurrentError,
         });
+        const rebuild = createOneTalkHistoryRebuildFlow({
+            scope: pluginScope,
+            bright,
+            sync: engine,
+            onError: reportCurrentError,
+        });
         const router = createOneTalkServiceWorkerFrameRouter({
             sync: engine,
             ...(activeProfile === undefined ? {} : { profile: activeProfile }),
             ...(activeBuyer === undefined ? {} : { buyer: activeBuyer }),
             send,
+            rebuild,
         });
         const unsubscribeRouter = bright.subscribe((frame) => {
             if (currentRevision === this.revision) router.handle(frame);

apps/chrome-extension/src/onetalk/service-worker/flows/frame-writers.ts

@@ -30,6 +30,7 @@ type WriterOptions = {
     send: OneTalkFrameSend;
     createRequestId: (kind: string) => string;
     historyGeneration?: (conversationId: string) => string | undefined;
+    rebuildId?: (conversationId: string) => string | undefined;
 };
 
 const isNonEmptyString = (value: string): boolean => value.length > 0;
@@ -194,6 +195,7 @@ export const createOneTalkSyncFrameWriter = (options: WriterOptions): OneTalkSyn
 
     const sendSyncComplete: OneTalkSyncFrameWriter["sendSyncComplete"] = (input) => {
         const historyGeneration = options.historyGeneration?.(input.conversationId);
+        const rebuildId = options.rebuildId?.(input.conversationId);
         if (
             !isNonEmptyString(input.conversationId) ||
             !historyGeneration ||
@@ -215,6 +217,7 @@ export const createOneTalkSyncFrameWriter = (options: WriterOptions): OneTalkSyn
             payload: {
                 conversationId: input.conversationId,
                 historyGeneration,
+                ...(rebuildId === undefined ? {} : { rebuildId }),
                 mode: input.mode,
                 historyComplete: input.historyComplete,
                 result: input.result,

apps/chrome-extension/src/onetalk/service-worker/flows/history-rebuild-flow.ts

@@ -0,0 +1,101 @@
+// 执行 Bright 发起的单会话账本清理与 full-sync release。
+
+import {
+    createOneTalkHistorySyncAckFrame,
+    createOneTalkStorageDeleteAckFrame,
+    type OneTalkFrame,
+    type OneTalkPluginScope,
+} from "@trade-message-center/onetalk-contract";
+
+import type { OneTalkSyncEngine } from "../sync-engine.ts";
+
+export type OneTalkHistoryRebuildFlow = {
+    handle: (
+        frame: Extract<OneTalkFrame, { type: "storage.delete.command" | "history.sync.command" }>,
+    ) => void;
+};
+
+const stableReason = (error: unknown): string => {
+    const code = error instanceof Error ? error.message : "";
+    return new Set(["runtime_disposed", "scope_mismatch", "rebuild_in_progress"]).has(code)
+        ? code
+        : "history_rebuild_failed";
+};
+
+/** 仅执行固定 conversation_history target；ACK 一律等待 durable clear 或 start outcome。 */
+export const createOneTalkHistoryRebuildFlow = (options: {
+    scope: OneTalkPluginScope;
+    bright: Pick<{ send: (frame: OneTalkFrame) => boolean }, "send">;
+    sync: Pick<OneTalkSyncEngine, "clearConversationHistory" | "startRebuildSync">;
+    onError: (error: unknown) => void;
+}): OneTalkHistoryRebuildFlow => {
+    const handleStorageDelete = async (
+        frame: Extract<OneTalkFrame, { type: "storage.delete.command" }>,
+    ): Promise<void> => {
+        const { rebuildId, target } = frame.payload;
+        try {
+            await options.sync.clearConversationHistory(
+                options.scope.channelAccountId,
+                target.conversationId,
+                rebuildId,
+            );
+            options.bright.send(
+                createOneTalkStorageDeleteAckFrame(
+                    { connectionType: "plugin", requestId: frame.requestId, scope: options.scope },
+                    { rebuildId, conversationId: target.conversationId, status: "cleared" },
+                ),
+            );
+        } catch (error) {
+            options.onError(error);
+            options.bright.send(
+                createOneTalkStorageDeleteAckFrame(
+                    { connectionType: "plugin", requestId: frame.requestId, scope: options.scope },
+                    {
+                        rebuildId,
+                        conversationId: target.conversationId,
+                        status: "rejected",
+                        reason: stableReason(error),
+                    },
+                ),
+            );
+        }
+    };
+
+    const handleHistorySync = async (
+        frame: Extract<OneTalkFrame, { type: "history.sync.command" }>,
+    ): Promise<void> => {
+        const result = await options.sync.startRebuildSync({
+            channelAccountId: options.scope.channelAccountId,
+            conversationId: frame.payload.conversationId,
+            rebuildId: frame.payload.rebuildId,
+            historyGeneration: frame.payload.historyGeneration,
+        });
+        options.bright.send(
+            createOneTalkHistorySyncAckFrame(
+                { connectionType: "plugin", requestId: frame.requestId, scope: options.scope },
+                result.status === "started"
+                    ? {
+                          rebuildId: frame.payload.rebuildId,
+                          conversationId: frame.payload.conversationId,
+                          status: "started",
+                      }
+                    : {
+                          rebuildId: frame.payload.rebuildId,
+                          conversationId: frame.payload.conversationId,
+                          status: "rejected",
+                          reason: result.reason ?? result.status,
+                      },
+            ),
+        );
+    };
+
+    return {
+        handle: (frame) => {
+            if (frame.type === "storage.delete.command") {
+                void handleStorageDelete(frame);
+                return;
+            }
+            void handleHistorySync(frame).catch(options.onError);
+        },
+    };
+};

apps/chrome-extension/src/onetalk/service-worker/routing/frame-router.ts

@@ -5,6 +5,7 @@ import { ONETALK_ERROR_CODES, type OneTalkFrame } from "@trade-message-center/on
 import type { OneTalkBuyerFactCoordinator } from "../buyer-fact-coordinator.ts";
 import type { OneTalkContactProfileCoordinator } from "../contact-profile-coordinator.ts";
 import type { OneTalkSendCommandFlow } from "../flows/send-command-flow.ts";
+import type { OneTalkHistoryRebuildFlow } from "../flows/history-rebuild-flow.ts";
 import type { OneTalkSyncEngine } from "../sync-engine.ts";
 
 type OneTalkBusinessRoute = {
@@ -36,6 +37,7 @@ export const createOneTalkServiceWorkerFrameRouter = (options: {
     profile?: Pick<OneTalkContactProfileCoordinator, "handleFrame">;
     buyer?: Pick<OneTalkBuyerFactCoordinator, "handleFrame">;
     send?: Pick<OneTalkSendCommandFlow, "handle">;
+    rebuild?: Pick<OneTalkHistoryRebuildFlow, "handle">;
 }): OneTalkServiceWorkerFrameRouter => {
     const routes = [
         defineOneTalkBusinessRoute("anchor.snapshot", (frame) =>
@@ -56,6 +58,12 @@ export const createOneTalkServiceWorkerFrameRouter = (options: {
         ),
         defineOneTalkBusinessRoute("buyer.facts.ack", (frame) => options.buyer?.handleFrame(frame)),
         defineOneTalkBusinessRoute("send.command", (frame) => options.send?.handle(frame)),
+        defineOneTalkBusinessRoute("storage.delete.command", (frame) =>
+            options.rebuild?.handle(frame),
+        ),
+        defineOneTalkBusinessRoute("history.sync.command", (frame) =>
+            options.rebuild?.handle(frame),
+        ),
     ];
     const handle = (frame: OneTalkFrame): void => {
         for (const route of routes) {

apps/chrome-extension/src/onetalk/service-worker/storage.ts

@@ -173,6 +173,7 @@ export type OneTalkSyncStore = {
         channelAccountId: string,
         conversationId?: string,
     ) => Promise<OneTalkSyncAnomaly[]>;
+    clearConversationHistory: (channelAccountId: string, conversationId: string) => Promise<void>;
 };
 
 export type OneTalkContactProfileStore = {
@@ -729,6 +730,41 @@ export const createOneTalkSyncStore = (
         );
     };
 
+    /** 只清除单个会话的消息同步账本；资料和账号级调度状态不在此事务内。 */
+    const clearConversationHistory = async (
+        channelAccountId: string,
+        conversationId: string,
+    ): Promise<void> => {
+        const database = await getDatabase();
+        const storeNames = [
+            ONE_TALK_MESSAGE_STORE_NAME,
+            ONE_TALK_CHECKPOINT_STORE_NAME,
+            ONE_TALK_CANDIDATE_STORE_NAME,
+            ONE_TALK_ANOMALY_STORE_NAME,
+        ] as const;
+        const transaction = database.transaction([...storeNames], "readwrite");
+        const completion = transactionResult(transaction);
+        for (const storeName of storeNames) {
+            const store = transaction.objectStore(storeName);
+            const request = store.getAll();
+            request.onsuccess = () => {
+                for (const record of request.result as Array<{
+                    key: string;
+                    channelAccountId: string;
+                    conversationId?: string;
+                }>) {
+                    if (
+                        record.channelAccountId === channelAccountId &&
+                        record.conversationId === conversationId
+                    ) {
+                        store.delete(record.key);
+                    }
+                }
+            };
+        }
+        await completion;
+    };
+
     return {
         getCheckpoint,
         listCheckpoints,
@@ -741,6 +777,7 @@ export const createOneTalkSyncStore = (
         updateCandidate,
         recordAnomaly,
         listAnomalies,
+        clearConversationHistory,
     };
 };
 

apps/chrome-extension/src/onetalk/service-worker/sync-engine.ts

@@ -61,6 +61,7 @@ class OneTalkSyncEngineImpl implements OneTalkSyncEngine {
             createRequestId: this.createRequestId,
             historyGeneration: (conversationId) =>
                 this.lifecycle?.getHistoryGeneration(conversationId),
+            rebuildId: (conversationId) => this.lifecycle?.getRebuildId(conversationId),
         });
         let notifyQueueChange = (): void => undefined;
         this.queue = new ConversationQueue(() => notifyQueueChange());
@@ -148,6 +149,90 @@ class OneTalkSyncEngineImpl implements OneTalkSyncEngine {
         return this.bootstrap.startSync(input);
     }
 
+    public async clearConversationHistory(
+        channelAccountId: string,
+        conversationId: string,
+        rebuildId = "local-rebuild",
+    ): Promise<void> {
+        if (this.lifecycle.isDisposed()) throw new Error("runtime_disposed");
+        if (channelAccountId !== this.options.scope.channelAccountId)
+            throw new Error("scope_mismatch");
+        if (!this.lifecycle.quiesceConversation(conversationId, rebuildId))
+            throw new Error("rebuild_in_progress");
+        await this.observations.waitForConversationObservations(channelAccountId, conversationId);
+        await this.queue.enqueue(channelAccountId, conversationId, async () => {
+            await this.observations.waitForConversationObservations(
+                channelAccountId,
+                conversationId,
+            );
+            this.ack.clearConversation(conversationId);
+            this.checkpoints.resetConversation(channelAccountId, conversationId);
+            await this.options.store.clearConversationHistory(channelAccountId, conversationId);
+        });
+    }
+
+    public async startRebuildSync(input: {
+        channelAccountId: string;
+        conversationId: string;
+        rebuildId: string;
+        historyGeneration: string;
+    }): Promise<OneTalkSyncStartResult> {
+        if (this.lifecycle.isDisposed())
+            return {
+                status: "failed",
+                conversationId: input.conversationId,
+                reason: "runtime_disposed",
+            };
+        if (input.channelAccountId !== this.options.scope.channelAccountId) {
+            return {
+                status: "failed",
+                conversationId: input.conversationId,
+                reason: "scope_mismatch",
+            };
+        }
+        if (!this.lifecycle.isConversationQuiesced(input.conversationId)) {
+            return {
+                status: "failed",
+                conversationId: input.conversationId,
+                reason: "rebuild_not_quiesced",
+            };
+        }
+        const discovery = await this.options.pageRuntime.routePageCommand({
+            channelAccountId: input.channelAccountId,
+            requestId: this.createRequestId("rebuild-discover"),
+            command: { action: "onetalk.discover-conversations" },
+        });
+        if (discovery.status !== "completed") {
+            return {
+                status: "failed",
+                conversationId: input.conversationId,
+                reason:
+                    typeof discovery.reason === "string"
+                        ? discovery.reason
+                        : "conversation_cache_missing",
+            };
+        }
+        if (
+            !this.lifecycle.releaseConversation(
+                input.conversationId,
+                input.rebuildId,
+                input.historyGeneration,
+            )
+        ) {
+            return {
+                status: "failed",
+                conversationId: input.conversationId,
+                reason: "stale_rebuild_command",
+            };
+        }
+        return this.bootstrap.startSync({
+            channelAccountId: input.channelAccountId,
+            conversationId: input.conversationId,
+            mode: "full",
+            rebuildId: input.rebuildId,
+        });
+    }
+
     public resume(conversationId?: string): Promise<OneTalkSyncStartResult[]> {
         return this.bootstrap.resume(conversationId);
     }
@@ -240,7 +325,10 @@ class OneTalkSyncEngineImpl implements OneTalkSyncEngine {
     private handleBrightFrame(frame: OneTalkFrame): void {
         if (this.lifecycle.isDisposed()) return;
         if (frame.type === "anchor.snapshot") {
-            const epoch = this.lifecycle.acceptAnchorSnapshot(frame.payload.anchors);
+            const epoch = this.lifecycle.acceptAnchorSnapshot(
+                frame.payload.anchors,
+                frame.requestId,
+            );
             if (epoch === undefined) return;
             this.ack.acceptAnchorSnapshot();
             this.bootstrap.acceptAnchorSnapshot();

apps/chrome-extension/src/onetalk/service-worker/sync-engine/ack-completion.ts

@@ -94,6 +94,28 @@ export class AckCompletionCoordinator {
         this.pendingIncompletes.clear();
     }
 
+    /** 清掉单会话的易失 ACK/discovery 状态，不能影响其它会话。 */
+    public clearConversation(conversationId: string): void {
+        const key = scopedKey(this.scope.channelAccountId, conversationId);
+        const candidatePrefix = `${JSON.stringify([
+            this.scope.channelAccountId,
+            conversationId,
+        ]).slice(0, -1)},`;
+        this.discovered.delete(key);
+        this.explicitlySentCandidates.forEach((candidateKey) => {
+            if (candidateKey.startsWith(candidatePrefix))
+                this.explicitlySentCandidates.delete(candidateKey);
+        });
+        for (const [requestId, request] of this.requestCandidates) {
+            if (request.conversationId === conversationId) this.requestCandidates.delete(requestId);
+        }
+        for (const [requestId, pending] of this.pendingDiscoveries) {
+            if (pending.conversationId === conversationId)
+                this.pendingDiscoveries.delete(requestId);
+        }
+        this.pendingIncompletes.delete(key);
+    }
+
     /** 新 snapshot 使旧 discovery request 的 generation 失去写入资格。 */
     public acceptAnchorSnapshot(): void {
         this.pendingDiscoveries.clear();
@@ -247,6 +269,7 @@ export class AckCompletionCoordinator {
     ): Promise<void> {
         if (
             this.lifecycle.isDisposed() ||
+            this.lifecycle.isConversationQuiesced(conversationId) ||
             (!this.lifecycle.isBusinessFrameGateOpen() && !explicitStart)
         ) {
             return;
@@ -299,7 +322,8 @@ export class AckCompletionCoordinator {
         forceResend = false,
         explicitStart = false,
     ): Promise<void> {
-        if (this.lifecycle.isDisposed()) return;
+        if (this.lifecycle.isDisposed() || this.lifecycle.isConversationQuiesced(conversationId))
+            return;
         if (!this.lifecycle.getHistoryGeneration(conversationId)) return;
         const checkpoint = await this.checkpoints.getCheckpoint(channelAccountId, conversationId);
         if (!checkpoint || !checkpoint.historyComplete || checkpoint.phase !== "uploading") return;
@@ -355,7 +379,7 @@ export class AckCompletionCoordinator {
         if (this.lifecycle.isDisposed()) return;
         const request = this.requestCandidates.get(frame.requestId);
         this.requestCandidates.delete(frame.requestId);
-        if (!request) return;
+        if (!request || this.lifecycle.isConversationQuiesced(request.conversationId)) return;
         const results = frame.type === "messages.ack" ? frame.payload.results : [frame.payload];
         for (const [index, result] of results.entries()) {
             if (

apps/chrome-extension/src/onetalk/service-worker/sync-engine/bootstrap-coordinator.ts

@@ -269,10 +269,23 @@ export class BootstrapCoordinator {
                 if (scanning.pageTimeStamp !== null) {
                     resumeFrom[input.conversationId] = scanning.pageTimeStamp;
                 }
+                const pageRequestId = this.createRequestId("page");
+                if (
+                    input.rebuildId !== undefined &&
+                    !this.lifecycle.bindRebuildPageCommand(
+                        input.conversationId,
+                        input.rebuildId,
+                        this.lifecycle.getHistoryGeneration(input.conversationId) ?? "",
+                        pageRequestId,
+                    )
+                ) {
+                    result = failedResult(input.conversationId, "stale_rebuild_command");
+                    return;
+                }
                 const route: OneTalkPageCommandRoute = {
                     channelAccountId: input.channelAccountId,
                     conversationId: input.conversationId,
-                    requestId: this.createRequestId("page"),
+                    requestId: pageRequestId,
                     command: {
                         ...pageCommandFor(effectiveMode, commandAnchors, commandModes, resumeFrom),
                         action: "onetalk.sync.conversation",
@@ -282,6 +295,18 @@ export class BootstrapCoordinator {
                 };
                 try {
                     const pageResult = await this.pageRuntime.routePageCommand(route);
+                    if (
+                        input.rebuildId !== undefined &&
+                        !this.lifecycle.isCurrentRebuildPageCommand(
+                            input.conversationId,
+                            input.rebuildId,
+                            this.lifecycle.getHistoryGeneration(input.conversationId) ?? "",
+                            pageRequestId,
+                        )
+                    ) {
+                        result = failedResult(input.conversationId, "stale_rebuild_command");
+                        return;
+                    }
                     result = await this.processPageResults(
                         input.channelAccountId,
                         input.conversationId,

apps/chrome-extension/src/onetalk/service-worker/sync-engine/checkpoint-coordinator.ts

@@ -99,6 +99,12 @@ export class CheckpointCoordinator {
         this.activeAnchors.clear();
     }
 
+    public resetConversation(channelAccountId: string, conversationId: string): void {
+        const key = scopedKey(channelAccountId, conversationId);
+        this.activeModes.delete(key);
+        this.activeAnchors.delete(key);
+    }
+
     public async loadOrCreateCheckpoint(
         channelAccountId: string,
         conversationId: string,

apps/chrome-extension/src/onetalk/service-worker/sync-engine/lifecycle.ts

@@ -49,6 +49,11 @@ export class OneTalkSyncLifecycle {
     private snapshotEpoch = 0;
     private snapshotFingerprint: string | undefined;
     private readonly anchors = new Map<string, OneTalkAnchor>();
+    private readonly quiescedConversations = new Set<string>();
+    private readonly rebuilds = new Map<
+        string,
+        { rebuildId: string; historyGeneration?: string; pageRequestId?: string }
+    >();
     private pageEpoch = 0;
     private pageConversationId: string | undefined;
     private lastError: string | undefined;
@@ -107,6 +112,116 @@ export class OneTalkSyncLifecycle {
         return this.anchors.get(conversationId)?.historyGeneration;
     }
 
+    public isConversationQuiesced(conversationId: string): boolean {
+        return this.quiescedConversations.has(conversationId);
+    }
+
+    /** 重建 clear 已开始后，旧页面观察和迟到 ACK 不得重新建立账本。 */
+    public quiesceConversation(conversationId: string, rebuildId: string): boolean {
+        const current = this.rebuilds.get(conversationId);
+        if (
+            current &&
+            current.rebuildId !== rebuildId &&
+            this.quiescedConversations.has(conversationId)
+        ) {
+            return false;
+        }
+        this.rebuilds.set(conversationId, { rebuildId });
+        this.quiescedConversations.add(conversationId);
+        return true;
+    }
+
+    /** 仅 post-commit history.sync.command 可以替换 generation 并解除静默。 */
+    public releaseConversation(
+        conversationId: string,
+        rebuildId: string,
+        historyGeneration: string,
+    ): boolean {
+        const current = this.rebuilds.get(conversationId);
+        if (
+            !current ||
+            current.rebuildId !== rebuildId ||
+            !this.quiescedConversations.has(conversationId)
+        ) {
+            return false;
+        }
+        this.anchors.set(conversationId, {
+            conversationId,
+            historyGeneration,
+            latestMessageId: null,
+        });
+        this.rebuilds.set(conversationId, { rebuildId, historyGeneration });
+        this.quiescedConversations.delete(conversationId);
+        return true;
+    }
+
+    public bindRebuildPageCommand(
+        conversationId: string,
+        rebuildId: string,
+        historyGeneration: string,
+        requestId: string,
+    ): boolean {
+        const current = this.rebuilds.get(conversationId);
+        if (
+            !current ||
+            current.rebuildId !== rebuildId ||
+            current.historyGeneration !== historyGeneration ||
+            this.quiescedConversations.has(conversationId)
+        ) {
+            return false;
+        }
+        this.rebuilds.set(conversationId, { ...current, pageRequestId: requestId });
+        return true;
+    }
+
+    public isCurrentRebuildPageCommand(
+        conversationId: string,
+        rebuildId: string,
+        historyGeneration: string,
+        requestId: string,
+    ): boolean {
+        const current = this.rebuilds.get(conversationId);
+        return (
+            current?.rebuildId === rebuildId &&
+            current.historyGeneration === historyGeneration &&
+            current.pageRequestId === requestId &&
+            !this.quiescedConversations.has(conversationId)
+        );
+    }
+
+    public acceptsPageObservation(conversationId: string, requestId: string | undefined): boolean {
+        const current = this.rebuilds.get(conversationId);
+        return (
+            current === undefined ||
+            (typeof current.pageRequestId === "string" &&
+                current.pageRequestId.length > 0 &&
+                current.pageRequestId === requestId)
+        );
+    }
+
+    public capturePageObservation(
+        conversationId: string,
+        requestId: string | undefined,
+    ): string | null {
+        if (!this.acceptsPageObservation(conversationId, requestId)) return null;
+        const current = this.rebuilds.get(conversationId);
+        return current
+            ? `${current.rebuildId}:${current.historyGeneration ?? ""}:${current.pageRequestId ?? ""}`
+            : "normal";
+    }
+
+    public pageObservationIsCurrent(
+        conversationId: string,
+        requestId: string | undefined,
+        token: string | null,
+    ): boolean {
+        return token !== null && this.capturePageObservation(conversationId, requestId) === token;
+    }
+
+    public getRebuildId(conversationId: string): string | undefined {
+        return this.rebuilds.get(conversationId)?.rebuildId;
+    }
+
     /** Discovery ACK 只能为当前 epoch 中尚无 snapshot 的会话补齐 generation，绝不覆盖它。 */
     public acceptHistoryGeneration(
         conversationId: string,
@@ -195,7 +310,7 @@ export class OneTalkSyncLifecycle {
         this.lastError = code;
     }
 
-    public acceptAnchorSnapshot(anchors: OneTalkAnchor[]): string | undefined {
+    public acceptAnchorSnapshot(anchors: OneTalkAnchor[], requestId?: string): string | undefined {
         if (!this.brightAuthenticated) {
             this.emitDiagnostic({
                 event: "sync_guard",
@@ -205,6 +320,15 @@ export class OneTalkSyncLifecycle {
             return undefined;
         }
         const nextFingerprint = JSON.stringify(anchors);
+        for (const [conversationId, rebuild] of this.rebuilds) {
+            const anchor = anchors.find((candidate) => candidate.conversationId === conversationId);
+            if (
+                requestId === rebuild.rebuildId &&
+                anchor?.historyGeneration === rebuild.historyGeneration
+            ) {
+                this.rebuilds.delete(conversationId);
+            }
+        }
         if (this.anchorSnapshotReceived && this.snapshotFingerprint === nextFingerprint) {
             return undefined;
         }
@@ -264,6 +388,8 @@ export class OneTalkSyncLifecycle {
             this.snapshotFingerprint = undefined;
             this.anchorSnapshotReceived = false;
             this.anchors.clear();
+            this.rebuilds.clear();
+            this.quiescedConversations.clear();
         } else if (nextStatus !== "authenticated") {
             this.connectionEpoch += 1;
         }

apps/chrome-extension/src/onetalk/service-worker/sync-engine/model.ts

@@ -93,6 +93,7 @@ export type OneTalkSyncStartInput = {
     mode?: "full" | "incremental";
     anchorMessageId?: string | null;
     resumeFromTimeStamp?: number | null;
+    rebuildId?: string;
 };
 
 export type OneTalkSyncStartResult =
@@ -118,6 +119,17 @@ export type OneTalkSyncEngine = {
         channelAccountId: string,
     ) => Promise<OneTalkSyncObservationResult>;
     startSync: (input: OneTalkSyncStartInput) => Promise<OneTalkSyncStartResult>;
+    clearConversationHistory: (
+        channelAccountId: string,
+        conversationId: string,
+        rebuildId?: string,
+    ) => Promise<void>;
+    startRebuildSync: (input: {
+        channelAccountId: string;
+        conversationId: string;
+        rebuildId: string;
+        historyGeneration: string;
+    }) => Promise<OneTalkSyncStartResult>;
     resume: (conversationId?: string) => Promise<OneTalkSyncStartResult[]>;
     handlePageReady: (channelAccountId: string, conversationId?: string) => Promise<void>;
     handlePageDisconnected: () => void;

apps/chrome-extension/src/onetalk/service-worker/sync-engine/observation-pipeline.ts

@@ -199,6 +199,7 @@ export class ObservationPipeline {
                             group.conversationId,
                             [item.message],
                             hasPageAnomalies,
+                            message.historyRequestId,
                         ),
                     );
                 }
@@ -216,7 +217,11 @@ export class ObservationPipeline {
             }),
         );
         if (message.historyProgress) {
-            await this.persistConversationProgress(channelAccountId, message.historyProgress);
+            await this.persistConversationProgress(
+                channelAccountId,
+                message.historyProgress,
+                message.historyRequestId,
+            );
         }
         return {
             candidates: results.flatMap((result) => result.candidates),
@@ -248,12 +253,21 @@ export class ObservationPipeline {
         conversationId: string,
         messages: OneTalkObservedMessage[],
         hasPageAnomalies = false,
+        historyRequestId?: string,
     ): Promise<OneTalkSyncObservationResult> {
         if (this.lifecycle.isDisposed()) return { candidates: [], anomalies: [] };
+        const token = this.lifecycle.capturePageObservation(conversationId, historyRequestId);
+        if (token === null) return { candidates: [], anomalies: [] };
         const existing = await this.checkpoints.getCheckpoint(
             input.channelAccountId,
             conversationId,
         );
+        if (!this.lifecycle.pageObservationIsCurrent(conversationId, historyRequestId, token))
+            return { candidates: [], anomalies: [] };
+        if (this.lifecycle.isConversationQuiesced(conversationId))
+            return { candidates: [], anomalies: [] };
+        if (!this.lifecycle.acceptsPageObservation(conversationId, historyRequestId))
+            return { candidates: [], anomalies: [] };
         const key = scopedKey(input.channelAccountId, conversationId);
         const source = this.normalizeObservationSource(input, existing);
         const mode = input.mode ?? existing?.mode ?? this.checkpoints.getActiveMode(key) ?? "full";
@@ -268,6 +282,10 @@ export class ObservationPipeline {
                 this.checkpoints.getActiveAnchor(key) ?? null,
             );
         if (!existing) await this.store.putCheckpoint(checkpoint);
+        if (!this.lifecycle.pageObservationIsCurrent(conversationId, historyRequestId, token))
+            return { candidates: [], anomalies: [] };
+        if (this.lifecycle.isConversationQuiesced(conversationId))
+            return { candidates: [], anomalies: [] };
         const candidateMode =
             checkpoint.mode === "incremental" &&
             (isHistoryObservation(source) || checkpoint.anchorState === "awaiting_anchor")
@@ -281,10 +299,16 @@ export class ObservationPipeline {
             mode: candidateMode,
             receivedAt: input.receivedAt,
         });
+        if (!this.lifecycle.pageObservationIsCurrent(conversationId, historyRequestId, token))
+            return { candidates: [], anomalies: [] };
+        if (this.lifecycle.isConversationQuiesced(conversationId))
+            return { candidates: [], anomalies: [] };
         const persistedCandidates = await this.store.listCandidates(
             input.channelAccountId,
             conversationId,
         );
+        if (!this.lifecycle.pageObservationIsCurrent(conversationId, historyRequestId, token))
+            return { candidates: [], anomalies: [] };
         const hasPendingCandidate = persistedCandidates.some(pendingCandidate);
         await this.checkpoints.applyObservationResult({
             checkpoint,
@@ -292,6 +316,8 @@ export class ObservationPipeline {
             hasPendingCandidate,
             hasAnomalies: hasPageAnomalies || result.anomalies.length > 0,
         });
+        if (!this.lifecycle.pageObservationIsCurrent(conversationId, historyRequestId, token))
+            return { candidates: [], anomalies: [] };
         if (checkpoint.mode === "incremental" && checkpoint.anchorState === "found") {
             // The ACK coordinator activates candidates after the anchor is known.
             await this.onAnchorFound(
@@ -299,6 +325,8 @@ export class ObservationPipeline {
                 conversationId,
                 checkpoint.anchorMessageId,
             );
+            if (!this.lifecycle.pageObservationIsCurrent(conversationId, historyRequestId, token))
+                return { candidates: [], anomalies: [] };
         }
         for (const anomaly of result.anomalies) {
             this.lifecycle.setLastError(anomaly.code);
@@ -311,6 +339,7 @@ export class ObservationPipeline {
         conversationId: string,
         messages: OneTalkObservedMessage[],
         hasPageAnomalies = false,
+        historyRequestId?: string,
     ): Promise<OneTalkSyncObservationResult> {
         if (this.lifecycle.isDisposed()) return { candidates: [], anomalies: [] };
         const key = scopedKey(input.channelAccountId, conversationId);
@@ -318,7 +347,7 @@ export class ObservationPipeline {
             this.observationWrites.get(key) ??
             Promise.resolve({ candidates: [], anomalies: [] } as OneTalkSyncObservationResult);
         const current = previous.then(() =>
-            this.persistGroup(input, conversationId, messages, hasPageAnomalies),
+            this.persistGroup(input, conversationId, messages, hasPageAnomalies, historyRequestId),
         );
         this.observationWrites.set(key, current);
         try {
@@ -331,12 +360,30 @@ export class ObservationPipeline {
     private async persistConversationProgress(
         channelAccountId: string,
         progress: Parameters<CheckpointCoordinator["applyHistoryProgress"]>[1],
+        historyRequestId?: string,
     ): Promise<void> {
-        if (this.lifecycle.isDisposed()) return;
+        if (
+            this.lifecycle.isDisposed() ||
+            this.lifecycle.isConversationQuiesced(progress.conversationId) ||
+            !this.lifecycle.acceptsPageObservation(progress.conversationId, historyRequestId)
+        )
+            return;
         const key = scopedKey(channelAccountId, progress.conversationId);
+        const token = this.lifecycle.capturePageObservation(progress.conversationId, historyRequestId);
+        if (token === null) return;
         const previous = this.observationWrites.get(key) ?? Promise.resolve();
         const current = previous.then(async () => {
+            if (!this.lifecycle.acceptsPageObservation(progress.conversationId, historyRequestId))
+                return;
             const result = await this.checkpoints.applyHistoryProgress(channelAccountId, progress);
+            if (
+                !this.lifecycle.pageObservationIsCurrent(
+                    progress.conversationId,
+                    historyRequestId,
+                    token,
+                )
+            )
+                return;
             await this.onHistoryProgress(channelAccountId, progress, result);
         });
         this.observationWrites.set(key, current);

apps/chrome-extension/src/onetalk/service-worker/sync-runtime.ts

@@ -7,6 +7,7 @@ import { createOneTalkServiceWorkerFrameRouter } from "./routing/frame-router.ts
 import { type OneTalkPageDiagnostic, type OneTalkServiceWorkerRuntime } from "./runtime.ts";
 import { OneTalkPageRuntimeHost } from "./page-runtime-host.ts";
 import { createOneTalkSendCommandFlow } from "./flows/send-command-flow.ts";
+import { createOneTalkHistoryRebuildFlow } from "./flows/history-rebuild-flow.ts";
 import {
     createOneTalkSyncEngine,
     type OneTalkSyncEngine,
@@ -89,10 +90,17 @@ export const createOneTalkServiceWorkerSyncRuntime = (
         onPageDiagnostic: options.onPageDiagnostic,
         onError: options.onError ?? (() => undefined),
     });
+    const rebuild = createOneTalkHistoryRebuildFlow({
+        scope: options.scope,
+        bright: options.bright,
+        sync: engine,
+        onError: options.onError ?? (() => undefined),
+    });
     const router = createOneTalkServiceWorkerFrameRouter({
         sync: engine,
         ...(profile === undefined ? {} : { profile }),
         send,
+        rebuild,
     });
     const unsubscribeRouter = options.bright.subscribe((frame) => router.handle(frame));
     return {

apps/chrome-extension/src/onetalk/service-worker/transport/bright-client.ts

@@ -288,7 +288,7 @@ export const createOneTalkBrightClient = (
     const now = options.now ?? Date.now;
     const createRequestId = options.createRequestId ?? defaultRequestId();
     const createSocket = options.webSocket ?? defaultWebSocket;
-    const requestedPermissions = options.requestedPermissions ?? ["read"];
+    const requestedPermissions = options.requestedPermissions ?? ["read", "rebuild"];
     const heartbeatInterval = options.heartbeatIntervalMs ?? DEFAULT_BRIGHT_HEARTBEAT_INTERVAL_MS;
     const reconnectDelay = options.reconnectDelayMs ?? DEFAULT_BRIGHT_RECONNECT_DELAY_MS;
     const autoReconnect = options.autoReconnect ?? true;

apps/chrome-extension/test/onetalk-bright-client.test.js

@@ -106,7 +106,7 @@ test("authenticates, receives anchors, uploads observations, and heartbeats", as
         type: "ws.hello",
         requestId: "hello-1",
         scope,
-        payload: { binding: "binding-1", requestedPermissions: ["read"] },
+        payload: { binding: "binding-1", requestedPermissions: ["read", "rebuild"] },
     });
 
     sockets[0].receive(accepted());
@@ -353,7 +353,7 @@ test("keeps ws.error visible and uses a browser-valid client close code", () =>
     });
     assert.equal(hello.protocolVersion, ONETALK_PROTOCOL_VERSION);
     assert.equal(hello.connectionType, "plugin");
-    assert.deepEqual(hello.requestedPermissions, ["read"]);
+    assert.deepEqual(hello.requestedPermissions, ["read", "rebuild"]);
     assert.equal(hello.bindingPresent, true);
     assert.ok(hello.redactedFields.includes("binding"));
     const close = diagnostics.find((event) => event.event === "socket_close");

apps/chrome-extension/test/onetalk-sync-engine.test.js

@@ -574,6 +574,91 @@ test("does not replace an anchor generation with a late discovery generation", (
     assert.equal(lifecycle.getHistoryGeneration("conversation-1"), "generation-2");
 });
 
+test("allows a second rebuild after release while rejecting the first rebuild's late sync command", () => {
+    const bright = new FakeBright();
+    const lifecycle = new OneTalkSyncLifecycle(
+        { bright, getActiveConversationCount: () => 0 },
+        { onBrightStatusChanged: () => undefined },
+    );
+
+    assert.equal(lifecycle.quiesceConversation("conversation-1", "rebuild-1"), true);
+    assert.equal(
+        lifecycle.releaseConversation("conversation-1", "rebuild-1", "generation-1"),
+        true,
+    );
+    assert.equal(lifecycle.quiesceConversation("conversation-1", "rebuild-2"), true);
+    assert.equal(
+        lifecycle.releaseConversation("conversation-1", "rebuild-1", "generation-1"),
+        false,
+    );
+    assert.equal(
+        lifecycle.releaseConversation("conversation-1", "rebuild-2", "generation-2"),
+        true,
+    );
+    assert.equal(lifecycle.getHistoryGeneration("conversation-1"), "generation-2");
+});
+
+test("clears a rebuild token from a duplicate matching completion snapshot", () => {
+    const bright = new FakeBright();
+    const lifecycle = new OneTalkSyncLifecycle(
+        { bright, getActiveConversationCount: () => 0 },
+        { onBrightStatusChanged: () => undefined },
+    );
+    lifecycle.start();
+    bright.emitStatus("authenticated");
+    const anchors = [
+        {
+            conversationId: "conversation-1",
+            historyGeneration: "generation-1",
+            latestMessageId: null,
+        },
+    ];
+    assert.equal(lifecycle.quiesceConversation("conversation-1", "rebuild-1"), true);
+    assert.equal(
+        lifecycle.releaseConversation("conversation-1", "rebuild-1", "generation-1"),
+        true,
+    );
+    assert.equal(
+        lifecycle.bindRebuildPageCommand("conversation-1", "rebuild-1", "generation-1", "page-1"),
+        true,
+    );
+    lifecycle.acceptAnchorSnapshot(anchors);
+    assert.equal(lifecycle.acceptsPageObservation("conversation-1", undefined), false);
+    lifecycle.acceptAnchorSnapshot(anchors, "rebuild-1");
+    assert.equal(lifecycle.acceptsPageObservation("conversation-1", undefined), true);
+});
+
+test("allows a rebuild retry after Bright disconnects before the sync release", () => {
+    const bright = new FakeBright();
+    const lifecycle = new OneTalkSyncLifecycle(
+        { bright, getActiveConversationCount: () => 0 },
+        { onBrightStatusChanged: () => undefined },
+    );
+    lifecycle.start();
+    bright.emitStatus("authenticated");
+
+    assert.equal(lifecycle.quiesceConversation("conversation-1", "rebuild-1"), true);
+    bright.emitStatus("offline");
+    bright.emitStatus("authenticated");
+
+    assert.equal(lifecycle.quiesceConversation("conversation-1", "rebuild-2"), true);
+});
+
+test("does not accept an untagged live observation while a released rebuild awaits its page command", () => {
+    const bright = new FakeBright();
+    const lifecycle = new OneTalkSyncLifecycle(
+        { bright, getActiveConversationCount: () => 0 },
+        { onBrightStatusChanged: () => undefined },
+    );
+
+    assert.equal(lifecycle.quiesceConversation("conversation-1", "rebuild-1"), true);
+    assert.equal(
+        lifecycle.releaseConversation("conversation-1", "rebuild-1", "generation-1"),
+        true,
+    );
+    assert.equal(lifecycle.acceptsPageObservation("conversation-1", undefined), false);
+});
+
 test("holds incremental candidates before the anchor and completes after ACK", async () => {
     const store = new MemoryStore();
     const bright = new FakeBright();

apps/chrome-extension/test/onetalk-sync-storage.test.js

@@ -58,6 +58,10 @@ class FakeStore {
         this.records.clear();
     }
 
+    delete(key) {
+        this.records.delete(key);
+    }
+
     openCursor() {
         const entries = [...this.records.entries()];
         const request = new FakeRequest(null, false);
@@ -272,6 +276,61 @@ test("preserves existing stores while adding the bootstrap marker store from v7"
     }
 });
 
+test("clears only one conversation history ledger after its transaction commits", async () => {
+    const factory = new FakeFactory();
+    const store = createOneTalkSyncStore(factory, () => 500);
+    await store.persistObservedBatch({
+        channelAccountId: "account-1",
+        conversationId: "conversation-1",
+        messages: [validMessage],
+        observationSource: "history",
+        mode: "full",
+    });
+    await store.persistObservedBatch({
+        channelAccountId: "account-1",
+        conversationId: "conversation-2",
+        messages: [{ ...validMessage, conversationId: "conversation-2", messageId: "message-2" }],
+        observationSource: "history",
+        mode: "full",
+    });
+    await store.putCheckpoint(checkpoint);
+    await store.putCheckpoint({
+        ...checkpoint,
+        key: JSON.stringify(["account-1", "conversation-2"]),
+        conversationId: "conversation-2",
+    });
+    await store.recordAnomaly({
+        key: "conversation-anomaly",
+        channelAccountId: "account-1",
+        conversationId: "conversation-1",
+        code: "invalid",
+        observationSource: "sync",
+        fields: [],
+        occurrenceCount: 1,
+        firstObservedAt: 1,
+        lastObservedAt: 1,
+    });
+    await store.recordAnomaly({
+        key: "account-anomaly",
+        channelAccountId: "account-1",
+        code: "invalid",
+        observationSource: "sync",
+        fields: [],
+        occurrenceCount: 1,
+        firstObservedAt: 1,
+        lastObservedAt: 1,
+    });
+
+    await store.clearConversationHistory("account-1", "conversation-1");
+
+    assert.equal(await store.getCheckpoint("account-1", "conversation-1"), null);
+    assert.deepEqual(await store.listCandidates("account-1", "conversation-1"), []);
+    assert.deepEqual(await store.listAnomalies("account-1", "conversation-1"), []);
+    assert.notEqual(await store.getCheckpoint("account-1", "conversation-2"), null);
+    assert.equal((await store.listCandidates("account-1", "conversation-2")).length, 1);
+    assert.equal((await store.listAnomalies("account-1")).length, 1);
+});
+
 test("stores bootstrap markers by account and stable migration identifier", async () => {
     const factory = new FakeFactory();
     const store = createOneTalkConversationBootstrapStore(factory);

apps/server/src/app.ts

@@ -127,6 +127,7 @@ export const createApp = (
         readService,
         mindPageOrigin,
         cutoverPolicy,
+        registry: oneTalkRegistry,
     });
     app.addHook("onClose", async () => database.close());
 

apps/server/src/http/onetalk/contract.ts

@@ -33,7 +33,12 @@ export type BrightReadErrorCode =
     | "invalid_cursor"
     | "invalid_limit"
     | "invalid_time_range"
-    | "scope_mismatch";
+    | "scope_mismatch"
+    | "rebuild_in_progress"
+    | "plugin_offline"
+    | "plugin_heartbeat_stale"
+    | "plugin_reset_rejected"
+    | "plugin_reset_timeout";
 
 export type BrightReadErrorResponse = {
     error: {

apps/server/src/http/onetalk/public-context.ts

@@ -6,14 +6,16 @@ import {
     type OneTalkAuthorizationDecision,
     type OneTalkAuthorizationReader,
     type OneTalkMindScope,
+    type OneTalkPermission,
 } from "@trade-message-center/onetalk-contract";
 
 import type { OneTalkCutoverPolicy } from "../../cutover-policy.ts";
 import type { OneTalkReadService } from "../../onetalk/index.ts";
 import { sendError } from "./read.ts";
+import type { OneTalkConnectionRegistry } from "../../websocket/registry.ts";
 
 const CORS_ALLOWED_REQUEST_HEADERS = new Set(["content-type"]);
-const CORS_ALLOWED_REQUEST_METHOD = "GET";
+const CORS_ALLOWED_REQUEST_METHODS = new Set(["GET", "POST"]);
 
 type AuthorizationResult =
     | { ok: true; scope: OneTalkMindScope }
@@ -38,6 +40,7 @@ export type BrightReadRouteOptions = {
     readService: OneTalkReadService;
     mindPageOrigin?: string;
     cutoverPolicy: OneTalkCutoverPolicy;
+    registry?: Pick<OneTalkConnectionRegistry, "requestHistoryRebuild">;
 };
 
 export type PublicReadContext = {
@@ -47,6 +50,12 @@ export type PublicReadContext = {
         reply: FastifyReply,
         channelAccountId: string,
     ) => Promise<AuthorizedReadScopeResult>;
+    authorizeRebuildScope: (
+        request: FastifyRequest,
+        reply: FastifyReply,
+        channelAccountId: string,
+    ) => Promise<AuthorizedReadScopeResult>;
+    registry?: Pick<OneTalkConnectionRegistry, "requestHistoryRebuild">;
     requestIsAdmitted: (request: FastifyRequest) => boolean;
     mindOriginGuard: (request: FastifyRequest, reply: FastifyReply) => Promise<void>;
 };
@@ -66,6 +75,7 @@ const authorizeRead = async (
     request: FastifyRequest,
     authorization: OneTalkAuthorizationReader,
     channelAccountId: string,
+    requiredPermission: OneTalkPermission,
 ): Promise<AuthorizationResult> => {
     if (channelAccountId.trim() === "") {
         return { ok: false, statusCode: 403, code: "scope_mismatch" };
@@ -75,7 +85,7 @@ const authorizeRead = async (
     try {
         decision = await authorization.authorize({
             connectionType: "mind_page",
-            operation: "read",
+            operation: requiredPermission === "rebuild" ? "rebuild" : "read",
             scope: { channelAccountId },
             ...(cookie === undefined ? {} : { cookie }),
         });
@@ -87,7 +97,7 @@ const authorizeRead = async (
     if (decision.mindScope.channelAccountId !== channelAccountId) {
         return { ok: false, statusCode: 403, code: "scope_mismatch" };
     }
-    if (!decision.permissions.includes("read")) {
+    if (!decision.permissions.includes(requiredPermission)) {
         return { ok: false, statusCode: 403, code: "authorization_rejected" };
     }
     return { ok: true, scope: decision.mindScope };
@@ -126,7 +136,7 @@ const allowsCorsRequestHeaders = (value: string | string[] | undefined): boolean
 
 const allowsCorsRequestMethod = (value: string | string[] | undefined): boolean => {
     if (value === undefined) return true;
-    return value === CORS_ALLOWED_REQUEST_METHOD;
+    return typeof value === "string" && CORS_ALLOWED_REQUEST_METHODS.has(value);
 };
 
 /** 安装 public 读取 CORS 预检路由。 */
@@ -149,7 +159,7 @@ export const installPublicPreflight = (
         reply.header("access-control-allow-origin", options.mindPageOrigin);
         reply.header("access-control-allow-credentials", "true");
         reply.header("vary", "Origin");
-        reply.header("access-control-allow-methods", "GET,OPTIONS");
+        reply.header("access-control-allow-methods", "GET,POST,OPTIONS");
         reply.header("access-control-allow-headers", "content-type");
         return reply.code(204).send();
     });
@@ -183,6 +193,30 @@ export const createPublicReadContext = (options: BrightReadRouteOptions): Public
             request,
             options.authorization,
             channelAccountId,
+            "read",
+        );
+        if (!authorizationResult.ok) {
+            return {
+                ok: false,
+                response: sendError(
+                    reply,
+                    authorizationResult.statusCode,
+                    authorizationResult.code,
+                ),
+            };
+        }
+        return { ok: true, scope: authorizationResult.scope };
+    };
+    const authorizeRebuildScope = async (
+        request: FastifyRequest,
+        reply: FastifyReply,
+        channelAccountId: string,
+    ): Promise<AuthorizedReadScopeResult> => {
+        const authorizationResult = await authorizeRead(
+            request,
+            options.authorization,
+            channelAccountId,
+            "rebuild",
         );
         if (!authorizationResult.ok) {
             return {
@@ -199,6 +233,8 @@ export const createPublicReadContext = (options: BrightReadRouteOptions): Public
     return {
         readService: options.readService,
         authorizeRequestScope,
+        authorizeRebuildScope,
+        ...(options.registry === undefined ? {} : { registry: options.registry }),
         requestIsAdmitted,
         mindOriginGuard,
     };

apps/server/src/http/onetalk/public.ts

@@ -5,6 +5,7 @@ import type { FastifyInstance } from "fastify";
 import { installConversationListRoute } from "./conversation-list.ts";
 import { installConversationRoute } from "./conversation.ts";
 import { installMessagesRoute } from "./messages.ts";
+import { installHistoryRebuildRoute } from "./rebuild.ts";
 import {
     createPublicReadContext,
     installPublicPreflight,
@@ -23,4 +24,5 @@ export const installOneTalkReadRoutes = (
     installConversationListRoute(app, context);
     installConversationRoute(app, context);
     installMessagesRoute(app, context);
+    installHistoryRebuildRoute(app, context);
 };

apps/server/src/http/onetalk/rebuild.ts

@@ -0,0 +1,51 @@
+// 注册 Mind 发起的单会话历史重建 HTTP 边界。
+
+import type { FastifyInstance } from "fastify";
+import { ONETALK_HISTORY_REBUILD_ROUTE } from "@trade-message-center/onetalk-contract";
+
+import type { PublicReadContext } from "./public-context.ts";
+import { sendError } from "./read.ts";
+
+type ConversationParams = { channelAccountId: string; conversationId: string };
+
+const statusFor = (reason: Parameters<typeof sendError>[2]): 404 | 409 | 503 => {
+    if (reason === "conversation_not_found") return 404;
+    if (reason === "rebuild_in_progress") return 409;
+    return 503;
+};
+
+/** 成功仅意味着 Bright reset 已提交；post-commit full sync 以 resync 字段单独呈现。 */
+export const installHistoryRebuildRoute = (
+    app: FastifyInstance,
+    context: PublicReadContext,
+): void => {
+    app.post<{ Params: ConversationParams }>(
+        ONETALK_HISTORY_REBUILD_ROUTE,
+        { preHandler: context.mindOriginGuard },
+        async (request, reply) => {
+            const authorized = await context.authorizeRebuildScope(
+                request,
+                reply,
+                request.params.channelAccountId,
+            );
+            if (!authorized.ok) return authorized.response;
+            if (!context.requestIsAdmitted(request))
+                return sendError(reply, 503, "authorization_unavailable");
+            if (!context.registry) return sendError(reply, 503, "database_unavailable");
+            const result = await context.registry.requestHistoryRebuild(
+                authorized.scope,
+                request.params.conversationId,
+            );
+            if (!context.requestIsAdmitted(request))
+                return sendError(reply, 503, "authorization_unavailable");
+            if (!result.ok) return sendError(reply, statusFor(result.reason), result.reason);
+            return reply.send({
+                scope: authorized.scope,
+                conversationId: request.params.conversationId,
+                rebuildId: result.rebuildId,
+                status: "server_reset_committed",
+                resync: result.resync,
+            });
+        },
+    );
+};

apps/server/src/onetalk/model.ts

@@ -83,6 +83,7 @@ export type OneTalkAnomalyInput = {
 export type OneTalkSyncCompletionInput = {
     conversationId: string;
     historyGeneration: OneTalkHistoryGeneration;
+    rebuildId?: string;
     mode: OneTalkSyncMode;
     historyComplete: boolean;
     result: OneTalkSyncResult;
@@ -184,6 +185,13 @@ export type OneTalkRepository = {
         guard: OneTalkCommitGuard,
     ) => Promise<OneTalkObservationResult[]>;
     guardedRecordAnomaly: (input: OneTalkAnomalyInput, guard: OneTalkCommitGuard) => Promise<void>;
+    guardedRecordSyncAnomaly?: (
+        context: OneTalkSourceContext,
+        conversationId: string,
+        historyGeneration: OneTalkHistoryGeneration,
+        input: OneTalkAnomalyInput,
+        guard: OneTalkCommitGuard,
+    ) => Promise<boolean>;
     recordAnomaly: (input: OneTalkAnomalyInput) => Promise<void>;
     updateSyncState: (
         context: OneTalkSourceContext,
@@ -196,6 +204,12 @@ export type OneTalkRepository = {
         conversationId: string,
         guard: OneTalkCommitGuard,
     ) => Promise<OneTalkConversationState | null>;
+    resetConversationHistory?: (
+        context: OneTalkSourceContext,
+        conversationId: string,
+        historyGeneration: OneTalkHistoryGeneration,
+        guard: OneTalkCommitGuard,
+    ) => Promise<OneTalkConversationState | null>;
 };
 
 export type OneTalkMessageNormalization =
@@ -247,6 +261,12 @@ export type OneTalkService = {
         cursor: OneTalkHistoryCursor | null,
         limit: number,
     ) => Promise<OneTalkHistoryPage | null>;
+    resetConversationHistory?: (
+        context: OneTalkSourceContext,
+        conversationId: string,
+        historyGeneration: OneTalkHistoryGeneration,
+        commitGuard: OneTalkCommitGuard,
+    ) => Promise<OneTalkConversationState | null>;
     observeMessage: (
         context: OneTalkSourceContext,
         observationSource: OneTalkObservationSource,

apps/server/src/onetalk/repository.ts

@@ -665,6 +665,43 @@ const guardedRecordAnomaly = async (
     });
 };
 
+/** 在同一 conversation 行锁内验证 generation 后写入 sync anomaly。 */
+const guardedRecordSyncAnomaly = async (
+    database: Database,
+    context: OneTalkSourceContext,
+    conversationId: string,
+    historyGeneration: OneTalkHistoryGeneration,
+    input: OneTalkAnomalyInput,
+    guard: OneTalkCommitGuard,
+): Promise<boolean> => {
+    return database.transaction(async (transaction) => {
+        guard.assertValid();
+        const rows = await transaction
+            .select({ historyGeneration: onetalkConversation.historyGeneration })
+            .from(onetalkConversation)
+            .where(conversationCondition(context, conversationId))
+            .limit(1)
+            .for("update");
+        guard.assertValid();
+        if (rows[0]?.historyGeneration !== historyGeneration) return false;
+        const now = new Date();
+        await transaction
+            .insert(onetalkMessageAnomaly)
+            .values(anomalyValues(input))
+            .onConflictDoUpdate({
+                target: onetalkMessageAnomaly.fingerprint,
+                set: {
+                    lastSeenAt: now,
+                    occurrenceCount: sql`${onetalkMessageAnomaly.occurrenceCount} + 1`,
+                    missingFields: [...input.missingFields],
+                    payload: toJsonValue(input.payload),
+                },
+            });
+        guard.assertValid();
+        return true;
+    });
+};
+
 const updateSyncState = async (
     database: Database,
     context: OneTalkSourceContext,
@@ -724,6 +761,64 @@ const guardedUpdateSyncState = (
 ): Promise<OneTalkConversationState | null> =>
     updateSyncState(database, context, update, conversationId, guard);
 
+/** 在 conversation 行锁内删除目标历史事实并切换持久化 generation。 */
+const resetConversationHistory = async (
+    database: Database,
+    context: OneTalkSourceContext,
+    conversationId: string,
+    historyGeneration: OneTalkHistoryGeneration,
+    guard: OneTalkCommitGuard,
+): Promise<OneTalkConversationState | null> => {
+    const now = new Date();
+    return database.transaction(async (transaction) => {
+        guard.assertValid();
+        const conversations = await transaction
+            .select({ conversationKind: onetalkConversation.conversationKind })
+            .from(onetalkConversation)
+            .where(conversationCondition(context, conversationId))
+            .limit(1)
+            .for("update");
+        guard.assertValid();
+        if (conversations.length === 0 || conversations[0]?.conversationKind !== "direct")
+            return null;
+        await transaction
+            .delete(onetalkMessage)
+            .where(
+                and(
+                    eq(onetalkMessage.channelAccountId, context.channelAccountId),
+                    eq(onetalkMessage.conversationId, conversationId),
+                ),
+            );
+        guard.assertValid();
+        await transaction
+            .delete(onetalkMessageAnomaly)
+            .where(
+                and(
+                    eq(onetalkMessageAnomaly.channelAccountId, context.channelAccountId),
+                    eq(onetalkMessageAnomaly.conversationId, conversationId),
+                ),
+            );
+        guard.assertValid();
+        const rows = await transaction
+            .update(onetalkConversation)
+            .set({
+                historyGeneration,
+                lastMessageAtMs: null,
+                latestMessageId: null,
+                historyComplete: false,
+                messageCount: 0,
+                anchorUpdatedAt: null,
+                syncPhase: "initial",
+                syncResult: "incomplete",
+                lastObservedAt: now,
+            })
+            .where(conversationCondition(context, conversationId))
+            .returning();
+        guard.assertValid();
+        return rows[0] ? toConversation(rows[0]) : null;
+    });
+};
+
 /** 创建使用 Drizzle 的 OneTalk repository。 */
 export const createOneTalkRepository = (database: Database): OneTalkRepository => {
     const withDatabaseError = async <T>(operation: () => Promise<T>): Promise<T> => {
@@ -807,11 +902,32 @@ export const createOneTalkRepository = (database: Database): OneTalkRepository =
         recordAnomaly: (input) => withDatabaseError(() => recordAnomaly(database, input)),
         guardedRecordAnomaly: (input, guard) =>
             withDatabaseError(() => guardedRecordAnomaly(database, input, guard)),
+        guardedRecordSyncAnomaly: (context, conversationId, historyGeneration, input, guard) =>
+            withDatabaseError(() =>
+                guardedRecordSyncAnomaly(
+                    database,
+                    context,
+                    conversationId,
+                    historyGeneration,
+                    input,
+                    guard,
+                ),
+            ),
         updateSyncState: (context, update, conversationId) =>
             withDatabaseError(() => updateSyncState(database, context, update, conversationId)),
         guardedUpdateSyncState: (context, update, conversationId, guard) =>
             withDatabaseError(() =>
                 guardedUpdateSyncState(database, context, update, conversationId, guard),
             ),
+        resetConversationHistory: (context, conversationId, historyGeneration, guard) =>
+            withDatabaseError(() =>
+                resetConversationHistory(
+                    database,
+                    context,
+                    conversationId,
+                    historyGeneration,
+                    guard,
+                ),
+            ),
     };
 };

apps/server/src/onetalk/service.ts

@@ -5,6 +5,7 @@ import { createHash } from "node:crypto";
 import type {
     OneTalkConversationDiscoveryEntry,
     OneTalkConversationType,
+    OneTalkHistoryGeneration,
     OneTalkJsonValue,
     OneTalkMessage,
     OneTalkObservedMessage,
@@ -14,6 +15,7 @@ import { isOneTalkMessage } from "@trade-message-center/onetalk-contract";
 
 import {
     type OneTalkAnomalyInput,
+    OneTalkDatabaseError,
     type OneTalkCommitGuard,
     type OneTalkConversationState,
     type OneTalkHistoryCursor,
@@ -149,8 +151,21 @@ const completeSync = async (
 
     const syncAnomaly = syncAnomalyFor(context, completion);
     if (syncAnomaly) {
-        if (commitGuard) await repository.guardedRecordAnomaly(syncAnomaly, commitGuard);
-        else await repository.recordAnomaly(syncAnomaly);
+        if (commitGuard) {
+            if (!repository.guardedRecordSyncAnomaly)
+                return { status: "rejected", reason: "conversation_not_discovered" };
+            if (
+                !(await repository.guardedRecordSyncAnomaly(
+                    context,
+                    completion.conversationId,
+                    completion.historyGeneration,
+                    syncAnomaly,
+                    commitGuard,
+                ))
+            ) {
+                return { status: "rejected", reason: "conversation_not_discovered" };
+            }
+        } else await repository.recordAnomaly(syncAnomaly);
         commitGuard?.assertValid();
     }
 
@@ -210,8 +225,21 @@ const completeSync = async (
                     historyComplete: completion.historyComplete,
                 },
             };
-            if (commitGuard) await repository.guardedRecordAnomaly(missingLatest, commitGuard);
-            else await repository.recordAnomaly(missingLatest);
+            if (commitGuard) {
+                if (!repository.guardedRecordSyncAnomaly)
+                    return { status: "rejected", reason: "conversation_not_discovered" };
+                if (
+                    !(await repository.guardedRecordSyncAnomaly(
+                        context,
+                        completion.conversationId,
+                        completion.historyGeneration,
+                        missingLatest,
+                        commitGuard,
+                    ))
+                ) {
+                    return { status: "rejected", reason: "conversation_not_discovered" };
+                }
+            } else await repository.recordAnomaly(missingLatest);
             commitGuard?.assertValid();
             result = "incomplete";
             historyComplete = false;
@@ -316,6 +344,22 @@ export const createOneTalkService = (repository: OneTalkRepository): OneTalkServ
         return repository.listMessagesForRead(scope, conversationId, cursor, limit);
     };
 
+    const resetConversationHistory = (
+        context: OneTalkSourceContext,
+        conversationId: string,
+        historyGeneration: OneTalkHistoryGeneration,
+        commitGuard: OneTalkCommitGuard,
+    ): Promise<OneTalkConversationState | null> => {
+        if (!repository.resetConversationHistory)
+            throw new OneTalkDatabaseError("reset unavailable");
+        return repository.resetConversationHistory(
+            context,
+            conversationId,
+            historyGeneration,
+            commitGuard,
+        );
+    };
+
     const observeMessage = async (
         context: OneTalkSourceContext,
         observationSource: OneTalkObservationSource,
@@ -383,6 +427,7 @@ export const createOneTalkService = (repository: OneTalkRepository): OneTalkServ
         listConversations,
         readConversation,
         readHistory,
+        resetConversationHistory,
         observeMessage,
         observeMessages,
         completeSync: (context, completion, commitGuard) =>

apps/server/src/websocket/authenticated-router.ts

@@ -99,6 +99,18 @@ export const ONETALK_AUTHENTICATED_ROUTE_DEFINITIONS = [
         authorization: "session",
         operation: "sync",
     },
+    {
+        type: "storage.delete.ack",
+        connectionType: "plugin",
+        authorization: "session",
+        operation: "rebuild",
+    },
+    {
+        type: "history.sync.ack",
+        connectionType: "plugin",
+        authorization: "session",
+        operation: "rebuild",
+    },
     {
         type: "send.request",
         connectionType: "mind_page",

apps/server/src/websocket/history-rebuild-coordinator.ts

@@ -0,0 +1,365 @@
+// 串行化单会话 rebuild 的 plugin clear、Bright reset 与 post-commit release。
+
+import { randomUUID } from "node:crypto";
+
+import {
+    createOneTalkAnchorSnapshotFrame,
+    createOneTalkHistorySyncCommandFrame,
+    createOneTalkStorageDeleteCommandFrame,
+    isSameOneTalkScope,
+    type OneTalkAuthorizationReader,
+    type OneTalkFrame,
+    type OneTalkMindScope,
+} from "@trade-message-center/onetalk-contract";
+
+import type {
+    OneTalkConversationState,
+    OneTalkService,
+    OneTalkSourceContext,
+    OneTalkSyncCompletionInput,
+} from "../onetalk/index.ts";
+import type { OneTalkCommitGuard, OneTalkRegisteredConnection } from "./connection-store.ts";
+
+type RebuildFailure =
+    | "conversation_not_found"
+    | "rebuild_in_progress"
+    | "plugin_offline"
+    | "plugin_heartbeat_stale"
+    | "plugin_reset_rejected"
+    | "plugin_reset_timeout"
+    | "database_unavailable";
+
+export type OneTalkHistoryRebuildResult =
+    | {
+          ok: true;
+          rebuildId: string;
+          historyGeneration: string;
+          resync: { status: "started" } | { status: "failed"; reason: string };
+      }
+    | { ok: false; rebuildId: string; reason: RebuildFailure };
+
+type PendingAck = {
+    connection: OneTalkPluginConnection;
+    conversationId: string;
+    stage: "clear" | "sync";
+    settle: (result: "accepted" | "rejected" | "timeout") => void;
+};
+
+type OneTalkPluginConnection = Extract<OneTalkRegisteredConnection, { connectionType: "plugin" }>;
+
+const sourceContext = (connection: OneTalkRegisteredConnection): OneTalkSourceContext | null => {
+    if (connection.connectionType !== "plugin") return null;
+    return {
+        binding: connection.binding,
+        mindUserId: connection.mindScope.mindUserId,
+        workspaceId: connection.mindScope.workspaceId,
+        channelAccountId: connection.scope.channelAccountId,
+        deviceId: connection.scope.deviceId,
+    };
+};
+
+/** 进程内 owner：每个 account/conversation 同时最多一个重建，重启后允许安全重试。 */
+export const createOneTalkHistoryRebuildCoordinator = (options: {
+    authorization: OneTalkAuthorizationReader;
+    service: OneTalkService;
+    findFreshPlugin: (scope: OneTalkMindScope) => OneTalkPluginConnection | null;
+    isCurrentPlugin: (connection: OneTalkPluginConnection) => boolean;
+    sendPluginFrame: (connection: OneTalkPluginConnection, frame: OneTalkFrame) => boolean;
+    createCommitGuard: (connection: OneTalkPluginConnection) => OneTalkCommitGuard;
+    publishStatus: (input: {
+        scope: OneTalkMindScope;
+        conversationId: string;
+        rebuildId: string;
+        stage: Extract<OneTalkFrame, { type: "rebuild.status" }>["payload"]["stage"];
+        reason?: Extract<OneTalkFrame, { type: "rebuild.status" }>["payload"]["reason"];
+    }) => Promise<void>;
+    timeoutMs?: number;
+}): {
+    request: (
+        scope: OneTalkMindScope,
+        conversationId: string,
+    ) => Promise<OneTalkHistoryRebuildResult>;
+    handleFrame: (connection: OneTalkRegisteredConnection, frame: OneTalkFrame) => void;
+    handleCompletion: (
+        connection: OneTalkRegisteredConnection,
+        input: { completion: OneTalkSyncCompletionInput; conversation: OneTalkConversationState },
+    ) => Promise<void>;
+} => {
+    const active = new Set<string>();
+    const pending = new Map<string, PendingAck>();
+    const committed = new Map<
+        string,
+        { rebuildId: string; historyGeneration: string; connection: OneTalkPluginConnection }
+    >();
+    const timeoutMs = options.timeoutMs ?? 10_000;
+    const operationKey = (scope: OneTalkMindScope, conversationId: string) =>
+        JSON.stringify([scope.channelAccountId, conversationId]);
+    const authorizePlugin = async (connection: OneTalkRegisteredConnection): Promise<boolean> => {
+        if (connection.connectionType !== "plugin" || !options.isCurrentPlugin(connection))
+            return false;
+        try {
+            const decision = await options.authorization.authorize({
+                connectionType: "plugin",
+                operation: "rebuild",
+                scope: connection.scope,
+                binding: connection.binding,
+            });
+            return (
+                decision.allowed &&
+                isSameOneTalkScope(decision.mindScope, connection.mindScope) &&
+                decision.binding === connection.binding &&
+                decision.authorizationVersion === connection.authorizationVersion &&
+                decision.permissions.includes("rebuild") &&
+                connection.permissions.includes("rebuild") &&
+                options.isCurrentPlugin(connection)
+            );
+        } catch {
+            return false;
+        }
+    };
+    const awaitAck = (
+        rebuildId: string,
+        connection: OneTalkPluginConnection,
+        conversationId: string,
+        stage: PendingAck["stage"],
+    ): Promise<"accepted" | "rejected" | "timeout"> =>
+        new Promise((resolve) => {
+            const timer = setTimeout(() => {
+                pending.delete(rebuildId);
+                resolve("timeout");
+            }, timeoutMs);
+            pending.set(rebuildId, {
+                connection,
+                conversationId,
+                stage,
+                settle: (result) => {
+                    clearTimeout(timer);
+                    pending.delete(rebuildId);
+                    resolve(result);
+                },
+            });
+        });
+    const publishFailure = async (
+        scope: OneTalkMindScope,
+        conversationId: string,
+        rebuildId: string,
+        reason: RebuildFailure,
+    ): Promise<void> =>
+        options.publishStatus({ scope, conversationId, rebuildId, stage: "rejected", reason });
+    const publishPostCommitFailure = async (
+        scope: OneTalkMindScope,
+        conversationId: string,
+        rebuildId: string,
+    ): Promise<void> => {
+        try {
+            await options.publishStatus({
+                scope,
+                conversationId,
+                rebuildId,
+                stage: "sync_failed",
+                reason: "sync_start_failed",
+            });
+        } catch {
+            // Post-commit status is observational and cannot change HTTP success.
+        }
+    };
+
+    const request = async (
+        scope: OneTalkMindScope,
+        conversationId: string,
+    ): Promise<OneTalkHistoryRebuildResult> => {
+        const rebuildId = randomUUID();
+        const key = operationKey(scope, conversationId);
+        if (active.has(key)) return { ok: false, rebuildId, reason: "rebuild_in_progress" };
+        active.add(key);
+        try {
+            const connection = options.findFreshPlugin(scope);
+            if (!connection) {
+                await publishFailure(scope, conversationId, rebuildId, "plugin_offline");
+                return { ok: false, rebuildId, reason: "plugin_offline" };
+            }
+            if (!(await authorizePlugin(connection))) {
+                await publishFailure(scope, conversationId, rebuildId, "plugin_heartbeat_stale");
+                return { ok: false, rebuildId, reason: "plugin_heartbeat_stale" };
+            }
+            const clearFrame = createOneTalkStorageDeleteCommandFrame(
+                { connectionType: "plugin", requestId: rebuildId, scope: connection.scope },
+                { rebuildId, target: { kind: "conversation_history", conversationId } },
+            );
+            const clearAck = awaitAck(rebuildId, connection, conversationId, "clear");
+            if (!options.sendPluginFrame(connection, clearFrame)) {
+                pending.delete(rebuildId);
+                await publishFailure(scope, conversationId, rebuildId, "plugin_offline");
+                return { ok: false, rebuildId, reason: "plugin_offline" };
+            }
+            const cleared = await clearAck;
+            if (cleared !== "accepted") {
+                const reason =
+                    cleared === "timeout" ? "plugin_reset_timeout" : "plugin_reset_rejected";
+                await publishFailure(scope, conversationId, rebuildId, reason);
+                return { ok: false, rebuildId, reason };
+            }
+            if (!(await authorizePlugin(connection))) {
+                await publishFailure(scope, conversationId, rebuildId, "plugin_heartbeat_stale");
+                return { ok: false, rebuildId, reason: "plugin_heartbeat_stale" };
+            }
+            await options.publishStatus({
+                scope,
+                conversationId,
+                rebuildId,
+                stage: "storage_cleared",
+            });
+            if (!(await authorizePlugin(connection))) {
+                await publishFailure(scope, conversationId, rebuildId, "plugin_heartbeat_stale");
+                return { ok: false, rebuildId, reason: "plugin_heartbeat_stale" };
+            }
+            const context = sourceContext(connection);
+            if (!context) return { ok: false, rebuildId, reason: "plugin_offline" };
+            const historyGeneration = randomUUID();
+            if (!options.service.resetConversationHistory) {
+                await publishFailure(scope, conversationId, rebuildId, "database_unavailable");
+                return { ok: false, rebuildId, reason: "database_unavailable" };
+            }
+            const conversation = await options.service.resetConversationHistory(
+                context,
+                conversationId,
+                historyGeneration,
+                options.createCommitGuard(connection),
+            );
+            if (!conversation) {
+                await publishFailure(scope, conversationId, rebuildId, "conversation_not_found");
+                return { ok: false, rebuildId, reason: "conversation_not_found" };
+            }
+            committed.set(key, { rebuildId, historyGeneration, connection });
+            await options.publishStatus({
+                scope,
+                conversationId,
+                rebuildId,
+                stage: "server_reset_committed",
+            });
+            if (!(await authorizePlugin(connection))) {
+                await publishPostCommitFailure(scope, conversationId, rebuildId);
+                return {
+                    ok: true,
+                    rebuildId,
+                    historyGeneration,
+                    resync: { status: "failed", reason: "sync_start_failed" },
+                };
+            }
+            const syncFrame = createOneTalkHistorySyncCommandFrame(
+                { connectionType: "plugin", requestId: rebuildId, scope: connection.scope },
+                { rebuildId, historyGeneration, conversationId, mode: "full" },
+            );
+            const syncAck = awaitAck(rebuildId, connection, conversationId, "sync");
+            if (!options.sendPluginFrame(connection, syncFrame)) {
+                pending.delete(rebuildId);
+                await options.publishStatus({
+                    scope,
+                    conversationId,
+                    rebuildId,
+                    stage: "sync_failed",
+                    reason: "sync_start_failed",
+                });
+                return {
+                    ok: true,
+                    rebuildId,
+                    historyGeneration,
+                    resync: { status: "failed", reason: "sync_start_failed" },
+                };
+            }
+            if ((await syncAck) === "accepted") {
+                await options.publishStatus({
+                    scope,
+                    conversationId,
+                    rebuildId,
+                    stage: "sync_started",
+                });
+                return { ok: true, rebuildId, historyGeneration, resync: { status: "started" } };
+            }
+            await options.publishStatus({
+                scope,
+                conversationId,
+                rebuildId,
+                stage: "sync_failed",
+                reason: "sync_start_failed",
+            });
+            return {
+                ok: true,
+                rebuildId,
+                historyGeneration,
+                resync: { status: "failed", reason: "sync_start_failed" },
+            };
+        } catch {
+            await publishFailure(scope, conversationId, rebuildId, "database_unavailable");
+            return { ok: false, rebuildId, reason: "database_unavailable" };
+        } finally {
+            active.delete(key);
+        }
+    };
+    const handleFrame = (connection: OneTalkRegisteredConnection, frame: OneTalkFrame): void => {
+        if (frame.type !== "storage.delete.ack" && frame.type !== "history.sync.ack") return;
+        const pendingAck = pending.get(frame.payload.rebuildId);
+        if (
+            !pendingAck ||
+            pendingAck.connection !== connection ||
+            pendingAck.conversationId !== frame.payload.conversationId ||
+            (pendingAck.stage === "clear" && frame.type !== "storage.delete.ack") ||
+            (pendingAck.stage === "sync" && frame.type !== "history.sync.ack")
+        )
+            return;
+        pendingAck.settle(
+            (
+                pendingAck.stage === "clear"
+                    ? frame.payload.status === "cleared"
+                    : frame.payload.status === "started"
+            )
+                ? "accepted"
+                : "rejected",
+        );
+    };
+    const handleCompletion = async (
+        connection: OneTalkRegisteredConnection,
+        input: { completion: OneTalkSyncCompletionInput; conversation: OneTalkConversationState },
+    ): Promise<void> => {
+        const { completion, conversation } = input;
+        if (!completion.rebuildId || connection.connectionType !== "plugin") return;
+        const token = committed.get(operationKey(connection.mindScope, completion.conversationId));
+        if (
+            !token ||
+            token.connection !== connection ||
+            token.rebuildId !== completion.rebuildId ||
+            token.historyGeneration !== completion.historyGeneration ||
+            conversation.historyGeneration !== completion.historyGeneration ||
+            !conversation.historyComplete ||
+            (conversation.syncResult !== "succeeded" &&
+                conversation.syncResult !== "succeeded_with_anomalies")
+        )
+            return;
+        const context = sourceContext(connection);
+        if (!context || !(await authorizePlugin(connection))) return;
+        const anchors = await options.service.listAnchors(context);
+        if (!(await authorizePlugin(connection))) return;
+        if (
+            !options.sendPluginFrame(
+                connection,
+                createOneTalkAnchorSnapshotFrame(
+                    {
+                        connectionType: "plugin",
+                        requestId: completion.rebuildId,
+                        scope: connection.scope,
+                    },
+                    anchors,
+                ),
+            )
+        )
+            return;
+        await options.publishStatus({
+            scope: connection.mindScope,
+            conversationId: completion.conversationId,
+            rebuildId: completion.rebuildId,
+            stage: "sync_completed",
+        });
+        committed.delete(operationKey(connection.mindScope, completion.conversationId));
+    };
+    return { request, handleFrame, handleCompletion };
+};

apps/server/src/websocket/index.ts

@@ -142,6 +142,7 @@ export const installWebsocket = (
         options.registry ??
         createOneTalkConnectionRegistry({
             authorization,
+            service: options.service,
             cutoverPolicy: options.cutoverPolicy,
             heartbeatIntervalMs: options.heartbeatIntervalMs,
             heartbeatTimeoutMs: options.heartbeatTimeoutMs,

apps/server/src/websocket/plugin/flows/sync-flows.ts

@@ -23,6 +23,8 @@ import {
     type OneTalkObservationResult,
     type OneTalkService,
     type OneTalkSourceContext,
+    type OneTalkSyncCompletionInput,
+    type OneTalkConversationState,
 } from "../../../onetalk/index.ts";
 import { OneTalkInvalidDiscoveryBatchError } from "../../../onetalk/model.ts";
 import type { OneTalkAuthenticatedClientFrame } from "../../authenticated-router.ts";
@@ -78,6 +80,10 @@ type OneTalkSyncFlowsOptions = {
         moveToTop: boolean;
         policyEpoch: number;
     }) => Promise<void>;
+    handleHistoryRebuildCompletion: (input: {
+        completion: OneTalkSyncCompletionInput;
+        conversation: OneTalkConversationState;
+    }) => Promise<void>;
 };
 
 export type OneTalkSyncFlows = {
@@ -464,6 +470,10 @@ export const createOneTalkSyncFlows = (options: OneTalkSyncFlowsOptions): OneTal
                 moveToTop: false,
                 policyEpoch: request.policyEpoch,
             });
+            await options.handleHistoryRebuildCompletion({
+                completion: frame.payload,
+                conversation: result.conversation,
+            });
         } catch (error: unknown) {
             if (!options.isPolicyCurrent(request.policyEpoch)) {
                 options.closeForPause();

apps/server/src/websocket/plugin/index.ts

@@ -132,6 +132,8 @@ export const createOneTalkPluginWebSocketHandler = (
                     },
                     publishSyncStatus: context.options.registry.publishSyncStatus,
                     publishMessageCreated: context.options.registry.publishMessageCreated,
+                    handleHistoryRebuildCompletion: (input) =>
+                        context.options.registry.handleHistoryRebuildCompletion(socket, input),
                     publishConversationUpdated: async (input) =>
                         publishConversationUpdated(
                             context,
@@ -207,7 +209,11 @@ export const createOneTalkPluginWebSocketHandler = (
                     const code = sessionAuthorizationFailure(
                         state,
                         decision,
-                        route.operation === "heartbeat" ? null : "read",
+                        route.operation === "heartbeat"
+                            ? null
+                            : route.operation === "rebuild"
+                              ? "rebuild"
+                              : "read",
                     );
                     if (code === null) return epoch;
                     const authorizationDenied = !decision.allowed;
@@ -296,6 +302,16 @@ export const createOneTalkPluginWebSocketHandler = (
                                 policyEpoch: epoch,
                             });
                         }),
+                        defineOneTalkEndpointRouteHandler("storage.delete.ack", async (frame) => {
+                            const epoch = await authorizeSessionRoute(frame);
+                            if (epoch === null) return;
+                            context.options.registry.handleHistoryRebuildFrame(socket, frame);
+                        }),
+                        defineOneTalkEndpointRouteHandler("history.sync.ack", async (frame) => {
+                            const epoch = await authorizeSessionRoute(frame);
+                            if (epoch === null) return;
+                            context.options.registry.handleHistoryRebuildFrame(socket, frame);
+                        }),
                         defineOneTalkEndpointRouteHandler(
                             "contact.profile.observed",
                             async (frame) => {

apps/server/src/websocket/registry.ts

@@ -11,8 +11,17 @@ import type {
     OneTalkSendRequestFrame,
     OneTalkSendResultFrame,
     OneTalkSyncStatusPayload,
+    OneTalkFrame,
+    OneTalkRebuildStatusFrame,
+} from "@trade-message-center/onetalk-contract";
+import {
+    createOneTalkRebuildStatusFrame,
+    isSameOneTalkScope,
 } from "@trade-message-center/onetalk-contract";
 import type { OneTalkCutoverPolicy } from "../cutover-policy.ts";
+import type { OneTalkService } from "../onetalk/index.ts";
+import type { OneTalkSyncCompletionInput } from "../onetalk/index.ts";
+import type { OneTalkConversationState } from "../onetalk/index.ts";
 import {
     createOneTalkConnectionStore,
     type OneTalkCommitGuard,
@@ -23,6 +32,10 @@ import {
     createOneTalkPendingSendCoordinator,
     type OneTalkPendingSend,
 } from "./pending-send-coordinator.ts";
+import {
+    createOneTalkHistoryRebuildCoordinator,
+    type OneTalkHistoryRebuildResult,
+} from "./history-rebuild-coordinator.ts";
 
 export type { OneTalkRegisteredConnection } from "./connection-store.ts";
 export type { OneTalkPendingSend, OneTalkSendAttemptPhase } from "./pending-send-coordinator.ts";
@@ -72,6 +85,15 @@ export type OneTalkConnectionRegistry = OneTalkPluginPresenceReader & {
             message: NonNullable<OneTalkSendConfirmationFrame["payload"]["message"]>,
         ) => Promise<{ status: "accepted" | "duplicate" | "unknown"; message?: OneTalkMessage }>,
     ) => Promise<void>;
+    requestHistoryRebuild: (
+        scope: OneTalkMindScope,
+        conversationId: string,
+    ) => Promise<OneTalkHistoryRebuildResult>;
+    handleHistoryRebuildFrame: (socket: WebSocket, frame: OneTalkFrame) => void;
+    handleHistoryRebuildCompletion: (
+        socket: WebSocket,
+        input: { completion: OneTalkSyncCompletionInput; conversation: OneTalkConversationState },
+    ) => Promise<void>;
 };
 
 /** 创建兼容 handler 的 OneTalk 连接注册表 façade。 */
@@ -80,15 +102,19 @@ export const createOneTalkConnectionRegistry = (options: {
     cutoverPolicy?: OneTalkCutoverPolicy;
     heartbeatIntervalMs?: number;
     heartbeatTimeoutMs?: number;
+    now?: () => number;
     onPublishFailure: OneTalkPublishFailureSink;
     sendTimeoutMs?: number;
     scheduleTimeout?: typeof setTimeout;
     cancelTimeout?: typeof clearTimeout;
+    service?: OneTalkService;
+    historyRebuildTimeoutMs?: number;
 }): OneTalkConnectionRegistry => {
     const store = createOneTalkConnectionStore({
         cutoverPolicy: options.cutoverPolicy,
         heartbeatIntervalMs: options.heartbeatIntervalMs,
         heartbeatTimeoutMs: options.heartbeatTimeoutMs,
+        now: options.now,
     });
     const publisher = createOneTalkMindPublisher({
         authorization: options.authorization,
@@ -102,6 +128,125 @@ export const createOneTalkConnectionRegistry = (options: {
         scheduleTimeout: options.scheduleTimeout,
         cancelTimeout: options.cancelTimeout,
     });
+    const findFreshPlugin = (
+        scope: OneTalkMindScope,
+    ): Extract<OneTalkRegisteredConnection, { connectionType: "plugin" }> | null => {
+        const connection = store
+            .getConnections()
+            .find(
+                (candidate) =>
+                    candidate.connectionType === "plugin" &&
+                    isSameOneTalkScope(candidate.mindScope, scope) &&
+                    candidate.permissions.includes("rebuild") &&
+                    store.isPluginLeaseFresh(candidate),
+            );
+        return connection?.connectionType === "plugin" ? connection : null;
+    };
+    const isCurrentPlugin = (
+        connection: Extract<OneTalkRegisteredConnection, { connectionType: "plugin" }>,
+    ): boolean => {
+        return (
+            connection.connectionType === "plugin" &&
+            store.getCanonicalConnection(connection.socket) === connection &&
+            connection.permissions.includes("rebuild") &&
+            store.isPluginLeaseFresh(connection)
+        );
+    };
+    const sendPluginFrame = (
+        connection: Extract<OneTalkRegisteredConnection, { connectionType: "plugin" }>,
+        frame: OneTalkFrame,
+    ): boolean => {
+        if (!isCurrentPlugin(connection) || connection.socket.readyState !== 1) return false;
+        try {
+            connection.socket.send(JSON.stringify(frame));
+            return true;
+        } catch {
+            return false;
+        }
+    };
+    const publishRebuildStatus = async (input: {
+        scope: OneTalkMindScope;
+        conversationId: string;
+        rebuildId: string;
+        stage: OneTalkRebuildStatusFrame["payload"]["stage"];
+        reason?: OneTalkRebuildStatusFrame["payload"]["reason"];
+    }): Promise<void> => {
+        for (const connection of store.getConnections()) {
+            if (
+                connection.connectionType !== "mind_page" ||
+                !isSameOneTalkScope(connection.mindScope, input.scope) ||
+                !connection.permissions.includes("rebuild") ||
+                connection.socket.readyState !== 1
+            )
+                continue;
+            const policyEpoch = store.currentEpoch();
+            try {
+                const decision = await (connection.sessionAuthorization
+                    ? connection.sessionAuthorization.authorize(connection.scope, "rebuild")
+                    : options.authorization.authorize({
+                          connectionType: "mind_page",
+                          operation: "rebuild",
+                          scope: connection.scope,
+                      }));
+                if (
+                    !decision.allowed ||
+                    !isSameOneTalkScope(decision.mindScope, connection.mindScope) ||
+                    decision.binding !== connection.binding ||
+                    decision.authorizationVersion !== connection.authorizationVersion ||
+                    !decision.permissions.includes("rebuild") ||
+                    store.getCanonicalConnection(connection.socket) !== connection ||
+                    connection.socket.readyState !== 1 ||
+                    !store.epochIsCurrent(policyEpoch) ||
+                    !store.policyAdmits("mind_page")
+                ) {
+                    continue;
+                }
+                connection.socket.send(
+                    JSON.stringify(
+                        createOneTalkRebuildStatusFrame(
+                            {
+                                connectionType: "mind_page",
+                                requestId: input.rebuildId,
+                                scope: connection.scope,
+                            },
+                            {
+                                rebuildId: input.rebuildId,
+                                conversationId: input.conversationId,
+                                stage: input.stage,
+                                ...(input.reason === undefined ? {} : { reason: input.reason }),
+                            },
+                        ),
+                    ),
+                );
+            } catch {
+                // A failed status publish never changes a committed reset.
+            }
+        }
+    };
+    const rebuild = options.service
+        ? createOneTalkHistoryRebuildCoordinator({
+              authorization: options.authorization,
+              service: options.service,
+              findFreshPlugin,
+              isCurrentPlugin,
+              sendPluginFrame,
+              createCommitGuard: (connection) => {
+                  const guard = store.createCommitGuard(connection);
+                  return {
+                      assertValid: () => {
+                          guard.assertValid();
+                          if (!isCurrentPlugin(connection)) {
+                              throw new Error("connection_commit_invalid");
+                          }
+                      },
+                  };
+              },
+              publishStatus: publishRebuildStatus,
+              ...(options.historyRebuildTimeoutMs === undefined
+                  ? {}
+                  : { timeoutMs: options.historyRebuildTimeoutMs }),
+          })
+        : undefined;
 
     return {
         getCanonicalConnection: store.getCanonicalConnection,
@@ -116,5 +261,18 @@ export const createOneTalkConnectionRegistry = (options: {
         sweepExpired: store.sweepExpired,
         requestSend: pendingSends.requestSend,
         handleSendConfirmation: pendingSends.handleSendConfirmation,
+        requestHistoryRebuild: async (scope, conversationId) => {
+            if (!rebuild)
+                return { ok: false, rebuildId: "unavailable", reason: "database_unavailable" };
+            return rebuild.request(scope, conversationId);
+        },
+        handleHistoryRebuildFrame: (socket, frame) => {
+            const connection = store.getCanonicalConnection(socket);
+            if (connection) rebuild?.handleFrame(connection, frame);
+        },
+        handleHistoryRebuildCompletion: async (socket, input) => {
+            const connection = store.getCanonicalConnection(socket);
+            if (connection) await rebuild?.handleCompletion(connection, input);
+        },
     };
 };

apps/server/test/onetalk-domain.test.ts

@@ -560,3 +560,54 @@ test("rejects stale history generation for observations and sync completion", as
     });
     assert.equal(harness.syncUpdates.length, 0);
 });
+
+test("does not leave a sync anomaly when reset wins after the initial generation read", async () => {
+    const harness = createRepositoryHarness();
+    await harness.repository.discoverConversation(context, "conversation-1", "direct");
+    const baseFindConversation = harness.repository.findConversation;
+    const baseGuardedRecordAnomaly = harness.repository.guardedRecordAnomaly;
+    const baseGuardedUpdateSyncState = harness.repository.guardedUpdateSyncState;
+    const repository: OneTalkRepository = {
+        ...harness.repository,
+        findConversation: async (...args) => baseFindConversation(...args),
+        guardedRecordAnomaly: async (input, guard) => {
+            const key = `${context.channelAccountId}:${input.conversationId}`;
+            const current = harness.conversations.get(key);
+            if (current) {
+                harness.conversations.set(key, {
+                    ...current,
+                    historyGeneration: "reset-generation",
+                });
+            }
+            await baseGuardedRecordAnomaly(input, guard);
+        },
+        guardedUpdateSyncState: async (sourceContext, update, conversationId, guard) => {
+            const current = harness.conversations.get(
+                `${sourceContext.channelAccountId}:${conversationId}`,
+            );
+            if (current?.historyGeneration !== update.historyGeneration) return null;
+            return baseGuardedUpdateSyncState(sourceContext, update, conversationId, guard);
+        },
+    };
+    const service = createOneTalkService(repository);
+
+    const result = await service.completeSync(
+        context,
+        {
+            conversationId: "conversation-1",
+            historyGeneration: "initial",
+            mode: "full",
+            historyComplete: true,
+            result: "succeeded",
+            latestMessageId: null,
+            anomalyCode: "incremental_anchor_not_found",
+        },
+        { assertValid: () => undefined },
+    );
+
+    assert.deepEqual(result, {
+        status: "rejected",
+        reason: "conversation_not_discovered",
+    });
+    assert.equal(harness.anomalies.size, 0);
+});

apps/server/test/onetalk-history-rebuild-coordinator.test.ts

@@ -0,0 +1,311 @@
+// 验证 rebuild 的 commit 前 ACK 栅栏与 post-commit best-effort 语义。
+
+import assert from "node:assert/strict";
+import test from "node:test";
+
+import type {
+    OneTalkAuthorizationReader,
+    OneTalkFrame,
+    OneTalkMindScope,
+} from "@trade-message-center/onetalk-contract";
+
+import type { OneTalkService } from "../src/onetalk/index.ts";
+import type { OneTalkRegisteredConnection } from "../src/websocket/registry.ts";
+import { createOneTalkHistoryRebuildCoordinator } from "../src/websocket/history-rebuild-coordinator.ts";
+import { createOneTalkConnectionRegistry } from "../src/websocket/registry.ts";
+
+const scope: OneTalkMindScope = {
+    mindUserId: "mind-1",
+    workspaceId: "workspace-1",
+    channelAccountId: "account-1",
+};
+
+const connection = {
+    connectionType: "plugin",
+    socket: { readyState: 1 },
+    scope: { channelAccountId: "account-1", deviceId: "device-1" },
+    mindScope: scope,
+    binding: "binding-1",
+    authorizationVersion: "version-1",
+    permissions: ["read", "rebuild"],
+} as unknown as Extract<OneTalkRegisteredConnection, { connectionType: "plugin" }>;
+
+const authorization: OneTalkAuthorizationReader = {
+    authorize: async () => ({
+        allowed: true,
+        mindScope: scope,
+        binding: "binding-1",
+        authorizationVersion: "version-1",
+        permissions: ["read", "rebuild"],
+    }),
+    readAuthorizationVersion: async () => "version-1",
+};
+
+const service = (calls: string[]): OneTalkService => ({
+    discoverConversation: async () => {
+        throw new Error("not used");
+    },
+    discoverConversations: async () => [],
+    listAnchors: async () => [],
+    listConversations: async () => [],
+    readConversation: async () => null,
+    readHistory: async () => null,
+    observeMessage: async () => ({ status: "rejected", reason: "conversation_not_discovered" }),
+    observeMessages: async () => [],
+    completeSync: async () => ({ status: "rejected", reason: "conversation_not_discovered" }),
+    resetConversationHistory: async (_context, conversationId, historyGeneration, guard) => {
+        guard.assertValid();
+        calls.push(`reset:${conversationId}`);
+        return {
+            channelAccountId: scope.channelAccountId,
+            conversationId,
+            conversationKind: "direct",
+            syncPhase: "initial",
+            syncResult: "incomplete",
+            latestMessageId: null,
+            historyComplete: false,
+            messageCount: 0,
+            historyGeneration,
+        };
+    },
+});
+
+test("commits only after the matching clear ACK and treats sync rejection as post-commit", async () => {
+    const calls: string[] = [];
+    const stages: string[] = [];
+    let coordinator: ReturnType<typeof createOneTalkHistoryRebuildCoordinator>;
+    coordinator = createOneTalkHistoryRebuildCoordinator({
+        authorization,
+        service: service(calls),
+        findFreshPlugin: () => connection,
+        isCurrentPlugin: () => true,
+        createCommitGuard: () => ({ assertValid: () => undefined }),
+        sendPluginFrame: (_connection, frame) => {
+            calls.push(frame.type);
+            queueMicrotask(() => {
+                if (frame.type === "storage.delete.command") {
+                    coordinator.handleFrame(connection, {
+                        ...frame,
+                        type: "storage.delete.ack",
+                        payload: {
+                            rebuildId: frame.payload.rebuildId,
+                            conversationId: frame.payload.target.conversationId,
+                            status: "cleared",
+                        },
+                    } as OneTalkFrame);
+                }
+                if (frame.type === "history.sync.command") {
+                    coordinator.handleFrame(connection, {
+                        ...frame,
+                        type: "history.sync.ack",
+                        payload: {
+                            rebuildId: frame.payload.rebuildId,
+                            conversationId: frame.payload.conversationId,
+                            status: "rejected",
+                            reason: "conversation_cache_missing",
+                        },
+                    } as OneTalkFrame);
+                }
+            });
+            return true;
+        },
+        publishStatus: async ({ stage }) => void stages.push(stage),
+    });
+
+    const result = await coordinator.request(scope, "conversation-1");
+
+    assert.deepEqual(calls, [
+        "storage.delete.command",
+        "reset:conversation-1",
+        "history.sync.command",
+    ]);
+    assert.equal(result.ok, true);
+    if (!result.ok) return;
+    assert.deepEqual(result.resync, { status: "failed", reason: "sync_start_failed" });
+    assert.deepEqual(stages, ["storage_cleared", "server_reset_committed", "sync_failed"]);
+});
+
+test("does not create a reset when no fresh canonical plugin exists", async () => {
+    const calls: string[] = [];
+    const coordinator = createOneTalkHistoryRebuildCoordinator({
+        authorization,
+        service: service(calls),
+        findFreshPlugin: () => null,
+        isCurrentPlugin: () => false,
+        createCommitGuard: () => ({ assertValid: () => undefined }),
+        sendPluginFrame: () => false,
+        publishStatus: async () => undefined,
+    });
+
+    const result = await coordinator.request(scope, "conversation-1");
+
+    assert.deepEqual(result.ok ? [] : [result.reason], ["plugin_offline"]);
+    assert.deepEqual(calls, []);
+});
+
+test("reports sync_failed without changing committed success when post-commit reauth fails", async () => {
+    const calls: string[] = [];
+    const stages: string[] = [];
+    let authorizationChecks = 0;
+    let coordinator: ReturnType<typeof createOneTalkHistoryRebuildCoordinator>;
+    coordinator = createOneTalkHistoryRebuildCoordinator({
+        authorization: {
+            ...authorization,
+            authorize: async () => {
+                authorizationChecks += 1;
+                return authorizationChecks < 4
+                    ? await authorization.authorize({
+                          connectionType: "plugin",
+                          operation: "rebuild",
+                          scope: connection.scope,
+                          binding: connection.binding,
+                      })
+                    : { allowed: false, code: "binding_revoked" as const };
+            },
+        },
+        service: service(calls),
+        findFreshPlugin: () => connection,
+        isCurrentPlugin: () => true,
+        createCommitGuard: () => ({ assertValid: () => undefined }),
+        sendPluginFrame: (_connection, frame) => {
+            if (frame.type === "storage.delete.command") {
+                queueMicrotask(() =>
+                    coordinator.handleFrame(connection, {
+                        ...frame,
+                        type: "storage.delete.ack",
+                        payload: {
+                            rebuildId: frame.payload.rebuildId,
+                            conversationId: frame.payload.target.conversationId,
+                            status: "cleared",
+                        },
+                    } as OneTalkFrame),
+                );
+            }
+            return true;
+        },
+        publishStatus: async ({ stage }) => void stages.push(stage),
+    });
+
+    const result = await coordinator.request(scope, "conversation-1");
+
+    assert.equal(result.ok, true);
+    if (!result.ok) return;
+    assert.deepEqual(result.resync, { status: "failed", reason: "sync_start_failed" });
+    assert.deepEqual(stages, ["storage_cleared", "server_reset_committed", "sync_failed"]);
+    assert.equal(calls.includes("reset:conversation-1"), true);
+});
+
+test("does not commit reset when the post-clear commit guard expires", async () => {
+    const calls: string[] = [];
+    let coordinator: ReturnType<typeof createOneTalkHistoryRebuildCoordinator>;
+    coordinator = createOneTalkHistoryRebuildCoordinator({
+        authorization,
+        service: service(calls),
+        findFreshPlugin: () => connection,
+        isCurrentPlugin: () => true,
+        createCommitGuard: () => ({
+            assertValid: () => {
+                throw new Error("connection_commit_invalid");
+            },
+        }),
+        sendPluginFrame: (_connection, frame) => {
+            if (frame.type === "storage.delete.command") {
+                queueMicrotask(() =>
+                    coordinator.handleFrame(connection, {
+                        ...frame,
+                        type: "storage.delete.ack",
+                        payload: {
+                            rebuildId: frame.payload.rebuildId,
+                            conversationId: frame.payload.target.conversationId,
+                            status: "cleared",
+                        },
+                    } as OneTalkFrame),
+                );
+            }
+            return true;
+        },
+        publishStatus: async () => undefined,
+    });
+
+    const result = await coordinator.request(scope, "conversation-1");
+
+    assert.deepEqual(result.ok ? [] : [result.reason], ["database_unavailable"]);
+    assert.equal(calls.includes("reset:conversation-1"), false);
+});
+
+test("does not publish rebuild status to a Mind connection revoked after handshake", async () => {
+    const calls: string[] = [];
+    const mindFrames: string[] = [];
+    const registry = createOneTalkConnectionRegistry({
+        authorization,
+        service: service(calls),
+        onPublishFailure: () => undefined,
+    });
+    const mindSocket = {
+        readyState: 1,
+        send: (payload: string) => mindFrames.push(payload),
+        close: () => undefined,
+    };
+    const pluginSocket = {
+        readyState: 1,
+        send: (payload: string) => {
+            const frame = JSON.parse(payload) as OneTalkFrame;
+            if (frame.type === "storage.delete.command") {
+                registry.handleHistoryRebuildFrame(
+                    pluginSocket as never,
+                    {
+                        ...frame,
+                        type: "storage.delete.ack",
+                        payload: {
+                            rebuildId: frame.payload.rebuildId,
+                            conversationId: frame.payload.target.conversationId,
+                            status: "cleared",
+                        },
+                    } as OneTalkFrame,
+                );
+            }
+            if (frame.type === "history.sync.command") {
+                registry.handleHistoryRebuildFrame(
+                    pluginSocket as never,
+                    {
+                        ...frame,
+                        type: "history.sync.ack",
+                        payload: {
+                            rebuildId: frame.payload.rebuildId,
+                            conversationId: frame.payload.conversationId,
+                            status: "started",
+                        },
+                    } as OneTalkFrame,
+                );
+            }
+        },
+    };
+    registry.register({
+        socket: mindSocket as never,
+        connectionType: "mind_page",
+        scope,
+        mindScope: scope,
+        binding: "binding-1",
+        authorizationVersion: "version-1",
+        permissions: ["rebuild"],
+        sessionAuthorization: {
+            authorize: async () => ({ allowed: false, code: "binding_revoked" as const }),
+            clear: () => undefined,
+        },
+    });
+    registry.register({
+        ...connection,
+        socket: pluginSocket as never,
+    });
+
+    const result = await registry.requestHistoryRebuild(scope, "conversation-1");
+
+    assert.equal(result.ok, true);
+    assert.equal(
+        mindFrames.some(
+            (payload) => (JSON.parse(payload) as { type: string }).type === "rebuild.status",
+        ),
+        false,
+    );
+    assert.deepEqual(calls, ["reset:conversation-1"]);
+});

apps/server/test/onetalk-http.test.ts

@@ -19,6 +19,7 @@ import {
     type OneTalkReadConversation,
     type OneTalkReadService,
 } from "../src/onetalk/index.ts";
+import type { OneTalkConnectionRegistry } from "../src/websocket/registry.ts";
 
 const testConfig = {
     host: "127.0.0.1",
@@ -135,6 +136,7 @@ const conversationsUrl = (): string =>
 const conversationUrl = (): string => conversationsUrl() + "/" + conversation.conversationId;
 const messagesUrl = (): string => conversationUrl() + "/messages";
 const historyUrl = (): string => conversationUrl() + "/history";
+const rebuildUrl = (): string => historyUrl() + "/rebuild";
 
 const headers = (): Record<string, string> => ({
     cookie: "mind_session=opaque",
@@ -154,6 +156,79 @@ const createInternalSummaryApp = (
     return app;
 };
 
+test("rejects a read-only Mind session before invoking the rebuild registry", async () => {
+    let rebuildCalls = 0;
+    const registry = {
+        requestHistoryRebuild: async () => {
+            rebuildCalls += 1;
+            return { ok: false as const, rebuildId: "unused", reason: "plugin_offline" as const };
+        },
+    } as unknown as OneTalkConnectionRegistry;
+    const app = createApp(testConfig, {
+        database: createDatabaseStub(),
+        authorization: createMockAuthorizationReader([authorizationRecord]),
+        readService: createReadService(),
+        oneTalkRegistry: registry,
+    });
+
+    try {
+        const response = await app.inject({
+            method: "POST",
+            url: rebuildUrl(),
+            headers: headers(),
+        });
+        assert.equal(response.statusCode, 403);
+        assert.deepEqual(response.json(), { error: { code: "authorization_rejected" } });
+        assert.equal(rebuildCalls, 0);
+    } finally {
+        await closeApp(app);
+    }
+});
+
+test("returns the exact committed rebuild response and preserves post-commit resync failure", async () => {
+    const rebuildCalls: Array<{ scope: typeof mindScope; conversationId: string }> = [];
+    const registry = {
+        requestHistoryRebuild: async (scope: typeof mindScope, conversationId: string) => {
+            rebuildCalls.push({ scope, conversationId });
+            return {
+                ok: true as const,
+                rebuildId: "rebuild-1",
+                historyGeneration: "generation-1",
+                resync: { status: "failed" as const, reason: "sync_start_failed" },
+            };
+        },
+    } as unknown as OneTalkConnectionRegistry;
+    const app = createApp(testConfig, {
+        database: createDatabaseStub(),
+        authorization: createMockAuthorizationReader([
+            { ...authorizationRecord, permissions: ["read", "rebuild"] },
+        ]),
+        readService: createReadService(),
+        oneTalkRegistry: registry,
+    });
+
+    try {
+        const response = await app.inject({
+            method: "POST",
+            url: rebuildUrl(),
+            headers: headers(),
+        });
+        assert.equal(response.statusCode, 200);
+        assert.deepEqual(response.json(), {
+            scope: mindScope,
+            conversationId: conversation.conversationId,
+            rebuildId: "rebuild-1",
+            status: "server_reset_committed",
+            resync: { status: "failed", reason: "sync_start_failed" },
+        });
+        assert.deepEqual(rebuildCalls, [
+            { scope: mindScope, conversationId: conversation.conversationId },
+        ]);
+    } finally {
+        await closeApp(app);
+    }
+});
+
 test("returns the HTTP customer profile projection for list and detail", async () => {
     const app = createApp(testConfig, {
         database: createDatabaseStub(),
@@ -607,7 +682,7 @@ test("preflight rejects retired summary headers and rejected origins cannot invo
         });
         assert.equal(disallowedHeader.statusCode, 403);
 
-        const disallowedMethod = await app.inject({
+        const rebuildMethod = await app.inject({
             method: "OPTIONS",
             url: historyUrl(),
             headers: {
@@ -615,7 +690,7 @@ test("preflight rejects retired summary headers and rejected origins cannot invo
                 "access-control-request-method": "POST",
             },
         });
-        assert.equal(disallowedMethod.statusCode, 403);
+        assert.equal(rebuildMethod.statusCode, 204);
     } finally {
         await closeApp(app);
     }

apps/server/test/onetalk-websocket.test.ts

@@ -49,6 +49,8 @@ const CLIENT_FRAME_COVERAGE_BASELINES = {
     "buyer.facts.observed": ["buyer.facts.ack"],
     "message.observed": ["message.ack", "message.created", "conversation.updated"],
     "messages.observed": ["messages.ack"],
+    "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[]>;
@@ -424,7 +426,9 @@ test("executes a wire assertion for every declared OneTalk client frame", async
     }));
     const app = createApp(testConfig, {
         database: createDatabaseStub(),
-        authorization: createMockAuthorizationReader([authorizationRecord]),
+        authorization: createMockAuthorizationReader([
+            { ...authorizationRecord, permissions: ["read", "send", "rebuild"] },
+        ]),
         oneTalkService: service,
         profileService: {
             ingestProfiles: async ({ profiles }) => ({
@@ -486,7 +490,7 @@ test("executes a wire assertion for every declared OneTalk client frame", async
                 type: "ws.hello",
                 requestId: "matrix-mind-hello",
                 scope: mindScope,
-                payload: { requestedPermissions: ["read", "send"] },
+                payload: { requestedPermissions: ["read", "send", "rebuild"] },
             }),
         );
         const [mindAccepted, initialPluginStatus] = await mindHandshake;
@@ -502,7 +506,10 @@ test("executes a wire assertion for every declared OneTalk client frame", async
                 type: "ws.hello",
                 requestId: "matrix-plugin-hello",
                 scope: pluginScope,
-                payload: { binding: "binding-1", requestedPermissions: ["read", "send"] },
+                payload: {
+                    binding: "binding-1",
+                    requestedPermissions: ["read", "send", "rebuild"],
+                },
             }),
         );
         const handshakeFrames = await pluginHandshake;
@@ -528,6 +535,40 @@ test("executes a wire assertion for every declared OneTalk client frame", async
         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(
@@ -1652,6 +1693,55 @@ test("does not publish live facts to a Mind connection that did not request read
     );
 });
 
+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) => {

packages/onetalk-contract/src/conversation-sync.ts

@@ -93,6 +93,7 @@ export type OneTalkSyncCompleteFrame = OneTalkBaseFrame<
     {
         conversationId: string;
         historyGeneration: OneTalkHistoryGeneration;
+        rebuildId?: string;
         mode: OneTalkSyncMode;
         historyComplete: boolean;
         result: OneTalkSyncResult;
@@ -251,6 +252,7 @@ export const isValidOneTalkConversationSyncPayload = (
                 hasExactKeys(value, [
                     "conversationId",
                     "historyGeneration",
+                    ...(value.rebuildId === undefined ? [] : ["rebuildId"]),
                     "mode",
                     "historyComplete",
                     "result",
@@ -259,6 +261,7 @@ export const isValidOneTalkConversationSyncPayload = (
                     ...(value.anomalyCode === undefined ? [] : ["anomalyCode"]),
                 ]) &&
                 isOneTalkHistoryGeneration(value.historyGeneration) &&
+                (value.rebuildId === undefined || isNonEmptyString(value.rebuildId)) &&
                 isNonEmptyString(value.conversationId) &&
                 typeof value.mode === "string" &&
                 ONETALK_SYNC_MODES.includes(value.mode as never) &&

packages/onetalk-contract/src/wire.ts

@@ -95,6 +95,8 @@ export const ONETALK_CLIENT_FRAME_TYPES = [
     "messages.observed",
     "send.request",
     "send.confirmation",
+    "storage.delete.ack",
+    "history.sync.ack",
 ] as const satisfies readonly OneTalkFrameType[];
 
 export const ONETALK_SERVER_FRAME_TYPES = [