From d7eb01652a54b946a7c46b37ee751ea7a684c8ed Mon Sep 17 00:00:00 2001 From: Huarch Date: Mon, 9 Nov 2026 18:49:32 +0800 Subject: [PATCH] fix: deduplicate workbench requests and remove unused map paths --- src/features/agent/api/client.test.ts | 32 +- src/features/agent/api/client.ts | 58 +--- .../agent/hooks/use-agent-message-inbox.ts | 17 +- .../components/map-dev-panel.test.tsx | 3 - .../workbench/hooks/use-operational-tasks.ts | 100 +++--- .../workbench/hooks/use-workbench-agent.ts | 5 +- .../use-workbench-communication.test.tsx | 298 ++++++++++++++++++ .../hooks/use-workbench-runtime-data.ts | 4 + src/features/workbench/map-workbench-page.tsx | 9 - src/features/workbench/map/scada-analysis.ts | 15 +- src/features/workbench/map/sources.ts | 5 - .../map/workbench-map-controller.test.ts | 42 +-- .../workbench/map/workbench-map-controller.ts | 136 +------- 13 files changed, 403 insertions(+), 321 deletions(-) create mode 100644 src/features/workbench/hooks/use-workbench-communication.test.tsx diff --git a/src/features/agent/api/client.test.ts b/src/features/agent/api/client.test.ts index 8d0fdd5..7b0ccb5 100644 --- a/src/features/agent/api/client.test.ts +++ b/src/features/agent/api/client.test.ts @@ -233,35 +233,11 @@ describe("Agent API client sessions", () => { }); }); - it("falls back when a candidate is not the agent service", async () => { - const fetchMock = vi - .fn() - .mockResolvedValueOnce(new Response(JSON.stringify({ message: "not found" }), { status: 404 })) - .mockResolvedValueOnce( - new Response( - JSON.stringify({ - schema_version: "agent-ui-registry@1", - chart_grammars: [], - components: [], - actions: [] - }), - { status: 200 } - ) - ); + it("uses the configured Agent address once and propagates its error", async () => { + const fetchMock = vi.fn().mockResolvedValue(Response.json({ message: "unavailable" }, { status: 503 })); vi.stubGlobal("fetch", fetchMock); - - await expect(createAgentApiClient(["http://not-agent.local", "http://agent.local"]).getUiRegistry()).resolves.toEqual({ - schema_version: "agent-ui-registry@1", - chart_grammars: [], - components: [], - actions: [] - }); - expect(fetchMock).toHaveBeenNthCalledWith( - 1, - "http://not-agent.local/api/v1/ui-registry", - undefined - ); - expect(fetchMock).toHaveBeenNthCalledWith(2, "http://agent.local/api/v1/ui-registry", undefined); + await expect(createAgentApiClient("http://agent.local/prefix/").getUiRegistry()).rejects.toThrow("unavailable"); + expect(fetchMock).toHaveBeenCalledExactlyOnceWith("http://agent.local/prefix/api/v1/ui-registry", undefined); }); it("sends title update payloads using backend field names", async () => { diff --git a/src/features/agent/api/client.ts b/src/features/agent/api/client.ts index 9ce04dc..2df9ded 100644 --- a/src/features/agent/api/client.ts +++ b/src/features/agent/api/client.ts @@ -220,24 +220,16 @@ export type AgentApiClientOptions = { getProjectId?: () => Promise; }; -const AGENT_API_BASE_URLS = [env.TJWATER_AGENT_URL.replace(/\/$/, "")]; - const API_PATH = "/api/v1"; export function createAgentApiClient( - baseUrls: string | string[] = AGENT_API_BASE_URLS, + baseUrl: string = env.TJWATER_AGENT_URL, options: AgentApiClientOptions = {} ): AgentApiClient { - const candidates = (Array.isArray(baseUrls) ? baseUrls : [baseUrls]).map((item) => item.replace(/\/$/, "")); - let activeBaseUrl = candidates[0] ?? ""; - const setActiveBaseUrl = (baseUrl: string) => { - activeBaseUrl = baseUrl; - }; + const normalizedBaseUrl = baseUrl.replace(/\/$/, ""); const request = (path: string, init?: RequestInit) => - fetchWithFallback( - candidates, - activeBaseUrl, - setActiveBaseUrl, + fetchAgent( + normalizedBaseUrl, path, init, options.getAccessToken, @@ -265,7 +257,7 @@ export function createAgentApiClient( return token; } : undefined; const answerStateRequest = async (path: string, init?: RequestInit) => { - const response = await fetchWithFallback(candidates, activeBaseUrl, setActiveBaseUrl, path, init, + const response = await fetchAgent(normalizedBaseUrl, path, init, checkedAnswerStateToken, options.getProjectId); // An identity change while awaiting the network also invalidates the result. await checkedAnswerStateToken?.(); @@ -275,7 +267,7 @@ export function createAgentApiClient( return { ...createInteractionRequestApi(answerStateRequest), async getHealth(signal) { - const response = await fetchWithFallback(candidates, activeBaseUrl, setActiveBaseUrl, "/health", { signal, cache: "no-store" }, undefined, undefined, ""); + const response = await fetchAgent(normalizedBaseUrl, "/health", { signal, cache: "no-store" }, undefined, undefined, ""); if (!response.ok) throw new Error("Agent 服务暂不可用"); const data: unknown = await response.json(); if (!isRecord(data) || data.ok !== true || data.ready !== true) throw new Error("Agent 服务尚未就绪"); @@ -579,19 +571,14 @@ export function createAgentApiClient( }; } -async function fetchWithFallback( - baseUrls: string[], - activeBaseUrl: string, - setActiveBaseUrl: (baseUrl: string) => void, +async function fetchAgent( + baseUrl: string, path: string, init?: RequestInit, getAccessToken?: AccessTokenProvider, getProjectId?: () => Promise, prefix = API_PATH ) { - const orderedBaseUrls = [activeBaseUrl, ...baseUrls.filter((item) => item !== activeBaseUrl)]; - let lastError: unknown; - let lastResponse: Response | null = null; const accessToken = await getAccessToken?.(); const projectId = await getProjectId?.(); const hasContextHeaders = Boolean(accessToken || projectId); @@ -600,34 +587,7 @@ async function fetchWithFallback( if (projectId) headers?.set("X-Project-Id", projectId); const requestInit = hasContextHeaders ? { ...init, headers } : init; - for (const baseUrl of orderedBaseUrls) { - try { - const response = await fetch(`${baseUrl}${prefix}${path}`, requestInit); - if (response.ok) { - setActiveBaseUrl(baseUrl); - return response; - } - - lastResponse = response; - if (shouldFallbackOnHttpStatus(response.status) && baseUrl !== orderedBaseUrls[orderedBaseUrls.length - 1]) { - await response.body?.cancel(); - continue; - } - return response; - } catch (error) { - lastError = error; - } - } - - if (lastResponse) { - return lastResponse; - } - - throw lastError instanceof Error ? lastError : new Error("Agent API unavailable"); -} - -function shouldFallbackOnHttpStatus(status: number) { - return status === 404 || status === 405 || status === 502 || status === 503 || status === 504; + return fetch(`${baseUrl}${prefix}${path}`, requestInit); } function isRecord(value: unknown): value is Record { diff --git a/src/features/agent/hooks/use-agent-message-inbox.ts b/src/features/agent/hooks/use-agent-message-inbox.ts index dd3978b..5802307 100644 --- a/src/features/agent/hooks/use-agent-message-inbox.ts +++ b/src/features/agent/hooks/use-agent-message-inbox.ts @@ -4,7 +4,7 @@ import type { AgentApiClient } from "../api/client"; export function useAgentMessageInbox(client: AgentApiClient, sessionId: string | null) { const inbox = useSWR(sessionId ? ["agent-message-inbox", sessionId] : null, - ([, id]) => client.listAgentMessages(id), { revalidateOnFocus: true, shouldRetryOnError: false }); + ([, id]) => client.listAgentMessages(id), { revalidateOnMount: false, revalidateOnFocus: true, shouldRetryOnError: false }); const [loadingMore, setLoadingMore] = useState(false); const mutate = inbox.mutate; const refresh = useCallback(() => { void mutate().catch(() => undefined); }, [mutate]); @@ -15,20 +15,27 @@ export function useAgentMessageInbox(client: AgentApiClient, sessionId: string | const controller = new AbortController(); let timer: ReturnType | undefined; let retryDelay = 1500; + let initialized = false; + // Prefer the initial stream notification, but do not wait indefinitely for its first frame. + const initialLoadTimer = setTimeout(() => { initialized = true; refreshRef.current(); }, 1500); function subscribe() { - void client.streamAgentMessages(sessionId!, { signal: controller.signal, onMessage: () => { retryDelay = 1500; refreshRef.current(); } }) - .catch(() => { retryDelay = Math.min(retryDelay * 2, 30_000); }) + void client.streamAgentMessages(sessionId!, { signal: controller.signal, onMessage: () => { initialized = true; clearTimeout(initialLoadTimer); retryDelay = 1500; refreshRef.current(); } }) + .catch(() => { + // A failed subscription must not prevent the initial list from loading. + if (!initialized && !controller.signal.aborted) { initialized = true; clearTimeout(initialLoadTimer); refreshRef.current(); } + retryDelay = Math.min(retryDelay * 2, 30_000); + }) .finally(() => { if (!controller.signal.aborted) timer = setTimeout(subscribe, retryDelay); }); } subscribe(); - return () => { controller.abort(); clearTimeout(timer); }; + return () => { controller.abort(); clearTimeout(timer); clearTimeout(initialLoadTimer); }; }, [client, sessionId]); return { messages: inbox.data?.items ?? [], hasMore: Boolean(inbox.data?.next_cursor), - loading: inbox.isLoading, + loading: Boolean(sessionId) && (inbox.isLoading || (!inbox.data && !inbox.error)), error: inbox.error ? "任务消息暂时无法加载" : null, refresh, loadingMore, diff --git a/src/features/workbench/components/map-dev-panel.test.tsx b/src/features/workbench/components/map-dev-panel.test.tsx index 0872699..3aa5ffa 100644 --- a/src/features/workbench/components/map-dev-panel.test.tsx +++ b/src/features/workbench/components/map-dev-panel.test.tsx @@ -138,8 +138,6 @@ function createCommands() { clearValueLabel: vi.fn(), applyLayerGroupStyle: vi.fn(), resetLayerGroupStyle: vi.fn(), - renderScadaAnalysis: vi.fn(), - clearScadaAnalysis: vi.fn(), resetAll: vi.fn() } satisfies WorkbenchMapCommands; } @@ -163,7 +161,6 @@ function createControllerState( invalid: 0 }, styledGroups: [], - scadaAnalysis: null, ...patch }; } diff --git a/src/features/workbench/hooks/use-operational-tasks.ts b/src/features/workbench/hooks/use-operational-tasks.ts index c6109c9..0b5f9b9 100644 --- a/src/features/workbench/hooks/use-operational-tasks.ts +++ b/src/features/workbench/hooks/use-operational-tasks.ts @@ -79,7 +79,7 @@ export function useOperationalTasks( const [error, setError] = useState(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) => { + 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" diff --git a/src/features/workbench/hooks/use-workbench-agent.ts b/src/features/workbench/hooks/use-workbench-agent.ts index 336e127..2a962b0 100644 --- a/src/features/workbench/hooks/use-workbench-agent.ts +++ b/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 { diff --git a/src/features/workbench/hooks/use-workbench-communication.test.tsx b/src/features/workbench/hooks/use-workbench-communication.test.tsx new file mode 100644 index 0000000..ed52ca5 --- /dev/null +++ b/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()), + 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()), + 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) => ( + + cache, revalidateOnFocus: false, shouldRetryOnError: false }} + > + {children} + + + ); +} +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((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((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((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); +}); diff --git a/src/features/workbench/hooks/use-workbench-runtime-data.ts b/src/features/workbench/hooks/use-workbench-runtime-data.ts index aeb7b9c..b1d5dfe 100644 --- a/src/features/workbench/hooks/use-workbench-runtime-data.ts +++ b/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, diff --git a/src/features/workbench/map-workbench-page.tsx b/src/features/workbench/map-workbench-page.tsx index 7a085e9..7378e17 100644 --- a/src/features/workbench/map-workbench-page.tsx +++ b/src/features/workbench/map-workbench-page.tsx @@ -956,19 +956,15 @@ function MapWorkbenchContent({ } async function handleStartNewAgentSession() { - mapController.clearScadaAnalysis(); await agent.startNewSession(); } async function handleLoadAgentHistorySession(sessionId: string) { - const changesSession = sessionId !== agent.sessionId; const loaded = await agent.loadHistorySession(sessionId); if (!loaded) throw new Error("会话打开已取消"); - if (changesSession) mapController.clearScadaAnalysis(); } async function handleOpenTaskCollaboration(taskId: string, discussion?: InvestigationDiscussion) { - mapController.clearScadaAnalysis(); const id = await agent.openTaskCollaboration(taskId, discussion); if (discussion?.reference && id) agent.setContentReference(discussion.reference); readingWorkspace?.close(); @@ -980,7 +976,6 @@ function MapWorkbenchContent({ target: TaskExecutionHistoryTarget, knownRunId?: string ) { - mapController.clearScadaAnalysis(); let runId = knownRunId; if (!runId) { const runs = await fetchOperationalTaskRuns(taskId, getAccessToken); @@ -999,12 +994,8 @@ function MapWorkbenchContent({ } async function handleDeleteAgentHistorySession(sessionId: string) { - const clearsActiveAnalysis = sessionId === agent.sessionId; const deleted = await agent.deleteHistorySession(sessionId); if (!deleted) throw new Error("会话删除失败，请重试"); - if (deleted && clearsActiveAnalysis) { - mapController.clearScadaAnalysis(); - } } async function handleAgentMapAction( diff --git a/src/features/workbench/map/scada-analysis.ts b/src/features/workbench/map/scada-analysis.ts index c26e77d..0c565a3 100644 --- a/src/features/workbench/map/scada-analysis.ts +++ b/src/features/workbench/map/scada-analysis.ts @@ -1,5 +1,5 @@ import type { Feature, FeatureCollection, Point } from "geojson"; -import type { ExpressionSpecification, Map as MapLibreMap, StyleSpecification } from "maplibre-gl"; +import type { ExpressionSpecification, StyleSpecification } from "maplibre-gl"; import { MAP_STYLE_TOKENS } from "./map-colors"; export const SCADA_ANALYSIS_SOURCE_ID = "agent-scada-analysis"; @@ -121,19 +121,6 @@ export const scadaAnalysisLayers: Layer[] = [ } ]; -export function ensureScadaAnalysisLayers( - map: Pick -): void { - scadaAnalysisLayers.forEach((layer, index) => { - if (map.getLayer(layer.id)) return; - const beforeId = scadaAnalysisLayers - .slice(index + 1) - .map((candidate) => candidate.id) - .find((candidateId) => Boolean(map.getLayer(candidateId))); - map.addLayer(layer, beforeId); - }); -} - export function createEmptyScadaAnalysisCollection(): FeatureCollection { return { type: "FeatureCollection", features: [] }; } diff --git a/src/features/workbench/map/sources.ts b/src/features/workbench/map/sources.ts index 9b278aa..566fe7d 100644 --- a/src/features/workbench/map/sources.ts +++ b/src/features/workbench/map/sources.ts @@ -2,7 +2,6 @@ import type { StyleSpecification, VectorSourceSpecification } from "maplibre-gl" import { GEOSERVER_WORKSPACE, MAP_URL } from "./geoserver-config"; import { MAP_STYLE_TOKENS } from "./map-colors"; -import { SCADA_ANALYSIS_SOURCE_ID, createEmptyScadaAnalysisCollection } from "./scada-analysis"; import { SCADA_SOURCE_ID } from "./scada"; export const WATER_NETWORK_GLOBAL_VIEW = { @@ -325,10 +324,6 @@ export function createBaseStyle(mapboxToken?: string, tiandituToken?: string): S type: "geojson", promoteId: "device_id", data: { type: "FeatureCollection", features: [] } - }, - [SCADA_ANALYSIS_SOURCE_ID]: { - type: "geojson", - data: createEmptyScadaAnalysisCollection() } }; const layers: StyleSpecification["layers"] = [ diff --git a/src/features/workbench/map/workbench-map-controller.test.ts b/src/features/workbench/map/workbench-map-controller.test.ts index 30163b4..0481191 100644 --- a/src/features/workbench/map/workbench-map-controller.test.ts +++ b/src/features/workbench/map/workbench-map-controller.test.ts @@ -20,29 +20,23 @@ describe("WorkbenchMapController", () => { expect(map.easeTo).not.toHaveBeenCalled(); expect(controller.getSnapshot().targets).toEqual([]); }); - it("clears every controller overlay, preserves layer styles, and invalidates pending analysis", async () => { + it("clears controller overlays and preserves layer styles", async () => { const map = createMap(); map.getLayer.mockReturnValue({} as never); const setData = vi.fn(); map.getSource.mockReturnValue({ setData }); - let resolveFeatures = (_value: FeatureCollection) => {}; - const pendingAnalysis = new Promise((resolve) => { resolveFeatures = resolve; }); const controller = new WorkbenchMapController({ getMap: () => map as unknown as MapLibreMap, isReady: () => true, getPadding: () => ({ top: 0, right: 0, bottom: 0, left: 0 }), - fetchFeatures: (source) => source === "scada" ? pendingAnalysis : Promise.resolve(pointCollection) + fetchFeatures: async () => pointCollection }); await controller.highlight({ sourceId: "junctions", featureId: "J-1" }); controller.applyLayerGroupStyle("pipes", { color: "#123456" }); const showingFlow = controller.setFlowVisible(true); - const analysis = controller.renderScadaAnalysis([{ device_id: "J-1", level: "high" }]); controller.clearOverlays(); await showingFlow; - resolveFeatures(pointCollection); - await expect(analysis).rejects.toThrow("ACTION_SUPERSEDED"); - expect(setData).toHaveBeenCalledWith({ type: "FeatureCollection", features: [] }); expect(controller.getSnapshot()).toMatchObject({ - target: null, targets: [], targetResult: null, scadaAnalysis: null, + target: null, targets: [], targetResult: null, flowVisible: false, pending: false, valueLabels: {}, styledGroups: ["pipes"] }); expect(map.removeFeatureState).toHaveBeenCalledWith(expect.objectContaining({ id: "J-1" }), "highlighted"); @@ -552,37 +546,7 @@ describe("WorkbenchMapController", () => { expect(map.removeFeatureState).not.toHaveBeenCalled(); }); - it("renders SCADA analysis through the registered GeoJSON source", async () => { - const map = createMap(); - const source = { setData: vi.fn() }; - map.getSource.mockReturnValue(source); - const controller = createController(map, { - type: "FeatureCollection", - features: [ - { - type: "Feature", - geometry: { type: "Point", coordinates: [120.7, 28] }, - properties: { device_id: "MP01" } - } - ] - }); - await expect( - controller.renderScadaAnalysis([{ device_id: "MP01", level: "high" }]) - ).resolves.toMatchObject({ - rendered_ids: ["MP01"], - fitted: true - }); - expect(source.setData).toHaveBeenCalled(); - expect(controller.getSnapshot()).toMatchObject({ - errorCode: null, - pending: false, - scadaAnalysis: { - rendered_ids: ["MP01"], - fitted: true - } - }); - }); }); const pointCollection: FeatureCollection = { type: "FeatureCollection", features: [{ type: "Feature", geometry: { type: "Point", coordinates: [120.7, 28] }, properties: { id: "J-1" } }] }; diff --git a/src/features/workbench/map/workbench-map-controller.ts b/src/features/workbench/map/workbench-map-controller.ts index 6c703ee..709ac89 100644 --- a/src/features/workbench/map/workbench-map-controller.ts +++ b/src/features/workbench/map/workbench-map-controller.ts @@ -1,5 +1,5 @@ import type { Feature, FeatureCollection, Geometry, Position } from "geojson"; -import type { GeoJSONSource, Map as MapLibreMap, PaddingOptions } from "maplibre-gl"; +import type { Map as MapLibreMap, PaddingOptions } from "maplibre-gl"; import { SOURCE_LAYERS, type WaterNetworkSourceId } from "./sources"; import { PRESENTATION_FEATURE_SOURCE } from "./presentation-highlight-style"; import { @@ -36,14 +36,7 @@ import { type ValueLabelSourceId, type ValueLabelStyle } from "./value-label"; -import { - SCADA_ANALYSIS_SOURCE_ID, - createEmptyScadaAnalysisCollection, - createScadaAnalysisCollection, - ensureScadaAnalysisLayers, - type ScadaAnalysisItem, - type ScadaAnalysisRenderResult -} from "./scada-analysis"; + export type FeatureTarget = { sourceId: WaterNetworkSourceId; featureId: string }; export type FeatureBatchTarget = { sourceId: WaterNetworkSourceId; featureIds: string[] }; @@ -56,7 +49,6 @@ export type WorkbenchMapErrorCode = | "INVALID_FEATURE_IDS" | "WFS_UNAVAILABLE" | "SCADA_API_UNAVAILABLE" - | "SCADA_FEATURES_NOT_FOUND" | "ACTION_SUPERSEDED" | "FLOW_UNAVAILABLE" | "INVALID_FLOW_CALCULATION" @@ -72,7 +64,6 @@ export type WorkbenchMapControllerState = { flowVisible: boolean; flowSummary: PipeFlowSummary; styledGroups: LayerGroupId[]; - scadaAnalysis: ScadaAnalysisRenderResult | null; }; export type ValueLabelRequest = Omit & { unit?: string }; @@ -89,8 +80,6 @@ export type WorkbenchMapCommands = { clearValueLabel: (sourceId?: ValueLabelSourceId) => void; applyLayerGroupStyle: (groupId: LayerGroupId, patch: LayerGroupStylePatch) => void; resetLayerGroupStyle: (groupId: LayerGroupId) => void; - renderScadaAnalysis: (items: ScadaAnalysisItem[], signal?: AbortSignal) => Promise; - clearScadaAnalysis: () => { cleared: true }; resetAll: () => void; }; @@ -114,15 +103,13 @@ export class WorkbenchMapController implements WorkbenchMapCommands { errorCode: null, flowVisible: false, flowSummary: createEmptyPipeFlowSummary(), - styledGroups: [], - scadaAnalysis: null + styledGroups: [] }; private listeners = new Set<() => void>(); private cachedFeatures = new Map(); private flowState: FlowOverlayState = createFlowOverlayState(); private flowUpdateRevision = 0; private flowVisibilityRevision = 0; - private scadaAnalysisRevision = 0; private targetRequestRevision = 0; private pendingTargetSourceId: WaterNetworkSourceId | null = null; private availableSourceIds: ReadonlySet | null = null; @@ -399,125 +386,13 @@ export class WorkbenchMapController implements WorkbenchMapCommands { this.patchState({ styledGroups: this.state.styledGroups.filter((item) => item !== groupId), errorCode: null }); } - async renderScadaAnalysis( - items: ScadaAnalysisItem[], - signal?: AbortSignal - ): Promise { - const revision = ++this.scadaAnalysisRevision; - const map = this.options.getMap(); - if (!map || !this.options.isReady()) { - this.setError("MAP_NOT_READY"); - throw new Error("MAP_NOT_READY"); - } - this.patchState({ pending: true, errorCode: null }); - - let sourceCollection: FeatureCollection; - try { - sourceCollection = await this.fetchFeatures( - "scada", - items.map((item) => item.device_id), - signal - ); - } catch { - if (revision === this.scadaAnalysisRevision) { - this.patchState({ pending: false, errorCode: "SCADA_API_UNAVAILABLE" }); - } - throw new Error( - revision === this.scadaAnalysisRevision - ? "SCADA_API_UNAVAILABLE" - : "ACTION_SUPERSEDED" - ); - } - if (revision !== this.scadaAnalysisRevision) { - throw new Error("ACTION_SUPERSEDED"); - } - - const next = createScadaAnalysisCollection(sourceCollection, items); - if (next.collection.features.length === 0) { - this.patchState({ - pending: false, - errorCode: "SCADA_FEATURES_NOT_FOUND" - }); - throw new Error("SCADA_FEATURES_NOT_FOUND"); - } - const source = map.getSource( - SCADA_ANALYSIS_SOURCE_ID - ) as GeoJSONSource | undefined; - if (!source) { - this.patchState({ pending: false, errorCode: "MAP_NOT_READY" }); - throw new Error("MAP_NOT_READY"); - } - - try { - ensureScadaAnalysisLayers(map); - source.setData(next.collection); - } catch { - this.patchState({ pending: false, errorCode: "MAP_NOT_READY" }); - throw new Error("MAP_NOT_READY"); - } - - const result: ScadaAnalysisRenderResult = { - ...next.result, - fitted: true - }; - this.patchState({ - pending: false, - errorCode: null, - scadaAnalysis: result - }); - const coordinates = next.collection.features.map( - (feature) => feature.geometry.coordinates - ); - if (coordinates.length === 1) { - map.easeTo({ - center: [coordinates[0][0], coordinates[0][1]], - zoom: 18, - padding: this.options.getPadding(), - duration: this.options.reducedMotion?.() ? 0 : 600 - }); - } else { - const bounds = coordinates.reduce<[number, number, number, number]>( - (value, coordinate) => [ - Math.min(value[0], coordinate[0]), - Math.min(value[1], coordinate[1]), - Math.max(value[2], coordinate[0]), - Math.max(value[3], coordinate[1]) - ], - [Infinity, Infinity, -Infinity, -Infinity] - ); - map.fitBounds( - [ - [bounds[0], bounds[1]], - [bounds[2], bounds[3]] - ], - { - padding: this.options.getPadding(), - maxZoom: 18, - duration: this.options.reducedMotion?.() ? 0 : 600 - } - ); - } - return result; - } - - clearScadaAnalysis() { - this.scadaAnalysisRevision += 1; - const map = this.options.getMap(); - (map?.getSource(SCADA_ANALYSIS_SOURCE_ID) as GeoJSONSource | undefined)?.setData( - createEmptyScadaAnalysisCollection() - ); - this.patchState({ scadaAnalysis: null, pending: false, errorCode: null }); - return { cleared: true as const }; - } - clearOverlays = () => { this.clearHighlight(); this.clearValueLabel(); - this.clearScadaAnalysis(); this.flowVisibilityRevision += 1; const map = this.options.getMap(); if (map) hideFlowOverlay(map, this.flowState); - this.patchState({ flowVisible: false }); + this.patchState({ flowVisible: false, pending: false, errorCode: null }); }; resetAll = () => { @@ -536,8 +411,7 @@ export class WorkbenchMapController implements WorkbenchMapCommands { errorCode: null, flowVisible: false, flowSummary: createEmptyPipeFlowSummary(), - styledGroups: [], - scadaAnalysis: null + styledGroups: [] }); };