From 692bbf609cc704fbab3d5de9445997e221bd1ccd Mon Sep 17 00:00:00 2001 From: Huarch Date: Mon, 9 Nov 2026 17:55:00 +0800 Subject: [PATCH] refactor: consolidate requests and finalize agent tool states Share project JSON transport, interaction replies and report selectors. Settle tool progress consistently for direct streams, reconnects and history, preserving failure reasons and stopped state. --- README.md | 1 + package.json | 1 + scripts/export-answer-ui-contract.mjs | 17 ++-- src/features/agent/api/client.ts | 4 + .../agent/components/agent-command-panel.tsx | 5 ++ .../components/agent-progress-status-icon.tsx | 3 +- src/features/agent/index.ts | 1 + src/features/agent/session-state.test.ts | 4 +- src/features/agent/session-state.ts | 27 +++++-- src/features/agent/types.ts | 3 +- .../workbench/api/gis-properties-client.ts | 39 ++-------- .../workbench/api/operational-task-client.ts | 12 +-- .../workbench/api/place-search-client.ts | 24 ++---- src/features/workbench/api/project-request.ts | 11 +-- .../workbench/api/realtime-flow-client.ts | 19 +---- .../workbench/api/request-json.test.ts | 60 ++++++++++++++ src/features/workbench/api/request-json.ts | 38 +++++++++ src/features/workbench/api/scada-client.ts | 13 +--- .../workbench/api/scada-replay-client.ts | 16 ++-- .../api/workbench-analysis-history-client.ts | 22 +----- .../workbench/api/workbench-runtime-client.ts | 19 +---- .../workbench/api/workbench-sync-client.ts | 9 +-- .../components/agent-execution-plan-panel.tsx | 15 +--- .../components/investigation-report-view.tsx | 5 +- .../components/investigation-task-report.tsx | 14 +--- .../hooks/agent-interaction-replies.test.ts | 25 ++++++ .../hooks/agent-interaction-replies.ts | 60 ++++++++++++++ .../hooks/agent-session-message-state.test.ts | 73 +++++++++++++++++ .../hooks/agent-session-message-state.ts | 74 ++++++++++++++---- .../hooks/agent-session-runtime.test.ts | 37 ++++++++- .../workbench/hooks/agent-session-runtime.ts | 25 ++++-- .../workbench/hooks/use-workbench-agent.ts | 78 +++---------------- .../scada-monitoring-notice.tsx | 3 +- .../scada-monitoring-rules.tsx | 5 +- .../scada-replay-section.tsx | 4 +- .../utils/report-presentation-state.test.ts | 27 +++++++ .../utils/report-presentation-state.ts | 23 ++++++ tests/browser/agent-tool-terminal.e2e.ts | 69 ++++++++++++++++ 38 files changed, 605 insertions(+), 280 deletions(-) create mode 100644 src/features/workbench/api/request-json.test.ts create mode 100644 src/features/workbench/api/request-json.ts create mode 100644 src/features/workbench/hooks/agent-interaction-replies.test.ts create mode 100644 src/features/workbench/hooks/agent-interaction-replies.ts create mode 100644 src/features/workbench/utils/report-presentation-state.test.ts create mode 100644 src/features/workbench/utils/report-presentation-state.ts create mode 100644 tests/browser/agent-tool-terminal.e2e.ts 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())); + } + }); +}