feat: add node coverage and improve investigation presentation

This commit is contained in:
2026-11-02 15:23:34 +08:00
parent 05183aff93
commit 4a7dcbab92
25 changed files with 588 additions and 21 deletions
+1
View File
@@ -58,6 +58,7 @@ McpSnapshotSource = Literal[
'run_burst_detection', 'run_burst_detection',
'run_burst_location', 'run_burst_location',
'analyze_valve_isolation', 'analyze_valve_isolation',
'generate_node_coverage',
'run_leakage_identification', 'run_leakage_identification',
'get_element_properties', 'get_element_properties',
'list_elements', 'list_elements',
+1 -1
View File
@@ -252,7 +252,7 @@ def _undeclared_diagram_node_refs(diagram: InvestigationDiagram | None) -> list[
for node in diagram.node_refs: for node in diagram.node_refs:
escaped = re.escape(node.node_id) escaped = re.escape(node.node_id)
if not re.search( if not re.search(
rf"(?<![A-Za-z0-9_-]){escaped}(?=\s*(?:[\[({{@;]|--|==|-\.|$))", rf"(?<![A-Za-z0-9_-]){escaped}(?=\s*(?:[\[({{@;]|:::|--|==|-\.|$))",
diagram.mermaid, diagram.mermaid,
re.MULTILINE, re.MULTILINE,
): ):
+31 -1
View File
@@ -4,7 +4,7 @@ from datetime import datetime, timedelta
from typing import Annotated, Literal from typing import Annotated, Literal
from uuid import UUID from uuid import UUID
from pydantic import Field, model_validator from pydantic import BaseModel, ConfigDict, Field, model_validator
from app.mcp.models import Id, StrictInput from app.mcp.models import Id, StrictInput
@@ -19,6 +19,36 @@ class CalculationInput(StrictInput):
) )
class NodeCoverageGroup(BaseModel):
model_config = ConfigDict(extra="forbid")
group_id: Id = Field(max_length=80)
title: str = Field(min_length=1, max_length=200)
node_ids: list[Id] = Field(min_length=1, max_length=10000)
class NodeCoverageInput(CalculationInput):
groups: list[NodeCoverageGroup] = Field(min_length=1, max_length=8,
description="已核实的模型节点集合;各组为需要比较的业务或空间集合,可重叠,不接受任意坐标")
concave_ratio: float = Field(default=0.25, ge=0, le=1, allow_inf_nan=False,
description="凹包比例;越接近1越平滑接近凸包,参数只决定示意形状")
buffer_m: float = Field(default=25, gt=0, le=1000, allow_inf_nan=False,
description="展示缓冲距离,米;不是服务半径或影响距离")
split_by_connectivity: bool = Field(default=False, strict=True,
description="需要按模型连接分组时设true;地理分布比较默认false,不能据此认定实际供水关系")
excluded_link_ids: list[Id] = Field(default_factory=list, max_length=200,
description="连通分组时额外视为断开的管段/阀门,例如已有隔离建议中的关阀ID;仅影响绘图分组")
@model_validator(mode="after")
def bounded_groups(self):
if len({group.group_id for group in self.groups}) != len(self.groups):
raise ValueError("coverage group IDs must be distinct")
if len({node for group in self.groups for node in group.node_ids}) > 10000:
raise ValueError("coverage supports at most 10000 distinct nodes")
if self.excluded_link_ids and not self.split_by_connectivity:
raise ValueError("excluded_link_ids requires split_by_connectivity")
return self
class BurstDetectionInput(CalculationInput): class BurstDetectionInput(CalculationInput):
target_time: datetime = Field(description="带时区的侦测时刻;读取此前15天同周期压力数据,不是单小时定位") target_time: datetime = Field(description="带时区的侦测时刻;读取此前15天同周期压力数据,不是单小时定位")
sensor_nodes: list[Id] = Field(min_length=5, max_length=20, description="已核实的压力监测节点ID,至少5个完整传感器") sensor_nodes: list[Id] = Field(min_length=5, max_length=20, description="已核实的压力监测节点ID,至少5个完整传感器")
+4
View File
@@ -71,6 +71,10 @@ def _calculate(name: str, parameters: dict[str, Any], network: str, username: st
from app.services.valve_isolation import analyze_valve_isolation from app.services.valve_isolation import analyze_valve_isolation
return analyze_valve_isolation(network=network, **parameters) return analyze_valve_isolation(network=network, **parameters)
if name == "generate_node_coverage":
from app.services.node_coverage import generate_node_coverage
return generate_node_coverage(network=network, **parameters)
raise ValueError("Calculation is not registered") raise ValueError("Calculation is not registered")
finally: finally:
_slots.release() _slots.release()
+4
View File
@@ -441,6 +441,10 @@ class McpQueryApplication:
"resource_uri": uri, "resource_uri": uri,
"preview": "The complete query snapshot is saved in task evidence", "preview": "The complete query snapshot is saved in task evidence",
} }
from app.mcp.presentation_preview import saved_presentation_preview
presentation = saved_presentation_preview(result)
if presentation is not None:
output["data"]["presentation"] = presentation
output["truncated"] = True output["truncated"] = True
output["data_gaps"] = [ output["data_gaps"] = [
*result.data_gaps, *result.data_gaps,
+2 -1
View File
@@ -6,7 +6,7 @@ from mcp import types
from app.mcp import models as m from app.mcp import models as m
from app.mcp import calculation_models as jobs from app.mcp import calculation_models as jobs
from app.mcp import extended_models as e from app.mcp import extended_models as e
from app.mcp.analysis_models import BurstDetectionInput, BurstLocationInput, ValveIsolationInput, LeakageIdentificationInput from app.mcp.analysis_models import BurstDetectionInput, BurstLocationInput, ValveIsolationInput, LeakageIdentificationInput, NodeCoverageInput
from app.mcp.analysis_models import ScenarioInput, ContaminantSimulationInput, ValveClosureInput, FlushingInput, PressureRegulationInput, SensorPlacementInput from app.mcp.analysis_models import ScenarioInput, ContaminantSimulationInput, ValveClosureInput, FlushingInput, PressureRegulationInput, SensorPlacementInput
@@ -61,6 +61,7 @@ _specs = [
("run_burst_detection", "爆管异常侦测:使用至少5个完整压力传感器此前15天数据;只提示压力异常,不定位、不证明爆管", BurstDetectionInput, "burst.run", "analysis", False, True), ("run_burst_detection", "爆管异常侦测:使用至少5个完整压力传感器此前15天数据;只提示压力异常,不定位、不证明爆管", BurstDetectionInput, "burst.run", "analysis", False, True),
("run_burst_location", "爆管候选定位:对比异常与正常压力,可辅助流量;burst_leakage必需,可使用有依据的估计或明确给定的情景假设,不要求现场确诊;本工具不反演总漏水量。压力节点可省略,自动读取项目SCADA压力测点;monitoring正常窗可成对省略,默认前一天同时间段,须说明该对照假设。simulation须有已有分析运行UUID并读取同窗模型基线,不是实测;定位内部会执行水力计算,不创建新的事件方案", BurstLocationInput, "burst.run", "analysis", False, True), ("run_burst_location", "爆管候选定位:对比异常与正常压力,可辅助流量;burst_leakage必需,可使用有依据的估计或明确给定的情景假设,不要求现场确诊;本工具不反演总漏水量。压力节点可省略,自动读取项目SCADA压力测点;monitoring正常窗可成对省略,默认前一天同时间段,须说明该对照假设。simulation须有已有分析运行UUID并读取同窗模型基线,不是实测;定位内部会执行水力计算,不创建新的事件方案", BurstLocationInput, "burst.run", "analysis", False, True),
("analyze_valve_isolation", "按事故候选要素分析拓扑隔离边界、建议关闭阀门和受影响节点;候选分析无需等待现场确诊。输出不是污染扩散或水力影响范围;现场操作前须核实阀门映射及可用性,本工具不执行关阀", ValveIsolationInput, "simulation.run", "analysis", False, True), ("analyze_valve_isolation", "按事故候选要素分析拓扑隔离边界、建议关闭阀门和受影响节点;候选分析无需等待现场确诊。输出不是污染扩散或水力影响范围;现场操作前须核实阀门映射及可用性,本工具不执行关阀", ValveIsolationInput, "simulation.run", "analysis", False, True),
("generate_node_coverage", "为当前项目已核实的模型节点集合生成空间覆盖示意,适用于候选分布、监测覆盖比较、区域成员、隔离或模拟结果节点等,不限调查任务。支持最多8组、合计10000个不同节点,从真实GIS查询坐标;监测设备、投诉或地址须先关联模型节点。可按连接分组,记录凹包/缓冲参数、缺失节点和display_only;不计算故障、污染、实际停水或服务边界。返回result.map_payload及coverage,任务中用set_overlay引用本次evidence_id、省略data展示;生成不自动操作地图", NodeCoverageInput, "webgis.view", "analysis", False, True),
( (
"get_element_properties", "get_element_properties",
"读取当前项目的单个管网元素属性", "读取当前项目的单个管网元素属性",
+31
View File
@@ -0,0 +1,31 @@
"""Keep a saved geometry discoverable even when its inline response is shortened."""
from uuid import NAMESPACE_URL, uuid5
from app.mcp.models import QueryResult
def saved_presentation_preview(result: QueryResult) -> dict | None:
if result.source != "generate_node_coverage" or not result.evidence_id:
return None
analysis = result.data.get("result") if isinstance(result.data, dict) else None
if not isinstance(analysis, dict):
return None
payload = analysis.get("map_payload")
if not isinstance(payload, dict) or not payload.get("features"):
return None
coverage = analysis.get("coverage", {})
return {
"command": {
"action": "set_overlay", "mode": "geometry", "evidence_id": result.evidence_id,
"overlay_id": f"node-coverage-{uuid5(NAMESPACE_URL, result.evidence_id)}",
"title": "节点空间覆盖示意",
},
"display_only": True,
"coverage_status": coverage.get("status"),
"groups": {role: {
"node_count": len(group.get("node_ids", [])),
"mapped_node_count": group.get("mapped_node_count", 0),
"missing_node_count": len(group.get("missing_node_ids", [])),
} for role, group in coverage.get("groups", {}).items()},
}
+1
View File
@@ -179,6 +179,7 @@ async def execute_query(
"run_burst_detection": "Pressure anomalies alone do not confirm a burst or locate a pipe.", "run_burst_detection": "Pressure anomalies alone do not confirm a burst or locate a pipe.",
"run_burst_location": "Candidate ranking depends on the leakage assumption, model and sensor coverage; field confirmation is required.", "run_burst_location": "Candidate ranking depends on the leakage assumption, model and sensor coverage; field confirmation is required.",
"analyze_valve_isolation": "Topology-only isolation advice; valve operability, hydraulic impact and field actions are not verified or executed.", "analyze_valve_isolation": "Topology-only isolation advice; valve operability, hydraulic impact and field actions are not verified or executed.",
"generate_node_coverage": "Derived node coverage for display only; geometry does not establish a fault, contamination, outage or service boundary. Missing GIS positions are reported separately.",
"calculate_demand_allocation": "Pipe-length-weighted allocation only; model demands were not modified. Units follow the current model.", "calculate_demand_allocation": "Pipe-length-weighted allocation only; model demands were not modified. Units follow the current model.",
"run_burst_simulation": "Forward hydraulic simulation for assumed orifice areas, not detection or confirmed burst location; only requested elements returned.", "run_burst_simulation": "Forward hydraulic simulation for assumed orifice areas, not detection or confirmed burst location; only requested elements returned.",
}[name]] }[name]]
+29
View File
@@ -5,6 +5,35 @@ from typing import Any
from ..core.database import read, read_all from ..core.database import read, read_all
def get_node_geographic_positions(name: str, node_ids: list[str]) -> list[Any]:
"""Resolve only the selected project's stored node geometries to WGS84."""
if not node_ids:
return []
return read_all(name, """
SELECT node_id AS id,
ST_X(ST_Transform(geom,4326)) AS longitude,
ST_Y(ST_Transform(geom,4326)) AS latitude
FROM gis.node_geometries
WHERE node_id = ANY(%s) AND geom IS NOT NULL AND NOT ST_IsEmpty(geom)
ORDER BY node_id
""", (node_ids,))
def get_node_coverage_links(name: str, node_ids: list[str]) -> list[Any]:
"""Read the selected subgraph and initial statuses without changing the model."""
if not node_ids:
return []
return read_all(name, """
SELECT l.id, l.start_node_id, l.end_node_id,
COALESCE(s.status::text, p.status::text) AS status
FROM network.links l
LEFT JOIN network.pipes p ON p.link_id = l.id
LEFT JOIN network.link_initial_settings s ON s.link_id = l.id
WHERE l.start_node_id = ANY(%s) AND l.end_node_id = ANY(%s)
ORDER BY l.id
""", (node_ids, node_ids))
def _node_coord_rows( def _node_coord_rows(
rows: list[Any], rows: list[Any],
) -> dict[str, dict[str, Any]]: ) -> dict[str, dict[str, Any]]:
+128
View File
@@ -0,0 +1,128 @@
"""Reusable geometric coverage of verified project nodes, independent of business analysis."""
import json
from math import isfinite
from typing import Any
from pyproj import CRS, Transformer
from shapely import MultiPoint, concave_hull, get_num_coordinates
from shapely.geometry import mapping
from shapely.ops import transform, unary_union
from app.native.wndb.gis.network_views import get_node_coverage_links, get_node_geographic_positions
CONCAVE_RATIO = 0.25
BUFFER_METERS = 25.0
SIMPLIFY_METERS = 2.0
MAX_COVERAGE_NODES = 10000
def _components(node_ids: list[str], links: list[dict[str, Any]], closed: set[str]) -> list[list[str]]:
parent = {node_id: node_id for node_id in node_ids}
def root(node: str) -> str:
while parent[node] != node:
parent[node] = parent[parent[node]]
node = parent[node]
return node
for link in links:
start, end = link["start_node_id"], link["end_node_id"]
if link["id"] not in closed and link.get("status") != "CLOSED" and start in parent and end in parent:
parent[root(start)] = root(end)
groups: dict[str, list[str]] = {}
for node_id in node_ids:
groups.setdefault(root(node_id), []).append(node_id)
return list(groups.values())
def build_node_coverage(
groups: list[dict[str, Any]], positions: list[dict[str, Any]],
links: list[dict[str, Any]], *, split_by_connectivity: bool = False,
excluded_link_ids: list[str] | None = None,
concave_ratio: float = CONCAVE_RATIO, buffer_m: float = BUFFER_METERS,
) -> dict[str, Any]:
"""Keep caller-defined comparison groups and derive geometry without assigning business meaning."""
coordinates = {}
for row in positions:
longitude, latitude = row.get("longitude"), row.get("latitude")
if (isinstance(longitude, (int, float)) and not isinstance(longitude, bool)
and isinstance(latitude, (int, float)) and not isinstance(latitude, bool)
and isfinite(longitude) and isfinite(latitude)
and -180 <= longitude <= 180 and -90 < latitude < 90):
coordinates[str(row["id"])] = (longitude, latitude)
features, groups_metadata = [], {}
closed = set(excluded_link_ids or [])
simplify_m = min(SIMPLIFY_METERS, buffer_m / 4)
for group in groups:
group_id, label = group["group_id"], group["title"]
ids = sorted(set(group["node_ids"]))
mapped = [node for node in ids if node in coordinates]
missing = [node for node in ids if node not in coordinates]
metadata = {"node_ids": ids, "mapped_node_count": len(mapped), "missing_node_ids": missing}
groups_metadata[group_id] = {"title": label, **metadata}
metadata = groups_metadata[group_id]
if not mapped:
metadata["component_count"] = 0
continue
points = [coordinates[node] for node in mapped]
longitude = sum(point[0] for point in points) / len(points)
latitude = sum(point[1] for point in points) / len(points)
local = CRS.from_proj4(
f"+proj=aeqd +lat_0={latitude} +lon_0={longitude} +datum=WGS84 +units=m"
)
project = Transformer.from_crs(4326, local, always_xy=True).transform
unproject = Transformer.from_crs(local, 4326, always_xy=True).transform
envelopes = []
components = _components(ids, links, closed) if split_by_connectivity else [ids]
for nodes in components:
component_points = [coordinates[node] for node in nodes if node in coordinates]
if not component_points:
continue
cloud = transform(project, MultiPoint(component_points))
# A point or collinear cloud remains drawable through the same metric buffer.
envelope = concave_hull(cloud, ratio=concave_ratio).buffer(buffer_m)
envelopes.append(envelope)
metadata["component_count"] = len(envelopes)
geometry = unary_union(envelopes).simplify(simplify_m, preserve_topology=True)
if geometry.is_empty or not geometry.is_valid or get_num_coordinates(geometry) > 20000:
raise ValueError("Invalid derived coverage")
feature = {
"type": "Feature", "id": group_id,
"properties": {
"coverage_group": group_id, "label": label, "derived": True, "display_only": True,
"node_count": len(ids), "mapped_node_count": len(mapped),
"missing_node_count": len(missing), "component_count": len(envelopes),
},
"geometry": mapping(transform(unproject, geometry)),
}
features.append(feature)
return {
"status": "partial" if any(group["missing_node_ids"] for group in groups_metadata.values())
else "ready" if features else "empty",
"method": "concave_hull_buffer",
"parameters": {"concave_ratio": concave_ratio, "buffer_m": buffer_m,
"split_by_connectivity": split_by_connectivity,
"excluded_link_ids": sorted(closed),
"simplify_m": simplify_m, "projection": "local_azimuthal_equidistant"},
"output_crs": "EPSG:4326", "display_only": True,
"position_source": "gis.node_geometries",
"boundary": "节点空间覆盖示意;不推断故障、污染、停水、水力影响或用户服务边界。",
"groups": groups_metadata,
"map_payload": {"type": "FeatureCollection", "features": features},
}
def generate_node_coverage(network: str, groups: list[dict[str, Any]], **parameters: Any) -> dict[str, Any]:
node_ids = sorted({node for group in groups for node in group["node_ids"]})
if len(node_ids) > MAX_COVERAGE_NODES:
raise ValueError("Node coverage supports at most 10000 distinct nodes")
positions = get_node_geographic_positions(network, node_ids)
links = get_node_coverage_links(network, node_ids) if parameters.get("split_by_connectivity") else []
coverage = build_node_coverage(groups, positions, links, **parameters)
payload = coverage.pop("map_payload")
if len(json.dumps(payload, ensure_ascii=False).encode("utf-8")) > 400 * 1024:
raise ValueError("Coverage exceeds presentation budget; narrow the node selection")
return {"coverage": coverage, "map_payload": payload}
+7
View File
@@ -51,12 +51,19 @@ def _geometry(value):
def resolve_overlay(snapshot: TaskDataSnapshot, mode: str) -> dict: def resolve_overlay(snapshot: TaskDataSnapshot, mode: str) -> dict:
"""Use recorded GeoJSON or recorded coordinates; never geocode or infer a boundary.""" """Use recorded GeoJSON or recorded coordinates; never geocode or infer a boundary."""
data = snapshot.data data = snapshot.data
# MCP persists its complete QueryResult envelope, unlike direct data queries.
if (isinstance(data, dict) and data.get("source") == snapshot.source
and isinstance(data.get("data"), (dict, list))):
data = data["data"]
candidates = [data] candidates = [data]
if isinstance(data, dict): if isinstance(data, dict):
candidates.extend([data.get("map_payload"), data.get("drawing_payload")]) candidates.extend([data.get("map_payload"), data.get("drawing_payload")])
latest = data.get("latest_analysis") latest = data.get("latest_analysis")
if isinstance(latest, dict): if isinstance(latest, dict):
candidates.append(latest.get("map_payload")) candidates.append(latest.get("map_payload"))
result = data.get("result")
if isinstance(result, dict):
candidates.append(result.get("map_payload"))
collection = next((item for item in candidates collection = next((item for item in candidates
if isinstance(item, dict) and item.get("type") == "FeatureCollection"), None) if isinstance(item, dict) and item.get("type") == "FeatureCollection"), None)
if collection is not None: if collection is not None:
+1 -1
View File
@@ -3,7 +3,7 @@
"contracts": { "contracts": {
"server": { "server": {
"file": "server-v1.openapi.json", "file": "server-v1.openapi.json",
"sha256": "fa5b186d0f1e4257352a812e578cb9901baf47beaca33c388e876f1a7b4c42ae" "sha256": "2c1d757ccf06ae72b10c592e1e9b3d32b4f72cda8489e224cf2c5f2690fc6972"
} }
} }
} }
+1
View File
@@ -8305,6 +8305,7 @@
"run_burst_detection", "run_burst_detection",
"run_burst_location", "run_burst_location",
"analyze_valve_isolation", "analyze_valve_isolation",
"generate_node_coverage",
"run_leakage_identification", "run_leakage_identification",
"get_element_properties", "get_element_properties",
"list_elements", "list_elements",
@@ -77,6 +77,38 @@ Server 校验当前 run、绑定会话、有效租约和幂等身份,记录 `o
没有坐标的地址、空数据、非法几何、模型自绘区域均不能生成覆盖层。该接口不做地址解析、 没有坐标的地址、空数据、非法几何、模型自绘区域均不能生成覆盖层。该接口不做地址解析、
范围推算或污染传播计算。展示事件总大小上限为 512 KiB,超限返回 422。 范围推算或污染传播计算。展示事件总大小上限为 512 KiB,超限返回 422。
`generate_node_coverage` 是独立的节点空间覆盖工具,适用于普通项目聊天和后台调查,
不绑定爆管、阀门隔离或固定执行顺序。调用者提供需要比较的节点分组,Server 从当前项目
`gis.node_geometries` 读取真实位置。监测设备、投诉、地址等须先关联模型节点,
不接受模型自由生成的坐标。具体适用范围和参数见[节点覆盖工具](node-coverage-tool.md)。
计算先投影到局部米制坐标,取凹包(默认 ratio=0.25)、加展示缓冲(默认 25 m),
再以 2 m 容差简化并转回 WGS84;凹包比例与缓冲距离可在有界参数内选择。
默认展示各组的地理包络,不判断连接关系;需要时可按模型连接拆分,额外指定视为断开的连接。
少量点、重合点和共线点同样通过缓冲形成面。多组结果一起保存在 `result.map_payload`,
避免连续 `set_overlay` 覆盖前一组;`result.coverage` 保存来源节点、映射完整性及算法参数。
该面不推断故障、污染、实际停水、水力影响或用户服务边界,缓冲距离也不是供水服务半径。
缺失几何时标记 `partial`,没有可用位置时不生成面。输入最多 8 组、合计 10000 个不同节点;
几何输出遵守覆盖层顶点和大小限制,超限要求缩小范围,不静默抽样或改变原分析结果。
MCP 保存完整查询响应作为不可变证据,展示解析器同时支持响应信封中的
`data.result.map_payload`。Agent 使用 `set_overlay` 引用本次 `evidence_id` 并省略 `data`。
超过内联返回预算时,响应的 `data.presentation.command` 仍保留可发布的展示命令,
`data.presentation.groups` 保留覆盖完整性计数,
完整几何留在证据中,不能把缩略返回误认为没有空间成果。调查按实际步骤同步发布展示:
地址得到明确坐标后缩放定位,取得候选资产后定位候选,检查影响节点后生成并展示覆盖示意,
形成或调整关阀方案后定位阀门,读取监测对照后展示点位及序列。
图表同样随结果返回发布:监测趋势与事件对照用折线图,可比候选评价或方案用柱状图,
水力及污染传播等模拟按实际返回的时刻、对象和指标画图。压力、流量、浓度等不同单位分图,
注明来源、范围与实测或模型属性;没有同口径基准时不计算或绘制改善量,缺测不补零。
过程图表通过 `open_chart` 提供 `chart_id/chart/evidence_refs`,每图最多100个横轴值、8组序列,
超限时分图或明确说明抽样、聚合范围;单值或无可比数据用文字或表格。
关键趋势与比较结论保留到报告中,最终报告的图表不替代过程发布。
业务 Skill 说明本步骤应展示什么,调查指令统一说明发布方法、非空几何检查、去重和失败处理。
同一阶段批量展示,数据或方案没有变化时不重复请求;不为地图动作增加无关业务分析。
报告的 `map_targets` 仅为手动定位入口。关键步骤的数据缺失或展示失败需记录具体限制,
没有浏览器连接也不阻止分析或报告收尾。
`GET /api/v1/operational-tasks/{task_id}/presentation-cursor` 在同一事务中读取 `GET /api/v1/operational-tasks/{task_id}/presentation-cursor` 在同一事务中读取
`task_id/execution_generation/active_run/event_cursor`,其中活跃 run 仅返回 `run_id/generation/status`。 `task_id/execution_generation/active_run/event_cursor`,其中活跃 run 仅返回 `run_id/generation/status`。
浏览器打开 Dev 联动、切换任务或断线重连时,以此建立新基线,只消费之后的展示事件。 浏览器打开 Dev 联动、切换任务或断线重连时,以此建立新基线,只消费之后的展示事件。
+43
View File
@@ -0,0 +1,43 @@
# 节点空间覆盖工具
`generate_node_coverage` 增加 Agent 可选择的空间表达能力,将已取得的模型节点集合转为
可解释、可复核的覆盖示意。它可以在项目聊天、任务协作和后台调查中使用,
不要求爆管任务、隔离成功或特定报告。生成、发布展示和业务分析是三个独立操作。
## 适用范围
| 问题 | 输入依据 | 输出含义 |
| --- | --- | --- |
| 候选对象分布在哪里 | 已核实的候选模型节点 | 候选节点空间包络 |
| 两组监测点覆盖有何不同 | 已核对的设备到模型节点映射 | 已接入节点分布比较,不证明监测可观测性 |
| 区域成员分布在哪里 | 已查询的区域成员节点 | 成员节点包络,不替代已有行政区或 DMA 边界 |
| 哪些节点属于模型影响集合 | 已有拓扑或模拟结果中的节点集合 | 结果节点的空间示意,不重新计算影响 |
| 投诉关联的节点如何分布 | 已解析并核对的投诉到模型节点关联 | 已关联节点分布,不证明投诉或污染真实边界 |
只有地址或未关联设备时,先查询位置和资产关联;已有权威区域几何时优先展示原始边界。
本工具不接收任意坐标、HTML、SQL、脚本或调用者指定的项目身份,不执行关阀、模拟或方案保存。
## 参数和结果
- `groups`:最多 8 组,每组提供唯一 `group_id`、`title` 和 `node_ids`;组之间可重叠,
合计最多 10000 个不同节点。不按组名自动赋予隔离、污染或服务覆盖语义。
- `concave_ratio`:0–1,默认 0.25;越接近 1 越接近凸包。分散的点取单一包络可能包含较大空白,
应按实际比较目标分组,并明确示意含义。
- `buffer_m`:大于 0、最多 1000 m,默认 25 m,仅是展示缓冲。
- `split_by_connectivity`:默认 false。true 时按所选节点的模型连接拆分,并排除初始 CLOSED 连接。
`excluded_link_ids` 可额外指定最多 200 个断开连接,适用于已有关阀建议等有依据的绘图情景。
这不会修改模型,也不能替代完整拓扑分析。
`result.map_payload` 是多组共存的 GeoJSON;`result.coverage` 记录算法参数、节点 ID、
已映射数量、缺失几何节点、分组数量和 `display_only` 边界。缺少部分位置时返回 `partial`,
不补造坐标。单点、重合点或共线点通过米制缓冲生成可绘制面。
## 与展示工具配合
生成工具不会移动地图。项目聊天可用 `present_view` 展示本次返回的几何;后台调查用
`publish_task_presentation` 的 `set_overlay`,引用生成结果自己的 `evidence_id` 并省略 `data`,
Server 从不可变证据解析完整几何。大结果内联缩略时,`data.presentation.command` 保留可发布命令。
`recorded` 仍只表示请求已记录,浏览器显示和后台调查完成各自独立。
本次爆管报告只是一个验证样例:可将 24 个直接隔离节点、295 个供水影响节点作为两组,
按已有三阀建议设置连通分组;不将这种调用固定进 `analyze_valve_isolation`。
@@ -2,7 +2,7 @@
name: appointment-follow-up name: appointment-follow-up
description: 核查关联预约工单的审批、提交、现场执行和反馈,分别判断预约履约与原事件闭环。 description: 核查关联预约工单的审批、提交、现场执行和反馈,分别判断预约履约与原事件闭环。
metadata: metadata:
version: "2.0.2" version: "2.0.4"
--- ---
# 预约任务跟进 # 预约任务跟进
@@ -18,13 +18,15 @@ metadata:
1. 将要求与来源工单最新版本逐项对照,区分预约、派发、到场、完成和原事件发生时间,不用创建或更新时间替代现场时间,也不因预约已过推定已经执行或失约。 1. 将要求与来源工单最新版本逐项对照,区分预约、派发、到场、完成和原事件发生时间,不用创建或更新时间替代现场时间,也不因预约已过推定已经执行或失约。
2. 核对动作的审批、提交、执行状态和回执。`approved` 只表示允许提交,`submitted` 只表示已接收,`succeeded` 的建单动作只证明工单创建成功;这些状态均不证明采样、通知、抢修、冲洗或复测完成。`unknown` 由 Server 复核原执行请求,不重复下发。 2. 核对动作的审批、提交、执行状态和回执。`approved` 只表示允许提交,`submitted` 只表示已接收,`succeeded` 的建单动作只证明工单创建成功;这些状态均不证明采样、通知、抢修、冲洗或复测完成。`unknown` 由 Server 复核原执行请求,不重复下发。
3. 用 `query_peripheral_data` 查询外围工单、点位和现场观察,按来源 ID、关联动作及版本核对;目录首屏不能证明没有反馈。区分缺失、重复、旧版本和相互冲突的记录。 3. 用 `query_peripheral_data` 查询外围工单、点位和现场观察,按来源 ID、关联动作及版本核对;目录首屏不能证明没有反馈。区分缺失、重复、旧版本和相互冲突的记录。
4. 按原目标查询位置、资产和工作前后的监测。核实设备与模型要素、时间覆盖、单位和数据质量;缺测不是零值,局部正常不证明整项工作完成。 4. 按原目标查询位置、资产和工作前后的监测。地点和资产明确后展示正在核查的工作位置;取得监测对照后展示核查点位,并画出实际取得的工作前后趋势或对比图,核对现场发生时间与数据覆盖,不能把计划时间当实际执行分界;已有影响结果时展示其真实几何或节点覆盖示意。核实设备与模型要素、时间覆盖、单位和数据质量;缺测不是零值,局部正常不证明整项工作完成。
5. 将现场描述、来源系统附件或照片说明、监测变化和原工作要求交叉核对。没有图像查看能力时不声称已检查照片内容。每项要求标为已核验、待核验或不适用并给出依据;状态与证据冲突时明确列出。 5. 将现场描述、来源系统附件或照片说明、监测变化和原工作要求交叉核对。没有图像查看能力时不声称已检查照片内容。每项要求标为已核验、待核验或不适用并给出依据;状态与证据冲突时明确列出。
## 结果与后续 ## 结果与后续
报告分别说明原预约要求、审批及派发事实、已核实的现场工作和效果、冲突或缺口、当前责任事项,以及它与原事件的关系。等待回执不等于调查完成,更不等于现场工作完成。 报告分别说明原预约要求、审批及派发事实、已核实的现场工作和效果、冲突或缺口、当前责任事项,以及它与原事件的关系。等待回执不等于调查完成,更不等于现场工作完成。
位置、资产、监测和已有影响结果随核查步骤同步展示,遵循后台地图与图表规则,不用最终报告代替过程展示;单个数值或无可比数据时用文字、指标或表格,不为凑图新增业务分析。无数据或展示失败时说明具体限制,调查不等待浏览器。
只有证据支持新的具体工作时才提议动作。先核对已有有效动作和关联工单,优先沿原工单的反馈与人工修改流程推进;不要因缺少一项反馈就新建催办或复查工单。新增人工工作使用关联的 `work_order_type=general` 草稿,写明地点、内容、依赖、执行前核验和回执要求。 只有证据支持新的具体工作时才提议动作。先核对已有有效动作和关联工单,优先沿原工单的反馈与人工修改流程推进;不要因缺少一项反馈就新建催办或复查工单。新增人工工作使用关联的 `work_order_type=general` 草稿,写明地点、内容、依赖、执行前核验和回执要求。
本流程不自动发起隔离停水、污染水排放或设备命令。发现新的爆管或水质风险时,列出证据、紧迫事项和建议,由原事件的合法调查继续专业分析;不要改写当前任务类型或递归创建同类事件。 本流程不自动发起隔离停水、污染水排放或设备命令。发现新的爆管或水质风险时,列出证据、紧迫事项和建议,由原事件的合法调查继续专业分析;不要改写当前任务类型或递归创建同类事件。
@@ -2,7 +2,7 @@
name: burst-report name: burst-report
description: 调查供水爆管上报,形成有依据的候选位置、隔离阀门、影响范围和后续处置建议。 description: 调查供水爆管上报,形成有依据的候选位置、隔离阀门、影响范围和后续处置建议。
metadata: metadata:
version: "2.0.3" version: "2.0.6"
--- ---
# 爆管上报处置 # 爆管上报处置
@@ -17,18 +17,18 @@ metadata:
根据现有线索自主选择必要步骤: 根据现有线索自主选择必要步骤:
1. 核对已有管段或节点。只有地点描述时,先解析位置,再查询附近资产、属性和连接。空间邻近只支持候选,不能单独证明故障。 1. 核对已有管段或节点。只有地点描述时,先解析位置,再查询附近资产、属性和连接。地点明确后展示调查位置,候选资产取得后展示正在核查的管段或节点。空间邻近只支持候选,不能单独证明故障。
2. 核对 SCADA 设备与模型节点映射,用 `query_scada_readings` 或 `query_element_history` 查询事件前后及正常对照。先用少量对象和合适时间窗,确认返回时间实际覆盖事件;必要时缩窗或翻页。保留单位、缺测、真实零值和覆盖限制。 2. 核对 SCADA 设备与模型节点映射,用 `query_scada_readings` 或 `query_element_history` 查询事件前后及正常对照。先用少量对象和合适时间窗,确认返回时间实际覆盖事件;必要时缩窗或翻页。同步展示核查点位,并将取得的事件、对照压力或流量序列画成过程曲线,分清指标和单位;保留缺测、真实零值和覆盖限制。
3. 有异常观测时评估爆管定位输入。`run_burst_location` 必须有 `burst_leakage`(m³/s);先查询已有估计及方法依据,或结合边界流量、用水和储量变化判断能否估计。不能把单表增量、压力下降或任意数值直接当漏量。 3. 有异常观测时评估爆管定位输入。`run_burst_location` 必须有 `burst_leakage`(m³/s);先查询已有估计及方法依据,或结合边界流量、用水和储量变化判断能否估计。不能把单表增量、压力下降或任意数值直接当漏量。
4. 有依据的漏量估计或明确的情景假设可用于条件定位,须说明来源、方法和不确定性。`pressure_scada_ids` 使用模型节点 ID;监测模式读取 SCADA,模拟模式须引用已有分析运行。压力点和历史数据满足工具要求时可选 `run_burst_detection`,但侦测不代替定位,评价值也不表述为概率。缺少定位参数时说明已查来源和影响,继续其他可独立完成的分析。 4. 有依据的漏量估计或明确的情景假设可用于条件定位,须说明来源、方法和不确定性。`pressure_scada_ids` 使用模型节点 ID;监测模式读取 SCADA,模拟模式须引用已有分析运行。取得多个可比候选的评价值后,用柱状图展示实际排序及评价含义。压力点和历史数据满足工具要求时可选 `run_burst_detection`,但侦测不代替定位,评价值也不表述为概率。缺少定位参数时说明已查来源和影响,继续其他可独立完成的分析。
5. 对已确认或有依据的候选资产使用 `analyze_valve_isolation` 评估阀门隔离,无需等待现场确诊。逐候选比较;多个事故 ID 表示联合场景。保留可隔离性、必关阀、可选阀及受影响节点。未知阀门状态不能写成可用,不可隔离或空结果也不能解释为零影响;已知阀门失效时评估替代边界。 5. 对已确认或有依据的候选资产使用 `analyze_valve_isolation` 评估阀门隔离,无需等待现场确诊。逐候选比较;多个事故 ID 表示联合场景。保留可隔离性、必关阀、可选阀及受影响节点。结果取得后展示直接隔离与供水影响节点的覆盖示意,形成或调整关阀方案时展示对应阀门;区分两种影响及必关、可选阀。未知阀门状态不能写成可用,不可隔离或空结果也不能解释为零影响;已知阀门失效时评估替代边界。
6. 输入具备时用 `analyze_valve_closure` 分析关阀水力影响,核对阀门组合、起算时间、时长和输出对象。区分拓扑范围、水力影响和有依据的用户影响,不虚构户数,不用局部结果代表全网。 6. 输入具备时用 `analyze_valve_closure` 分析关阀水力影响,核对阀门组合、起算时间、时长和输出对象。结果返回后,用曲线展示请求节点的压力、请求管段的流量;有同口径方案或基准时再画对比图,不自造前后变化。不同单位分图,标明局部模型结果。区分拓扑范围、水力影响和有依据的用户影响,不虚构户数,不用局部结果代表全网。
查询错误、权限拒绝或接口故障不是缺数证据。必要分析无法执行时,说明失败步骤和已经完成的部分,不借有限摘要宣布调查完成。 查询错误、权限拒绝或接口故障不是缺数证据。必要分析无法执行时,说明失败步骤和已经完成的部分,不借有限摘要宣布调查完成。
## 处置与报告 ## 处置与报告
报告分别给出位置候选及依据、隔离阀门及可用性、替代方案、拓扑与水力影响、用户影响边界,以及处置和闭环缺口。只有初步结果时明确范围,不宣称已经定位、确诊或闭环。按需展示候选、阀门、监测和影响,但展示不替代证据。 报告分别给出位置候选及依据、隔离阀门及可用性、替代方案、拓扑与水力影响、用户影响边界,以及处置和闭环缺口。只有初步结果时明确范围,不宣称已经定位、确诊或闭环。位置、候选、监测、影响和阀门方案按实际调查步骤同步展示,遵循后台展示规则,不用报告定位链接代替过程展示。覆盖面保留分组、缺失几何和示意边界,不能当作实际停水、水力影响或用户服务边界。无数据或展示失败时说明具体限制;展示不替代证据,也不要求等待浏览器。
形成具体建议后,先保存阶段结论,再提议必要动作。人工核验、通知、定位隔离、抢修、适用的清洗复测和恢复回访使用关联的 `work_order_type=general` 草稿;原爆管事件保持独立。草稿写明位置、任务、依赖、执行前核验和回执要求。没有通知适配器时明确由人工实施。设备命令仅用于真实可控目标,审批和执行由 Server 控制。 形成具体建议后,先保存阶段结论,再提议必要动作。人工核验、通知、定位隔离、抢修、适用的清洗复测和恢复回访使用关联的 `work_order_type=general` 草稿;原爆管事件保持独立。草稿写明位置、任务、依赖、执行前核验和回执要求。没有通知适配器时明确由人工实施。设备命令仅用于真实可控目标,审批和执行由 Server 控制。
@@ -2,7 +2,7 @@
name: water-quality-complaint name: water-quality-complaint
description: 调查供水水质投诉,核查投诉与监测覆盖,评估候选原因、影响和采样处置建议。 description: 调查供水水质投诉,核查投诉与监测覆盖,评估候选原因、影响和采样处置建议。
metadata: metadata:
version: "2.0.2" version: "2.0.4"
--- ---
# 水质投诉处置 # 水质投诉处置
@@ -18,17 +18,17 @@ metadata:
根据当前证据选择必要步骤: 根据当前证据选择必要步骤:
1. 核实投诉位置及相近时段的同类投诉,用 `query_peripheral_data` 按实际时间、位置和来源标识核对外围工单或观察。目录首屏不能证明没有其他投诉;未知点位先查目录,不猜编号或把演示点当真实资产。 1. 核实投诉位置及相近时段的同类投诉,用 `query_peripheral_data` 按实际时间、位置和来源标识核对外围工单或观察。目录首屏不能证明没有其他投诉;未知点位先查目录,不猜编号或把演示点当真实资产。
2. 解析未知地址,查询附近资产、属性和连接,区分上游、管网、末梢、二次供水和用户端的实际覆盖。拓扑连接不等于实际流向,方向判断需要同窗模型或监测依据。 2. 解析未知地址,查询附近资产、属性和连接,区分上游、管网、末梢、二次供水和用户端的实际覆盖。地点明确后展示投诉位置,取得关联资产后展示正在核查的管段或节点。拓扑连接不等于实际流向,方向判断需要同窗模型或监测依据。
3. 核对 SCADA 指标、单位和模型映射,用 `query_scada_readings` 或 `query_element_history` 查询投诉前后及对照,检查时间覆盖、在线状态、漂移、长期缺测和质量标识。缺测不是零值,上游正常不能排除末梢问题,压力正常不能证明水质合格。 3. 核对 SCADA 指标、单位和模型映射,用 `query_scada_readings` 或 `query_element_history` 查询投诉前后及对照,检查时间覆盖、在线状态、漂移、长期缺测和质量标识。同步展示核查点位,将实际取得的水质、压力等指标画成趋势或对比图,不同单位分图。缺测不是零值,上游正常不能排除末梢问题,压力正常不能证明水质合格。
4. 综合投诉分布、监测、供水关联和实际采样,列出候选原因、支持证据、反证和区分原因所需的核验。未证实污染源时保留假设,不把感官描述换算成浓度。水质限值只能引用已取得且适用的依据,不能自行承诺安全饮用。 4. 综合投诉分布、监测、供水关联和实际采样,列出候选原因、支持证据、反证和区分原因所需的核验。未证实污染源时保留假设,不把感官描述换算成浓度。水质限值只能引用已取得且适用的依据,不能自行承诺安全饮用。
5. 有合理目标管段或节点时,可用 `analyze_valve_isolation` 和 `analyze_valve_closure` 评估候选隔离及关阀影响,无需等待污染确诊,但不能据此直接执行停水。逐候选区分可隔离性、必关阀、可选阀、受影响节点和未知阀门状态。拓扑隔离区、水力影响、污染范围和用户影响分别说明;无用户资料时不虚构户数。 5. 有合理目标管段或节点时,可用 `analyze_valve_isolation` 和 `analyze_valve_closure` 评估候选隔离及关阀影响,无需等待污染确诊,但不能据此直接执行停水。逐候选区分可隔离性、必关阀、可选阀、受影响节点和未知阀门状态。取得影响节点后展示覆盖示意,形成或调整关阀方案时展示对应阀门。拓扑隔离区、水力影响、污染范围和用户影响分别说明;节点覆盖面不能当作污染传播边界,无用户资料时不虚构户数。
6. 只有假设源节点、浓度及必要模型输入有依据时,才用 `run_contaminant_simulation` 正向比较传播;它不是源头反演,未模拟范围不能宣布安全。具备实际排放节点、流量和前置条件时才用 `analyze_flushing` 评估冲洗。冲洗模拟不证明消毒合格,排放位置和承接条件仍需现场核验与审批。 6. 只有假设源节点、浓度及必要模型输入有依据时,才用 `run_contaminant_simulation` 正向比较传播;它不是源头反演,未模拟范围不能宣布安全。具备实际排放节点、流量和前置条件时才用 `analyze_flushing` 评估冲洗。取得模拟结果后,按实际返回的对象、时刻和指标展示浓度、压力或流量曲线;同口径的隔离、冲洗方案可画对比图,标明情景假设和局部覆盖。冲洗模拟不证明消毒合格,排放位置和承接条件仍需现场核验与审批。
缺少源参数不阻止已有证据支持的采样、候选隔离或人工告知建议。查询错误、权限拒绝或接口故障属于执行异常,不能写成无数据或未见异常。 缺少源参数不阻止已有证据支持的采样、候选隔离或人工告知建议。查询错误、权限拒绝或接口故障属于执行异常,不能写成无数据或未见异常。
## 处置与报告 ## 处置与报告
报告分别说明投诉与监测覆盖、候选原因和未排除路径、隔离及影响、通知保障、采样或污染水处置,以及复测与恢复条件。每项给出实际结果或经查询确认的限制;区分当前风险判断与尚未发生的未来处置。按需展示投诉、监测、候选隔离和影响,但展示不替代证据。 报告分别说明投诉与监测覆盖、候选原因和未排除路径、隔离及影响、通知保障、采样或污染水处置,以及复测与恢复条件。每项给出实际结果或经查询确认的限制;区分当前风险判断与尚未发生的未来处置。投诉位置、监测、候选隔离和影响按实际步骤同步展示,遵循后台展示规则,不用报告定位链接代替过程展示。无数据或展示失败时说明具体限制;展示不替代证据,也不要求等待浏览器。
形成具体建议后,先保存阶段结论,再提议关联的 `work_order_type=general` 草稿。写明实际位置、采样或处置目标、时间要求、依赖、现场待核验条件和回执要求。通知、采样、清洗冲洗与复测按依赖合理组织,不机械逐节点建单,也不重复已有有效动作。没有客户通知适配器时明确由人工实施;调查消息不能当作用户通知。设备命令仅针对真实可控目标,审批与执行由 Server 控制。 形成具体建议后,先保存阶段结论,再提议关联的 `work_order_type=general` 草稿。写明实际位置、采样或处置目标、时间要求、依赖、现场待核验条件和回执要求。通知、采样、清洗冲洗与复测按依赖合理组织,不机械逐节点建单,也不重复已有有效动作。没有客户通知适配器时明确由人工实施;调查消息不能当作用户通知。设备命令仅针对真实可控目标,审批与执行由 Server 控制。
@@ -37,6 +37,18 @@ def test_unknown_node_state_is_rejected():
InvestigationDiagramNodeRef(node_id="A", label="查询", status="blocked") InvestigationDiagramNodeRef(node_id="A", label="查询", status="blocked")
def test_class_attached_nodes_are_valid_declarations_without_accepting_partial_ids():
payload = {"request_id": uuid4(), "source_session_id": "session-1", "progress": {
"summary": "核查", "diagram": {"title": "路线", "mermaid": "flowchart TD\nQUERY:::running --> REVIEW:::pending",
"node_refs": [{"node_id": "QUERY", "label": "核查", "status": "running"},
{"node_id": "REVIEW", "label": "复核", "status": "pending"}]},
}}
assert AgentArtifactsCommand.model_validate(payload).progress.diagram.node_refs[0].status == "running"
payload["progress"]["diagram"]["node_refs"][0]["node_id"] = "QUER"
with pytest.raises(ValidationError, match="QUER"):
AgentArtifactsCommand.model_validate(payload)
@pytest.mark.parametrize("artifact", ["progress", "report"]) @pytest.mark.parametrize("artifact", ["progress", "report"])
def test_artifact_write_rejects_node_refs_missing_from_mermaid(artifact): def test_artifact_write_rejects_node_refs_missing_from_mermaid(artifact):
diagram = { diagram = {
+3 -2
View File
@@ -35,6 +35,7 @@ INPUTS = {
"run_burst_detection": {"target_time": START, "sensor_nodes": ["1", "2", "3", "4", "5"]}, "run_burst_detection": {"target_time": START, "sensor_nodes": ["1", "2", "3", "4", "5"]},
"run_burst_location": {"data_source": "simulation", "simulation_run_id": str(uuid4()), "burst_leakage": 0.02, "pressure_scada_ids": ["1"], "scada_burst_start": START, "scada_burst_end": END}, "run_burst_location": {"data_source": "simulation", "simulation_run_id": str(uuid4()), "burst_leakage": 0.02, "pressure_scada_ids": ["1"], "scada_burst_start": START, "scada_burst_end": END},
"analyze_valve_isolation": {"accident_element": ["540672"]}, "analyze_valve_isolation": {"accident_element": ["540672"]},
"generate_node_coverage": {"groups": [{"group_id": "monitoring", "title": "监测节点分布", "node_ids": ["J1"]}]},
"run_leakage_identification": {"scada_start": START, "scada_end": END, "sensor_nodes": ["1"], "q_sum": 0.02, "start_time": 12, "dma_count": 1}, "run_leakage_identification": {"scada_start": START, "scada_end": END, "sensor_nodes": ["1"], "q_sum": 0.02, "start_time": 12, "dma_count": 1},
} }
@@ -129,7 +130,7 @@ def test_missing_timescale_route_stops_online_simulation_before_model_changes(mo
calculate.assert_not_awaited() calculate.assert_not_awaited()
@pytest.mark.parametrize("name", ["analyze_valve_isolation", "optimize_sensor_placement"]) @pytest.mark.parametrize("name", ["analyze_valve_isolation", "optimize_sensor_placement", "generate_node_coverage"])
def test_business_only_calculations_do_not_require_timescale(monkeypatch, routed_calculation_scope, name): def test_business_only_calculations_do_not_require_timescale(monkeypatch, routed_calculation_scope, name):
scope, _, events = routed_calculation_scope scope, _, events = routed_calculation_scope
@@ -192,7 +193,7 @@ def test_catalog_and_snapshots_accept_calculations_but_reject_model_owned_author
@pytest.mark.parametrize("name", INPUTS) @pytest.mark.parametrize("name", INPUTS)
def test_calculation_permission_and_reader_boundary(monkeypatch, name): def test_calculation_permission_and_reader_boundary(monkeypatch, name):
permissions = {"burst.run", "simulation.run", "simulation.view", "scada.view", "optimization.run"} permissions = {"burst.run", "simulation.run", "simulation.view", "scada.view", "optimization.run", "webgis.view"}
monkeypatch.setattr(query_service, "permissions_for_context", lambda _: permissions) monkeypatch.setattr(query_service, "permissions_for_context", lambda _: permissions)
scope = SimpleNamespace(project=SimpleNamespace(project_role="member")) scope = SimpleNamespace(project=SimpleNamespace(project_role="member"))
binding = SimpleNamespace(tool_cap=list(TOOLS), purpose="task_execution", permission_cap=list(permissions), task_id=uuid4(), task_scope_cap=["scada_readings"]) binding = SimpleNamespace(tool_cap=list(TOOLS), purpose="task_execution", permission_cap=list(permissions), task_id=uuid4(), task_scope_cap=["scada_readings"])
@@ -0,0 +1,36 @@
from datetime import datetime, timezone
from uuid import uuid4
from app.domain.schemas.operational_task import TaskPresentationEventCommand
from app.mcp.models import QueryResult
from app.mcp.presentation_preview import saved_presentation_preview
def test_abbreviated_coverage_keeps_publishable_command_without_copying_geometry():
result = QueryResult(
source="generate_node_coverage", source_ref="mcp:analysis", evidence_id="mcp:analysis",
queried_at=datetime.now(timezone.utc), semantics="analysis", origin="system_data", truncated=False,
data={"result": {"map_payload": {"type": "FeatureCollection", "features": [{"large": "geometry"}]},
"coverage": {"status": "partial", "groups": {"isolation": {
"node_ids": ["A", "B"], "mapped_node_count": 1, "missing_node_ids": ["B"],
}}}}},
)
hint = saved_presentation_preview(result)
assert hint == saved_presentation_preview(result)
assert hint["groups"]["isolation"] == {"node_count": 2, "mapped_node_count": 1, "missing_node_count": 1}
command = TaskPresentationEventCommand(
request_id=uuid4(), run_id=uuid4(), source_session_id="session", command=hint["command"],
)
assert command.command.evidence_id == result.evidence_id and command.command.data is None
assert "geometry" not in str(hint["groups"]) and result.data["result"]["map_payload"]["features"]
result.evidence_id = None
assert saved_presentation_preview(result) is None
def test_missing_geometry_does_not_offer_a_broken_overlay_command():
result = QueryResult(
source="generate_node_coverage", source_ref="mcp:analysis", evidence_id="mcp:analysis",
queried_at=datetime.now(timezone.utc), semantics="analysis", origin="system_data", truncated=False,
data={"result": {"coverage": {"status": "unavailable"}}},
)
assert saved_presentation_preview(result) is None
+36
View File
@@ -543,6 +543,42 @@ def test_evidence_is_saved_before_return_and_call_conflicts_fail(runtime, monkey
assert len(saved) == 1 assert len(saved) == 1
def test_large_node_coverage_retains_publishable_hint_and_saves_complete_geometry(runtime, monkeypatch):
app, client, _, _ = runtime
value = binding(task=True)
saved = []
monkeypatch.setattr(application, "verify_task", AsyncMock())
monkeypatch.setattr(contexts, "verify_task", AsyncMock())
class TaskService:
repo = SimpleNamespace(lock=lambda key: asyncio.sleep(0))
async def record_data_snapshot(self, task_id, snapshot):
saved.append(snapshot)
return SimpleNamespace(evidence_id=snapshot.source_ref), snapshot
monkeypatch.setattr(application, "task_service", lambda scope: TaskService())
geometry = {"type": "FeatureCollection", "features": [{
"type": "Feature", "properties": {"label": "x" * 13000},
"geometry": {"type": "Polygon", "coordinates": [[[117, 32], [118, 32], [118, 33], [117, 32]]]},
}]}
app.executor = AsyncMock(return_value=QueryResult(
source="generate_node_coverage", source_ref="mcp:before-persisting", queried_at=datetime.now(timezone.utc),
semantics="analysis", data={"result": {"map_payload": geometry, "coverage": {
"status": "ready", "groups": {"compare": {"node_ids": ["A"], "mapped_node_count": 1}},
}}},
))
output = request(client, "tools/call", {"name": "generate_node_coverage", "arguments": {
"context_handle": value.context_handle,
"groups": [{"group_id": "compare", "title": "节点比较", "node_ids": ["A"]}],
}}).json()["result"]["structuredContent"]
assert output["truncated"] is True
command = output["data"]["presentation"]["command"]
assert command["action"] == "set_overlay" and command["evidence_id"] == saved[0].source_ref
assert "data" not in command
assert saved[0].data["data"]["result"]["map_payload"] == geometry
def test_business_skill_is_delivered_over_mcp_without_recording_event_evidence(runtime, monkeypatch): def test_business_skill_is_delivered_over_mcp_without_recording_event_evidence(runtime, monkeypatch):
app, client, _, _ = runtime app, client, _, _ = runtime
value = binding(task=True) value = binding(task=True)
+118
View File
@@ -0,0 +1,118 @@
from datetime import datetime, timezone
import pytest
from fastapi.encoders import jsonable_encoder
from pydantic import ValidationError
from shapely.geometry import Point, shape
from app.domain.schemas.operational_data import TaskDataSnapshot
from app.mcp.analysis_models import NodeCoverageInput
from app.services.node_coverage import build_node_coverage
from app.services.operational_presentation import resolve_overlay
def position(node, longitude, latitude=32.75):
return {"id": node, "longitude": longitude, "latitude": latitude}
def group(ids, group_id="selected"):
return {"group_id": group_id, "title": "节点覆盖比较", "node_ids": ids}
def link(link_id, start, end, status=None):
return {"id": link_id, "start_node_id": start, "end_node_id": end, "status": status}
def test_coverage_can_compare_arbitrary_groups_without_assigning_isolation_semantics():
positions = [position("A", 117.24), position("B", 117.241), position("C", 117.26)]
coverage = build_node_coverage([group(["A", "B"], "monitoring"), group(["C"], "candidate")], positions, [])
assert coverage["status"] == "ready"
assert set(coverage["groups"]) == {"monitoring", "candidate"}
assert [f["properties"]["coverage_group"] for f in coverage["map_payload"]["features"]] == ["monitoring", "candidate"]
assert coverage["parameters"]["split_by_connectivity"] is False
def test_connectivity_option_does_not_bridge_disconnected_groups():
positions = [position("A", 117.24), position("B", 117.241),
position("C", 117.26), position("D", 117.261)]
coverage = build_node_coverage([group(["A", "B", "C", "D"])], positions,
[link("P1", "A", "B"), link("P2", "C", "D")],
split_by_connectivity=True)
feature = coverage["map_payload"]["features"][0]
geometry = shape(feature["geometry"])
assert geometry.geom_type == "MultiPolygon" and geometry.is_valid
assert feature["properties"]["component_count"] == 2
assert all(geometry.covers(Point(row["longitude"], row["latitude"])) for row in positions)
assert not geometry.covers(Point(117.25, 32.75))
@pytest.mark.parametrize("status,excluded", [("CLOSED", []), (None, ["V"])])
def test_connectivity_excludes_initially_closed_and_explicitly_excluded_links(status, excluded):
coverage = build_node_coverage([group(["A", "B"])], [position("A", 117.24), position("B", 117.26)],
[link("V", "A", "B", status)], split_by_connectivity=True,
excluded_link_ids=excluded)
assert coverage["groups"]["selected"]["component_count"] == 2
def test_partial_geometries_are_reported_and_single_point_buffer_uses_meters():
coverage = build_node_coverage([group(["A", "A", "missing", "invalid"])],
[position("A", 117.24), position("invalid", float("nan"))], [])
assert coverage["status"] == "partial"
metadata = coverage["groups"]["selected"]
assert metadata["mapped_node_count"] == 1 and metadata["missing_node_ids"] == ["invalid", "missing"]
feature = coverage["map_payload"]["features"][0]
geometry = shape(feature["geometry"])
assert geometry.covers(Point(117.24, 32.75))
assert 0.0004 < geometry.bounds[2] - geometry.bounds[0] < 0.001
assert feature["properties"]["display_only"] is True
def test_duplicate_and_collinear_positions_are_valid_and_parameters_are_preserved():
coverage = build_node_coverage([group(["A", "B", "C"])],
[position("A", 117.24), position("B", 117.24), position("C", 117.241)], [],
concave_ratio=0.5, buffer_m=40)
assert coverage["status"] == "ready"
assert coverage["parameters"]["concave_ratio"] == 0.5 and coverage["parameters"]["buffer_m"] == 40
assert shape(coverage["map_payload"]["features"][0]["geometry"]).is_valid
def test_missing_positions_do_not_fabricate_an_area():
coverage = build_node_coverage([group(["missing"])], [], [])
assert coverage["status"] == "partial" and coverage["map_payload"]["features"] == []
@pytest.mark.parametrize("mcp_envelope", [False, True])
def test_derived_payload_round_trips_through_saved_evidence_overlay(mcp_envelope):
coverage = build_node_coverage([group(["A"])], [position("A", 117.24)], [])
data = jsonable_encoder({"result": {"map_payload": coverage["map_payload"]}})
if mcp_envelope:
data = {"source": "generate_node_coverage", "source_ref": "mcp:coverage", "data": data}
snapshot = TaskDataSnapshot(source="generate_node_coverage", source_ref="mcp:coverage",
queried_at=datetime.now(timezone.utc), origin="system_data",
truncated=False, data=data)
overlay = resolve_overlay(snapshot, "geometry")
assert overlay["features"][0]["properties"]["coverage_group"] == "selected"
assert shape(overlay["features"][0]["geometry"]).covers(Point(117.24, 32.75))
def test_service_resolves_only_selected_project_nodes_and_reads_topology_only_when_requested(monkeypatch):
from app.services import node_coverage
def positions(network, ids):
assert network == "bound-project" and ids == ["A", "B"]
return [position("A", 117.24), position("B", 117.241)]
monkeypatch.setattr(node_coverage, "get_node_geographic_positions", positions)
monkeypatch.setattr(node_coverage, "get_node_coverage_links", lambda *args: pytest.fail("unnecessary topology query"))
result = node_coverage.generate_node_coverage("bound-project", [group(["A", "B"])])
assert result["coverage"]["status"] == "ready"
assert len(result["map_payload"]["features"]) == 1
@pytest.mark.parametrize("extra", [
{"groups": []}, {"groups": [group(["A"]), group(["B"])]},
{"buffer_m": 0}, {"buffer_m": 1001}, {"concave_ratio": float("nan")},
{"excluded_link_ids": ["V"]}, {"groups": [{**group(["A"]), "coordinates": [117, 32]}]},
{"groups": [group([str(i) for i in range(10000)]), group(["extra"], "other")]},
])
def test_tool_rejects_invalid_parameters_and_unbounded_or_arbitrary_coordinate_inputs(extra):
with pytest.raises(ValidationError):
NodeCoverageInput.model_validate({"context_handle": "x" * 43, "groups": [group(["A"])], **extra})
+36
View File
@@ -598,6 +598,42 @@ async def test_unified_commands_record_canonical_events_without_mutating_task(pa
assert len(repository.events) == 1 and repository.task["revision"] == task["revision"] assert len(repository.events) == 1 and repository.task["revision"] == task["revision"]
@pytest.mark.anyio
async def test_saved_mcp_coverage_envelope_publishes_groups_without_agent_geometry():
from fastapi.encoders import jsonable_encoder
from app.services.node_coverage import build_node_coverage
now = datetime(2026, 10, 10, tzinfo=timezone.utc)
context = _context()
task = _task_document(context, now, lease=now + timedelta(minutes=1))
repository = RunTaskRepository(task, _active_run(task, now))
service = OperationalTaskService(repository, context, set(), clock=lambda: now)
coverage = build_node_coverage(
[{"group_id": "one", "title": "第一组", "node_ids": ["A"]},
{"group_id": "two", "title": "第二组", "node_ids": ["A", "B"]}],
[{"id": "A", "longitude": 117.24, "latitude": 32.75},
{"id": "B", "longitude": 117.241, "latitude": 32.75}],
[],
)
snapshot = TaskDataSnapshot(
source="generate_node_coverage", source_ref="mcp:node-coverage", queried_at=now,
origin="system_data", truncated=False,
data=jsonable_encoder({"source": "generate_node_coverage", "data": {
"result": {"map_payload": coverage["map_payload"]},
}}),
)
evidence, _ = await service.record_data_snapshot(task["task_id"], snapshot)
await service.publish_presentation_event(task["task_id"], TaskPresentationEventCommand(
request_id=uuid4(), run_id=repository.run["run_id"], source_session_id="session-1",
command={"action": "set_overlay", "overlay_id": "isolation-coverage", "title": "节点覆盖示意",
"mode": "geometry", "evidence_id": evidence.evidence_id},
))
features = repository.events[-1]["metadata"]["command"]["data"]["features"]
assert {feature["properties"]["coverage_group"] for feature in features} == {"one", "two"}
assert all(feature["properties"]["display_only"] for feature in features)
assert [feature["properties"]["node_count"] for feature in features] == [1, 2]
@pytest.mark.anyio @pytest.mark.anyio
async def test_overlay_event_uses_stored_coordinates_and_rejects_untrusted_or_oversized_data(): async def test_overlay_event_uses_stored_coordinates_and_rejects_untrusted_or_oversized_data():
now = datetime(2026, 10, 10, tzinfo=timezone.utc) now = datetime(2026, 10, 10, tzinfo=timezone.utc)
+14
View File
@@ -1,4 +1,18 @@
from app.native.wndb.gis import network_views, region_geometry from app.native.wndb.gis import network_views, region_geometry
def test_selected_node_positions_use_bound_project_and_parameterized_geometry_query(monkeypatch):
calls = []
def read_all(network, statement, parameters):
calls.append((network, statement, parameters))
return [{"id": "A", "longitude": 117.24, "latitude": 32.75}]
monkeypatch.setattr(network_views, "read_all", read_all)
assert network_views.get_node_geographic_positions("project-a", []) == []
assert network_views.get_node_geographic_positions("project-a", ["A"])[0]["id"] == "A"
assert len(calls) == 1 and calls[0][0] == "project-a" and calls[0][2] == (["A"],)
assert "ST_Transform(geom,4326)" in calls[0][1] and "node_id = ANY(%s)" in calls[0][1]
from app.native.wndb.model import elements from app.native.wndb.model import elements