import { z } from "zod"; import type { FeatureCollection, Geometry, Position } from "geojson"; import type { AccessTokenProvider } from "@/shared/auth/keycloak-auth"; import { env } from "@/shared/config/env"; import type { ScheduledConditionRecord, ScheduledWorkOrderItem, WorkbenchAlert, 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. const runtimeConditionSchema = z.object({ id: z.string().trim().min(1), task_id: z.enum(["backend-analysis", "backend-schedule"]), session_id: z.string().trim().min(1), title: z.string().trim().min(1), summary: z.string().trim().min(1), status: z.enum(["pending", "running", "completed", "warning", "error"]), risk_level: z.enum(["normal", "attention", "critical"]), scheduled_at: z.string().datetime({ offset: true }), updated_at: z.string().datetime({ offset: true }), detail: z.string().trim().min(1), recommendation: z.string().trim().min(1).nullable(), evidence: z.array(z.string().trim().min(1)), model_name: z.enum(["TJWater analysis runtime", "TJWater scheduler"]) }); type RuntimeConditionPayload = z.infer; type OperationalRuntimeConditionPayload = RuntimeConditionPayload & { task_id: "backend-schedule"; }; const runtimeAlertSchema = z.object({ id: z.string().trim().min(1), condition_id: z.string().trim().min(1).nullable(), source: z.object({ task_type: z.enum(["scheduled_task", "analysis_task", "other_task"]), task_id: z.string().trim().min(1), run_id: z.string().trim().min(1) }), title: z.string().trim().min(1), description: z.string().trim().min(1), occurred_at: z.string().datetime({ offset: true }), severity: z.enum(["warning", "critical"]) }); const runtimeScenarioSchema = z.object({ id: z.string().trim().min(1), run_id: z.string().trim().min(1), name: z.string().trim().min(1), description: z.string().trim().min(1), status: z.enum(["draft", "active", "review"]), run_type: z.string().trim().min(1), run_type_label: z.string().trim().min(1), run_status: z.string().trim().min(1), created_at: z.string().datetime({ offset: true }) }); const runtimeAnalysisResultSchema = z.object({ result_type: z.string().trim().min(1), created_at: z.string().datetime({ offset: true }), facts: z.array(z.object({ label: z.string().trim().min(1), value: z.string().trim().min(1) })) }); const geoJsonFeatureCollectionSchema = z.custom>( isGeoJsonFeatureCollection ); const runtimeLatestAnalysisSchema = z.object({ run_id: z.string().trim().min(1), name: z.string().trim().min(1), run_type: z.string().trim().min(1), run_type_label: z.string().trim().min(1), status: z.string().trim().min(1), status_label: z.string().trim().min(1), created_by: z.string().trim().min(1), created_at: z.string().datetime({ offset: true }), started_at: z.string().datetime({ offset: true }), results: z.array(runtimeAnalysisResultSchema), map_payload: z.unknown().nullable() }); const runtimeWorkOrderStageSchema = z.object({ id: z.string().trim().min(1), sequence: z.number().int().positive(), title: z.string().trim().min(1), status: z.enum(["done", "current", "pending", "blocked", "skipped"]), owner: z.string().trim().min(1), time_label: z.string().trim().min(1), description: z.string().trim().min(1) }); const runtimeWorkOrderSubtaskSchema = z.object({ id: z.string().trim().min(1), title: z.string().trim().min(1), summary: z.string().trim().min(1), status: z.enum(["pending", "running", "blocked", "completed", "cancelled"]), scheduled_at: z.string().datetime({ offset: true }), location: z.string().trim().min(1), assignee: z.string().trim().min(1), duration_minutes: z.number().int().positive(), reply_window_minutes: z.number().int().positive(), stages: z.array(runtimeWorkOrderStageSchema).min(1), reply_requirements: z.array(z.string().trim().min(1)) }).superRefine((subtask, context) => { if (subtask.stages.some((stage, index) => stage.sequence !== index + 1)) { context.addIssue({ code: z.ZodIssueCode.custom, message: "子任务阶段序号必须从 1 连续递增", path: ["stages"] }); } if (new Set(subtask.stages.map((stage) => stage.id)).size !== subtask.stages.length) { context.addIssue({ code: z.ZodIssueCode.custom, message: "子任务阶段编号不能重复", path: ["stages"] }); } if (subtask.stages.filter((stage) => stage.status === "current").length > 1) { context.addIssue({ code: z.ZodIssueCode.custom, message: "单个子任务最多只能有一个当前阶段", path: ["stages"] }); } }); const runtimeWorkOrderSchema = z.object({ id: z.string().trim().min(1), code: z.string().trim().min(1), scheduled_at: z.string().datetime({ offset: true }), title: z.string().trim().min(1), summary: z.string().trim().min(1), status: z.enum(["pending", "running", "awaiting_reply", "completed"]), risk_level: z.enum(["normal", "attention", "critical"]), updated_at: z.string().datetime({ offset: true }), detail: z.string().trim().min(1), recommendation: z.string().trim().min(1).nullable(), duration_minutes: z.number().int().positive(), source: z.string().trim().min(1), dispatcher: z.string().trim().min(1), priority: z.enum(["normal", "urgent"]), subtasks: z.array(runtimeWorkOrderSubtaskSchema).min(1) }).superRefine((workOrder, context) => { if (new Set(workOrder.subtasks.map((subtask) => subtask.id)).size !== workOrder.subtasks.length) { context.addIssue({ code: z.ZodIssueCode.custom, message: "工单子任务编号不能重复", path: ["subtasks"] }); } }); const runtimeSnapshotSchema = z.object({ availability: z.enum(["ready", "not_configured", "degraded"]), source: z.enum(["none", "analysis_runs", "scheduled_tasks"]), scenario_availability: z.enum(["ready", "not_configured", "degraded"]), artifact_availability: z.enum(["ready", "not_configured", "degraded"]), schedule_availability: z.enum(["ready", "not_configured", "degraded"]), work_order_availability: z.enum(["ready", "not_configured", "degraded"]), alert_availability: z.enum(["ready", "not_configured", "degraded"]), generated_at: z.string().datetime({ offset: true }), conditions: z.array(runtimeConditionSchema), alerts: z.array(runtimeAlertSchema), scenarios: z.array(runtimeScenarioSchema), latest_analysis: runtimeLatestAnalysisSchema.nullable(), work_orders: z.array(runtimeWorkOrderSchema) }); type ProjectionAvailability = "ready" | "not_configured" | "degraded"; export type WorkbenchRuntimeSnapshot = { availability: ProjectionAvailability; source: "none" | "scheduled_tasks"; scenarioAvailability: ProjectionAvailability; artifactAvailability: ProjectionAvailability; scheduleAvailability: ProjectionAvailability; workOrderAvailability: ProjectionAvailability; alertAvailability: ProjectionAvailability; generatedAt: string; conditions: ScheduledConditionRecord[]; alerts: WorkbenchAlert[]; scenarios: WorkbenchScenario[]; latestAnalysisArtifact: AnalysisArtifact | null; simulationOverlay: FeatureCollection | null; workOrders: ScheduledWorkOrderItem[]; }; export type WorkbenchAnalysisProjection = { runId: string; artifact: AnalysisArtifact; simulationOverlay: FeatureCollection | null; }; const SERVER_API_BASE_URL = env.TJWATER_SERVER_API_BASE_URL.replace(/\/$/, ""); export async function fetchWorkbenchRuntimeSnapshot( getAccessToken?: AccessTokenProvider, signal?: AbortSignal ): Promise { const { project, accessToken } = await findOwnedWorkbenchProject(getAccessToken, signal); const response = await fetch(`${SERVER_API_BASE_URL}/api/v1/workbench/runtime`, { 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 snapshot = runtimeSnapshotSchema.parse(payload); const latestAnalysis = snapshot.latest_analysis ? projectWorkbenchAnalysis(snapshot.latest_analysis) : null; const operationalConditions = snapshot.conditions.filter(isOperationalRuntimeCondition); return { availability: snapshot.availability, source: resolveOperationalRuntimeSource(snapshot, operationalConditions), scenarioAvailability: snapshot.scenario_availability, artifactAvailability: snapshot.artifact_availability, scheduleAvailability: snapshot.schedule_availability, workOrderAvailability: snapshot.work_order_availability, alertAvailability: snapshot.alert_availability, generatedAt: snapshot.generated_at, conditions: operationalConditions.map((condition) => ({ id: condition.id, kind: "condition", taskId: condition.task_id, dataSource: getConditionDataSource(condition.task_id), sessionId: condition.session_id, title: condition.title, summary: condition.summary, status: condition.status, riskLevel: condition.risk_level, scheduledAt: condition.scheduled_at, updatedAt: Date.parse(condition.updated_at), detail: condition.detail, recommendation: condition.recommendation ?? undefined, evidence: condition.evidence, modelName: condition.model_name })), alerts: snapshot.alerts.map((alert) => ({ ...(alert.condition_id ? { conditionId: alert.condition_id } : {}), id: alert.id, title: alert.title, description: alert.description, source: { taskType: alert.source.task_type, taskId: alert.source.task_id, runId: alert.source.run_id }, severity: alert.severity, occurredAt: alert.occurred_at, time: formatRuntimeAlertTime(alert.occurred_at) })), scenarios: snapshot.scenarios.map((scenario) => ({ id: scenario.id, name: scenario.name, description: scenario.description, status: scenario.status, runId: scenario.run_id, runType: scenario.run_type, runStatus: scenario.run_status, createdAt: scenario.created_at })), latestAnalysisArtifact: latestAnalysis?.artifact ?? null, simulationOverlay: latestAnalysis?.simulationOverlay ?? null, workOrders: snapshot.work_orders.map((workOrder) => ({ id: workOrder.id, kind: "work_order", code: workOrder.code, scheduledAt: workOrder.scheduled_at, title: workOrder.title, summary: workOrder.summary, status: workOrder.status, riskLevel: workOrder.risk_level, updatedAt: Date.parse(workOrder.updated_at), detail: workOrder.detail, recommendation: workOrder.recommendation ?? undefined, durationMinutes: workOrder.duration_minutes, source: workOrder.source, dispatcher: workOrder.dispatcher, priority: workOrder.priority, subtasks: workOrder.subtasks.map((subtask) => ({ id: subtask.id, title: subtask.title, summary: subtask.summary, status: subtask.status, scheduledAt: subtask.scheduled_at, location: subtask.location, assignee: subtask.assignee, durationMinutes: subtask.duration_minutes, replyWindowMinutes: subtask.reply_window_minutes, stages: subtask.stages.map((stage) => ({ id: stage.id, sequence: stage.sequence, title: stage.title, status: stage.status, owner: stage.owner, timeLabel: stage.time_label, description: stage.description })), replyRequirements: subtask.reply_requirements })) })) }; } export async function fetchWorkbenchAnalysisRun( runId: string, getAccessToken?: AccessTokenProvider, signal?: AbortSignal ): 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 analysis = runtimeLatestAnalysisSchema.parse(payload); if (analysis.run_id !== normalizedRunId) { throw new Error("分析运行成果与请求 ID 不一致"); } return projectWorkbenchAnalysis(analysis); } function projectWorkbenchAnalysis( analysis: z.infer ): WorkbenchAnalysisProjection { const parsedSimulationOverlay = analysis.map_payload ? geoJsonFeatureCollectionSchema.safeParse(analysis.map_payload) : null; const simulationOverlay = parsedSimulationOverlay?.success ? parsedSimulationOverlay.data : null; return { runId: analysis.run_id, artifact: createPersistedAnalysisArtifact(analysis, simulationOverlay), simulationOverlay }; } function createPersistedAnalysisArtifact( analysis: z.infer, simulationOverlay: FeatureCollection | null ): AnalysisArtifact { const resultTypes = [...new Set(analysis.results.map((result) => result.result_type))]; const completed = ["completed", "succeeded"].includes(analysis.status); const lifecycle: ArtifactLifecycle = completed ? "reviewed" : "draft"; const artifactType = getArtifactType(analysis.run_type); const factCount = analysis.results.reduce((total, result) => total + result.facts.length, 0); const resultSummary = analysis.results.length > 0 ? `后端记录状态为${analysis.status_label},已持久化 ${analysis.results.length} 条分析结果。` : `后端记录状态为${analysis.status_label},当前未查询到独立结果记录。`; return { id: `persisted-analysis-${analysis.run_id}`, origin: "backend", type: artifactType, lifecycle, revision: 1, title: analysis.name, subtitle: `${analysis.run_type_label} · 项目后端分析运行`, summary: resultSummary, scope: `运行 ${analysis.run_id}`, confidence: "后端未提供", generatedAt: formatRuntimeDateTime(analysis.created_at), analyticalTimeRange: { start: analysis.started_at, end: analysis.created_at }, preferredView: simulationOverlay ? "map_split" : "focus", mapRelation: simulationOverlay ? "required" : "none", ...(simulationOverlay ? { spatialOverlay: simulationOverlay } : {}), blocks: [ { id: `persisted-analysis-${analysis.run_id}-metrics`, kind: "metric-grid", title: "运行摘要", span: 12, metrics: [ { label: "后端状态", value: analysis.status_label, detail: analysis.status, tone: getStatusTone(analysis.status) }, { label: "结果记录", value: String(analysis.results.length), detail: "analysis.results", tone: analysis.results.length > 0 ? "success" : "neutral" }, { label: "结果类型", value: String(resultTypes.length), detail: resultTypes.join("、") || "暂无", tone: "info" }, { label: "成果事实", value: String(factCount), detail: factCount > 0 ? "后端已识别字段" : "暂无已识别字段", tone: factCount > 0 ? "success" : "neutral" }, { label: "创建人", value: analysis.created_by, detail: analysis.run_type_label, tone: "neutral" } ] }, { id: `persisted-analysis-${analysis.run_id}-metadata`, kind: "narrative", title: "后端运行记录", span: 12, paragraphs: [ `运行 ID:${analysis.run_id}`, `分析类型:${analysis.run_type_label}(${analysis.run_type})`, `创建人:${analysis.created_by};状态:${analysis.status_label}。` ] }, ...(analysis.results.length > 0 ? [{ id: `persisted-analysis-${analysis.run_id}-results`, kind: "data-table" as const, title: "持久化结果索引", description: "来自项目 BizDB analysis.results 的结果记录", span: 12 as const, columns: [ { key: "type", label: "结果类型" }, { key: "createdAt", label: "写入时间" }, { key: "facts", label: "后端成果事实" } ], rows: analysis.results.map((result) => ({ type: result.result_type, createdAt: formatRuntimeDateTime(result.created_at), facts: result.facts.length > 0 ? result.facts.map((fact) => `${fact.label}:${fact.value}`).join(";") : "未提供已识别字段" })) }] : []) ], sources: [ "项目 BizDB · analysis.runs", ...(analysis.results.length > 0 ? ["项目 BizDB · analysis.results"] : []) ], limitations: [ factCount > 0 ? "当前投影展示后端持久化元数据、结果索引和已识别的成果字段;未透传任意结果载荷。" : "当前投影只展示后端持久化元数据与结果索引。", "后端记录没有提供综合置信度。" ], actions: [] }; } function getArtifactType(runType: string): ArtifactType { if (["burst_detection", "burst_location", "dma_leak_identification"].includes(runType)) { return "diagnosis"; } if (runType === "sensor_placement") return "metric_analysis"; return "simulation"; } function getStatusTone(status: string) { if (["completed", "succeeded"].includes(status)) return "success" as const; if (["failed", "error"].includes(status)) return "danger" as const; if (status === "running") return "info" as const; return "warning" as const; } function formatRuntimeAlertTime(value: string) { const date = new Date(value); return `${String(date.getMonth() + 1).padStart(2, "0")}-${String(date.getDate()).padStart(2, "0")} ${String(date.getHours()).padStart(2, "0")}:${String(date.getMinutes()).padStart(2, "0")}`; } function getConditionDataSource( _taskId: OperationalRuntimeConditionPayload["task_id"] ) { return "scheduled_tasks" as const; } function isOperationalRuntimeCondition( condition: RuntimeConditionPayload ): condition is OperationalRuntimeConditionPayload { return condition.task_id === "backend-schedule"; } function resolveOperationalRuntimeSource( snapshot: z.infer, conditions: OperationalRuntimeConditionPayload[] ): WorkbenchRuntimeSnapshot["source"] { if (snapshot.source === "scheduled_tasks") return "scheduled_tasks"; return conditions.length > 0 ? "scheduled_tasks" : "none"; } function formatRuntimeDateTime(value: string) { return new Intl.DateTimeFormat("zh-CN", { month: "2-digit", day: "2-digit", hour: "2-digit", minute: "2-digit", hour12: false }).format(new Date(value)); } function isGeoJsonFeatureCollection(value: unknown): value is FeatureCollection { if (!isRecord(value) || value.type !== "FeatureCollection" || !Array.isArray(value.features)) { return false; } return value.features.every((feature) => isRecord(feature) && feature.type === "Feature" && (feature.properties === null || isRecord(feature.properties)) && (feature.geometry === null || isGeoJsonGeometry(feature.geometry)) ); } function isGeoJsonGeometry(value: unknown): value is Geometry { if (!isRecord(value) || typeof value.type !== "string") return false; if (value.type === "GeometryCollection") { return Array.isArray(value.geometries) && value.geometries.every(isGeoJsonGeometry); } const coordinates = value.coordinates; if (value.type === "Point") return isPosition(coordinates); if (value.type === "MultiPoint" || value.type === "LineString") { return isPositionArray(coordinates); } if (value.type === "MultiLineString" || value.type === "Polygon") { return Array.isArray(coordinates) && coordinates.every(isPositionArray); } if (value.type === "MultiPolygon") { return Array.isArray(coordinates) && coordinates.every((polygon) => Array.isArray(polygon) && polygon.every(isPositionArray) ); } return false; } function isPosition(value: unknown): value is Position { return Array.isArray(value) && value.length >= 2 && value.every((coordinate) => typeof coordinate === "number" && Number.isFinite(coordinate)); } function isPositionArray(value: unknown): value is Position[] { return Array.isArray(value) && value.every(isPosition); } function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null && !Array.isArray(value); }