diff --git a/README.md b/README.md index 5ea7d80..b117c45 100644 --- a/README.md +++ b/README.md @@ -89,6 +89,7 @@ pnpm format # 使用 Prettier 格式化文件 pnpm test # 运行 Vitest 单元测试 pnpm test:watch # 监听模式运行单元测试 pnpm test:browser # 运行 Playwright 浏览器测试 +pnpm check:contracts # 在三端集成工作区只读检查前端生成的 Agent/Server 契约 ``` ## Docker diff --git a/package.json b/package.json index 480cf13..2a38e8b 100644 --- a/package.json +++ b/package.json @@ -10,6 +10,7 @@ "scripts": { "dev": "vite", "export:answer-ui-contract": "node scripts/export-answer-ui-contract.mjs", + "check:contracts": "node scripts/export-answer-ui-contract.mjs --check && node scripts/export-view-command-contract.mjs --check", "build": "tsc --noEmit && vite build && pnpm build:verify-css && pnpm build:tts", "build:verify-css": "node scripts/verify-production-css.mjs", "build:tts": "esbuild server/edge-tts-server.ts --bundle --platform=node --format=cjs --outfile=dist-server/edge-tts-server.cjs", diff --git a/scripts/export-answer-ui-contract.mjs b/scripts/export-answer-ui-contract.mjs index ac2a86c..75c4fe7 100644 --- a/scripts/export-answer-ui-contract.mjs +++ b/scripts/export-answer-ui-contract.mjs @@ -1,11 +1,13 @@ import process from "node:process"; -import { writeFile } from "node:fs/promises"; +import { readFile, writeFile } from "node:fs/promises"; import { resolve } from "node:path"; import { zodToJsonSchema } from "zod-to-json-schema"; import { answerUiCatalog, ANSWER_UI_VERSION } from "../src/features/agent/answer-ui/catalog.ts"; -const targets = process.argv[2] - ? [resolve(process.argv[2])] +const check = process.argv.includes("--check"); +const destination = process.argv.slice(2).find((arg) => arg !== "--check"); +const targets = destination + ? [resolve(destination)] : [ resolve("../next-tjwater-agent/src/chat/answer-ui-catalog.json"), resolve("../next-tjwater-server/contracts/answer-ui-catalog.json") @@ -23,7 +25,12 @@ const contract = { ]) ) }; +const content = `${JSON.stringify(contract, null, 2)}\n`; for (const target of targets) { - await writeFile(target, `${JSON.stringify(contract, null, 2)}\n`); - process.stdout.write(`Exported ${contract.components.length} components to ${target}\n`); + if (check) { + if (await readFile(target, "utf8") !== content) throw new Error(`Answer UI contract differs: ${target}`); + } else { + await writeFile(target, content); + } + process.stdout.write(`${check ? "Checked" : "Exported"} ${contract.components.length} components ${check ? "in" : "to"} ${target}\n`); } diff --git a/src/features/agent/api/client.ts b/src/features/agent/api/client.ts index 3fae7a7..9ce04dc 100644 --- a/src/features/agent/api/client.ts +++ b/src/features/agent/api/client.ts @@ -121,6 +121,7 @@ export type AgentSessionStreamDataEventType = | "todo_update" | "permission_request" | "permission_response" + | "run_status" | "interaction_request" | "question_request" | "question_response" @@ -131,6 +132,7 @@ export type AgentSessionStreamDataEventType = | "ui_envelope" | "session_title" | "done" + | "aborted" | "error" | "auth_required" | "credential_refresh_required" @@ -837,6 +839,7 @@ function isSessionStreamDataEventType(value: string): value is AgentSessionStrea value === "execution_plan" || value === "permission_request" || value === "permission_response" || + value === "run_status" || value === "interaction_request" || value === "question_request" || value === "question_response" || @@ -847,6 +850,7 @@ function isSessionStreamDataEventType(value: string): value is AgentSessionStrea value === "ui_envelope" || value === "session_title" || value === "done" || + value === "aborted" || value === "error" || value === "credential_refresh_required" || value === "credential_refreshed" || diff --git a/src/features/agent/components/agent-command-panel.tsx b/src/features/agent/components/agent-command-panel.tsx index fcbdc91..ab1d46a 100644 --- a/src/features/agent/components/agent-command-panel.tsx +++ b/src/features/agent/components/agent-command-panel.tsx @@ -1066,6 +1066,11 @@ function BackendMessage({ ) ) : null} + {message.role === "assistant" && message.runError && message.content !== `错误：${message.runError}` ? ( +

+ 错误：{message.runError} +

