perf(web): share document parsers and skip covered snapshot reads (#920)

Author: tt-a1i

docs/development/OPENPI_WEB_DEVELOPMENT.md

@@ -134,7 +134,15 @@ bun run check
 bun run test
 ```
 
-修改 `web/ui/` 或 Web 构建配置后，需要同时提交重新生成的 `web/dist/`。CI 会重新构建并检查该目录；如果源码和提交的产物不一致，门禁会失败。不要手工编辑 `web/dist/`。
+修改 `web/ui/` 或 Web 构建配置后，需要同时提交重新生成的 `web/dist/`。CI 会重新构建并检查该目录；如果源码和提交的产物不一致，门禁会失败。如果合并或 rebase 时 `web/dist/` 有冲突，先解决源码和锁文件的冲突，再从仓库根目录运行：
+
+```bash
+bun install --frozen-lockfile
+bun run build:web
+git add -A -- web/dist
+```
+
+Vite 会清空旧的输出目录。最后一步会同时暂存新产物和已失效的 chunk 删除记录。
 
 CI 在 Node `22.19.0` 上运行一次 Chromium E2E，覆盖项目声明的最低 Node 版本；普通 `check` 和 `test` 继续同时覆盖 Node `22.19.0` 与 Node `24`。
 

tests/web/file-attachments.spec.ts

@@ -30,6 +30,7 @@ describe("composer file staging boundary", () => {
     await vi.waitFor(() => expect(StagingWorker.instances).toHaveLength(1));
     const worker = StagingWorker.instances[0]!;
     expect(worker.postMessage.mock.calls[0]?.[0]).toMatchObject({
+      operation: "extract",
       kind: "text",
     });
     worker.onmessage?.({

tests/web/file-text-worker.spec.ts

@@ -13,9 +13,9 @@ async function extract(bytes: Uint8Array, kind: string) {
     postMessage: vi.fn(),
   };
   vi.stubGlobal("self", worker);
-  await import("../../web/ui/src/features/composer/file-text-worker.ts");
+  await import("../../web/ui/src/features/files/document-worker.ts");
   worker.onmessage?.({
-    data: { data: Uint8Array.from(bytes).buffer, kind },
+    data: { operation: "extract", data: Uint8Array.from(bytes).buffer, kind },
   } as MessageEvent);
   await vi.waitFor(() => expect(worker.postMessage).toHaveBeenCalledOnce());
   return worker.postMessage.mock.calls[0]![0].result;

tests/web/office-preview-worker.spec.ts

@@ -6,13 +6,19 @@ type WorkerReply = { result?: OfficePreview; error?: string };
 let receive: ((reply: WorkerReply) => void) | undefined;
 const worker = {
   onmessage: null as
-    | ((event: MessageEvent<{ data: ArrayBuffer; extension: string }>) => void)
+    | ((
+        event: MessageEvent<{
+          operation: "preview";
+          data: ArrayBuffer;
+          extension: string;
+        }>,
+      ) => void)
     | null,
   postMessage: (reply: WorkerReply) => receive?.(reply),
 };
 beforeAll(async () => {
   vi.stubGlobal("self", worker);
-  await import("../../web/ui/src/features/files/office-worker.ts");
+  await import("../../web/ui/src/features/files/document-worker.ts");
 });
 afterAll(() => vi.unstubAllGlobals());
 
@@ -25,7 +31,9 @@ async function preview(workbook: ExcelJS.Workbook) {
   });
   expect(worker.onmessage).toBeTypeOf("function");
   worker.onmessage?.(
-    new MessageEvent("message", { data: { data, extension: "xlsx" } }),
+    new MessageEvent("message", {
+      data: { operation: "preview", data, extension: "xlsx" },
+    }),
   );
   const reply = await response;
   expect(reply.error).toBeUndefined();
@@ -68,3 +76,21 @@ it("marks a budget cut within the last worksheet row, but not an exactly complet
   expect(clipped.sheets[1].rows[1][19]).toBe("2,20");
   expect(clipped.sheets[1].truncated).toBe(true);
 });
+
+it("reports invalid preview documents through the preview error response", async () => {
+  const response = new Promise<WorkerReply>((resolve) => {
+    receive = resolve;
+  });
+  worker.onmessage?.(
+    new MessageEvent("message", {
+      data: {
+        operation: "preview",
+        data: new TextEncoder().encode("not an Office archive").buffer,
+        extension: "xlsx",
+      },
+    }),
+  );
+  const reply = await response;
+  expect(reply.result).toBeUndefined();
+  expect(reply.error).toBeTypeOf("string");
+});

tests/web/web-store.spec.ts

@@ -3014,6 +3014,121 @@ describe("OpenPI Web store", () => {
     store.getState().actions.stop();
   });
 
+  it.each([
+    { snapshotCursor: 6, expectedReads: 2 },
+    { snapshotCursor: 5, expectedReads: 3 },
+  ])(
+    "keeps only uncovered event refreshes after an in-flight snapshot at cursor $snapshotCursor",
+    async ({ snapshotCursor, expectedReads }) => {
+      vi.useFakeTimers();
+      const client = new FakeClient();
+      const reading = deferred<WebSnapshot>();
+      const canonical = { ...snapshot("Canonical"), cursor: snapshotCursor };
+      client.snapshots.push(
+        Promise.resolve(snapshot()),
+        reading.promise,
+        Promise.resolve({ ...canonical, cursor: 6 }),
+      );
+      const stream = eventStreamHarness();
+      const store = createWebStore(client, {
+        consumeEvents: stream.consumeEvents,
+      });
+      await store.getState().actions.refreshSnapshot();
+      store.getState().actions.start();
+      try {
+        stream.emit(runtimeEvent(5, "agent_start", { sessionId: "session-1" }));
+        await vi.advanceTimersByTimeAsync(160);
+        expect(client.snapshotPaths).toHaveLength(2);
+        stream.emit(
+          runtimeEvent(6, "message_end", {
+            messageKey: "answer",
+            message: { role: "assistant", content: "Finished answer" },
+          }),
+        );
+        // Live projection remains immediate while canonical history is read.
+        expect(store.getState().liveMessages[0]?.message.content).toBe(
+          "Finished answer",
+        );
+        reading.resolve(canonical);
+        await vi.advanceTimersByTimeAsync(320);
+        expect(client.snapshotPaths).toHaveLength(expectedReads);
+        expect(store.getState().snapshot?.cursor).toBe(6);
+        expect(store.getState().snapshot?.sessions[0]?.name).toBe("Canonical");
+      } finally {
+        store.getState().actions.stop();
+      }
+    },
+  );
+
+  it("does not let a superseded snapshot suppress refreshes for newer events", async () => {
+    vi.useFakeTimers();
+    const client = new FakeClient();
+    const older = deferred<WebSnapshot>();
+    client.snapshots.push(
+      Promise.resolve(snapshot()),
+      older.promise,
+      Promise.resolve({ ...snapshot(), cursor: 5 }),
+      Promise.resolve({ ...snapshot("Newer event confirmed"), cursor: 6 }),
+    );
+    const stream = eventStreamHarness();
+    const store = createWebStore(client, {
+      consumeEvents: stream.consumeEvents,
+    });
+    await store.getState().actions.refreshSnapshot();
+    store.getState().actions.start();
+    try {
+      stream.emit(runtimeEvent(5, "agent_start", { sessionId: "session-1" }));
+      await vi.advanceTimersByTimeAsync(160);
+      stream.emit(
+        runtimeEvent(6, "session_renamed", { sessionId: "session-1" }),
+      );
+      await store.getState().actions.refreshSnapshot();
+      // This cursor would cover the event, but the newer request superseded it.
+      older.resolve({ ...snapshot("Superseded"), cursor: 6 });
+      await vi.advanceTimersByTimeAsync(320);
+      expect(client.snapshotPaths).toHaveLength(4);
+      expect(store.getState().snapshot?.sessions[0]?.name).toBe(
+        "Newer event confirmed",
+      );
+    } finally {
+      store.getState().actions.stop();
+    }
+  });
+
+  it("retains heartbeat reconciliation even when a snapshot covers pending events", async () => {
+    vi.useFakeTimers();
+    const client = new FakeClient();
+    client.snapshots.push(
+      Promise.resolve(snapshot()),
+      Promise.resolve({ ...snapshot(), cursor: 5 }),
+      Promise.resolve({ ...snapshot("Externally updated"), cursor: 5 }),
+    );
+    let heartbeat: (() => void) | undefined;
+    const stream = eventStreamHarness();
+    const store = createWebStore(client, {
+      consumeEvents: (options) => {
+        heartbeat = options.onHeartbeat;
+        return stream.consumeEvents(options);
+      },
+    });
+    await store.getState().actions.refreshSnapshot();
+    store.getState().actions.start();
+    try {
+      stream.emit(
+        runtimeEvent(5, "session_renamed", { sessionId: "session-1" }),
+      );
+      await store.getState().actions.refreshSnapshot();
+      heartbeat?.();
+      await vi.advanceTimersByTimeAsync(160);
+      expect(client.snapshotPaths).toHaveLength(3);
+      expect(store.getState().snapshot?.sessions[0]?.name).toBe(
+        "Externally updated",
+      );
+    } finally {
+      store.getState().actions.stop();
+    }
+  });
+
   it("refreshes Session metadata events emitted by another browser tab", async () => {
     vi.useFakeTimers();
     const client = new FakeClient();