fix: deduplicate workbench requests and remove unused map paths

src/features/workbench/hooks/use-operational-tasks.ts

@@ -79,7 +79,7 @@ export function useOperationalTasks(
   const [error, setError] = useState<string | null>(null);
   const [loading, setLoading] = useState(true);
   const [refreshGeneration, setRefreshGeneration] = useState(0);
-  const [eventRetryGeneration, setEventRetryGeneration] = useState(0);
+  const enqueueEvents = useRef<(events: WorkbenchEvent[]) => void>(() => {});
   const eventRetryMsRef = useRef(1_000);
   const forceRefreshRef = useRef(false);
   const detailResetGenerationRef = useRef(sync.detailResetGeneration);
@@ -131,50 +131,76 @@ export function useOperationalTasks(
   }, [getAccessToken, refreshGeneration, sync.detailResetGeneration, sync.snapshot]);
 
   useEffect(() => {
-    const events = syncEvents;
-    if (events.length === 0) return;
     const controller = new AbortController();
     let retryTimer: number | undefined;
-    const lastSequence = events.at(-1)!.event_seq;
-    const { taskIds, deletedTaskIds, notificationsChanged } = collectOperationalTaskEventChanges(events);
-    if (deletedTaskIds.size > 0) {
+    let pendingEvents: WorkbenchEvent[] = [];
+    let appliedSequence = 0;
+    let processing = false;
+    const removeDeletedTasks = (deletedTaskIds: Set<string>) => {
+      if (deletedTaskIds.size === 0) return;
       tasksRef.current = tasksRef.current.filter((task) => !deletedTaskIds.has(task.task_id));
       setTasks(tasksRef.current);
       setNotifications((current) => current.filter((item) => !deletedTaskIds.has(item.task_id)));
-    }
-    void Promise.all([
-      notificationsChanged
-        ? fetchOperationalTaskNotifications(getAccessToken, controller.signal)
-        : Promise.resolve(null),
-      Promise.all([...taskIds].map((taskId) =>
-        fetchOperationalTask(taskId, getAccessToken, controller.signal)
-      ))
-    ]).then(([nextNotifications, changedTasks]) => {
-      if (controller.signal.aborted) return;
-      if (nextNotifications) setNotifications(nextNotifications);
-      if (changedTasks.length > 0) {
-        const changed = new Map(changedTasks.map((task) => [task.task_id, task]));
-        tasksRef.current = [
-          ...changedTasks,
-          ...tasksRef.current.filter((task) => !changed.has(task.task_id))
-        ];
-        setTasks(tasksRef.current);
-      }
-      acknowledgeEventsThrough(lastSequence);
-      eventRetryMsRef.current = 1_000;
-      setError(null);
-    }).catch((nextError) => {
-      if (!controller.signal.aborted) {
-        setError(nextError instanceof Error ? nextError.message : "工况任务增量刷新失败。");
-        retryTimer = window.setTimeout(() => setEventRetryGeneration((current) => current + 1), eventRetryMsRef.current);
-        eventRetryMsRef.current = Math.min(eventRetryMsRef.current * 2, 30_000);
-      }
-    });
+    };
+    const load = async () => {
+      if (processing || controller.signal.aborted) return;
+      processing = true;
+      window.clearTimeout(retryTimer);
+      try {
+        while (pendingEvents.length > 0 && !controller.signal.aborted) {
+          const events = pendingEvents;
+          const lastSequence = events.at(-1)!.event_seq;
+          const { taskIds, notificationsChanged } = collectOperationalTaskEventChanges(events);
+          const [nextNotifications, changedTasks] = await Promise.all([
+            notificationsChanged
+              ? fetchOperationalTaskNotifications(getAccessToken, controller.signal)
+              : Promise.resolve(null),
+            Promise.all([...taskIds].map((taskId) =>
+              fetchOperationalTask(taskId, getAccessToken, controller.signal)
+            ))
+          ]);
+          if (controller.signal.aborted) return;
+          // A deletion received during this read must not be undone by its response.
+          const { deletedTaskIds } = collectOperationalTaskEventChanges(pendingEvents);
+          if (nextNotifications) setNotifications(nextNotifications.filter((item) => !deletedTaskIds.has(item.task_id)));
+          const visibleTasks = changedTasks.filter((task) => !deletedTaskIds.has(task.task_id));
+          if (visibleTasks.length > 0) {
+            const changed = new Map(visibleTasks.map((task) => [task.task_id, task]));
+            tasksRef.current = [
+              ...visibleTasks,
+              ...tasksRef.current.filter((task) => !changed.has(task.task_id))
+            ];
+            setTasks(tasksRef.current);
+          }
+          appliedSequence = lastSequence;
+          pendingEvents = pendingEvents.filter((event) => event.event_seq > appliedSequence);
+          acknowledgeEventsThrough(lastSequence);
+          eventRetryMsRef.current = 1_000;
+          setError(null);
+        }
+      } catch (nextError) {
+        if (!controller.signal.aborted) {
+          setError(nextError instanceof Error ? nextError.message : "工况任务增量刷新失败。");
+          retryTimer = window.setTimeout(() => void load(), eventRetryMsRef.current);
+          eventRetryMsRef.current = Math.min(eventRetryMsRef.current * 2, 30_000);
+        }
+      } finally { processing = false; }
+    };
+    enqueueEvents.current = (events) => {
+      pendingEvents = events.filter((event) => event.event_seq > appliedSequence);
+      removeDeletedTasks(collectOperationalTaskEventChanges(pendingEvents).deletedTaskIds);
+      void load();
+    };
     return () => {
       controller.abort();
-      if (retryTimer !== undefined) window.clearTimeout(retryTimer);
+      enqueueEvents.current = () => {};
+      window.clearTimeout(retryTimer);
     };
-  }, [acknowledgeEventsThrough, eventRetryGeneration, getAccessToken, syncEvents]);
+  }, [acknowledgeEventsThrough, getAccessToken, sync.detailResetGeneration]);
+
+  useEffect(() => {
+    enqueueEvents.current(syncEvents);
+  }, [acknowledgeEventsThrough, getAccessToken, sync.detailResetGeneration, syncEvents]);
 
   const conditions = useMemo(() => tasks.map((task) => ({ ...toOperationalTaskCondition(task),
     hasNewReport: notifications.some((item) => item.task_id === task.task_id && item.kind === "report_updated"

src/features/workbench/hooks/use-workbench-agent.ts

@@ -531,12 +531,15 @@ export function useWorkbenchAgent({
     () => clientRef.current.listResponseInputs(sessionId!),
     { refreshInterval: streaming ? 1000 : 0, shouldRetryOnError: false }
   );
+  const handledRecovery = useRef({ scopeKey, version: 0 });
   useEffect(() => {
+    if (handledRecovery.current.scopeKey === scopeKey && handledRecovery.current.version === recoveryVersion) return;
+    handledRecovery.current = { scopeKey, version: recoveryVersion };
     if (!recoveryVersion) return;
     void refreshBootstrap();
     void refreshInteractionRequests();
     void refreshPendingInputs();
-  }, [recoveryVersion, refreshBootstrap, refreshInteractionRequests, refreshPendingInputs]);
+  }, [scopeKey, recoveryVersion, refreshBootstrap, refreshInteractionRequests, refreshPendingInputs]);
   const updateInputDelivery = useCallback(async (id: string, delivery: AgentInputDelivery) => {
     if (!sessionIdRef.current) return;
     try {

src/features/workbench/hooks/use-workbench-communication.test.tsx

@@ -0,0 +1,298 @@
+import { act, cleanup, renderHook, waitFor } from "@testing-library/react";
+import { afterEach, expect, it, vi } from "vitest";
+import { SWRConfig } from "swr";
+const harness = vi.hoisted(() => ({
+  api: null as any,
+  recovery: 0,
+  flow: vi.fn(),
+  task: vi.fn(),
+  check: vi.fn()
+}));
+vi.mock("@/features/agent", async (original) => ({
+  ...(await original<any>()),
+  createAgentApiClient: () => harness.api
+}));
+vi.mock("@/features/agent/hooks/use-agent-connection", () => ({
+  useAgentConnection: () => ({
+    connectionStatus: "online",
+    recoveryVersion: harness.recovery,
+    recheck: harness.check
+  })
+}));
+vi.mock("@/features/workbench/api/realtime-flow-client", () => ({
+  fetchRealtimePipeFlows: harness.flow
+}));
+vi.mock("@/features/workbench/api/operational-task-client", async (original) => ({
+  ...(await original<any>()),
+  fetchOperationalTask: harness.task,
+  fetchOperationalTaskNotifications: async () => []
+}));
+import { useWorkbenchRuntimeData } from "@/features/workbench/hooks/use-workbench-runtime-data";
+import { useWorkbenchAgent } from "@/features/workbench/hooks/use-workbench-agent";
+import { useOperationalTasks } from "@/features/workbench/hooks/use-operational-tasks";
+import { WorkbenchSessionProvider } from "@/features/workbench/hooks/workbench-session-provider";
+import { useAgentMessageInbox } from "@/features/agent/hooks/use-agent-message-inbox";
+const access: any = {
+  project: { project_id: "p", gs_workspace: "tjwater_next", status: "active" },
+  getAccessToken: async () => "test"
+};
+const sync: any = {
+  snapshot: null,
+  events: [],
+  status: "live",
+  error: null,
+  detailResetGeneration: 0,
+  acknowledgeEventsThrough: vi.fn(),
+  refresh: vi.fn(),
+  subscribePresentationEvents: () => () => {}
+};
+afterEach(() => {
+  cleanup();
+  vi.useRealTimers();
+  vi.restoreAllMocks();
+  harness.flow.mockReset();
+  harness.task.mockReset();
+  harness.recovery = 0;
+});
+function wrapperFactory() {
+  const cache = new Map();
+  return ({ children }: any) => (
+    <WorkbenchSessionProvider owner={access}>
+      <SWRConfig
+        value={{ provider: () => cache, revalidateOnFocus: false, shouldRetryOnError: false }}
+      >
+        {children}
+      </SWRConfig>
+    </WorkbenchSessionProvider>
+  );
+}
+it("keeps one flow request and one polling timer when visibility changes during a request", async () => {
+  vi.useFakeTimers();
+  let visibility = "visible";
+  vi.spyOn(document, "visibilityState", "get").mockImplementation(() => visibility as any);
+  const done: Array<() => void> = [];
+  harness.flow.mockImplementation(
+    () => new Promise((resolve) => done.push(() => resolve({ records: [], observedAt: null })))
+  );
+  renderHook(() => useWorkbenchRuntimeData(access, sync));
+  expect(harness.flow).toHaveBeenCalledTimes(1);
+  await act(async () => {
+    visibility = "hidden";
+    document.dispatchEvent(new Event("visibilitychange"));
+    visibility = "visible";
+    document.dispatchEvent(new Event("visibilitychange"));
+  });
+  expect(harness.flow).toHaveBeenCalledTimes(1);
+  await act(async () => {
+    done[0]();
+  });
+  await act(async () => {
+    await vi.advanceTimersByTimeAsync(15000);
+  });
+  expect(harness.flow).toHaveBeenCalledTimes(2);
+  await act(async () => {
+    done[1]();
+    await vi.advanceTimersByTimeAsync(15000);
+  });
+  expect(harness.flow).toHaveBeenCalledTimes(3);
+});
+it("queues new task events without cancelling or repeating the current detail request", async () => {
+  const resolve: Array<(value: unknown) => void> = [];
+  harness.task.mockImplementation(() => new Promise((done) => resolve.push(done)));
+  const event = (seq: number, id: string) => ({
+    event_seq: seq,
+    topic: "operational.progress.changed",
+    aggregate_type: "operational_task",
+    aggregate_id: id,
+    metadata: { task_id: id }
+  });
+  const first = event(1, "task-a");
+  const { result, rerender, unmount } = renderHook(
+    ({ events }) => useOperationalTasks(access, { ...sync, events }),
+    { initialProps: { events: [first] } }
+  );
+  rerender({ events: [first, event(2, "task-b")] });
+  expect(harness.task.mock.calls.map((call) => call[0])).toEqual(["task-a"]);
+  expect(harness.task.mock.calls[0][2].aborted).toBe(false);
+  await act(async () => {
+    resolve[0]({
+      task_id: "task-a",
+      title: "A",
+      trigger: { kind: "manual" },
+      source: { kind: "operator" },
+      task_type: "event_investigation",
+      evidence: [],
+      updated_at: "2026-11-09T00:00:00Z"
+    });
+  });
+  expect(harness.task.mock.calls.map((call) => call[0])).toEqual(["task-a", "task-b"]);
+  await act(async () => {
+    resolve[1]({
+      task_id: "task-b",
+      title: "B",
+      trigger: { kind: "manual" },
+      source: { kind: "operator" },
+      task_type: "event_investigation",
+      evidence: [],
+      updated_at: "2026-11-09T00:00:00Z"
+    });
+  });
+  expect(result.current.tasks.map((task) => task.task_id).sort()).toEqual(["task-a", "task-b"]);
+  expect(sync.acknowledgeEventsThrough).toHaveBeenLastCalledWith(2);
+  unmount();
+});
+it("refreshes once per health recovery and uses normal loads on session changes", async () => {
+  const api: any = {
+    getModels: vi.fn(async () => ({ models: [], defaultModel: "" })),
+    getUiRegistry: vi.fn(async () => ({})),
+    getFrontendActionRegistry: vi.fn(async () => ({})),
+    listSessionPage: vi.fn(async () => ({ sessions: [], hasMore: false, nextCursor: null })),
+    listTopics: vi.fn(async () => []),
+    listInteractionRequests: vi.fn(async () => ({ items: [] })),
+    listResponseInputs: vi.fn(async () => []),
+    listAgentMessages: vi.fn(async () => ({ items: [], next_cursor: null })),
+    streamAgentMessages: vi.fn(
+      (_id: any, options: any) =>
+        new Promise<void>((resolve) =>
+          options.signal.addEventListener("abort", () => resolve(), { once: true })
+        )
+    ),
+    loadSession: vi.fn(async (id: string) => ({
+      id,
+      sessionId: id,
+      title: id,
+      purpose: "ordinary_chat",
+      messages: [],
+      createdAt: 1,
+      updatedAt: 1,
+      runStatus: "completed",
+      isStreaming: false,
+      isTitleManuallyEdited: false
+    }))
+  };
+  harness.api = api;
+  const { result, rerender } = renderHook(
+    () =>
+      useWorkbenchAgent({
+        getAccessToken: access,
+        onUiEnvelope: () => {},
+        onFrontendAction: async () => ({})
+      }),
+    { wrapper: wrapperFactory() }
+  );
+  await waitFor(() => expect(api.getModels).toHaveBeenCalledTimes(1));
+  await act(async () => {
+    await result.current.loadHistorySession("a");
+  });
+  await waitFor(() =>
+    expect(api.listInteractionRequests.mock.calls.filter((c: any) => c[0] === "a").length).toBe(1)
+  );
+  harness.recovery = 1;
+  rerender();
+  await waitFor(() => expect(api.getModels).toHaveBeenCalledTimes(2));
+  const before = api.getModels.mock.calls.length;
+  await act(async () => {
+    await result.current.loadHistorySession("b");
+  });
+  await waitFor(() => expect(api.getModels.mock.calls.length).toBe(before));
+  await waitFor(() =>
+    expect(api.listInteractionRequests.mock.calls.filter((c: any) => c[0] === "b").length).toBe(1)
+  );
+  const pending = api.listResponseInputs.mock.calls.filter((c: any) => c[0] === "b").length;
+  expect(pending).toBe(1);
+});
+it("loads the inbox once when the stream is ready and refreshes on later notifications", async () => {
+  const list = vi.fn(async () => ({ items: [], next_cursor: null }));
+  const stream = vi.fn((_id: any, options: any) => {
+    options.onMessage();
+    return new Promise<void>((resolve) =>
+      options.signal.addEventListener("abort", () => resolve(), { once: true })
+    );
+  });
+  const api = { listAgentMessages: list, streamAgentMessages: stream } as any;
+  renderHook(() => useAgentMessageInbox(api, "s"), { wrapper: wrapperFactory() });
+  await waitFor(() => expect(list).toHaveBeenCalledTimes(1));
+  await act(async () => {
+    stream.mock.calls[0][1].onMessage();
+  });
+  await waitFor(() => expect(list).toHaveBeenCalledTimes(2));
+});
+
+it("keeps newer task changes and does not revive a task deleted during a read", async () => {
+  const complete: Array<(value: unknown) => void> = [];
+  harness.task.mockImplementation(() => new Promise((resolve) => complete.push(resolve)));
+  const event = (seq: number, deleted = false) => ({
+    event_seq: seq,
+    topic: "operational.task.changed",
+    aggregate_type: "operational_task",
+    aggregate_id: "a",
+    metadata: { task_id: "a", ...(deleted ? { change: "deleted" } : {}) }
+  });
+  const a = event(1);
+  const { result, rerender } = renderHook(
+    ({ events }) => useOperationalTasks(access, { ...sync, events }),
+    {
+      initialProps: { events: [a] }
+    }
+  );
+  rerender({ events: [a, event(2)] });
+  const task = {
+    task_id: "a",
+    title: "A",
+    trigger: { kind: "manual" },
+    source: { kind: "operator" },
+    task_type: "event_investigation",
+    evidence: [],
+    updated_at: "2026-11-09T00:00:00Z"
+  };
+  await act(async () => {
+    complete[0](task);
+  });
+  expect(harness.task).toHaveBeenCalledTimes(2);
+  rerender({ events: [a, event(2), event(3, true)] });
+  expect(result.current.tasks).toEqual([]);
+  await act(async () => {
+    complete[1](task);
+  });
+  await waitFor(() => expect(sync.acknowledgeEventsThrough).toHaveBeenLastCalledWith(3));
+  expect(result.current.tasks).toEqual([]);
+  expect(harness.task).toHaveBeenCalledTimes(2);
+});
+
+it("loads the initial inbox even when its stream cannot connect", async () => {
+  const list = vi.fn(async () => ({ items: [], next_cursor: null }));
+  const api = {
+    listAgentMessages: list,
+    streamAgentMessages: vi.fn(async () => {
+      throw new Error("offline");
+    })
+  } as any;
+  const { result } = renderHook(() => useAgentMessageInbox(api, "s"), {
+    wrapper: wrapperFactory()
+  });
+  await waitFor(() => expect(result.current.loading).toBe(false));
+  expect(list).toHaveBeenCalledTimes(1);
+});
+
+it("loads the inbox when its stream remains pending without a first frame", async () => {
+  vi.useFakeTimers();
+  const list = vi.fn(async () => ({ items: [], next_cursor: null }));
+  const api = {
+    listAgentMessages: list,
+    streamAgentMessages: vi.fn(
+      (_id: string, options: { signal: AbortSignal }) =>
+        new Promise<void>((resolve) => {
+          options.signal.addEventListener("abort", () => resolve(), { once: true });
+        })
+    )
+  } as any;
+  const { result } = renderHook(() => useAgentMessageInbox(api, "s"), {
+    wrapper: wrapperFactory()
+  });
+  expect(list).not.toHaveBeenCalled();
+  await act(async () => {
+    await vi.advanceTimersByTimeAsync(1500);
+  });
+  expect(list).toHaveBeenCalledTimes(1);
+  expect(result.current.loading).toBe(false);
+});

src/features/workbench/hooks/use-workbench-runtime-data.ts

@@ -177,12 +177,15 @@ export function useWorkbenchRuntimeData(
     const controller = new AbortController();
     let timeoutId: number | undefined;
 
+    let refreshing = false;
     const refreshRealtimePipeFlows = async () => {
+      if (controller.signal.aborted || refreshing) return;
       if (document.visibilityState === "hidden") {
         timeoutId = window.setTimeout(refreshRealtimePipeFlows, BACKEND_RUNTIME_REFRESH_INTERVAL_MS);
         return;
       }
       try {
+        refreshing = true;
         const snapshot = await fetchRealtimePipeFlows(getAccessToken, controller.signal);
         if (!controller.signal.aborted) {
           setRealtimePipeFlows(snapshot.records.length > 0 ? snapshot : null);
@@ -190,6 +193,7 @@ export function useWorkbenchRuntimeData(
       } catch {
         if (!controller.signal.aborted) setRealtimePipeFlows(null);
       } finally {
+        refreshing = false;
         if (!controller.signal.aborted) {
           timeoutId = window.setTimeout(
             refreshRealtimePipeFlows,