+ ) : null} {message.role === "assistant" && processComplete && hasProcessError ? (

部分查询或工具未完成，可展开处理过程查看原因。 diff --git a/src/features/agent/components/agent-progress-status-icon.tsx b/src/features/agent/components/agent-progress-status-icon.tsx index 3c3c870..3135d09 100644 --- a/src/features/agent/components/agent-progress-status-icon.tsx +++ b/src/features/agent/components/agent-progress-status-icon.tsx @@ -1,9 +1,10 @@ -import { CheckCircle2, LoaderCircle, XCircle } from "lucide-react"; +import { Ban, CheckCircle2, LoaderCircle, XCircle } from "lucide-react"; import { StatusIcon, type StatusTone } from "@/shared/ui/status"; import type { AgentChatProgress } from "../types"; const states = { completed: { label: "已完成", tone: "success", Icon: CheckCircle2 }, + cancelled: { label: "已中止", tone: "neutral", Icon: Ban }, error: { label: "失败", tone: "danger", Icon: XCircle }, running: { label: "进行中", tone: "info", Icon: LoaderCircle } } satisfies Record< diff --git a/src/features/agent/index.ts b/src/features/agent/index.ts index 0c9e12e..97f70c7 100644 --- a/src/features/agent/index.ts +++ b/src/features/agent/index.ts @@ -55,6 +55,7 @@ export { cancelRunningTodos, completeRunningProgress, finalizeAssistantMessageAfterAbort, + finalizeAssistantMessageAfterError, toTodoUpdate, upsertPermission, upsertProgress, diff --git a/src/features/agent/session-state.test.ts b/src/features/agent/session-state.test.ts index e527717..0d1d5d4 100644 --- a/src/features/agent/session-state.test.ts +++ b/src/features/agent/session-state.test.ts @@ -73,6 +73,8 @@ describe("agent session state", () => { requestId: "permission-1", status: "approved_once" }); + expect(applyPermissionResponse(permissions, { request_id: "permission-1", status: "error", error: "native unavailable" })?.[0]) + .toMatchObject({ status: "error", error: "native unavailable" }); }); it("dedupes question requests by tool call id", () => { @@ -152,7 +154,7 @@ describe("agent session state", () => { const next = finalizeAssistantMessageAfterAbort(message); expect(next.content).toBe("请求已中止。"); - expect(next.progress?.[0].status).toBe("completed"); + expect(next.progress?.[0].status).toBe("cancelled"); expect(next.permissions?.[0].status).toBe("aborted"); }); }); diff --git a/src/features/agent/session-state.ts b/src/features/agent/session-state.ts index 80ca998..e5fe827 100644 --- a/src/features/agent/session-state.ts +++ b/src/features/agent/session-state.ts @@ -47,7 +47,10 @@ export function upsertProgress(progress: AgentChatProgress[] | undefined, data: return next; } -export function completeRunningProgress(progress: AgentChatProgress[] | undefined) { +export function completeRunningProgress( + progress: AgentChatProgress[] | undefined, + status: "completed" | "error" | "cancelled" = "completed" +) { return progress?.map((item) => { if (item.status !== "running") { return item; @@ -56,7 +59,7 @@ export function completeRunningProgress(progress: AgentChatProgress[] | undefine const endedAt = Date.now(); return { ...item, - status: "completed" as const, + status, endedAt, elapsedMs: undefined, elapsedSnapshotAt: undefined, @@ -154,9 +157,9 @@ export function applyPermissionResponse( permission.requestId === requestId ? { ...permission, - status: toPermissionStatus(readString(payload.reply)), + status: payload.status === "error" ? "error" as const : toPermissionStatus(readString(payload.reply)), repliedAt: Date.now(), - error: undefined + error: payload.status === "error" ? readString(payload.error) : undefined } : permission ); @@ -235,7 +238,7 @@ export function applyQuestionResponse(questions: AgentQuestionRequest[] | undefi } export function finalizeAssistantMessageAfterAbort(message: AgentChatMessage): AgentChatMessage { - const progress = completeRunningProgress(message.progress); + const progress = completeRunningProgress(message.progress, "cancelled"); const todos = cancelRunningTodos(message.todos); return { @@ -266,6 +269,18 @@ export function finalizeAssistantMessageAfterAbort(message: AgentChatMessage): A }; } +export function finalizeAssistantMessageAfterError(message: AgentChatMessage, reason: string): AgentChatMessage { + const settled = finalizeAssistantMessageAfterAbort(message); + return { + ...settled, + content: message.content || `错误：${reason}`, + runError: reason, + progress: settled.progress?.map((item, index) => message.progress?.[index].status === "running" + ? { ...item, status: "error", detail: item.detail ? `${item.detail}\n${reason}` : reason } + : item) + }; +} + function toPermissionStatus(reply: string | undefined): AgentPermissionStatus { if (reply === "always") { return "approved_always"; @@ -340,7 +355,7 @@ function toToolRef(value: unknown): AgentInteractionToolRef | undefined { } function readProgressStatus(value: unknown): AgentChatProgress["status"] { - return value === "completed" || value === "error" ? value : "running"; + return value === "completed" || value === "error" || value === "cancelled" ? value : "running"; } function readTodoStatus(value: unknown): AgentTodoItem["status"] | undefined { diff --git a/src/features/agent/types.ts b/src/features/agent/types.ts index 45aaaca..63dba1e 100644 --- a/src/features/agent/types.ts +++ b/src/features/agent/types.ts @@ -2,6 +2,7 @@ export type AgentChatMessage = { id: string; role: "user" | "assistant"; content: string; + runError?: string; answerParts?: import("./answer-ui/contract").AnswerUiPart[]; finalAnswer?: { content: string; uiBlockIds: string[] }; progress?: AgentChatProgress[]; @@ -24,7 +25,7 @@ export type AgentChatProgress = { tool?: { name: string; callId?: string; input?: Record }; id: string; phase: string; - status: "running" | "completed" | "error"; + status: "running" | "completed" | "error" | "cancelled"; title: string; detail?: string; startedAt?: number; diff --git a/src/features/workbench/api/gis-properties-client.ts b/src/features/workbench/api/gis-properties-client.ts index 7dbd3e3..542d84e 100644 --- a/src/features/workbench/api/gis-properties-client.ts +++ b/src/features/workbench/api/gis-properties-client.ts @@ -1,13 +1,7 @@ -import { getWorkbenchAccessToken } from "./request-context"; +import { requestWorkbenchJson } from "./request-json"; import { z } from "zod"; import type { WorkbenchAccess } from "./request-context"; -import { env } from "@/shared/config/env"; import type { WaterNetworkSourceId } from "../map/sources"; -import { - createWorkbenchHeaders, - findOwnedWorkbenchProject, - getServerError -} from "./project-client"; const gisAttributeGroupSchema = z.object({ dataset_key: z.string().min(1), @@ -43,7 +37,6 @@ const sourceGisFeaturePropertiesSchema = z.object({ export type GisAttributeGroup = z.infer; -const SERVER_API_BASE_URL = env.TJWATER_SERVER_API_BASE_URL.replace(/\/$/, ""); export async function fetchElementGisProperties( elementType: WaterNetworkSourceId, @@ -52,26 +45,15 @@ export async function fetchElementGisProperties( getAccessToken?: WorkbenchAccess, signal?: AbortSignal ): Promise { - const ownedProject = projectId - ? null - : await findOwnedWorkbenchProject(getAccessToken, signal); - const accessToken = ownedProject?.accessToken ?? await getWorkbenchAccessToken(getAccessToken); - const resolvedProjectId = projectId ?? ownedProject?.project.project_id; - if (!resolvedProjectId) throw new Error("未找到 GIS 台账所属项目"); const path = [ "/api/v1/elements", encodeURIComponent(elementType), encodeURIComponent(elementId), "gis-properties" ].join("/"); - const response = await fetch(`${SERVER_API_BASE_URL}${path}`, { - headers: createWorkbenchHeaders(accessToken, resolvedProjectId), - signal + const payload = await requestWorkbenchJson(path, "GIS 台账属性查询", { + access: getAccessToken, projectId, init: { signal } }); - const payload: unknown = await response.json().catch(() => null); - if (!response.ok) { - throw new Error(getServerError(payload, response.status, "GIS 台账属性查询")); - } const result = elementGisPropertiesSchema.parse(payload); if (result.element_type !== elementType || result.element_id !== elementId) { throw new Error("GIS 台账属性与所选元素不一致"); @@ -86,12 +68,6 @@ export async function fetchSourceGisFeatureProperties( getAccessToken?: WorkbenchAccess, signal?: AbortSignal ): Promise { - const ownedProject = projectId - ? null - : await findOwnedWorkbenchProject(getAccessToken, signal); - const accessToken = ownedProject?.accessToken ?? await getWorkbenchAccessToken(getAccessToken); - const resolvedProjectId = projectId ?? ownedProject?.project.project_id; - if (!resolvedProjectId) throw new Error("未找到 GIS 台账所属项目"); const path = [ "/api/v1/source-gis/layers", encodeURIComponent(layerKey), @@ -99,14 +75,9 @@ export async function fetchSourceGisFeatureProperties( encodeURIComponent(String(sourceFid)), "properties" ].join("/"); - const response = await fetch(`${SERVER_API_BASE_URL}${path}`, { - headers: createWorkbenchHeaders(accessToken, resolvedProjectId), - signal + const payload = await requestWorkbenchJson(path, "源 GIS 属性查询", { + access: getAccessToken, projectId, init: { signal } }); - const payload: unknown = await response.json().catch(() => null); - if (!response.ok) { - throw new Error(getServerError(payload, response.status, "源 GIS 属性查询")); - } const result = sourceGisFeaturePropertiesSchema.parse(payload); if (result.layer_key !== layerKey || result.source_fid !== sourceFid) { throw new Error("源 GIS 属性与所选要素不一致"); diff --git a/src/features/workbench/api/operational-task-client.ts b/src/features/workbench/api/operational-task-client.ts index fc12eb0..f73505a 100644 --- a/src/features/workbench/api/operational-task-client.ts +++ b/src/features/workbench/api/operational-task-client.ts @@ -1,9 +1,8 @@ +import { requestWorkbenchJson } from "./request-json"; import { z } from "zod"; import { investigationReportSchema, investigationDiagramSchema } from "./investigation-report"; import { investigationPresentationCursorSchema, investigationPresentationHistoryPageSchema, type InvestigationPresentationHistoryEvent } from "./investigation-presentation"; import type { WorkbenchAccess } from "./request-context"; -import { env } from "@/shared/config/env"; -import { createWorkbenchHeaders, findOwnedWorkbenchProject, getServerError } from "./project-client"; const taskStatusSchema = z.enum(["queued", "running", "waiting_human", "completed", "failed", "cancelled"]); const taskPrioritySchema = z.enum(["normal", "urgent"]); @@ -432,16 +431,9 @@ export type CreateOperationalTaskInput = { agent_session_id?: string; }; -const SERVER_API_BASE_URL = env.TJWATER_SERVER_API_BASE_URL.replace(/\/$/, ""); async function operationalTaskRequest(path: string, getAccessToken?: WorkbenchAccess, init?: RequestInit) { - const { project, accessToken } = await findOwnedWorkbenchProject(getAccessToken, init?.signal ?? undefined); - const headers = new Headers(createWorkbenchHeaders(accessToken, project.project_id)); - if (init?.body) headers.set("Content-Type", "application/json"); - const response = await fetch(`${SERVER_API_BASE_URL}/api/v1${path}`, { ...init, headers }); - const payload: unknown = await response.json().catch(() => null); - if (!response.ok) throw new Error(getServerError(payload, response.status, "工况任务请求")); - return payload; + return requestWorkbenchJson(`/api/v1${path}`, "工况任务请求", { access: getAccessToken, init }); } export async function fetchOperationalTask(taskId: string, getAccessToken?: WorkbenchAccess, signal?: AbortSignal) { diff --git a/src/features/workbench/api/place-search-client.ts b/src/features/workbench/api/place-search-client.ts index 743b98e..a663266 100644 --- a/src/features/workbench/api/place-search-client.ts +++ b/src/features/workbench/api/place-search-client.ts @@ -1,8 +1,7 @@ -import { getWorkbenchAccessToken } from "./request-context"; +import { requestWorkbenchJson } from "./request-json"; import { z } from "zod"; import type { WorkbenchAccess } from "./request-context"; -import { env } from "@/shared/config/env"; -import { createWorkbenchHeaders, findOwnedWorkbenchProject, getServerError, type PlaceSearchConfiguration } from "./project-client"; +import { type PlaceSearchConfiguration } from "./project-client"; export const MAX_PLACE_SEARCH_RESULTS = 10; @@ -71,21 +70,8 @@ async function requestPlaceData( getAccessToken?: WorkbenchAccess, signal?: AbortSignal ): Promise { - const requestSignal = signal - ? AbortSignal.any([signal, AbortSignal.timeout(20_000)]) - : AbortSignal.timeout(20_000); - const ownedProject = projectId ? null : await findOwnedWorkbenchProject(getAccessToken, requestSignal); - const accessToken = ownedProject?.accessToken ?? await getWorkbenchAccessToken(getAccessToken); - requestSignal.throwIfAborted(); - const resolvedProjectId = projectId ?? ownedProject?.project.project_id; - if (!resolvedProjectId) throw new Error("未找到地名搜索所属项目"); - const response = await fetch(`${env.TJWATER_SERVER_API_BASE_URL.replace(/\/$/, "")}/api/v1/${path}`, { - method: "POST", - headers: createWorkbenchHeaders(accessToken, resolvedProjectId, true), - body: JSON.stringify(body), - signal: requestSignal + return requestWorkbenchJson(`/api/v1/${path}`, "地名搜索", { + access: getAccessToken, projectId, timeoutMs: 20_000, + init: { method: "POST", body: JSON.stringify(body), signal } }); - const payload: unknown = await response.json().catch(() => null); - if (!response.ok) throw new Error(getServerError(payload, response.status, "地名搜索")); - return payload; } diff --git a/src/features/workbench/api/project-request.ts b/src/features/workbench/api/project-request.ts index ad727ff..a241f1c 100644 --- a/src/features/workbench/api/project-request.ts +++ b/src/features/workbench/api/project-request.ts @@ -48,18 +48,19 @@ export async function withProjectRequest( baseUrl: string, operation: string, request: (signal: AbortSignal) => Promise, - signal?: AbortSignal + signal?: AbortSignal, + timeoutMs: number | null = PROJECT_REQUEST_TIMEOUT_MS ): Promise { - const deadline = AbortSignal.timeout(PROJECT_REQUEST_TIMEOUT_MS); - const combined = signal ? AbortSignal.any([signal, deadline]) : deadline; + const deadline = timeoutMs == null ? undefined : AbortSignal.timeout(timeoutMs); + const combined = signal && deadline ? AbortSignal.any([signal, deadline]) : signal ?? deadline ?? new AbortController().signal; try { combined.throwIfAborted(); return await request(combined); } catch (error) { if (signal?.aborted && !isTimeout(signal.reason)) throw signal.reason; - if (isTimeout(error) || (combined.aborted && isTimeout(combined.reason))) { + if (timeoutMs != null && (isTimeout(error) || (combined.aborted && isTimeout(combined.reason)))) { const diagnosis = await describeServerReadiness(baseUrl); - throw new Error(`${operation}超时（已等待 15 秒）。${diagnosis}`, { cause: error }); + throw new Error(`${operation}超时（已等待 ${timeoutMs / 1000} 秒）。${diagnosis}`, { cause: error }); } if (error instanceof TypeError || (error instanceof DOMException && error.name === "NetworkError")) { throw new Error(`${operation}失败：无法连接后端服务。请检查网络和后端服务状态后重试。`, { cause: error }); diff --git a/src/features/workbench/api/realtime-flow-client.ts b/src/features/workbench/api/realtime-flow-client.ts index 8d3eebb..6247943 100644 --- a/src/features/workbench/api/realtime-flow-client.ts +++ b/src/features/workbench/api/realtime-flow-client.ts @@ -1,8 +1,7 @@ +import { requestWorkbenchJson } from "./request-json"; import { z } from "zod"; import type { WorkbenchAccess } from "./request-context"; -import { env } from "@/shared/config/env"; import type { PipeFlowCalculation } from "../map/flow-overlay"; -import { createWorkbenchHeaders, findOwnedWorkbenchProject, getServerError } from "./project-client"; const realtimeLinkSchema = z.object({ time: z.string().datetime({ offset: true }), @@ -11,7 +10,6 @@ const realtimeLinkSchema = z.object({ }); const realtimeLinksSchema = z.array(realtimeLinkSchema); -const SERVER_API_BASE_URL = env.TJWATER_SERVER_API_BASE_URL.replace(/\/$/, ""); const FLOW_LOOKBACK_MS = 15 * 60 * 1000; export type RealtimePipeFlowSnapshot = { @@ -24,22 +22,13 @@ export async function fetchRealtimePipeFlows( signal?: AbortSignal, now = new Date() ): Promise { - const { project, accessToken } = await findOwnedWorkbenchProject(getAccessToken, signal); const query = new URLSearchParams({ start_time: new Date(now.getTime() - FLOW_LOOKBACK_MS).toISOString(), end_time: now.toISOString() }); - const response = await fetch( - `${SERVER_API_BASE_URL}/api/v1/timeseries/realtime/links?${query.toString()}`, - { - headers: createWorkbenchHeaders(accessToken, project.project_id), - signal - } - ); - const payload: unknown = await response.json().catch(() => null); - if (!response.ok) { - throw new Error(getServerError(payload, response.status, "实时管道流速请求")); - } + const payload = await requestWorkbenchJson(`/api/v1/timeseries/realtime/links?${query}`, "实时管道流速请求", { + access: getAccessToken, init: { signal } + }); const rows = realtimeLinksSchema.parse(payload); const latestByPipe = new Map>(); diff --git a/src/features/workbench/api/request-json.test.ts b/src/features/workbench/api/request-json.test.ts new file mode 100644 index 0000000..37cacf5 --- /dev/null +++ b/src/features/workbench/api/request-json.test.ts @@ -0,0 +1,60 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { findOwnedWorkbenchProject } from "./project-client"; +import { requestWorkbenchJson } from "./request-json"; + +vi.mock("./project-client", async (original) => ({ + ...await original(), findOwnedWorkbenchProject: vi.fn() +})); +afterEach(() => { vi.restoreAllMocks(); vi.unstubAllGlobals(); vi.clearAllMocks(); }); + +describe("project JSON transport", () => { + it("reuses resolved credentials and preserves request headers without choosing another project", async () => { + const fetcher = vi.fn().mockResolvedValue(Response.json({ ok: true })); + vi.stubGlobal("fetch", fetcher); + await expect(requestWorkbenchJson("/api/v1/probe", "查询", { + credentials: { projectId: "project-a", accessToken: "token-a" }, + init: { headers: { "X-Request-Id": "request-a" } } + })).resolves.toEqual({ ok: true }); + const headers = new Headers(fetcher.mock.calls[0][1].headers); + expect(headers.get("Authorization")).toBe("Bearer token-a"); + expect(headers.get("X-Project-Id")).toBe("project-a"); + expect(headers.get("X-Request-Id")).toBe("request-a"); + expect(findOwnedWorkbenchProject).not.toHaveBeenCalled(); + }); + + it("rejects an explicitly empty project instead of falling back to a default", async () => { + const fetcher = vi.fn(); vi.stubGlobal("fetch", fetcher); + await expect(requestWorkbenchJson("/api/v1/probe", "查询", { projectId: "" })).rejects.toThrow("所属项目不能为空"); + expect(findOwnedWorkbenchProject).not.toHaveBeenCalled(); + expect(fetcher).not.toHaveBeenCalled(); + }); + + it("preserves caller cancellation before fetching and while reading the body", async () => { + const controller = new AbortController(); + const reason = new DOMException("changed project", "AbortError"); + const fetcher = vi.fn().mockImplementation(async () => ({ ok: true, json: async () => { + controller.abort(reason); throw reason; + } })); + vi.stubGlobal("fetch", fetcher); + const options = { credentials: { projectId: "project-a", accessToken: null }, init: { signal: controller.signal } }; + await expect(requestWorkbenchJson("/api/v1/probe", "查询", options)).rejects.toBe(reason); + await expect(requestWorkbenchJson("/api/v1/probe", "查询", options)).rejects.toBe(reason); + expect(fetcher).toHaveBeenCalledTimes(1); + }); + + it("keeps deadlines opt-in and never retries a failed write", async () => { + const deadline = vi.spyOn(AbortSignal, "timeout"); + const fetcher = vi.fn().mockResolvedValue(Response.json({ detail: "服务繁忙" }, { status: 503 })); + vi.stubGlobal("fetch", fetcher); + await expect(requestWorkbenchJson("/api/v1/probe", "提交", { + credentials: { projectId: "project-a", accessToken: null }, init: { method: "POST", body: "{}" } + })).rejects.toThrow("服务繁忙"); + expect(fetcher).toHaveBeenCalledTimes(1); + expect(deadline).not.toHaveBeenCalled(); + fetcher.mockResolvedValue(Response.json({})); + await requestWorkbenchJson("/api/v1/probe", "地名搜索", { + credentials: { projectId: "project-a", accessToken: null }, timeoutMs: 20_000 + }); + expect(deadline).toHaveBeenCalledWith(20_000); + }); +}); diff --git a/src/features/workbench/api/request-json.ts b/src/features/workbench/api/request-json.ts new file mode 100644 index 0000000..ff9d98f --- /dev/null +++ b/src/features/workbench/api/request-json.ts @@ -0,0 +1,38 @@ +import { env } from "@/shared/config/env"; +import { createWorkbenchHeaders, findOwnedWorkbenchProject, getServerError } from "./project-client"; +import { withProjectRequest } from "./project-request"; +import { getWorkbenchAccessToken, type WorkbenchAccess } from "./request-context"; + +type ProjectCredentials = { projectId: string | undefined; accessToken: string | null | undefined }; + +/** Project-scoped JSON transport. Business validation and write retries belong to callers. */ +export async function requestWorkbenchJson(path: string, operation: string, options: { + access?: WorkbenchAccess; + projectId?: string | null; + credentials?: ProjectCredentials; + init?: RequestInit; + timeoutMs?: number; +} = {}): Promise { + const baseUrl = env.TJWATER_SERVER_API_BASE_URL.replace(/\/$/, ""); + return withProjectRequest(baseUrl, operation, async (signal) => { + let credentials = options.credentials; + if (!credentials) { + if (options.projectId != null) { + if (!options.projectId) throw new Error(`${operation}所属项目不能为空`); + credentials = { projectId: options.projectId, accessToken: await getWorkbenchAccessToken(options.access) }; + } else { + const { project, accessToken } = await findOwnedWorkbenchProject(options.access, signal); + credentials = { projectId: project.project_id, accessToken }; + } + } + signal.throwIfAborted(); + const headers = new Headers(options.init?.headers); + new Headers(createWorkbenchHeaders(credentials.accessToken, credentials.projectId)).forEach((value, key) => headers.set(key, value)); + if (options.init?.body != null && !headers.has("Content-Type")) headers.set("Content-Type", "application/json"); + const response = await fetch(`${baseUrl}${path}`, { ...options.init, headers, signal }); + const payload: unknown = await response.json().catch(() => null); + signal.throwIfAborted(); + if (!response.ok) throw new Error(getServerError(payload, response.status, operation)); + return payload; + }, options.init?.signal ?? undefined, options.timeoutMs ?? null); +} diff --git a/src/features/workbench/api/scada-client.ts b/src/features/workbench/api/scada-client.ts index a9c0444..6548705 100644 --- a/src/features/workbench/api/scada-client.ts +++ b/src/features/workbench/api/scada-client.ts @@ -1,8 +1,8 @@ +import { requestWorkbenchJson } from "./request-json"; import type { Feature, FeatureCollection, Point } from "geojson"; import { getWorkbenchAccessToken, type WorkbenchAccess } from "./request-context"; -import { env } from "@/shared/config/env"; import { z } from "zod"; -import { createWorkbenchHeaders, findOwnedWorkbenchProject, getServerError } from "./project-client"; +import { findOwnedWorkbenchProject } from "./project-client"; const scadaDeviceSchema = z.object({ device_id: z.string().trim().min(1), @@ -68,7 +68,6 @@ export type ScadaSnapshot = { telemetryStatus: "ready" | "empty" | "unavailable"; }; -const SERVER_API_BASE_URL = env.TJWATER_SERVER_API_BASE_URL.replace(/\/$/, ""); const SCADA_LOOKBACK_MS = 15 * 60 * 1000; const SCADA_READING_BATCH_SIZE = 100; @@ -244,11 +243,5 @@ async function requestJson( projectId: string | undefined, signal: AbortSignal | undefined ) { - const headers = createWorkbenchHeaders(accessToken, projectId); - const response = await fetch(`${SERVER_API_BASE_URL}${path}`, { headers, signal }); - const payload: unknown = await response.json().catch(() => null); - if (!response.ok) { - throw new Error(getServerError(payload, response.status, "SCADA API请求")); - } - return payload; + return requestWorkbenchJson(path, "SCADA API请求", { credentials: { accessToken, projectId }, init: { signal } }); } diff --git a/src/features/workbench/api/scada-replay-client.ts b/src/features/workbench/api/scada-replay-client.ts index 479a5ea..1465179 100644 --- a/src/features/workbench/api/scada-replay-client.ts +++ b/src/features/workbench/api/scada-replay-client.ts @@ -1,6 +1,6 @@ +import { requestWorkbenchJson } from "./request-json"; import { z } from "zod"; -import { env } from "@/shared/config/env"; -import { createWorkbenchHeaders, findOwnedWorkbenchProject, getServerError } from "./project-client"; +import { findOwnedWorkbenchProject } from "./project-client"; import { getWorkbenchAccessToken, type WorkbenchAccess } from "./request-context"; const ruleSchema = z.object({ @@ -128,14 +128,10 @@ export class ScadaReplayClient { this.projectId ??= findOwnedWorkbenchProject(this.access).then(({ project }) => project.project_id) .catch((error) => { this.projectId = null; throw error; }); const [projectId, token] = await Promise.all([this.projectId, getWorkbenchAccessToken(this.access)]); - const headers = createWorkbenchHeaders(token, projectId); - if (body !== undefined) headers.set("Content-Type", "application/json"); - const response = await fetch(`${env.TJWATER_SERVER_API_BASE_URL.replace(/\/$/, "")}/api/v1/scada-replays${path}`, { - method: body === undefined ? "GET" : "POST", headers, signal, - ...(body === undefined ? {} : { body: JSON.stringify(body) }) + return requestWorkbenchJson(`/api/v1/scada-replays${path}`, "SCADA 回放", { + credentials: { projectId, accessToken: token }, + init: { method: body === undefined ? "GET" : "POST", signal, + ...(body === undefined ? {} : { body: JSON.stringify(body) }) } }); - const payload: unknown = await response.json().catch(() => null); - if (!response.ok) throw new Error(getServerError(payload, response.status, "SCADA 回放")); - return payload; } } diff --git a/src/features/workbench/api/workbench-analysis-history-client.ts b/src/features/workbench/api/workbench-analysis-history-client.ts index b72beb1..512d3f5 100644 --- a/src/features/workbench/api/workbench-analysis-history-client.ts +++ b/src/features/workbench/api/workbench-analysis-history-client.ts @@ -1,7 +1,6 @@ +import { requestWorkbenchJson } from "./request-json"; import { z } from "zod"; import type { WorkbenchAccess } from "./request-context"; -import { env } from "@/shared/config/env"; -import { createWorkbenchHeaders, findOwnedWorkbenchProject, getServerError } from "./project-client"; const analysisHistoryItemSchema = z.object({ run_id: z.string().trim().min(1), @@ -41,7 +40,6 @@ export type WorkbenchAnalysisHistoryPage = { nextOffset: number | null; }; -const SERVER_API_BASE_URL = env.TJWATER_SERVER_API_BASE_URL.replace(/\/$/, ""); export async function fetchWorkbenchAnalysisHistory( options: { @@ -59,25 +57,13 @@ export async function fetchWorkbenchAnalysisHistory( if (!Number.isInteger(offset) || offset < 0) { throw new Error("分析成果历史偏移量无效"); } - const { project, accessToken } = await findOwnedWorkbenchProject( - options.getAccessToken, - options.signal - ); const search = new URLSearchParams({ limit: String(limit), offset: String(offset) }); - const response = await fetch( - `${SERVER_API_BASE_URL}/api/v1/workbench/analysis-runs?${search}`, - { - headers: createWorkbenchHeaders(accessToken, project.project_id), - signal: options.signal - } - ); - const payload: unknown = await response.json().catch(() => null); - if (!response.ok) { - throw new Error(getServerError(payload, response.status, "分析成果历史请求")); - } + const payload = await requestWorkbenchJson(`/api/v1/workbench/analysis-runs?${search}`, "分析成果历史请求", { + access: options.getAccessToken, init: { signal: options.signal } + }); const page = analysisHistoryPageSchema.parse(payload); return { items: page.items.map((item) => ({ diff --git a/src/features/workbench/api/workbench-runtime-client.ts b/src/features/workbench/api/workbench-runtime-client.ts index f3d7c84..2a3d30a 100644 --- a/src/features/workbench/api/workbench-runtime-client.ts +++ b/src/features/workbench/api/workbench-runtime-client.ts @@ -1,7 +1,7 @@ +import { requestWorkbenchJson } from "./request-json"; import { z } from "zod"; import type { FeatureCollection, Geometry, Position } from "geojson"; import type { WorkbenchAccess } from "./request-context"; -import { env } from "@/shared/config/env"; import { formatDateTime } from "@/shared/utils/date-time"; import type { ScheduledConditionRecord, @@ -10,7 +10,6 @@ import type { WorkbenchScenario } from "../types"; import type { AnalysisArtifact, ArtifactLifecycle, ArtifactType } from "../workspace/workspace-model"; -import { createWorkbenchHeaders, findOwnedWorkbenchProject, getServerError } from "./project-client"; // Keep legacy analysis values parseable while old servers are rolling forward; // they are filtered before the operational workbench model is created. @@ -159,7 +158,6 @@ export type WorkbenchAnalysisProjection = { simulationOverlay: FeatureCollection | null; }; -const SERVER_API_BASE_URL = env.TJWATER_SERVER_API_BASE_URL.replace(/\/$/, ""); export function parseWorkbenchRuntimeSnapshot(payload: unknown): WorkbenchRuntimeSnapshot { const snapshot = runtimeSnapshotSchema.parse(payload); @@ -264,18 +262,9 @@ export async function fetchWorkbenchAnalysisRun( ): Promise { const normalizedRunId = runId.trim(); if (!normalizedRunId) throw new Error("分析运行 ID 不能为空"); - const { project, accessToken } = await findOwnedWorkbenchProject(getAccessToken, signal); - const response = await fetch( - `${SERVER_API_BASE_URL}/api/v1/workbench/analysis-runs/${encodeURIComponent(normalizedRunId)}`, - { - headers: createWorkbenchHeaders(accessToken, project.project_id), - signal - } - ); - const payload: unknown = await response.json().catch(() => null); - if (!response.ok) { - throw new Error(getServerError(payload, response.status, "分析运行成果请求")); - } + const payload = await requestWorkbenchJson(`/api/v1/workbench/analysis-runs/${encodeURIComponent(normalizedRunId)}`, "分析运行成果请求", { + access: getAccessToken, init: { signal } + }); const analysis = runtimeLatestAnalysisSchema.parse(payload); if (analysis.run_id !== normalizedRunId) { throw new Error("分析运行成果与请求 ID 不一致"); diff --git a/src/features/workbench/api/workbench-sync-client.ts b/src/features/workbench/api/workbench-sync-client.ts index c8a4309..71f6d74 100644 --- a/src/features/workbench/api/workbench-sync-client.ts +++ b/src/features/workbench/api/workbench-sync-client.ts @@ -1,3 +1,4 @@ +import { requestWorkbenchJson } from "./request-json"; import { z } from "zod"; import type { WorkbenchAccess } from "./request-context"; import { env } from "@/shared/config/env"; @@ -63,13 +64,9 @@ export async function fetchWorkbenchSyncSnapshot( getAccessToken?: WorkbenchAccess, signal?: AbortSignal ): Promise { - const { project, accessToken } = await findOwnedWorkbenchProject(getAccessToken, signal); - const response = await fetch(`${SERVER_API_BASE_URL}/api/v1/workbench/sync-snapshot`, { - headers: createWorkbenchHeaders(accessToken, project.project_id), - signal + const payload = await requestWorkbenchJson("/api/v1/workbench/sync-snapshot", "工作台同步快照请求", { + access: getAccessToken, init: { signal } }); - const payload: unknown = await response.json().catch(() => null); - if (!response.ok) throw new Error(getServerError(payload, response.status, "工作台同步快照请求")); const envelope = snapshotEnvelopeSchema.parse(payload); return { eventCursor: envelope.event_cursor, diff --git a/src/features/workbench/components/agent-execution-plan-panel.tsx b/src/features/workbench/components/agent-execution-plan-panel.tsx index 93285c2..c87c637 100644 --- a/src/features/workbench/components/agent-execution-plan-panel.tsx +++ b/src/features/workbench/components/agent-execution-plan-panel.tsx @@ -1,3 +1,4 @@ +import { getReportPresentationState, snapshotRound } from "../utils/report-presentation-state"; import { InvestigationReportView } from "./investigation-report-view"; import { formatDateTime as formatDisplayDateTime } from "@/shared/utils/date-time"; import { InvestigationTaskReport } from "./investigation-task-report"; @@ -110,18 +111,8 @@ export function AgentExecutionPlanPanel({ />; const plan = task.latest_plan?.plan; const report = task.latest_report?.report; - const planGeneration = task.latest_plan?.execution_generation; - const planState = planGeneration == null - ? "historical" - : planGeneration === task.execution_generation - ? "current" - : "previous"; - const reportGeneration = task.latest_report?.execution_generation; - const reportState = reportGeneration == null - ? "historical" - : reportGeneration === task.execution_generation - ? "current" - : "previous"; + const planState = snapshotRound(task.latest_plan?.execution_generation, task.execution_generation); + const { round: reportState } = getReportPresentationState(task, task.latest_report); const visiblePlanNodeIds = plan && plan.schema === "execution-plan@2" ? activePlanGraph(plan).visible : null; const visiblePlanNodes = plan && visiblePlanNodeIds diff --git a/src/features/workbench/components/investigation-report-view.tsx b/src/features/workbench/components/investigation-report-view.tsx index 67eeb09..cc7fbcc 100644 --- a/src/features/workbench/components/investigation-report-view.tsx +++ b/src/features/workbench/components/investigation-report-view.tsx @@ -1,3 +1,4 @@ +import { getReportPresentationState } from "../utils/report-presentation-state"; import { FileText } from "lucide-react"; import { StatusBadge } from "@/shared/ui/status"; import { formatDateTime } from "@/shared/utils/date-time"; @@ -7,9 +8,7 @@ import type { OperationalTask } from "../api/operational-task-client"; export function InvestigationReportView({ task, reportLabel }: { task: OperationalTask; reportLabel: string }) { const snapshot = task.latest_report; const report = snapshot?.report; - const insufficient = report?.outcome === "insufficient_data"; - const current = snapshot?.execution_generation === task.execution_generation; - const ended = current && task.lifecycle_status === "completed"; + const { insufficient, current, ended } = getReportPresentationState(task, snapshot); const insufficientLabel = ended ? "本轮已结束 · 无法形成结论" : task.lifecycle_status === "failed" ? "证据不足 · 本轮执行失败" : task.lifecycle_status === "cancelled" ? "证据不足 · 本轮已取消" diff --git a/src/features/workbench/components/investigation-task-report.tsx b/src/features/workbench/components/investigation-task-report.tsx index 814b10c..e2adb49 100644 --- a/src/features/workbench/components/investigation-task-report.tsx +++ b/src/features/workbench/components/investigation-task-report.tsx @@ -1,3 +1,4 @@ +import { getReportPresentationState } from "../utils/report-presentation-state"; import { useEffect, useRef, useState, type ReactNode } from "react"; import { InvestigationQuestions } from "./investigation-questions"; import { History, LoaderCircle, MessageSquare } from "lucide-react"; @@ -179,22 +180,13 @@ export function InvestigationTaskReport({ // A current final report owns its archived graph. During investigation, a // progress snapshot owns its whole graph, including an explicit null. A prior // round's final report must never hide the new round's progress. - const currentFinalReport = snapshot?.execution_generation === task.execution_generation - && report?.stage === "final"; + const { currentFinalReport, previousRound, olderRevision, insufficient } = getReportPresentationState(task, snapshot); const diagram = live ? progress && !currentFinalReport ? progress.progress.diagram : investigationDiagram(task, snapshot) : selection.diagram; const diagramRunning = Boolean(live && !currentFinalReport && progress && task.active_run?.run_id === progress.run_id && task.active_run.status === "running"); - const previousRound = Boolean( - snapshot && snapshot.execution_generation !== task.execution_generation - ); - const olderRevision = Boolean( - snapshot && task.latest_report && snapshot.revision !== task.latest_report.revision - ); const latestAvailable = !live; - const insufficient = report?.outcome === "insufficient_data"; - const currentInsufficient = task.latest_report?.execution_generation === task.execution_generation - && task.latest_report?.report.outcome === "insufficient_data"; + const { currentInsufficient } = getReportPresentationState(task, task.latest_report); const pendingActions = task.actions.filter(isPendingInvestigationAction); const completedActions = task.actions.filter((action) => !isPendingInvestigationAction(action)); const evidence = task.evidence.filter((item) => report?.evidence_refs.includes(item.evidence_id)); diff --git a/src/features/workbench/hooks/agent-interaction-replies.test.ts b/src/features/workbench/hooks/agent-interaction-replies.test.ts new file mode 100644 index 0000000..ec147dc --- /dev/null +++ b/src/features/workbench/hooks/agent-interaction-replies.test.ts @@ -0,0 +1,25 @@ +import { expect, it, vi } from "vitest"; +import type { AgentQuestionRequest } from "@/features/agent"; +import { createAgentInteractionReplies } from "./agent-interaction-replies"; +import type { AgentSessionRuntime } from "./agent-session-runtime"; +import { WorkbenchSessionResources } from "./workbench-session-resources"; + +it("keeps a reply bound to its original session when the user switches conversations", async () => { + let resolve: () => void = () => {}; + const request = vi.fn(() => new Promise((done) => { resolve = done; })); + const original = { sessionId: "original", client: { replyQuestion: request }, questionOverrides: {} } as unknown as AgentSessionRuntime; + const next = { sessionId: "next", questionOverrides: {} } as AgentSessionRuntime; + const resources = new WorkbenchSessionResources(); resources.activeAgent = original; + const setters = { permission: vi.fn(), question: vi.fn() }; + const replies = createAgentInteractionReplies(resources, setters); + const question = { sessionId: "original", requestId: "question" } as AgentQuestionRequest; + const submitted = replies.replyQuestion(question, [["A"]]); + expect(original.questionOverrides.question.status).toBe("submitting"); + resources.activeAgent = next; setters.question.mockClear(); + resolve(); await submitted; + expect(original.questionOverrides.question).toEqual({ status: "answered", answers: [["A"]] }); + expect(next.questionOverrides).toEqual({}); + expect(setters.question).not.toHaveBeenCalled(); + await replies.replyQuestion(question, [["B"]]); + expect(request).toHaveBeenCalledTimes(1); +}); diff --git a/src/features/workbench/hooks/agent-interaction-replies.ts b/src/features/workbench/hooks/agent-interaction-replies.ts new file mode 100644 index 0000000..ad3ff17 --- /dev/null +++ b/src/features/workbench/hooks/agent-interaction-replies.ts @@ -0,0 +1,60 @@ +import type { AgentPermissionReply, AgentPermissionRequest, AgentQuestionRequest } from "@/features/agent"; +import { showMapNotice } from "@/features/map/core"; +import { toPermissionStatus, type PermissionOverride, type QuestionOverride } from "./agent-session-message-state"; +import type { AgentSessionRuntime } from "./agent-session-runtime"; +import type { WorkbenchSessionResources } from "./workbench-session-resources"; + +export function createAgentInteractionReplies(resources: WorkbenchSessionResources, setters: { + permission: (value: Record) => void; + question: (value: Record) => void; +}) { + const permissionUpdate = (target: AgentSessionRuntime, id: string, value: PermissionOverride) => { + target.permissionOverrides = { ...target.permissionOverrides, [id]: value }; + if (resources.activeAgent === target) setters.permission(target.permissionOverrides); + resources.notify(); + }; + const questionUpdate = (target: AgentSessionRuntime, id: string, value: QuestionOverride) => { + target.questionOverrides = { ...target.questionOverrides, [id]: value }; + if (resources.activeAgent === target) setters.question(target.questionOverrides); + resources.notify(); + }; + const submit = async ( + request: { sessionId: string; requestId: string }, + update: (target: AgentSessionRuntime, id: string, value: T | { status: "submitting" } | { status: "error"; error: string }) => void, + execute: (target: AgentSessionRuntime) => Promise, + success: T, + failure: { id: string; title: string; message: string } + ) => { + const target = resources.activeAgent; + if (!target || target.sessionId !== request.sessionId) return; + const apply = (value: T | { status: "submitting" } | { status: "error"; error: string }) => update(target, request.requestId, value); + apply({ status: "submitting" }); + try { + await execute(target); + apply(success); + } catch (error) { + const message = error instanceof Error ? error.message : failure.message; + apply({ status: "error", error: message }); + showMapNotice({ id: failure.id, tone: "error", title: failure.title, message }); + } + }; + const questionFailure = { id: "agent-question-status", title: "Agent 问题提交失败", message: "问题回复提交失败。" }; + return { + replyPermission: (permission: AgentPermissionRequest, reply: AgentPermissionReply) => submit( + permission, permissionUpdate, + (target) => target.client.replyPermission(permission.requestId, { sessionId: permission.sessionId, reply }), + { status: toPermissionStatus(reply) }, + { id: "agent-permission-status", title: "Agent 权限提交失败", message: "权限请求提交失败。" } + ), + replyQuestion: (question: AgentQuestionRequest, answers: string[][]) => submit( + question, questionUpdate, + (target) => target.client.replyQuestion(question.requestId, { sessionId: question.sessionId, answers }), + { status: "answered", answers }, questionFailure + ), + rejectQuestion: (question: AgentQuestionRequest) => submit( + question, questionUpdate, + (target) => target.client.rejectQuestion(question.requestId, { sessionId: question.sessionId }), + { status: "rejected" }, { ...questionFailure, message: "问题跳过失败。" } + ) + }; +} diff --git a/src/features/workbench/hooks/agent-session-message-state.test.ts b/src/features/workbench/hooks/agent-session-message-state.test.ts index 107561d..bd63cbf 100644 --- a/src/features/workbench/hooks/agent-session-message-state.test.ts +++ b/src/features/workbench/hooks/agent-session-message-state.test.ts @@ -125,3 +125,76 @@ it("keeps a malformed persisted data payload from crashing the conversation", () expect(toAgentChatMessages(messages, {}, {})[0].content).toBe("这段结构化内容无法显示。"); } }); + +it("restores an aborted run's terminal state from canonical history", () => { + const messages = toAgentUiMessages([{ id: "answer", role: "assistant", parts: [ + { type: "text", text: "已核查的部分结果" }, + { type: "data-progress", data: { id: "query", title: "查询", status: "running" } }, + { type: "data-run_status", data: { status: "aborted", message: "请求已中断" } } + ] }]); + expect(toAgentChatMessages(messages, {}, {})[0]).toMatchObject({ content: "已核查的部分结果", progress: [{ status: "cancelled" }] }); +}); + +describe("terminal tool progress", () => { + const parts = [ + { type: "text", text: "已核查的部分结果" }, + { type: "data-progress", data: { id: "tool", phase: "tool", title: "正在处理", status: "running", tool: { name: "query", input: { id: 12 } } } }, + { type: "data-todo_update", data: { session_id: "s", todos: [{ id: "todo", content: "核查", status: "in_progress" }] } }, + { type: "data-permission_request", data: { request_id: "permission", session_id: "s", permission: "read", patterns: [] } }, + { type: "data-question_request", data: { request_id: "question", session_id: "s", questions: [{ question: "继续吗？", options: [] }] } } + ]; + const initial = () => toAgentUiMessages([{ id: "answer", role: "assistant", parts }]); + + it.each(["done", "error", "aborted"] as const)("settles running tool rows after resumed %s", (type) => { + const messages = applySessionStreamEvent(initial(), { type, data: { message: "模型请求失败" } }); + const answer = toAgentChatMessages(messages, {}, {})[0]; + expect(answer.progress?.[0]).toMatchObject({ status: type === "done" ? "completed" : type === "aborted" ? "cancelled" : "error", tool: { name: "query", input: { id: 12 } } }); + expect(answer.todos?.todos[0].status).toBe("cancelled"); + expect(answer.content).toBe("已核查的部分结果"); + if (type === "error") expect(answer.runError).toBe("模型请求失败"); + }); + + it("preserves the reason when reopening a failed answer with partial structured output", () => { + const messages = toAgentUiMessages([{ id: "answer", role: "assistant", parts: [...parts, + { type: "data-final_answer", data: { content: "已核查的部分结果" } }, + { type: "data-run_status", data: { status: "error", message: "模型请求失败" } } + ] }]); + const answer = toAgentChatMessages(messages, {}, {})[0]; + expect(answer).toMatchObject({ content: "已核查的部分结果", runError: "模型请求失败", progress: [{ status: "error", detail: "模型请求失败" }] }); + }); + + it("does not let an old reply override revive stopped cards", () => { + const messages = applySessionStreamEvent(initial(), { type: "run_status", data: { status: "aborted" } }); + const answer = toAgentChatMessages(messages, { permission: { status: "error", error: "late HTTP response" } }, + { question: { status: "submitting" } })[0]; + expect(answer.permissions?.[0].status).toBe("aborted"); + expect(answer.questions?.[0].status).toBe("rejected"); + const stopped = applySessionStreamEvent(messages, { type: "error", data: { message: "请求已中断" } }); + expect(toAgentChatMessages(stopped, {}, {})[0].runError).toBeUndefined(); + expect(toAgentChatMessages(stopped, {}, {})[0].progress?.[0].status).toBe("cancelled"); + }); + + it("settles legacy history using the session status while leaving active tools running", () => { + for (const status of ["completed", "aborted", "error", "waiting_input", "running"] as const) { + const messages = toAgentUiMessages(initial(), { runStatus: status, isStreaming: status === "running" }); + expect(toAgentChatMessages(messages, {}, {})[0].progress?.[0].status) + .toBe(status === "running" ? "running" : status === "error" ? "error" : status === "aborted" ? "cancelled" : "completed"); + } + const state = applySessionStreamEvent(initial(), { type: "state", sessionId: "s", isStreaming: false, messages: initial() }); + expect(toAgentChatMessages(state, {}, {})[0].progress?.[0].status).toBe("completed"); + }); + + it("keeps a saved failure authoritative when a legacy session defaults to completed", () => { + const saved = applySessionStreamEvent(initial(), { type: "error", data: { message: "历史错误" } }); + const answer = toAgentChatMessages(toAgentUiMessages(saved, { runStatus: "completed", isStreaming: false }), {}, {})[0]; + expect(answer.runError).toBe("历史错误"); + }); + + it("routes historical question replies to the owning assistant without changing the current turn", () => { + const messages = [...initial(), ...toAgentUiMessages([{ id: "current", role: "assistant", parts: [{ type: "text", text: "新一轮" }] }])]; + const updated = applySessionStreamEvent(messages, { type: "question_response", data: { request_id: "question", rejected: true } }); + const answers = toAgentChatMessages(updated, {}, {}); + expect(answers[0].questions?.[0].status).toBe("rejected"); + expect(updated[1]).toEqual(messages[1]); + }); +}); diff --git a/src/features/workbench/hooks/agent-session-message-state.ts b/src/features/workbench/hooks/agent-session-message-state.ts index 44a67c8..5026b0b 100644 --- a/src/features/workbench/hooks/agent-session-message-state.ts +++ b/src/features/workbench/hooks/agent-session-message-state.ts @@ -2,6 +2,10 @@ import type { UIMessage } from "ai"; import type { Dispatch, SetStateAction } from "react"; import { parseAnswerUiBlock, + completeRunningProgress, + cancelRunningTodos, + finalizeAssistantMessageAfterAbort, + finalizeAssistantMessageAfterError, readInteractionReply, getFinalAnswerTranscript, type AnswerUiPart, @@ -17,7 +21,8 @@ import { type AgentPermissionStatus, type AgentQuestionRequest, type AgentSessionStreamEvent, - type AgentStreamRenderState + type AgentStreamRenderState, + type AgentRunStatus } from "@/features/agent"; type AgentUiDataParts = { @@ -26,6 +31,7 @@ type AgentUiDataParts = { task_context: Record; execution_plan: Record; progress: Record; + run_status: Record; final_answer: Record; credential_refresh_required: Record; credential_refreshed: Record; @@ -137,11 +143,14 @@ export function toAgentChatMessages( ...(answerParts ? { answerParts } : {}) }; + let runState: Record | undefined; for (const part of message.parts) { if (!isAgentDataPart(part)) { continue; } - if (part.type === "data-interaction_reply" && message.role === "user") { + if (part.type === "data-run_status") { + runState = part.data; + } else if (part.type === "data-interaction_reply" && message.role === "user") { const reply = readInteractionReply(part.data, message.id); if (reply) next = { ...next, interactionReply: reply }; } else if (part.type === "data-task_context") { @@ -193,23 +202,44 @@ export function toAgentChatMessages( if (next.finalAnswer) { next.content = getFinalAnswerTranscript(next); } + if (runState?.status === "aborted") { + next = finalizeAssistantMessageAfterAbort(next); + } else if (runState?.status === "error") { + next = finalizeAssistantMessageAfterError(next, getDataString(runState, "message") ?? "Agent response failed"); + } else if (runState?.status === "completed" || runState?.status === "waiting_input") { + next = { ...next, progress: completeRunningProgress(next.progress), + ...(runState.status === "completed" ? { todos: cancelRunningTodos(next.todos) } : {}) }; + } return [applyQuestionOverrides(applyPermissionOverrides(next, permissionOverrides), questionOverrides)]; }); } -export function toAgentUiMessages(messages: unknown[]): AgentUiMessage[] { - return messages.flatMap((message) => { +export function toAgentUiMessages( + messages: unknown[], + session?: { runStatus?: AgentRunStatus; isStreaming?: boolean } +): AgentUiMessage[] { + const normalized = messages.flatMap((message) => { if (isAgentUiMessage(message)) { return [message]; } const legacyMessage = toAgentUiMessageFromLegacy(message); return legacyMessage ? [legacyMessage] : []; }); + if (!session) return normalized; + const lastAssistant = normalized.reduce((last, message, index) => message.role === "assistant" ? index : last, -1); + return normalized.map((message, index) => { + if (message.role !== "assistant" || message.parts.some((part) => part.type === "data-run_status")) return message; + const status = index === lastAssistant + ? session.runStatus ?? (session.isStreaming === false ? "completed" : undefined) + : "completed"; + if (status === "completed" && !message.parts.some((part) => part.type === "data-progress" || part.type === "data-todo_update")) return message; + return status && status !== "running" ? upsertUiDataPart(message, "run_status", { id: "run-status", status }) : message; + }); } export function applySessionStreamEvent(messages: AgentUiMessage[], event: AgentSessionStreamEvent): AgentUiMessage[] { if (event.type === "state") { - return toAgentUiMessages(event.messages); + return toAgentUiMessages(event.messages, event); } if (event.type === "user_input") { @@ -226,15 +256,34 @@ export function applySessionStreamEvent(messages: AgentUiMessage[], event: Agent return token ? updateLastAssistantUiMessage(messages, (message) => appendTextToUiMessage(message, token)) : messages; } + if (event.type === "done" || event.type === "error" || event.type === "aborted") { + return updateLastAssistantUiMessage(messages, (message) => { + const previous = [...message.parts].reverse().find((part) => part.type === "data-run_status"); + const data = previous && isAgentDataPart(previous) ? previous.data : {}; + return upsertUiDataPart(message, "run_status", { + ...data, id: "run-status", + status: data.status === "aborted" || event.type === "aborted" || event.data.status === "aborted" ? "aborted" + : event.type === "error" ? "error" : event.data.reason === "waiting_input" ? "waiting_input" : "completed", + ...(event.type === "error" || event.type === "aborted" ? { message: getDataString(event.data, "message") ?? getDataString(data, "message") ?? "Agent stream failed" } : {}) + }); + }); + } + + if (event.type === "permission_response" || event.type === "question_response") { + const requestType = event.type === "permission_response" ? "data-permission_request" : "data-question_request"; + return messages.map((message) => message.role === "assistant" && message.parts.some((part) => + part.type === requestType && isAgentDataPart(part) && part.data.request_id === event.data.request_id) + ? upsertUiDataPart(message, event.type, event.data) : message); + } + if ( + event.type === "run_status" || event.type === "final_answer" || event.type === "progress" || event.type === "execution_plan" || event.type === "todo_update" || event.type === "permission_request" || - event.type === "permission_response" || event.type === "question_request" || - event.type === "question_response" || event.type === "answer_ui" || event.type === "ui_envelope" || event.type === "session_title" @@ -244,13 +293,6 @@ export function applySessionStreamEvent(messages: AgentUiMessage[], event: Agent ); } - if (event.type === "error") { - const message = getDataString(event.data, "message") ?? "Agent stream failed"; - return updateLastAssistantUiMessage(messages, (item) => - appendTextToUiMessage(item, item.parts.some((part) => part.type === "text") ? `\n\n错误：${message}` : `错误：${message}`) - ); - } - return messages; } @@ -478,7 +520,7 @@ function applyPermissionOverrides( ...message, permissions: message.permissions.map((permission) => { const override = permissionOverrides[permission.requestId]; - return override + return override && ["pending", "submitting", "error"].includes(permission.status) ? { ...permission, status: override.status, @@ -502,7 +544,7 @@ function applyQuestionOverrides( ...message, questions: message.questions.map((question) => { const override = questionOverrides[question.requestId]; - return override + return override && ["checking", "pending", "submitting", "error"].includes(question.status) ? { ...question, status: override.status, diff --git a/src/features/workbench/hooks/agent-session-runtime.test.ts b/src/features/workbench/hooks/agent-session-runtime.test.ts index cd925bc..6c51da1 100644 --- a/src/features/workbench/hooks/agent-session-runtime.test.ts +++ b/src/features/workbench/hooks/agent-session-runtime.test.ts @@ -3,11 +3,42 @@ import type { ChatTransport, UIMessageChunk } from "ai"; import { AgentSessionStreamError, AgentStreamDisconnectedError, type AgentApiClient } from "@/features/agent"; import { WorkbenchSessionResources } from "./workbench-session-resources"; import { createAgentRuntime, mergeHistoryMessages, resumeAgentRuntime, runtimeRunning } from "./agent-session-runtime"; -import type { AgentUiMessage } from "./agent-session-message-state"; +import { toAgentChatMessages, type AgentUiMessage } from "./agent-session-message-state"; const message = (id: string, text = id): AgentUiMessage => ({ id, role: "assistant", parts: [{ type: "text", text }] }); describe("independent conversation runtimes", () => { + it.each(["completed", "error", "aborted"] as const)("settles direct tool progress when the response is %s without a final progress event", async (status) => { + const resources = new WorkbenchSessionResources(); + let controller!: ReadableStreamDefaultController; + const runtime = createAgentRuntime(resources, "project:a", {} as AgentApiClient, { + sendMessages: async ({ abortSignal }) => new ReadableStream({ start(stream) { + controller = stream; + abortSignal?.addEventListener("abort", () => stream.error(new DOMException("Aborted", "AbortError")), { once: true }); + } }), + reconnectToStream: async () => null, + }); + resources.activeAgent = runtime; + try { + const request = runtime.chat.sendMessage({ text: "核查工具状态" }); + await vi.waitFor(() => expect(controller).toBeDefined()); + controller.enqueue({ type: "start", messageId: "answer" }); + controller.enqueue({ type: "data-progress", data: { id: "tool", phase: "tool", title: "正在处理", status: "running" } }); + await vi.waitFor(() => expect(toAgentChatMessages(runtime.chat.messages, {}, {}).at(-1)?.progress?.[0].status).toBe("running")); + if (status === "aborted") await runtime.chat.stop(); + else { + if (status === "error") controller.enqueue({ type: "error", errorText: "工具执行失败" }); + controller.enqueue({ type: "finish", finishReason: status === "error" ? "error" : "stop" }); + controller.close(); + } + await request; + const answer = toAgentChatMessages(runtime.chat.messages, {}, {}).at(-1); + expect(answer?.progress?.[0].status).toBe(status === "aborted" ? "cancelled" : status); + expect(runtime.runStatus).toBe(status); + if (status === "error") expect(answer?.runError).toBe("工具执行失败"); + } finally { resources.dispose(); } + }); + it("retries a disconnected subscription and preserves the draft, pages and reading position", async () => { vi.useFakeTimers(); const resources = new WorkbenchSessionResources(); @@ -141,7 +172,9 @@ describe("independent conversation runtimes", () => { }); resumeAgentRuntime(resources, runtime); await vi.waitFor(() => expect(runtime.controller).toBeUndefined()); - expect(runtime.chat.messages).toEqual([message("a"), message("b", "new"), message("c")]); + expect(toAgentChatMessages(runtime.chat.messages, {}, {})).toMatchObject([ + { id: "a", content: "a" }, { id: "b", content: "new" }, { id: "c", content: "c" } + ]); expect(submit).toHaveBeenCalledWith("a", "action-a", expect.objectContaining({ status: "failed", error: expect.objectContaining({ code: "SESSION_NOT_ACTIVE" }), })); diff --git a/src/features/workbench/hooks/agent-session-runtime.ts b/src/features/workbench/hooks/agent-session-runtime.ts index 3071c12..4c4a94e 100644 --- a/src/features/workbench/hooks/agent-session-runtime.ts +++ b/src/features/workbench/hooks/agent-session-runtime.ts @@ -91,7 +91,7 @@ export function createAgentRuntime(resources: WorkbenchSessionResources, scope: const active = () => resources.activeAgent === runtime; const notify = () => queueMicrotask(() => resources.notify()); runtime.chat = new Chat({ - id: runtime.key, transport, messages: toAgentUiMessages(loaded?.messages ?? []), + id: runtime.key, transport, messages: toAgentUiMessages(loaded?.messages ?? [], loaded), onData: (part) => { if (resources.disposed) return; const data = part as AgentDataPart; @@ -99,6 +99,10 @@ export function createAgentRuntime(resources: WorkbenchSessionResources, scope: if (sessionId && runtime.sessionId && sessionId !== runtime.sessionId) return; if (sessionId) runtime.sessionId = sessionId; if (data.type === "data-interaction_request") runtime.runStatus = "waiting_input"; + if (data.type === "data-run_status") { + const status = data.data.status; + if (status === "completed" || status === "error" || status === "aborted" || status === "waiting_input") runtime.runStatus = status; + } if (data.type === "data-session_title") runtime.title = getDataString(data.data, "title") ?? runtime.title; if (data.type === "data-credential_refresh_required") { void refreshRuntimeCredential(runtime, data.data); @@ -118,10 +122,16 @@ export function createAgentRuntime(resources: WorkbenchSessionResources, scope: if (resources.disposed) return; runtime.loadedAt = Date.now(); runtime.updatedAt = Date.now(); - runtime.runStatus = isAbort ? "aborted" : isError + runtime.runStatus = isAbort || runtime.runStatus === "aborted" ? "aborted" : isError ? isPendingInteractionError(runtime.error) ? "waiting_input" : "error" : runtime.runStatus === "waiting_input" ? "waiting_input" : "completed"; const recovering = isError && !isAbort && runtime.transportInterrupted && runtime.sessionId; + if (!recovering) { + runtime.chat.messages = applySessionStreamEvent(runtime.chat.messages, { + type: runtime.runStatus === "error" || runtime.runStatus === "aborted" ? "error" : "done", + data: { status: runtime.runStatus, reason: runtime.runStatus, ...(runtime.error ? { message: runtime.error } : {}) } + }); + } if (active()) { if (!recovering) runtime.handlers?.onFinish(runtime.runStatus); } else if (!recovering) notifyBackgroundCompletion(runtime); if (recovering) { @@ -173,7 +183,7 @@ export function resumeAgentRuntime(resources: WorkbenchSessionResources, runtime const sessionId = event.type === "state" ? event.sessionId : event.sessionId ?? getDataString(event.data, "session_id"); if (sessionId && sessionId !== runtime.sessionId) return; if (event.type === "state") { - const tail = toAgentUiMessages(event.messages); + const tail = toAgentUiMessages(event.messages, event); const prior = runtime.chat.messages; // Start filling a disconnected interval before reconciling the live tail. if (prior.length && tail.length && !tail.some((message) => prior.some((old) => old.id === message.id))) { @@ -191,11 +201,16 @@ export function resumeAgentRuntime(resources: WorkbenchSessionResources, runtime if (event.type === "credential_refresh_required") void refreshRuntimeCredential(runtime, event.data); else if (resources.activeAgent !== runtime && event.type === "frontend_action") void rejectBackgroundAction(resources, runtime, event.data); runtime.chat.messages = applySessionStreamEvent(runtime.chat.messages, event); + if (event.type === "run_status") { + const status = event.data.status; + if (status === "aborted" || status === "error" || status === "completed" || status === "waiting_input") runtime.runStatus = status; + } if (event.type === "session_title") runtime.title = getDataString(event.data, "title") ?? runtime.title; - if (event.type === "done" || event.type === "error") { + if (event.type === "done" || event.type === "error" || event.type === "aborted") { terminal = true; runtime.resumed = false; - runtime.runStatus = event.type === "error" ? "error" : event.data.reason === "waiting_input" ? "waiting_input" : "completed"; + runtime.runStatus = event.type === "aborted" || event.data.status === "aborted" || runtime.runStatus === "aborted" ? "aborted" + : event.type === "error" ? "error" : event.data.reason === "waiting_input" ? "waiting_input" : "completed"; runtime.updatedAt = Date.now(); } } diff --git a/src/features/workbench/hooks/use-workbench-agent.ts b/src/features/workbench/hooks/use-workbench-agent.ts index c20731a..336e127 100644 --- a/src/features/workbench/hooks/use-workbench-agent.ts +++ b/src/features/workbench/hooks/use-workbench-agent.ts @@ -1,3 +1,4 @@ +import { createAgentInteractionReplies } from "./agent-interaction-replies"; import { getWorkbenchAccessToken, getWorkbenchTokenProvider, getWorkbenchProject } from "../api/request-context"; import { useCallback, useEffect, useLayoutEffect, useMemo, useRef, useState, useSyncExternalStore } from "react"; import { useChat } from "@ai-sdk/react"; @@ -33,9 +34,6 @@ import { type AgentEventTriggerContext, type AgentModelOption, type AgentInputDelivery, - type AgentPermissionReply, - type AgentPermissionRequest, - type AgentQuestionRequest, type AgentSessionStreamEvent, type AgentSessionPurpose, type AgentStreamRenderState, @@ -52,7 +50,6 @@ import { readBodyString, toAgentChatMessages, toAgentUiMessages, - toPermissionStatus, type AgentDataPart, type AgentUiMessage, type PermissionOverride, @@ -200,7 +197,7 @@ export function useWorkbenchAgent({ try { const loaded = await target.client.loadSession(owningSessionId, { messageLimit: 40 }); if (loaded && !resources.disposed && target.client === scopeClient) { - target.chat.messages = mergeHistoryMessages(target.chat.messages, toAgentUiMessages(loaded.messages)); + target.chat.messages = mergeHistoryMessages(target.chat.messages, toAgentUiMessages(loaded.messages, loaded)); resources.notify(); } } catch { /* A failed reconciliation leaves the accepted answer intact. */ } @@ -681,9 +678,9 @@ export function useWorkbenchAgent({ markLastAssistantStreamDone(chatRef.current.messages, setStreamRenderState); setStatusLabel(event.data.reason === "waiting_input" ? "等待补充信息" : "Agent 分析完成"); runtimeRef.current!.loadedAt = Date.now(); - } else if (event.type === "error") { + } else if (event.type === "aborted" || event.type === "error") { markLastAssistantStreamDone(chatRef.current.messages, setStreamRenderState); - setStatusLabel("Agent 后端请求失败"); + setStatusLabel(event.type === "aborted" || event.data.status === "aborted" ? "Agent 已停止" : "Agent 后端请求失败"); runtimeRef.current!.loadedAt = Date.now(); } @@ -732,7 +729,7 @@ export function useWorkbenchAgent({ if (generation !== loadingGenerationRef.current) return false; if (!target) target = createAgentRuntime(resources, scopeKey, scopeClient, transport, loaded); else { - target.chat.messages = mergeHistoryMessages(target.chat.messages, toAgentUiMessages(loaded.messages)); + target.chat.messages = mergeHistoryMessages(target.chat.messages, toAgentUiMessages(loaded.messages, loaded)); target.title = loaded.title; target.topicId = loaded.topicId; target.loadedAt = Date.now(); target.runStatus = loaded.runStatus; target.error = undefined; target.messagePage ??= loaded.messagePage; @@ -827,7 +824,7 @@ export function useWorkbenchAgent({ ); } else { target.client = projectClient; - target.chat.messages = toAgentUiMessages(loaded.messages); + target.chat.messages = toAgentUiMessages(loaded.messages, loaded); target.title = loaded.title; target.purpose = "task_execution"; target.taskId = normalizedTaskId; @@ -877,7 +874,7 @@ export function useWorkbenchAgent({ refreshing = false; return; } - target.chat.messages = mergeHistoryMessages(target.chat.messages, toAgentUiMessages(loaded.messages)); + target.chat.messages = mergeHistoryMessages(target.chat.messages, toAgentUiMessages(loaded.messages, loaded)); target.title = loaded.title; target.updatedAt = loaded.updatedAt; target.loadedAt = Date.now(); @@ -1045,7 +1042,7 @@ export function useWorkbenchAgent({ target.purpose = "task_collaboration"; target.taskId = binding.taskId; target.taskExecutionHistory = undefined; - target.chat.messages = mergeHistoryMessages(target.chat.messages, toAgentUiMessages(loaded.messages)); + target.chat.messages = mergeHistoryMessages(target.chat.messages, toAgentUiMessages(loaded.messages, loaded)); target.messagePage = loaded.messagePage; target.loadedAt = Date.now(); target.updatedAt = loaded.updatedAt; @@ -1139,62 +1136,9 @@ export function useWorkbenchAgent({ markLastAssistantStreamDone(chatRef.current.messages, setStreamRenderState); }, [chat, interactionRequests, refreshPendingInputs, stopSessionStreamSubscription]); - const replyPermission = useCallback(async (permission: AgentPermissionRequest, reply: AgentPermissionReply) => { - const target = resources.activeAgent; - if (!target || target.sessionId !== permission.sessionId) return; - const update = (value: PermissionOverride) => { - target.permissionOverrides = { ...target.permissionOverrides, [permission.requestId]: value }; - if (resources.activeAgent === target) setPermissionOverrides(target.permissionOverrides); - resources.notify(); - }; - update({ status: "submitting" }); - try { - await target.client.replyPermission(permission.requestId, { sessionId: permission.sessionId, reply }); - update({ status: toPermissionStatus(reply) }); - } catch (error) { - const message = error instanceof Error ? error.message : "权限请求提交失败。"; - update({ status: "error", error: message }); - showMapNotice({ id: "agent-permission-status", tone: "error", title: "Agent 权限提交失败", message }); - } - }, [resources]); - - const replyQuestion = useCallback(async (question: AgentQuestionRequest, answers: string[][]) => { - const target = resources.activeAgent; - if (!target || target.sessionId !== question.sessionId) return; - const update = (value: QuestionOverride) => { - target.questionOverrides = { ...target.questionOverrides, [question.requestId]: value }; - if (resources.activeAgent === target) setQuestionOverrides(target.questionOverrides); - resources.notify(); - }; - update({ status: "submitting" }); - try { - await target.client.replyQuestion(question.requestId, { sessionId: question.sessionId, answers }); - update({ status: "answered", answers }); - } catch (error) { - const message = error instanceof Error ? error.message : "问题回复提交失败。"; - update({ status: "error", error: message }); - showMapNotice({ id: "agent-question-status", tone: "error", title: "Agent 问题提交失败", message }); - } - }, [resources]); - - const rejectQuestion = useCallback(async (question: AgentQuestionRequest) => { - const target = resources.activeAgent; - if (!target || target.sessionId !== question.sessionId) return; - const update = (value: QuestionOverride) => { - target.questionOverrides = { ...target.questionOverrides, [question.requestId]: value }; - if (resources.activeAgent === target) setQuestionOverrides(target.questionOverrides); - resources.notify(); - }; - update({ status: "submitting" }); - try { - await target.client.rejectQuestion(question.requestId, { sessionId: question.sessionId }); - update({ status: "rejected" }); - } catch (error) { - const message = error instanceof Error ? error.message : "问题跳过失败。"; - update({ status: "error", error: message }); - showMapNotice({ id: "agent-question-status", tone: "error", title: "Agent 问题提交失败", message }); - } - }, [resources]); + const { replyPermission, replyQuestion, rejectQuestion } = useMemo(() => createAgentInteractionReplies(resources, { + permission: setPermissionOverrides, question: setQuestionOverrides + }), [resources]); return { apiClient: clientRef.current, diff --git a/src/features/workbench/peripheral-sample/scada-monitoring-notice.tsx b/src/features/workbench/peripheral-sample/scada-monitoring-notice.tsx index ec5c5d2..4b734a0 100644 --- a/src/features/workbench/peripheral-sample/scada-monitoring-notice.tsx +++ b/src/features/workbench/peripheral-sample/scada-monitoring-notice.tsx @@ -1,3 +1,4 @@ +import { formatLocalMonthDayTime } from "@/shared/utils/date-time"; import { useEffect, useState } from "react"; import { motion, useReducedMotion } from "motion/react"; import { ArrowUpRight, CheckCircle2, CircleHelp, Clock3, LoaderCircle, TriangleAlert } from "lucide-react"; @@ -54,7 +55,7 @@ export function ScadaMonitoringNotice({ result, animationId, processing, onOpenI 调查监测 {result ? : 尚无监测结果}

diff --git a/src/features/workbench/peripheral-sample/scada-monitoring-rules.tsx b/src/features/workbench/peripheral-sample/scada-monitoring-rules.tsx index f040a54..b7c6b01 100644 --- a/src/features/workbench/peripheral-sample/scada-monitoring-rules.tsx +++ b/src/features/workbench/peripheral-sample/scada-monitoring-rules.tsx @@ -1,3 +1,4 @@ +import { formatLocalMonthDayTime as dateTime } from "@/shared/utils/date-time"; import type { ScadaMonitoringCalibration, ScadaMonitoringConfig } from "../api/scada-replay-client"; export function ScadaMonitoringRules({ config, calibration, frozen }: { @@ -68,7 +69,3 @@ function thresholdLabel(ids: string[], profiles: ScadaMonitoringCalibration["pre const range = lower === upper ? `${direction}阈值 ${lower}%` : `${direction}阈值 ${lower}%～${upper}%（逐测点）`; return thresholds.length < ids.length ? `${range} · 已就绪 ${thresholds.length}/${ids.length} 个测点` : range; } - -function dateTime(value: string) { - return new Date(value).toLocaleString("zh-CN", { month: "2-digit", day: "2-digit", hour: "2-digit", minute: "2-digit", hour12: false }); -} diff --git a/src/features/workbench/peripheral-sample/scada-replay-section.tsx b/src/features/workbench/peripheral-sample/scada-replay-section.tsx index 2607109..d33d917 100644 --- a/src/features/workbench/peripheral-sample/scada-replay-section.tsx +++ b/src/features/workbench/peripheral-sample/scada-replay-section.tsx @@ -1,3 +1,4 @@ +import { formatLocalMonthDayTime as stamp } from "@/shared/utils/date-time"; import { useEffect, useMemo, useRef, useState } from "react"; import { Activity, @@ -430,9 +431,6 @@ export function ScadaReplaySection({ ); } -function stamp(value: string) { - return new Date(value).toLocaleString("zh-CN", { month: "2-digit", day: "2-digit", hour: "2-digit", minute: "2-digit", hour12: false }); -} function localTime(time: number) { const date = new Date(time); diff --git a/src/features/workbench/utils/report-presentation-state.test.ts b/src/features/workbench/utils/report-presentation-state.test.ts new file mode 100644 index 0000000..53cb3ec --- /dev/null +++ b/src/features/workbench/utils/report-presentation-state.test.ts @@ -0,0 +1,27 @@ +import { describe, expect, it } from "vitest"; +import { getReportPresentationState, snapshotRound } from "./report-presentation-state"; +import type { OperationalTaskReport } from "../api/operational-task-client"; + +const snapshot = (generation: number | null, revision = 1) => ({ + revision, execution_generation: generation, report: { stage: "final", outcome: "insufficient_data" } +}) as OperationalTaskReport; + +describe("report presentation state", () => { + it("keeps historical unknown generations distinct from current and previous rounds", () => { + expect(snapshotRound(null, 2)).toBe("historical"); + expect(snapshotRound(undefined, 2)).toBe("historical"); + expect(snapshotRound(1, 2)).toBe("previous"); + expect(snapshotRound(2, 2)).toBe("current"); + }); + it("does not let a previous final report own the current round's progress or conclusion", () => { + const previous = snapshot(1); + const task = { execution_generation: 2, latest_report: previous, lifecycle_status: "completed" as const }; + expect(getReportPresentationState(task, previous)).toMatchObject({ previousRound: true, currentFinalReport: false, currentInsufficient: false, ended: false }); + expect(getReportPresentationState(task, snapshot(null))).toMatchObject({ round: "historical", previousRound: true, currentFinalReport: false }); + expect(getReportPresentationState(task, null)).toMatchObject({ previousRound: false, current: false }); + }); + it("distinguishes an older revision within the current round and an ended insufficient result", () => { + const task = { execution_generation: 2, latest_report: snapshot(2, 3), lifecycle_status: "completed" as const }; + expect(getReportPresentationState(task, snapshot(2, 1))).toMatchObject({ olderRevision: true, currentFinalReport: true, currentInsufficient: true, ended: true }); + }); +}); diff --git a/src/features/workbench/utils/report-presentation-state.ts b/src/features/workbench/utils/report-presentation-state.ts new file mode 100644 index 0000000..1920690 --- /dev/null +++ b/src/features/workbench/utils/report-presentation-state.ts @@ -0,0 +1,23 @@ +import type { OperationalTask, OperationalTaskReport } from "../api/operational-task-client"; + +export function snapshotRound(generation: number | null | undefined, currentGeneration: number) { + return generation == null ? "historical" : generation === currentGeneration ? "current" : "previous"; +} + +export function getReportPresentationState( + task: Pick, + snapshot: OperationalTaskReport | null | undefined +) { + const round = snapshotRound(snapshot?.execution_generation, task.execution_generation); + const current = round === "current"; + const insufficient = snapshot?.report.outcome === "insufficient_data"; + return { + round, current, insufficient, + previousRound: Boolean(snapshot && !current), + olderRevision: Boolean(snapshot && task.latest_report && snapshot.revision !== task.latest_report.revision), + currentFinalReport: current && snapshot?.report.stage === "final", + currentInsufficient: current && insufficient, + ended: current && task.lifecycle_status === "completed", + stopped: task.lifecycle_status === "failed" || task.lifecycle_status === "cancelled" + }; +} diff --git a/tests/browser/agent-tool-terminal.e2e.ts b/tests/browser/agent-tool-terminal.e2e.ts new file mode 100644 index 0000000..f0cc2e5 --- /dev/null +++ b/tests/browser/agent-tool-terminal.e2e.ts @@ -0,0 +1,69 @@ +import { expect, test } from "@playwright/test"; +import { createServer } from "node:http"; +import type { AddressInfo } from "node:net"; +import { mockAgentApi } from "./support/mock-agent-api"; +import { mockWorkbenchContext } from "./support/mock-workbench-context"; + +for (const status of ["completed", "error", "aborted"] as const) { + test(`tool progress settles after reconnect and reopening (${status})`, async ({ page }, testInfo) => { + await page.emulateMedia({ reducedMotion: "reduce" }); + await mockWorkbenchContext(page); + const sessionId = `tool-terminal-${status}`; + const title = `工具终态核查 ${status}`; + await mockAgentApi(page, { sessions: [{ session_id: sessionId, title, created_at: Date.now(), updated_at: Date.now() }] }); + const headers = { "Access-Control-Allow-Origin": "*" }; + const messages = [{ id: "tool-answer", role: "assistant", parts: [ + { type: "text", text: "已查到部分压力记录。" }, + { type: "data-progress", data: { id: "tool", phase: "tool", title: "正在处理", status: "running", tool: { name: "query_pressure", input: { node_id: 12 } } } }, + ] }]; + let settled = false; + let finish: (() => void) | undefined; + const server = createServer((_request, response) => { + response.writeHead(200, { ...headers, "Content-Type": "text/event-stream" }); + response.write(`event: state\ndata: ${JSON.stringify({ session_id: sessionId, messages, is_streaming: true, run_status: "running" })}\n\n`); + finish = () => { + settled = true; + const type = status === "completed" ? "done" : status === "aborted" ? "aborted" : "error"; + const data = { session_id: sessionId, status, message: status === "error" ? "压力查询连接失败" : "请求已中断" }; + // Legacy subscriptions may terminate without a separate run_status event. + response.end(`event: ${type}\ndata: ${JSON.stringify(data)}\n\n`); + }; + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + await page.route(`**/api/v1/session/${sessionId}?**`, (route) => route.fulfill({ headers, json: { + session_id: sessionId, title, purpose: "ordinary_chat", status: "active", + is_streaming: !settled, run_status: settled ? status : "running", messages, + } })); + await page.route(`**/api/v1/session/${sessionId}/stream**`, (route) => route.continue({ url: `http://127.0.0.1:${(server.address() as AddressInfo).port}` })); + const panel = page.getByRole("complementary", { name: "Agent 命令面板" }); + const answer = panel.locator('[data-agent-message-id="tool-answer"]'); + const icon = answer.locator('[data-agent-tool] [data-progress-status]'); + const open = async () => { + await page.goto("/", { waitUntil: "domcontentloaded" }); + await page.getByRole("button", { name: /展开 Agent 助手面板/ }).click(); + await panel.getByRole("button", { name: "打开 Agent 历史记录" }).click(); + await page.locator("button.agent-history-session-trigger").filter({ hasText: title }).click(); + }; + try { + await open(); + await expect(icon).toHaveAttribute("data-progress-status", "running"); + await expect.poll(() => Boolean(finish)).toBe(true); + finish!(); + await answer.getByRole("button", { name: "查看处理过程", exact: true }).click(); + const expectedStatus = status === "aborted" ? "cancelled" : status; + await expect(icon).toHaveAttribute("data-progress-status", expectedStatus); + await expect(answer.locator('[data-progress-status="running"]')).toHaveCount(0); + if (status === "error") await expect(answer.getByRole("status").filter({ hasText: "压力查询连接失败" })).toBeVisible(); + await answer.screenshot({ path: testInfo.outputPath(`${status}-settled.png`), animations: "disabled" }); + await open(); + await answer.getByRole("button", { name: "查看处理过程", exact: true }).click(); + await expect(icon).toHaveAttribute("data-progress-status", expectedStatus); + await expect(answer.locator('[data-progress-status="running"]')).toHaveCount(0); + if (status === "aborted") await expect(icon).toHaveAttribute("aria-label", "已中止"); + await answer.screenshot({ path: testInfo.outputPath(`${status}-reopened.png`), animations: "disabled" }); + } finally { + server.closeAllConnections(); + await new Promise((resolve) => server.close(() => resolve())); + } + }); +}