fix(web): order concurrent sedimentation reads

memhub/apps/web/src/ApplicationsHome.test.tsx

@@ -17,16 +17,24 @@ import {
 import * as api from "./api";
 import { saveConnectionPreferences } from "./connection-preferences";
 import { resetConnectionReconcile } from "./connection-reconcile";
-import { queryClient } from "./query-client";
+import { queryClient, resetClient } from "./query-client";
+import { commitSedimentation } from "./sedimentation-state";
 
-const { summary, processing, history, liveSync, recentConversations } =
-  vi.hoisted(() => ({
-    summary: vi.fn(),
-    processing: vi.fn(),
-    history: vi.fn(),
-    liveSync: vi.fn(),
-    recentConversations: vi.fn(),
-  }));
+const {
+  summary,
+  processing,
+  sedimentation,
+  history,
+  liveSync,
+  recentConversations,
+} = vi.hoisted(() => ({
+  summary: vi.fn(),
+  processing: vi.fn(),
+  sedimentation: vi.fn(),
+  history: vi.fn(),
+  liveSync: vi.fn(),
+  recentConversations: vi.fn(),
+}));
 vi.mock("./call-history-client", async (original) => ({
   ...(await original<typeof import("./call-history-client")>()),
   getCallHistory: history,
@@ -35,6 +43,7 @@ vi.mock("./api", () => ({
   getHomeSummary: summary,
   getLiveSyncHealth: liveSync,
   getMemoryProcessing: processing,
+  getSedimentation: sedimentation,
   getRecentConversations: recentConversations,
   getSources: vi.fn().mockResolvedValue([]),
 }));
@@ -74,6 +83,11 @@ beforeEach(() => {
     observed_at: "2026-09-19T00:00:00Z",
   });
   processing.mockReset().mockResolvedValue({ running: 0, pending: 0 });
+  sedimentation.mockReset().mockResolvedValue({
+    enabled: true,
+    pending_retained: true,
+    imports_unaffected: true,
+  });
   liveSync.mockReset().mockResolvedValue([]);
   recentConversations.mockReset().mockResolvedValue({
     items: [],
@@ -538,6 +552,7 @@ it.each([
 
 afterEach(() => {
   cleanup();
+  resetClient();
   resetConnectionReconcile();
   window.localStorage.clear();
   vi.restoreAllMocks();
@@ -616,7 +631,7 @@ it("keeps authoritative home and processing data when only the source snapshot b
   expect(screen.getByText("601")).toBeTruthy();
   expect(screen.getByText("后台运行中")).toBeTruthy();
   expect(screen.getByText("2 个应用已接入")).toBeTruthy();
-  expect(screen.getAllByText("一切就绪")).toHaveLength(2);
+  expect(screen.getAllByText("整理中")).toHaveLength(2);
   expect(screen.queryByText("状态待核对")).toBeNull();
 });
 it("does not turn a failed home summary into zero or expose a private error", async () => {
@@ -647,11 +662,11 @@ it("never calls a recall event a success when only stored memories are available
 });
 
 it.each([
-  [{ running: 0, pending: 0 }, "一切就绪"],
-  [{ running: 0, pending: 0, waiting_for_model: 1 }, "等待模型"],
-  [{ running: 0, pending: 0, blocked_stages: [{ count: 1 }] }, "排队处理中"],
-  [{ running: 0, pending: 0, failed: 1 }, "需要处理"],
-])("does not confuse task state with readiness: %j", async (value, label) => {
+  { running: 0, pending: 0 },
+  { running: 0, pending: 0, waiting_for_model: 1 },
+  { running: 0, pending: 0, blocked_stages: [{ count: 1 }] },
+  { running: 0, pending: 0, failed: 1 },
+])("shows the enabled switch instead of task counters: %j", async (value) => {
   processing.mockResolvedValue(value);
   render(
     <ApplicationsHome
@@ -661,8 +676,51 @@ it.each([
       onSettings={() => {}}
     />,
   );
-  expect(await screen.findAllByText(label)).toHaveLength(2);
-  expect(screen.queryByText("已就绪")).toBeNull();
+  expect(await screen.findAllByText("整理中")).toHaveLength(2);
+});
+
+it("shows a paused switch immediately even when task rows still say running", async () => {
+  processing.mockResolvedValue({ running: 3, pending: 10 });
+  sedimentation.mockResolvedValue({
+    enabled: false,
+    pending_retained: true,
+    imports_unaffected: true,
+  });
+
+  render(
+    <ApplicationsHome
+      imports={imports}
+      onManage={() => {}}
+      onMemories={() => {}}
+      onSettings={() => {}}
+    />,
+  );
+
+  expect(await screen.findAllByText("已暂停")).toHaveLength(2);
+  expect(screen.getByText("已停止生成新记忆")).toBeTruthy();
+  expect(screen.queryByText("新的对话正在整理")).toBeNull();
+});
+
+it("updates an already mounted home as soon as settings commits the switch receipt", async () => {
+  render(
+    <ApplicationsHome
+      imports={imports}
+      onManage={() => {}}
+      onMemories={() => {}}
+      onSettings={() => {}}
+    />,
+  );
+  expect(await screen.findAllByText("整理中")).toHaveLength(2);
+
+  act(() => {
+    commitSedimentation({
+      enabled: false,
+      pending_retained: true,
+      imports_unaffected: true,
+    });
+  });
+
+  await waitFor(() => expect(screen.getAllByText("已暂停")).toHaveLength(2));
 });
 
 it("keeps the import-in-progress empty state stable without a progress link", async () => {

memhub/apps/web/src/ApplicationsHome.tsx

@@ -40,6 +40,7 @@ import {
 import { subscribePoll } from "./poll-scheduler";
 import { R6Button } from "./R6Controls";
 import { prefetchDataRuntime } from "./runtime-state";
+import { refreshSedimentation, useSedimentation } from "./sedimentation-state";
 import type { useImportSources } from "./use-import-sources";
 import "./applications-home.css";
 
@@ -192,6 +193,7 @@ export function ApplicationsHome({
   // A failed poll keeps the last answer on screen; the page must say it is old.
   const hostStale = hostQuery.isError && hostQuery.data !== undefined;
   const auto = useAutoConnect();
+  const sedimentation = useSedimentation();
   const [summary, setSummary] = useState<HomeSummary | null>(null);
   const [processing, setProcessing] = useState<MemoryProcessingSummary | null>(
     null,
@@ -554,6 +556,7 @@ export function ApplicationsHome({
         getMemoryProcessing(),
         getLiveSyncHealth(),
         getRecentConversations({ limit: 1, recentActivity: true }),
+        refreshSedimentation(),
       ]);
       busy = false;
       if (stopped) return;
@@ -651,19 +654,18 @@ export function ApplicationsHome({
   const apps = imports.sources.filter(
     (s) => s.authorized || s.detected === true || s.location_state === "found",
   );
-  const stateLabel = processing?.running
-    ? "整理中"
-    : processing?.blocked_stages?.some((stage) => stage.count > 0)
-      ? "排队处理中"
-      : processing?.waiting_for_model
-        ? "等待模型"
-        : processing?.failed
-          ? "需要处理"
-          : processing?.pending
-            ? "准备中"
-            : processing
-              ? "一切就绪"
-              : "待核对";
+  const stateLabel =
+    sedimentation?.enabled === true
+      ? "整理中"
+      : sedimentation?.enabled === false
+        ? "已暂停"
+        : "待核对";
+  const stateDetail =
+    sedimentation?.enabled === false
+      ? "已停止生成新记忆"
+      : sedimentation?.enabled === true
+        ? "新的对话正在整理"
+        : "正在核对整理状态";
   const latestClock = latest ? clockParts(latest.updated_at) : null;
   const time = latestClock?.clock ?? (latest ? "时间未知" : null);
   const checked = visible ? clockParts(visible.observed_at) : null;
@@ -835,13 +837,7 @@ export function ApplicationsHome({
               <div>
                 <p>沉淀状态</p>
                 <strong className="r6-home-word">{stateLabel}</strong>
-                <small>
-                  {processing?.running
-                    ? "新的对话正在整理"
-                    : processing?.pending
-                      ? "等待队列开始处理"
-                      : "以当前处理状态为准"}
-                </small>
+                <small>{stateDetail}</small>
               </div>
               <div>
                 <p>最近更新</p>

memhub/apps/web/src/R6Settings.test.tsx

@@ -5,6 +5,7 @@ import userEvent from "@testing-library/user-event";
 import type { Ref } from "react";
 import { afterEach, expect, it, vi } from "vitest";
 import { getModelProfiles, getSedimentation, setSedimentation } from "./api";
+import { queryClient, resetClient } from "./query-client";
 import { R6Settings } from "./R6Settings";
 
 vi.mock("./api", () => ({
@@ -54,6 +55,7 @@ vi.mock("./diagnostics-export", () => ({
 }));
 afterEach(() => {
   cleanup();
+  resetClient();
   vi.clearAllMocks();
   window.oneMemory = undefined;
 });
@@ -94,6 +96,11 @@ it("stops model sedimentation on one switch flip without touching the model conf
   await user.click(toggle);
 
   expect(setSedimentation).toHaveBeenCalledExactlyOnceWith(false);
+  expect(queryClient.getQueryData(["sedimentation"])).toEqual({
+    enabled: false,
+    pending_retained: true,
+    imports_unaffected: true,
+  });
   await screen.findByText(/停止后不再调用模型/);
 });
 

memhub/apps/web/src/R6Settings.tsx

@@ -8,7 +8,7 @@ import {
   useRef,
   useState,
 } from "react";
-import { getModelProfiles, getSedimentation, setSedimentation } from "./api";
+import { getModelProfiles, setSedimentation } from "./api";
 import { ConnectionFallback } from "./ConnectionFallback";
 import { exportDiagnosticsFile } from "./diagnostics-export";
 import { ModelSettings, type ModelSettingsActions } from "./ModelSettings";
@@ -18,6 +18,11 @@ import {
   setLoginPreference,
   useLoginPreference,
 } from "./runtime-state";
+import {
+  commitSedimentation,
+  refreshSedimentation,
+  useSedimentation,
+} from "./sedimentation-state";
 import "./r6-settings.css";
 
 type Group = "model" | "connections" | "help";
@@ -72,7 +77,8 @@ export function R6Settings({
         commit: "unknown",
       },
   );
-  const [sedimentation, setSedimentationState] = useState<boolean | null>(null);
+  const sedimentationState = useSedimentation();
+  const sedimentation = sedimentationState?.enabled ?? null;
   const [sedimentationPending, setSedimentationPending] = useState(false);
   const [sedimentationError, setSedimentationError] = useState("");
   const [sedimentationUnavailable, setSedimentationUnavailable] =
@@ -119,13 +125,11 @@ export function R6Settings({
 
   const readSedimentation = useCallback(async () => {
     try {
-      const state = await getSedimentation();
-      setSedimentationState(state.enabled === true);
+      await refreshSedimentation();
       setSedimentationUnavailable(false);
     } catch {
-      // Left unknown rather than assumed on: showing "已停止" when the switch
-      // could not be read would hide a running model.
-      setSedimentationState(null);
+      // Keep the last confirmed state visible, but disable writes until the
+      // server can confirm it again.
       setSedimentationUnavailable(true);
     }
   }, []);
@@ -141,7 +145,7 @@ export function R6Settings({
     setSedimentationError("");
     try {
       const state = await setSedimentation(enabled);
-      setSedimentationState(state.enabled === true);
+      commitSedimentation(state);
     } catch {
       setSedimentationError(
         enabled

memhub/apps/web/src/sedimentation-state.test.ts

@@ -0,0 +1,61 @@
+import { afterEach, expect, it, vi } from "vitest";
+import { queryClient, resetClient } from "./query-client";
+import {
+  commitSedimentation,
+  refreshSedimentation,
+} from "./sedimentation-state";
+
+const { getSedimentation } = vi.hoisted(() => ({
+  getSedimentation: vi.fn(),
+}));
+
+vi.mock("./api", () => ({ getSedimentation }));
+
+const enabled = {
+  enabled: true,
+  pending_retained: true,
+  imports_unaffected: true,
+};
+const paused = { ...enabled, enabled: false };
+
+afterEach(() => {
+  resetClient();
+  vi.clearAllMocks();
+});
+
+it("keeps a newer switch receipt when an older read lands later", async () => {
+  let finishRead: ((value: typeof enabled) => void) | undefined;
+  getSedimentation.mockImplementationOnce(
+    () =>
+      new Promise<typeof enabled>((resolve) => {
+        finishRead = resolve;
+      }),
+  );
+
+  const olderRead = refreshSedimentation();
+  commitSedimentation(paused);
+  finishRead?.(enabled);
+
+  expect(await olderRead).toEqual(paused);
+  expect(queryClient.getQueryData(["sedimentation"])).toEqual(paused);
+});
+
+it("keeps the newest read when concurrent reads finish out of order", async () => {
+  let finishFirst: ((value: typeof enabled) => void) | undefined;
+  getSedimentation
+    .mockImplementationOnce(
+      () =>
+        new Promise<typeof enabled>((resolve) => {
+          finishFirst = resolve;
+        }),
+    )
+    .mockResolvedValueOnce(paused);
+
+  const first = refreshSedimentation();
+  const second = refreshSedimentation();
+  expect(await second).toEqual(paused);
+
+  finishFirst?.(enabled);
+  expect(await first).toEqual(paused);
+  expect(queryClient.getQueryData(["sedimentation"])).toEqual(paused);
+});

memhub/apps/web/src/sedimentation-state.ts

@@ -0,0 +1,54 @@
+import { useQuery } from "@tanstack/react-query";
+import { getSedimentation, type SedimentationState } from "./api";
+import { onClientReset, queryClient } from "./query-client";
+
+const SEDIMENTATION_KEY = ["sedimentation"] as const;
+const FRESH_MS = 5_000;
+
+queryClient.setQueryDefaults(SEDIMENTATION_KEY, {
+  gcTime: Number.POSITIVE_INFINITY,
+  staleTime: FRESH_MS,
+});
+
+let revision = 0;
+let readsStarted = 0;
+let readsLanded = 0;
+onClientReset(() => {
+  revision = 0;
+  readsStarted = 0;
+  readsLanded = 0;
+});
+
+/**
+ * Reads the server switch without allowing an older request to overwrite a
+ * newer PUT receipt committed by the settings page.
+ */
+export async function refreshSedimentation(): Promise<SedimentationState> {
+  const order = ++readsStarted;
+  const startedAtRevision = revision;
+  const next = await getSedimentation();
+  const current =
+    queryClient.getQueryData<SedimentationState>(SEDIMENTATION_KEY);
+  if (startedAtRevision !== revision && current) return current;
+  if (order < readsLanded && current) return current;
+  readsLanded = order;
+  queryClient.setQueryData(SEDIMENTATION_KEY, next);
+  return next;
+}
+
+/** The PUT response is the latest confirmed server state. */
+export function commitSedimentation(value: SedimentationState) {
+  revision += 1;
+  queryClient.setQueryData(SEDIMENTATION_KEY, value);
+}
+
+export function useSedimentation() {
+  return useQuery(
+    {
+      queryKey: SEDIMENTATION_KEY,
+      queryFn: refreshSedimentation,
+      enabled: false,
+    },
+    queryClient,
+  ).data;
+}