Compare commits

..
Author SHA1 Message Date
jiang d3a4433386 test(replay): cover exploratory simulations without impact maps
Generic Container CI/CD / test-build-publish (push) Successful in 4m10s
Server CI/CD v2 / build-test-publish-and-deploy (push) Successful in 4m10s
2026-11-15 12:44:06 +08:00
jiang 747e9e53e0 fix(investigation): allow autonomous model baseline recovery
Generic Container CI/CD / test-build-publish (push) Failing after 2m11s
Server CI/CD v2 / build-test-publish-and-deploy (push) Failing after 2m11s
2026-11-15 12:37:08 +08:00
23 changed files with 331 additions and 101 deletions
+9
View File
@@ -116,6 +116,15 @@ async def calculate_replay_burst_location(
)
async def calculate_replay_hydraulic_scenario(name: str, args: BaseModel, *, network: str, username: str) -> dict[str, Any]:
"""Evaluate model-based hydraulic alternatives without live SCADA substitution."""
return await bounded_calculation(
calculate_scenario, name,
args.model_dump(exclude={"context_handle", "timeout_seconds"}, exclude_none=True),
network=network, username=username, model_only=True,
)
async def calculate_replay_burst_simulation(args: BaseModel, *, network: str, username: str) -> dict[str, Any]:
"""An independent current-model scenario, with no live SCADA substitution."""
return await bounded_calculation(
+1 -1
View File
@@ -63,7 +63,7 @@ class QueryTool:
_specs = [
("run_contaminant_source_trace", "水污染候选溯源:提交由事件位置或监测映射取得的候选模型节点及事件窗,快速返回确定的已核验节点。当前为临时稳定选择,正式反演算法后续接入;不生成置信度。取得located_node后先展示节点、保存阶段结论并提现场核查/采样处置工单;工单与污染传播模拟、accident_element_type=node的隔离和影响分析并行,结果补入同一调查,不因算法来源停止", ContaminantSourceTraceInput, "simulation.run", "analysis", False, True),
("run_simulation", "按水力步长在在线业务模型上执行指定时间段的实时模拟;更新该时段的模型条件及实时基线,返回指定节点和管段结果;duration单位为秒,输出为在线模型结果,正式分析成果通过方案工具保存", ScenarioInput, "simulation.run", "analysis", False, True),
("run_simulation", "按水力步长在在线业务模型上执行指定时间段的实时模拟,更新全网实时基线,返回指定节点和管段结果;调查发现同刻基线缺失时可自主补算,等待成功并读取结果后重试依赖分析。duration单位为秒,补单帧可用60秒;回放按当前模型补算,不读取实时SCADA,不替代冻结观测;正式分析成果通过方案工具保存", ScenarioInput, "simulation.run", "analysis", False, True),
("run_contaminant_simulation", "在隔离模型中模拟假设污染源的浓度传播;include_impact_coverage=true时汇总全网节点整个模拟时段超过impact_threshold_mg_l的累计范围、首个超阈值报告时刻和峰值,并保存地图证据。阈值为情景展示条件,非饮用水安全限值;回放按当前模型独立估计,不读取实时SCADA。定位后先提现场核查/采样处置工单,工单审核派发与本计算并行,结果补入同一调查", ContaminantSimulationInput, "simulation.run", "analysis", False, True),
("analyze_valve_closure", "在隔离模型中模拟关阀的水力影响,返回指定节点/管段结果;现场阀门操作按业务审批执行", ValveClosureInput, "simulation.run", "analysis", False, True),
("analyze_flushing", "在独立模型中评估指定排放节点、流量及明确开关阀组合的冲洗水力结果;未列阀门沿用当前模型,不继承上次模拟;不判定污染清除或恢复供水", FlushingInput, "simulation.run", "analysis", False, True),
-3
View File
@@ -15,7 +15,6 @@ from app.mcp.models import BindInput, BoundContext
from app.mcp.scope import QueryScope
from app.mcp.config import McpSettings
from mcp.server.auth.provider import AccessToken
from app.mcp.work_order_skills import skill_name_for_task
def task_repository(scope: QueryScope) -> OperationalTaskRepository:
@@ -32,8 +31,6 @@ async def verify_task(scope: QueryScope, binding: BoundContext) -> None:
if not task:
raise HTTPException(404, "Task is not in this project")
binding.replay_context = task.get("source", {}).get("replay_context")
if skill_name_for_task(task.get("workflow"), binding.replay_context) == "burst-event-handling":
binding.tool_cap = [name for name in binding.tool_cap if name != "run_burst_location"]
if binding.purpose == "task_collaboration":
return
if OPERATIONAL_TASK_EXECUTE not in permissions:
+24 -11
View File
@@ -79,20 +79,20 @@ async def authorize_query(
"get_element_properties", "list_elements", "query_nearby_elements", "get_network_options",
"get_node_connections", "get_element_gis_properties", "query_network_extent",
"list_regions", "get_region", "get_region_nodes",
"analyze_valve_isolation", "generate_node_coverage", "run_burst_simulation", "run_contaminant_simulation", "analyze_flushing",
"analyze_valve_isolation", "generate_node_coverage", *SCENARIO_TOOLS,
"calculate_demand_allocation", "optimize_sensor_placement",
"query_model_baseline_series", "get_model_baseline_at_time", "query_model_baseline_field",
"get_calculation_job", "get_calculation_result", "list_calculation_jobs"}
if tool.name not in allowed:
asset_query = tool.semantics == "asset" and not tool.mutating and not tool.computation
if tool.name not in allowed and not asset_query:
raise HTTPException(403, "回放调查仅开放冻结 SCADA、资产、拓扑分析及指定模型情景;时序观测请使用 query_scada_readings")
if tool.name == "run_burst_simulation":
if not args.include_impact_coverage:
raise HTTPException(422, "回放爆管情景必须启用影响覆盖,按当前模型估计并保存同刻在线基线")
check_replay_window(replay_context, args.start_time, args.start_time)
if tool.name == "run_contaminant_simulation":
if not args.include_impact_coverage:
raise HTTPException(422, "回放污染情景必须启用影响覆盖,按当前模型估计传播范围")
check_replay_window(replay_context, args.start_time, args.start_time)
if tool.name == "analyze_flushing":
if tool.name in SCENARIO_TOOLS:
check_replay_window(replay_context, args.start_time, args.start_time)
if tool.semantics == "model_baseline":
if tool.name == "query_model_baseline_series":
check_replay_window(replay_context, args.start, args.end)
else:
check_replay_window(replay_context, args.time, args.time)
if tool.name == "query_scada_readings":
check_replay_window(replay_context, args.start, args.end)
if tool.name == "run_contaminant_source_trace":
@@ -213,6 +213,17 @@ async def execute_query(
args.device_ids, args.start, args.end, args.limit + 1, args.offset)
data, page, truncated = page_result(rows, args)
origin = "synthetic"
elif name in {"run_simulation", "analyze_valve_closure", "analyze_pressure_regulation"} and replay_context:
from app.mcp.analysis_service import calculate_replay_hydraulic_scenario
await conn.commit()
async with scope.simulation_routing():
result = await calculate_replay_hydraulic_scenario(
name, args, network=scope.project.project_code, username=scope.user.username,
)
data = {"parameters": args.model_dump(mode="json", exclude={"context_handle"}), "result": result}
origin = "synthetic"
gaps = ["回放水力计算使用当前模型,不读取实时SCADA,不代表冻结观测或历史模型重建。在线模拟更新模型基线,关阀及调压仅评估隔离情景;未来模拟样本仅为预测。"]
elif name == "run_burst_simulation" and replay_context:
from app.mcp.analysis_service import calculate_replay_burst_simulation
@@ -632,6 +643,8 @@ async def execute_query(
truncated = snapshot.truncated
else:
raise ValueError("query handler is not registered")
if replay_context and tool.semantics == "model_baseline":
gaps.append("在线模型参考按本次查询时可用数据读取,非冻结SCADA观测;历史报告复用已保存证据。")
window = (
{"start": args.start, "end": args.end}
if hasattr(args, "start") and args.start is not None
+13 -26
View File
@@ -29,50 +29,36 @@ def calculate_scenario(name: str, parameters: dict[str, Any], *, network: str, u
impact_threshold = parameters.pop("impact_threshold_mg_l", 0)
common = dict(name=network, modify_pattern_start_time=start.isoformat(),
modify_total_duration=duration, persist_results=False, scheme_name=None)
boundary = {}
if model_only:
from app.infra.db.postgresql.scada import ScadaElementMappings
# Model scenarios may support replay decisions without consuming mutable
# live SCADA in place of the task's frozen observations.
boundary["scada_mappings"] = ScadaElementMappings(
reservoirs={}, tanks={}, fixed_pumps={}, variable_pumps={},
pressure={}, demand={}, quality={},
)
if name == "run_simulation":
from app.services.simulation import run_online_simulation
result = run_online_simulation(
network, start, duration, output_node_ids=nodes, output_link_ids=links,
network, start, duration, output_node_ids=nodes, output_link_ids=links, **boundary,
)
elif name == "run_burst_simulation":
boundary = {}
if model_only:
from app.infra.db.postgresql.scada import ScadaElementMappings
# Replay scenarios use current model assumptions, never mutable live
# SCADA readings in place of the investigation's frozen observations.
boundary["scada_mappings"] = ScadaElementMappings(
reservoirs={}, tanks={}, fixed_pumps={}, variable_pumps={},
pressure={}, demand={}, quality={},
)
result = scenarios.burst_analysis(**common, username=username,
burst_ID=parameters["burst_pipe_ids"], burst_size=parameters["orifice_areas_cm2"], **boundary)
elif name == "run_contaminant_simulation":
boundary = {}
if model_only:
from app.infra.db.postgresql.scada import ScadaElementMappings
boundary["scada_mappings"] = ScadaElementMappings(
reservoirs={}, tanks={}, fixed_pumps={}, variable_pumps={},
pressure={}, demand={}, quality={},
)
result = scenarios.contaminant_simulation(**common, username=username,
source=parameters["source"], concentration=parameters["concentration"], source_pattern=parameters.get("pattern"), **boundary)
elif name == "analyze_valve_closure":
result = scenarios.valve_close_analysis(**common,
modify_valve_opening={valve: 0.0 for valve in parameters["valves"]})
modify_valve_opening={valve: 0.0 for valve in parameters["valves"]}, **boundary)
elif name == "analyze_flushing":
# The service consumes m³/h (the older REST label incorrectly says L/s).
# Reject unsupported conversion rather than silently ignore the flow.
from app.native.wndb.model.options import get_option
if get_option(network).get("UNITS") not in {"LPS", "CMH"}:
raise ValueError("Flushing currently supports LPS or CMH models only")
boundary = {}
if model_only:
from app.infra.db.postgresql.scada import ScadaElementMappings
boundary["scada_mappings"] = ScadaElementMappings(
reservoirs={}, tanks={}, fixed_pumps={}, variable_pumps={},
pressure={}, demand={}, quality={},
)
result = scenarios.flushing_analysis(**common, username=username,
drainage_node_ID=parameters["drainage_node_id"], flushing_flow=parameters["flushing_flow_m3h"],
valve_control=flushing_valve_controls(parameters), **boundary)
@@ -87,7 +73,8 @@ def calculate_scenario(name: str, parameters: dict[str, Any], *, network: str, u
result = scenarios.pressure_regulation(**common,
modify_fixed_pump_pattern={key: value for key, value in controls.items() if key in mappings.fixed_pumps},
modify_variable_pump_pattern={key: value for key, value in controls.items() if key in mappings.variable_pumps},
modify_tank_initial_level=parameters["tank_init_level"], scada_mappings=mappings)
modify_tank_initial_level=parameters["tank_init_level"],
scada_mappings=boundary.get("scada_mappings", mappings))
else:
raise ValueError("Scenario is not registered")
if not isinstance(result, dict):
+7
View File
@@ -125,6 +125,13 @@ def estimate_burst_impact(network: str, start: datetime, simulation: dict[str, A
impact["below_threshold_node_count"] = len(comparisons) - affected
if not comparisons:
impact.update(status="unavailable", reason="no_exact_comparable_online_snapshot")
impact["baseline_recovery"] = {
"tool": "run_simulation",
"parameters": {
"start_time": start.isoformat(), "duration": 60, "output_node_ids": ids[:1],
},
"next_step": "自主补算同刻在线基线;等待作业成功并读取结果后,按原参数重跑爆管影响计算,再发布降压覆盖。返回节点仅用于核验,在线基线按全网写入。",
}
return output
if affected > MAX_COVERAGE_NODES:
impact["coverage_status"] = "unavailable"
+7 -3
View File
@@ -157,6 +157,7 @@ def run_online_simulation(
*,
output_node_ids: list[str] | None = None,
output_link_ids: list[str] | None = None,
scada_mappings: ScadaElementMappings | None = None,
) -> dict:
"""Recalculate online baseline snapshots without cloning a model database."""
if start_time.tzinfo is None or duration_seconds <= 0:
@@ -166,7 +167,10 @@ def run_online_simulation(
)
if step <= 0:
raise ValueError("HYDRAULIC TIMESTEP must be greater than 0.")
mappings = query_corresponding_element_id_and_query_id(name)
mappings = (
scada_mappings if scada_mappings is not None
else query_corresponding_element_id_and_query_id(name)
)
selected = {
"node": {node: [] for node in output_node_ids or []},
"link": {link: [] for link in output_link_ids or []},
@@ -179,13 +183,13 @@ def run_online_simulation(
output = run_simulation(
name=name,
simulation_type="realtime",
modify_pattern_start_time=current.isoformat(timespec="seconds"),
modify_pattern_start_time=current.isoformat(),
scada_mappings=mappings,
)
if not isinstance(output, dict):
raise RuntimeError("Online simulation did not return calculation results")
units = output.get("units", {})
sample_times.append(current.isoformat(timespec="seconds"))
sample_times.append(current.isoformat())
for kind, elements in selected.items():
if not elements:
continue
+2
View File
@@ -231,6 +231,7 @@ def valve_close_analysis(
_temporary_project: str | None = None,
persist_results: bool = True,
username: str = "system",
scada_mappings: ScadaElementMappings | None = None,
) -> dict | str | None:
"""
关阀模拟
@@ -291,6 +292,7 @@ def valve_close_analysis(
result_db_name=name,
persist_results=persist_results,
scheme_username=username,
scada_mappings=scada_mappings,
)
return result if not persist_results else None
# step 3. restore the base model
+11 -6
View File
@@ -2,7 +2,9 @@
工单触发的 `workflow=burst_report` 使用 [爆管事件处理 Skill](../resources/work-order-skills/burst-event-handling/SKILL.md)。Server 按持久化来源选择流程,Agent 接收同一任务的核验上下文;SCADA 异常和历史回放使用 [SCADA Skill](../resources/work-order-skills/scada-anomaly/SKILL.md)。
## 固定处理路线
## 参考处理路线
Agent围绕事件核实、影响评估和处置复核自主组织分析。图示说明常见依赖,可按证据补查、补算、调整方法或并行独立步骤;阶段名称与展示顺序不限制工具选择。
```mermaid
flowchart TD
@@ -12,8 +14,9 @@ flowchart TD
interim --> approval[立即发起工单草稿、人工审核派发]
interim --> burst[爆管模拟与同刻在线压力差]
burst --> coverage[保存并展示爆管后、处置前影响覆盖]
coverage --> isolation[分析隔离边界与阀门方案]
isolation --> report[补充阶段报告与处置建议]
interim --> isolation[分析隔离边界与阀门方案]
coverage --> report[补充阶段报告与处置建议]
isolation --> report
approval --> feedback[接收实际处置与恢复反馈]
feedback --> verify[复核影响与结案条件]
verify --> close[Server 完成任务并解除告警]
@@ -24,7 +27,7 @@ flowchart TD
2. **资产关联**:优先核对上报资产编号。坐标或地址通过地理编码、附近资产、属性和连接关系形成有依据的关联;位置歧义列为具体待确认项。
3. **并行工单与影响估计**:定位到管道后保存 interim/concluded 阶段报告,随即提出现场核查/抢修 general 工单草稿;复用同一 action_id,人工审核、确认派发保持现有权限和版本校验。待审核或现场处理期间继续独立的影响模拟,不等待模型完成才发起工单。
4. **隔离方案**:用 analyze_valve_isolation 分析关联资产,保留必关阀、可选阀、直接隔离与供水失源节点;已知失效阀门用于替代方案分析,现场可操作性列为执行条件。
5. **影响说明**:先保留爆管后、处置前的压力下降覆盖步骤,再展示隔离方案的拓扑范围;结合可用SCADA、水力与用户关联说明影响。
5. **影响说明**:分别保留事件降压与隔离失源范围,按证据就绪顺序展示;结合SCADA、水力与用户关联说明影响。缺口先评估可行补救,仍不可得再说明限制。
6. **报告与处置**:定位阶段已发起现场工单,后续复用原动作;更新报告后按新增需要提出通知、保障或恢复等动作草稿;关联现场任务采用 general 工单。审批、派发和现场完成分别核对回执。
7. **反馈复核**:已派发的关联工单保持 `waiting_work_order`,全部收到有效 `completed` 反馈后进入 `waiting_scada`。以现场反馈业务时间为 T,取严格晚于 T 的最近一个 SCADA 采样时刻进行复核;正常范围使用事件前15天同采样相位的正常包络。数据未到或仍异常时保持等待,后台工作进程定期复核;通过后持久保存恢复证据并接续 Agent。Agent 最终报告必须引用该证据,运行完成后才能完成任务并解除对应告警。现场资料缺项仅限定未核实细节,不新增本任务的关闭门槛。来源系统另有明确行政结案流程时,按其实际状态另记。
@@ -48,7 +51,7 @@ SCADA作为事件影响的辅助证据,按目标测点和事件/对照窗口
## 工具与入口约束
工单事件从上报位置开展资产关联。Agent会话策略隐藏 run_burst_location,Server在绑定与后续解析时同步移除其调用能力;普通聊天和SCADA候选定位沿用各自策略。持久计算在执行前重新校验任务绑定。
工单事件通常从上报位置关联资产;监测及模型证据需要时可自主调用run_burst_location补充定位,不按workflow隐藏分析工具。输入、角色、项目权限及任务绑定仍在执行前校验。
历史 burst_report 任务携带 replay_context 时,Server分发scada-anomaly,时序查询保持 knowledge_cutoff 与 batch_sequence,demo能力提示按冻结窗口返回。
@@ -80,7 +83,9 @@ SCADA作为事件影响的辅助证据,按目标测点和事件/对照窗口
爆管模型使用 PDA 需水方式和原有10模型压力单位的服务压力设定;在线模型方式、当前模型时态及在线压力单位推定来源一并保留。在线记录没有独立版本化单位元数据,不能把此估计称为严格历史同模型反事实或现场停水边界。
携带 `replay_context` 的调查只开放启用影响覆盖的爆管情景,起算时刻不得超过 `knowledge_cutoff`。使用当前模型的独立假设情景,初始化及扩展模拟均传入空SCADA边界映射,禁止从实时SCADA回填冻结观测;同刻在线结果仅作为计算时读取的模型参考,标明非冻结基线。未来样本属于预测,不能作为已发生的观测。
回放调查可自主选择在线补算、爆管、关阀、调压、污染传播及冲洗模型计算,起算时刻不超过knowledge_cutoff,使用当前模型和空SCADA边界映射。单独探索情景不强制生成覆盖;目标包含影响范围时应启用覆盖并核对结果。同刻在线结果为可变模型参考,未来样本属于预测,均不冒充冻结观测。
基线缺失时,impact.baseline_recovery提供run_simulation补算建议。Agent可按目标用同刻短窗补算,读取成功结果并核对模型基线,再重试依赖分析;新的工具调用可创建新作业,无需改变破口参数来绕过旧结果。可选择其他有依据的方法,并说明剩余限制。
完成持久计算后,Agent按回执 `data.presentation.command` 用证据编号发布覆盖,并保存三档计数、假设与缺口表。Server从任务所属的不可变证据解析几何,将完整展示命令写入事件。历史播放还原这份命令,前进到隔离范围和后退到事件影响均不重新查询在线模型,也不派单。工单审核、派发与反馈继续由通知和标准工单窗口记录;模型结果补入同一调查,不重复创建工单。
+2 -2
View File
@@ -33,7 +33,7 @@ Agent 原生工具统一使用 `mcp_tjwater_` 前缀。Server目录共67项:42
`persisted_baseline=true` 表示在线基线已重算,`persisted_scheme=false` 表示没有创建方案。
该工具的 MCP annotation 为 `readOnlyHint=false`、`destructiveHint=true`,因为重算会替换对应时刻的基线。
它不会创建 `analysis.run_id`,不能将本次计算冒充已有爆管事件运行;调查调用仍可保存证据。
计算读取现有模型和对应SCADA条件,不从摘要虚构观测,也不自动补齐数据或执行现场操作。
普通计算读取现有模型及对应SCADA条件;回放计算使用当前模型和空SCADA映射,不读取实时观测。Agent可自主补算缺失模型基线,不从摘要虚构观测或执行现场操作。
爆管定位 `run_burst_location` 默认使用 `data_source="monitoring"`。`pressure_scada_ids` 可省略,由后端从当前项目SCADA配置自动获取压力测点对应的模型节点;显式指定时仍限制为1至20个不同节点。正常对照时间窗 `scada_normal_start` / `scada_normal_end` 可成对省略,沿用后端默认的前一天同时间段;返回结果中的 `pressure_scada_ids` 和 `scada_window` 说明实际测点和时间窗。默认对照不等于已确认该时段正常。正常窗无有效压力数据或两窗没有共同有效测点时计算失败,不补造数据、不自动换窗。正常窗只有部分测点缺数时,沿用服务逻辑使用两窗共同有效测点,结合返回测点范围解释结果。
@@ -81,7 +81,7 @@ HTTP 预算由宿主从目录元数据读取,Server 默认 120 秒、Agent 默
`analyze_valve_isolation` 和 `calculate_demand_allocation` 保持同步。队列消费者独立于Uvicorn的4个API进程,默认全部配置项目合计并发2项,并通过元数据库单实例锁防止重复启动。取消、超时和租约失效需要停止计算子进程组;成功结果按原运行授权保存证据,未完成计算阻止调查提前结案。
在线重算逐个时刻写入基线,超时或中途失败可能已有部分时刻更新,不能把工具失败解释为整个窗口已回滚;不要自动重试整个窗口。
在线重算逐个时刻写入基线,超时或中途失败可能已有部分时刻更新。先核对实际作业、采样与基线,再自主补算必要范围,不能假定整个窗口已回滚而盲目重提。
队列接入需要在目标项目BizDB依次应用 `resources/sql/026_calculation_jobs.sql`、`resources/sql/027_calculation_job_lifecycle.sql` 并启动独立计算Worker。Server与Agent需配套发布,client模式还需重载原生连接。既有OAuth与项目权限保持不变;完整部署与回滚步骤见后台计算队列文档。
+4 -4
View File
@@ -24,7 +24,7 @@ flowchart TD
decision -->|是| final[分区恢复建议与最终报告]
```
图中连线表示业务与展示层次,先期保护、工单审核和独立计算可以并行。阀门操作均为方案,现场动作继续经过原审批与派发;阶段记录不能替代执行回执。首次展示保留处置前传播结果或明确的数据缺口,再展示隔离及清除方案。
图示为参考路径,Agent可按风险、证据和实际依赖调整方法、顺序及并行分析。各类结果按就绪顺序展示并分别标明性质,现场动作仍须审批与派发,阶段记录不能替代执行回执。
## 侦测入口
@@ -46,13 +46,13 @@ flowchart TD
阈值仅为模型展示条件,默认0表示模型正浓度,不是水质安全限值。沿用当前模型初始水质、已有源和反应条件;统计的是总模型浓度,不能称为本次源的独立贡献。源浓度、模式、业务起算时刻、预测时长及其假设必须有依据,不能把余氯下降或浊度换算成污染源浓度。
回放同样允许该工具,要求启用影响覆盖且起算不晚于知识截止;预测时段可以越过截止,但后续时刻只是模型预测。与爆管情景一样,使用当前模型和空 SCADA 映射传入临时模型计算,不读取可变实时监测,不重建历史模型。仍禁止借此放开未授权观测或其他水力工具。
回放允许自主选择传播及其他已开放模型计算,起算不晚于知识截止;预测时段可以越过截止,未来样本仅为模型预测。使用当前模型及空SCADA映射,不读取实时监测或重建历史模型。探索情景可不生成覆盖,影响范围目标应启用覆盖并核对结果。
工具沿用计算队列和 owner/run 校验,返回终态后通过 `get_calculation_result` 读取并持久保存证据。`presentation_preview` 提供基于 `evidence_id` 的地图请求;用正式调查展示事件保存模型曲线和覆盖,历史播放还原已保存证据。审批、派发可在计算排队或运行期间进行;存在未终态计算时继续保留原运行,不允许发布 final 或成功结束调查,已有工单不受模拟失败影响。
工具沿用计算队列和owner/run校验,终态后通过get_calculation_result读取并保存证据。data.presentation.command提供基于evidence_id的地图请求;正式事件保存曲线与覆盖,历史播放复用。审批、派发可与计算并行;未终态计算保留原运行,已有工单不受模拟失败影响。
## 展示、反馈与收尾
`water-pollution-event` 1.2.0 将观测核验、候选定位、处置前传播、源头控制、用户隔离、清除冲洗、现场反馈、水质复测和恢复复核分开。每次正式展示保存 `purpose`;有证据的阶段通过 `narration.stage` 记录,新增 `source_control`、`flushing_plan`、`clearance_verification`。历史保存格式继续使用原事件 metadata JSON,不增加数据库列。前端显示阶段和本步目的,逐步回看保留独立画面;同阶段兼容图表仍可合并。
`water-pollution-event` 1.2.1 将观测核验、候选定位、处置前传播、源头控制、用户隔离、清除冲洗、现场反馈、水质复测和恢复复核分开,作为可调整的参考路径。每次正式展示保存 `purpose`;有证据的阶段通过 `narration.stage` 记录,沿用 `source_control`、`flushing_plan`、`clearance_verification`。历史保存格式继续使用原事件 metadata JSON,不增加数据库列。前端显示阶段和本步目的,逐步回看保留独立画面;同阶段兼容图表仍可合并。
源头控制与保护下游用户分别决策。用已核实的连接和流向判断水力下游,不能按地理方位判断;源头未核实不得声称污染输入已停止。清除方法按污染物及现场条件选择,不统一推荐煮沸、加药或排水。通知保留,工单按执行职责和阶段复用或必要拆分,不能同时要求同一阀门开和关。
@@ -2,7 +2,7 @@
name: appointment-follow-up
description: 核查关联预约工单的审批、提交、现场执行和反馈,分别判断预约履约与原事件闭环。
metadata:
version: "2.0.4"
version: "2.0.5"
---
# 预约任务跟进
@@ -13,7 +13,7 @@ metadata:
读取当前任务、关联工单、已有报告、证据、动作和最新反馈,核对预约时间、地点、负责人、工作内容及回执要求;需要时追溯关联的投诉或爆管。新一轮优先检查新增事实和旧未决项,保留本流程名称和版本。
围绕原工作目标完成必要核查:
以下为参考核查路径,围绕原工作目标自主选择方法、调整顺序并补齐可查询或推导的资料。反馈缺项先核对已有来源和证据,不因首次缺数直接停止;新增现场事实仍以实际回执为准:
1. 将要求与来源工单最新版本逐项对照,区分预约、派发、到场、完成和原事件发生时间,不用创建或更新时间替代现场时间,也不因预约已过推定已经执行或失约。
2. 核对动作的审批、提交、执行状态和回执。`approved` 只表示允许提交,`submitted` 只表示已接收,`succeeded` 的建单动作只证明工单创建成功;这些状态均不证明采样、通知、抢修、冲洗或复测完成。`unknown` 由 Server 复核原执行请求,不重复下发。
@@ -2,12 +2,12 @@
name: burst-event-handling
description: 根据爆管上报工单处理事件,完成资产关联、隔离方案、影响说明、处置建议与反馈复核。
metadata:
version: "1.1.5"
version: "1.1.6"
---
# 爆管事件处理
适用于工单触发的 `workflow=burst_report`。以工单上报的事件位置和资产线索开展处置,依次完成下列步骤;每一步给出实际结果或具体待补项,已有证据可直接复用。SCADA 异常发现使用独立的 scada-anomaly 流程。
适用于工单触发的 `workflow=burst_report`。目标是核实事件、评估影响并提出处置与复核依据。以下为参考流程,按证据和依赖自主选择、调整或并行步骤,复用已有成果,补查或补算可取得的前置资料。SCADA 异常发现使用独立的 scada-anomaly 流程。
先按handling_context接续:等待阶段按第7节保存本轮阶段报告;verification_ready复用已有异常分析与资产、隔离证据,仅补反馈和恢复复核,不重跑初次调查步骤。真正新增的独立事实另行核查。
@@ -23,6 +23,8 @@ metadata:
爆管候选管段用 locate_features 传 event_kind="burst" 定位。在地图展示上报位置和关联资产,标明关联依据及现场确认状态。存在多个合理对象时保留候选及选择理由;关键位置需要人工补充时,先完成独立工作,再用 ask_task_user 提问。
上报位置与资产不足以区分候选时,可按现有监测和模型证据自主选择 run_burst_location 等补充分析;准备工具要求的输入,保留假设,不因工单来源限制分析方法。
定位到管道后立即保存 interim/concluded 阶段报告,明确“候选管段待现场核验”,调用 propose_task_action 提出 kind=create_work_order、work_order_type=general 的现场核查/抢修草稿,引用定位证据、地点和管段;沿用已有 action_id,不重复建单。人工审核、派发按 Server 权限与版本完成。审批或现场处理等待期间,继续独立的爆管影响模拟和隔离分析,不等模拟完成才发起工单;模拟结果补入同一调查与后续报告,不能把模型计算成功写成已派单。
## 3. 爆管后、处置前的影响范围
@@ -31,7 +33,7 @@ metadata:
工具在筛选展示节点前比较全网节点的起算同刻在线模拟压力,换算米水头,三档互斥:1≤下降<3、3≤下降<5、下降≥5。只用同刻有效值,不以代表节点、邻近时刻或缺失值代替;保留缺测、单位来源、模型时态、比较节点数及覆盖缺口。范围表示爆管后、关阀或抢修前的模型估计,单独于隔离失源范围。
读取计算结果后按 data.presentation.command 的 evidence_id 用 publish_task_presentation 发布“爆管后、处置前的预计影响范围”,并发布三档节点计数、比较时刻和假设表;无地图几何时说明原因。证据、地图及表格进入正式任务事件,历史播放还原已保存结果,不重新查询在线基线。本步 narration.stage 用 event_impact。三级结果 unavailable 时单列原因和待补项;先保留此步骤,再展示隔离方案,不得用关阀覆盖替代。
读取结果后按 data.presentation.command 发布预计降压范围和分档计数、比较时刻及假设,narration.stage 用 event_impact。基线缺失时可自主用 run_simulation 补算同刻数据;等待成功并读取结果后重试依赖计算,不靠扩大破口面积修复基线缺失。其他失败按原因选择补查、修正参数或替代方法,补救后仍不可用才列具体限制。事件降压与关阀范围分别保存,展示按结果就绪顺序发布;历史播放复用保存证据。
## 4. 隔离方案
@@ -2,14 +2,14 @@
name: burst-report
description: 爆管业务流程的历史入口索引,运行时按事件来源选择独立 Skill。
metadata:
version: "2.0.11"
version: "2.0.12"
---
# 爆管业务流程入口
Server 根据绑定任务的 workflow 和 replay_context 分发当前流程:
- 工单触发的 burst_report:使用 burst-event-handling,依次完成接收核对、地点与资产关联、隔离方案、影响说明、处置建议和反馈复核。
- 工单触发的 burst_report:使用 burst-event-handling,围绕事件核实、资产关联、影响评估、处置建议和反馈复核自主组织分析。
- SCADA 异常及带 replay_context 的历史回放任务:使用 scada-anomaly,完成冻结数据核验、诊断、候选定位与隔离影响分析。
通过 get_task_context 的 section="skill" 读取实际流程,保留返回的 name、version、content_sha256。工具范围由绑定任务的服务端策略及 Agent 会话策略共同确定。
@@ -2,12 +2,12 @@
name: scada-anomaly
description: 从 SCADA 指标异常开展数据核验、爆管诊断、候选定位与影响分析。
metadata:
version: "1.0.20"
version: "1.0.21"
---
# SCADA 数据异常调查
读取任务与 replay_context,核对设备、单位、异常时刻、knowledge_cutoff 和 batch_sequence。使用本轮已提交的 SCADA 观测开展调查,同一持续事件沿用已有调查。证据来源字段用于记录数据取得方式,不构成停止分析的条件。
目标是核验异常、判断原因和影响并形成处置依据。下文为参考流程,可按证据和依赖自主调整方法与顺序,补查、补算前置资料。核对任务、设备、单位、异常时刻、knowledge_cutoff及batch_sequence,使用本轮已提交观测,同一持续事件沿用原调查。证据来源字段用于记录数据取得方式,不构成停止分析的条件。
先按handling_context接续:待审批/工单/监测且无独立工作时,保存本轮interim/concluded及action_id后退出,不轮询。verification_ready复用原异常证据,只补恢复复核;初次分析按下文。
@@ -31,9 +31,9 @@ metadata:
取得候选即展示管道,发 interim/concluded 报告及现场工单草稿;复用action_id,待审批继续模拟。
run_burst_simulation设include_impact_coverage=true,用定位管道、异常时刻及有依据的破口面积cm²。全网同刻在线压力差按米水头分档1≤下降<3、3≤下降<5、下降≥5。按data.presentation发布地图及计数,narration.stage=event_impact;不可用时单列原因,禁用关阀范围替代。回放仅用当前模型、禁实时SCADA;在线参考非冻结观测,历史展示复用保存证据。
run_burst_simulation设include_impact_coverage=true,用候选管道、异常时刻及有依据的破口面积cm²,按米水头分档1≤下降<3、3≤下降<5、下降≥5。基线缺失可自主用run_simulation补算同刻数据,读取成功结果后重试。其他缺口先按原因补查、补算或修正参数,仍不可用才记录限制。按data.presentation发布地图及计数,stage=event_impact;禁用关阀范围替代。回放模型计算不读实时SCADA,非冻结观测;历史展示复用保存证据。
再按授权用 analyze_valve_isolation 分析隔离边界,保留必关阀、可选阀、affected_nodes、supply_affected_nodes 及计数;用 generate_node_coverage 展示关阀覆盖,stage=isolation_plan。候选与拓扑建议标注为待现场核验方案。爆管候选用locate_features传event_kind=burst。报告前核对定位结果,失败列明错误与缺口。
按需用analyze_valve_isolation分析隔离边界,保留必关阀、可选阀、affected_nodes、supply_affected_nodes及计数;generate_node_coverage展示关阀覆盖,stage=isolation_plan;analyze_valve_closure可补水力影响。候选与建议待现场核验,展示按就绪顺序。候选用locate_features传event_kind=burst。
## 对比展示与影响
@@ -43,22 +43,22 @@ run_burst_simulation设include_impact_coverage=true,用定位管道、异常
定位输入具备后继续定位。压力、流量按设备另发七天同期复合图,不占用或替代全量当前总览。chart_type=time_comparison:x_data是UTC+8的HH:mm或HH:mm:ss共享槽,不拼接历史日期;当前[t-3小时,t]用series.type=line,前7天每天t-30分钟、t-15分钟、t用scatter。15分钟数据13槽,历史七组仅末3槽有值。每组sample_times与data等长,保留真实带时区日期;有效值与对应槽的本地时刻一致,缺测值和时间均null。当前曲线共用窗口,历史每组取此前某天对应窗口,跨午夜顺序排列;普通图用line/bar,坐标轴标单位。七天按t的本地日期回退,不含当天。按实际周期保留缺口,按需翻页并报告参考点数;基线未冻结、读取失败或缺测时说明并保留当前趋势,不退化为前一天或相邻两点。
影响说明列出异常前参考、异常期间和已有恢复观测中的压力、流量或水质变化、持续时间、涉及测点及已核实资产。可比对象、单位、时段与运行条件下给出变化量,分别标明实际观测、模型预测和拓扑隔离。停水范围与漏量引用相应拓扑、水力、流量平衡或现场依据;恢复观测、用户关联或现场证据缺失时列为具体待核实项。
影响说明比较异常前、异常期间及已有恢复观测,列压力、流量或水质变化、持续时间、测点与资产。同对象、单位、时段及条件计算变化量,区分观测、模型和拓扑结果;停水及漏量保留依据,缺恢复、用户或现场资料时明确限制。
## 范围与报告
本轮时序分析保持任务冻结窗口,后续播放形成的观测由相应轮次处理。资产属性、空间查询和拓扑隔离使用项目当前模型,注明模型时态;水力工具按已开放的冻结输入能力调用。定位调用执行权限、设备与冻结观测校验;基线或覆盖缺口按实际返回记录。
观测保持本轮冻结窗口,后续观测由相应轮次处理。资产、空间与拓扑用当前模型;水力计算标明模型来源,定位保留权限、设备及冻结观测校验。
报告先写判断与关键依据,摘要建议 150 字、最多 300 字;多测点数值入 Table,保留单位、方向与缺测。区分各阶段测点数,不相加不同影响集合。现场事项按回执列待办。现场工单与设备操作遵守服务端动作权限,恢复复核通过后闭环。
报告先写判断与依据,摘要建议150字、最多300字;多测点数值入Table,保留单位、方向和缺测,不相加不同影响集合。现场事项按回执列待办。现场工单与设备操作遵守服务端动作权限,恢复复核通过后闭环。
现场工单用propose_task_action(kind=create_work_order、work_order_type=general),注明对象、条件、回执及回放来源;有权限者确认派发,通知保留,报告完成不等于现场完成;回放禁设备命令。
## 完工后的恢复复核
关联工单completed后,以handling_context.returned_at为实际完工业务时间,不用创建、预约或电脑当前时间代替。Server读取严格晚于该时刻的最近一帧及事件前15天同相位对照,独立保存恢复证据。回放来源工单同样适用,不推进回放,原knowledge_cutoff与batch_sequence不变。
工单completed后,以handling_context.returned_at为实际完工时间;Server取严格晚于该时刻的最近一帧及事件前15天同相位对照,保存恢复证据。回放同样适用,不推进回放或修改原冻结边界。
waiting_work_order等待完工;waiting_scada按last_check说明尚无后续采样、数据缺项或仍异常,等待监测复核。verification_ready时用get_task_evidence读取verification_evidence_id完整快照;工单completed且恢复证据status=passed时,发布final/concluded恢复闭环报告,同时引用原异常全量证据与当前恢复证据。Server完成运行后关闭任务并解除关联告警,phase=completed表示已闭环。阀位、漏点照片、现场复测数值等未提供时仅记为资料缺项,不新增现场核验、人工确认或更多采样的闭环条件。
恢复报告先写现场反馈与完工时间、采样时刻和结果,表格列全部复核测点及基线;边界说明单帧与模型限制,日常监测列建议。不重列旧定位参数或执行字段;更新原图中已解除的等待标记,保留路线与失败记录。
恢复报告列反馈、完工与采样时间、全部复核测点及基线,说明单帧与模型边界,日常监测列建议。更新已解除的等待标记,保留路线与失败记录。
恢复观测由Server独立保存,原冻结上限不阻断此证据的使用;query_scada_readings仍遵守冻结窗口。完工时间修订后使用新证据编号,旧证据仅作历史。确有独立工作时再提议新动作。
@@ -2,13 +2,15 @@
name: water-pollution-event
description: 水污染侦测后的观测核验、候选溯源、源头控制、用户隔离、清除冲洗与水质复测。
metadata:
version: "1.2.0"
version: "1.2.1"
---
# 水污染侦测事件调查
适用于 workflow=water_pollution,与 water_quality_complaint 投诉分开,不依赖爆管评分、漏量或压力流量异常。侦测信号、候选污染源、污染确诊和恢复供水分别说明。
目标是核验风险、来源、影响及处置恢复依据。以下为参考流程,可自主调整方法、顺序,补查参数并比较明确假设;缺口先补救,保护措施及时提出。
读取任务、信号、时间、handling_context 和已有成果。回放遵守 knowledge_cutoff、batch_sequence;阅读时间 Asia/Shanghai,查询与 sample_times 带时区。后续轮次复用成果,优先新增事实。
## 1. 核验观测与先期保护
@@ -25,15 +27,15 @@ query_scada_readings 省略 device_ids,读取异常前三小时全量 SCADA,
## 3. 处置前污染影响
run_contaminant_simulation 设 source=located_node、include_impact_coverage=true,采用事件起算时刻、明确预测时长和需画曲线的 output_node_ids。concentration 为 mg/L,须有来源或说明情景假设,不能把浊度、余氯下降或感官描述直接换算成污染源浓度。缺依据时列具体参数缺口,继续工单和独立分析。
run_contaminant_simulation 设 source=located_node、include_impact_coverage=true,采用事件起算时刻、明确预测时长和需画曲线的 output_node_ids。concentration 为 mg/L,须有来源或说明情景假设,不能把浊度、余氯下降或感官描述直接换算成污染源浓度。可自主补查来源、比较明确假设的浓度或时长;不可推断的现场事实才列具体缺口,继续工单和独立分析。
impact_threshold_mg_l 默认0、严格大于,仅供展示,非饮用水安全限值。全网整个时段汇总累计节点、首个超阈值采样时刻、峰值与覆盖;沿用初始水质和反应条件,非源独立贡献,范围外不等于安全。
queued/running 时继续独立工作,get_calculation_job 跟踪、get_calculation_result 读终态。未完计算不能 final,由宿主等待接续;待审批不阻止分析,模拟失败不撤回工单。
按 presentation_preview.evidence_id 发布“水污染传播预计影响范围(处置前情景)”、参数与计数表及浓度曲线;模型曲线不填观测专用 chart.scada。保存正式事件供历史复用;空范围、缺测、单位及几何限制单列。先保留污染传播步骤,再展示隔离范围,不以隔离覆盖替代传播范围。
按返回的data.presentation.command发布“水污染传播预计影响范围(处置前情景)”、参数与计数表及浓度曲线;模型曲线不填观测专用chart.scada。空范围、缺测、单位及几何问题先核对可行补救,再列剩余限制;传播与隔离分别保存,不以隔离覆盖替代传播范围。
回放允许传播及冲洗情景,起算不晚于知识截止,使用当前模型且不读取实时 SCADA;未来样本仅预测,非历史重建。
回放允许在线基线补算及传播、关阀、调压、冲洗情景,起算不晚于知识截止,使用当前模型且不读取实时SCADA;未来样本仅预测,非历史重建。
## 4. 源头控制与用户隔离
@@ -47,7 +49,7 @@ analyze_valve_isolation 设 accident_element=[located_node]、accident_element_t
先说明污染物性质、处理依据和未知项,再提出排空、冲洗、消毒或其他适用处理;不得默认任何污染都能冲洗清除。排放去向及污染水收集处理须有依据,不默认排入雨水、河道或普通排水口。
核对清洁水源、排放点、流量及许可条件后用 analyze_flushing。每阶段保存目的、前置条件、清洁进水依据、排放去向、完整开关阀清单、顺序、流量、时长、预计水量、监测和待办。隔离和冲洗不能共用固定关阀清单,进水阀打开不等于用户复供;缺参数仅停依赖分支。
核对清洁水源、排放点、流量及许可条件后用analyze_flushing。阶段记录目的、条件、进排水依据、完整阀态、顺序、流量、时长、水量、监测及待办。隔离和冲洗按各自目标选阀,进水阀打开不等于复供;缺参数先补查。
opened_valves 明确开阀、closed_valves 指定关闭阀门,两者互斥。每次是独立当前模型情景,未列阀门沿用模型状态,不继承上次模拟;重列本阶段必要阀态,不能称为连续污染清除。flushing_plan 保存方案及估计排放量,仅证明水力结果,不验证清洁进水、排放合规或现场执行。尚未耦合停止输入后的水质衰减,压力、流量或模型零浓度不能替代复测。
@@ -57,9 +59,9 @@ opened_valves 明确开阀、closed_valves 指定关闭阀门,两者互斥。
水质 chart.scada 填 metric=quality、unit、role、完整三小时 window_start/window_end,总览覆盖全部目标。逐设备 time_comparison 用当前 line、七天每天末三真实槽 scatter;series 带 device_id、day_offset、等长 sample_times,缺测 null。不同指标或单位分图,超限分段,不标成流量。
展示依次为观测、候选、传播、源头控制、隔离、冲洗、反馈、复测、恢复;计算可并行,首次隔离展示须在传播结果或缺口说明之后,通知和工单不等待展示。发布填写具体 purpose;有证据时带 narration(stage、summary、evidence_refs)。沿用原阶段,源头控制用 source_control、冲洗用 flushing_plan、水质复测用 clearance_verification。阶段不等于批准或已执行,不为凑阶段造证据或空操作。
展示随证据就绪发布,按观测、候选、传播、源头控制、隔离、冲洗、反馈、复测和恢复标明性质与依赖,无需等待其他独立分析或固定播放顺序。通知和工单不等待展示。发布填写purpose及有依据的narration(stage、summary、evidence_refs);源头控制用source_control、冲洗用flushing_plan、水质复测用clearance_verification。阶段不等于批准或已执行,不为凑阶段造证据或空操作。
报告依次写事件概况、观测、候选与传播、源头控制、用户隔离、清除冲洗、反馈复测及恢复条件;各项列结论、对象、预计或实际、证据和待办。操作表保留阶段、阀态、进排水、条件、反馈及复测依据。引用本轮全量 SCADA 和已取得的分析证据;权限拒绝不写成无数据。各阶段独立保存,回放不合并吞步骤,不派单或操作设备。
报告按风险与结论组织,覆盖观测、候选、传播、源头控制、隔离、清除及复测恢复,区分预计与实际。操作表保留阀态、进排水、条件和反馈依据。引用本轮全量SCADA与分析证据;权限拒绝不写成无数据。阶段独立保存,历史播放不派单或操作设备。
## 7. 工单反馈与复测恢复
@@ -2,7 +2,7 @@
name: water-quality-complaint
description: 调查供水水质投诉,核查投诉与监测覆盖,评估候选原因、影响和采样处置建议。
metadata:
version: "2.0.4"
version: "2.0.5"
---
# 水质投诉处置
@@ -15,7 +15,7 @@ metadata:
## 风险与工程分析
根据当前证据选择必要步骤:
以下为参考核查路径,根据当前证据选择必要步骤,自主调整顺序、查询范围、工具组合及模型情景。可通过补查、推导或有说明的假设解决的缺口先处理,确需现场事实或人工决策时再提问:
1. 核实投诉位置及相近时段的同类投诉,用 `query_peripheral_data` 按实际时间、位置和来源标识核对外围工单或观察。目录首屏不能证明没有其他投诉;未知点位先查目录,不猜编号或把演示点当真实资产。
2. 解析未知地址,查询附近资产、属性和连接,区分上游、管网、末梢、二次供水和用户端的实际覆盖。地点明确后展示投诉位置,取得关联资产后展示正在核查的管段或节点。拓扑连接不等于实际流向,方向判断需要同窗模型或监测依据。
+21
View File
@@ -99,6 +99,27 @@ def test_unknown_units_do_not_draw_an_impact_area(inputs):
inputs.assert_not_called()
def test_missing_online_baseline_offers_recalculation_then_coverage_can_be_retried(inputs):
inputs.side_effect = None
inputs.return_value = {}
native = simulation([2, 4, 6])
missing = service.estimate_burst_impact("own", START, native)
assert missing["impact"]["reason"] == "no_exact_comparable_online_snapshot"
recovery = missing["impact"]["baseline_recovery"]
assert recovery["tool"] == "run_simulation"
assert recovery["parameters"] == {
"start_time": START.isoformat(), "duration": 60, "output_node_ids": ["J0"],
}
assert "map_payload" not in missing
inputs.return_value = {
f"J{i}": [{"time": START.isoformat(), "value": 30}] for i in range(3)
}
retried = service.estimate_burst_impact("own", START, native)
assert retried["impact"]["status"] == "ready"
assert len(retried["map_payload"]["features"]) == 3
def test_full_native_network_compared_before_selected_series_filter(monkeypatch, inputs):
from app.mcp import scenario_service
from app.services import simulation_scenarios
@@ -42,27 +42,29 @@ def context(monkeypatch):
return scope, binding
def test_replay_prediction_can_extend_past_cutoff_without_reading_future_observations(monkeypatch):
@pytest.mark.parametrize("include_impact", [True, False])
def test_replay_prediction_can_extend_past_cutoff_without_reading_future_observations(monkeypatch, include_impact):
scope, binding = context(monkeypatch)
worker = AsyncMock(return_value={"impact": {"affected_node_count": 2}})
monkeypatch.setattr(analysis_service, "calculate_replay_contaminant_simulation", worker)
result = asyncio.run(query_service.execute_query(scope, binding, TOOLS["run_contaminant_simulation"], args()))
parameters = args(include_impact_coverage=include_impact)
result = asyncio.run(query_service.execute_query(scope, binding, TOOLS["run_contaminant_simulation"], parameters))
assert result.origin == "synthetic" and result.data["result"]["impact"]["affected_node_count"] == 2
assert "预测" in result.data_gaps[0]
scope.connection.commit.assert_awaited_once()
worker.assert_awaited_once()
assert worker.await_args.args[0].include_impact_coverage is include_impact
assert worker.await_args.kwargs == {"network": "project", "username": "actor"}
assert should_queue(TOOLS["run_contaminant_simulation"], args())
@pytest.mark.parametrize("problem", ["future_start", "no_impact", "reader", "permission"])
@pytest.mark.parametrize("problem", ["future_start", "reader", "permission"])
def test_replay_admission_keeps_time_permission_and_reader_boundaries(monkeypatch, problem):
scope, binding = context(monkeypatch)
parameters = args()
if problem == "future_start": parameters = args(start_time="2026-07-16T08:15:00+08:00")
if problem == "no_impact": parameters = args(include_impact_coverage=False)
if problem == "reader": binding.purpose = "task_reader"
if problem == "permission": binding.permission_cap.remove("scada.view")
with pytest.raises(HTTPException) as caught:
asyncio.run(query_service.authorize_query(scope, binding, TOOLS["run_contaminant_simulation"], parameters))
assert caught.value.status_code == (422 if problem in {"future_start", "no_impact"} else 403)
assert caught.value.status_code == (422 if problem == "future_start" else 403)
+9 -12
View File
@@ -373,7 +373,7 @@ def test_replay_binding_denies_future_and_allows_device_metadata(monkeypatch):
asyncio.run(query_service.authorize_query(SimpleNamespace(project=object()), bound, TOOLS["list_scada_devices"], SimpleNamespace()))
def test_replay_burst_scenario_requires_coverage_permission_and_known_start(monkeypatch):
def test_replay_burst_scenario_allows_exploration_and_checks_coverage_permission(monkeypatch):
bound = binding(task=True)
bound.replay_context = {"knowledge_cutoff": "2026-10-08T01:00:00Z"}
permissions = PERMISSIONS | {"simulation.run", "scada.view", "simulation.view"}
@@ -386,11 +386,12 @@ def test_replay_burst_scenario_requires_coverage_permission_and_known_start(monk
"include_impact_coverage": True,
})
asyncio.run(query_service.authorize_query(scope, bound, TOOLS["run_burst_simulation"], args))
for changed, status in (({"include_impact_coverage": False}, 422),
({"start_time": datetime(2026, 10, 8, 2, tzinfo=timezone.utc)}, 422)):
with pytest.raises(HTTPException) as error:
asyncio.run(query_service.authorize_query(scope, bound, TOOLS["run_burst_simulation"], args.model_copy(update=changed)))
assert error.value.status_code == status
asyncio.run(query_service.authorize_query(scope, bound, TOOLS["run_burst_simulation"],
args.model_copy(update={"include_impact_coverage": False})))
with pytest.raises(HTTPException) as error:
asyncio.run(query_service.authorize_query(scope, bound, TOOLS["run_burst_simulation"],
args.model_copy(update={"start_time": datetime(2026, 10, 8, 2, tzinfo=timezone.utc)})))
assert error.value.status_code == 422
bound.permission_cap = sorted(permissions - {"simulation.view"})
with pytest.raises(HTTPException) as error:
asyncio.run(query_service.authorize_query(scope, bound, TOOLS["run_burst_simulation"], args))
@@ -881,7 +882,7 @@ def test_old_task_run_and_reader_parent_are_rejected(monkeypatch):
@pytest.mark.parametrize("purpose", ["task_execution", "task_collaboration"])
@pytest.mark.parametrize("replay", [False, True])
def test_work_order_task_excludes_location_but_scada_replay_keeps_it(monkeypatch, purpose, replay):
def test_work_order_source_does_not_remove_authorized_analysis_tools(monkeypatch, purpose, replay):
monkeypatch.setattr(contexts, "permissions_for_context", lambda _: PERMISSIONS)
value = binding(task=True)
value.purpose = purpose
@@ -899,13 +900,9 @@ def test_work_order_task_excludes_location_but_scada_replay_keeps_it(monkeypatch
monkeypatch.setattr(contexts, "task_repository", lambda _: Tasks())
scope = SimpleNamespace(project=object(), connection=Connection())
asyncio.run(contexts.verify_task(scope, value))
assert ("run_burst_location" in value.tool_cap) is replay
assert "run_burst_location" in value.tool_cap
assert "analyze_valve_isolation" in value.tool_cap
assert "analyze_valve_closure" in value.tool_cap
if not replay:
with pytest.raises(HTTPException) as caught:
asyncio.run(query_service.authorize_query(scope, value, TOOLS["run_burst_location"], SimpleNamespace()))
assert caught.value.status_code == 403
def test_jwt_signature_issuer_audience_lifetime_and_key_cache():
+3 -3
View File
@@ -46,7 +46,7 @@ def test_work_order_handling_and_legacy_scada_replay_receive_separate_skills():
assert replay["workflow"] == "burst_report"
assert "影响说明" in handling["instructions"]
assert "supply_affected_nodes" in handling["instructions"]
assert "run_burst_location" not in handling["instructions"]
assert "run_burst_location" in handling["instructions"]
assert "run_burst_location" in replay["instructions"]
@@ -66,14 +66,14 @@ def test_pollution_instructions_offer_work_order_before_simulation_and_reuse_the
assert text.index("propose_task_action") < text.index("run_contaminant_simulation")
assert "include_impact_coverage=true" in text and "不等模拟完成才发起工单" in text
assert "同一 task_id" in text and "不覆盖已审核工单或绕过版本校验" in text
assert "模型曲线不填观测专用 chart.scada" in text
assert "模型曲线不填观测专用chart.scada" in text
assert "不能把浊度、余氯下降或感官描述直接换算成污染源浓度" in text
def test_pollution_workflow_distinguishes_protection_flushing_and_clearance():
skill = read_work_order_skill("water_pollution", include_content=True)
text = skill["instructions"]
assert skill["version"] == "1.2.0"
assert skill["version"] == "1.2.1"
assert text.index("## 4. 源头控制与用户隔离") < text.index("## 5. 污染物清除与分步冲洗")
assert "opened_valves" in text and "closed_valves" in text
assert "不继承上次模拟" in text and "排放去向" in text
+11
View File
@@ -46,6 +46,17 @@ def test_online_simulation_rejects_invalid_step_before_calculating(monkeypatch):
run.assert_not_called()
def test_online_baseline_preserves_fractional_start_for_exact_impact_comparison(monkeypatch):
start = START.replace(microsecond=123456)
monkeypatch.setattr(simulation, "get_time", lambda _: {"HYDRAULIC TIMESTEP": "0:15:00"})
monkeypatch.setattr(simulation, "query_corresponding_element_id_and_query_id", lambda _: object())
run = Mock(return_value={"node_results": [{"node": "J1", "result": [{"pressure": 30}]}]})
monkeypatch.setattr(simulation, "run_simulation", run)
result = simulation.run_online_simulation("online", start, 60, output_node_ids=["J1"])
assert run.call_args.kwargs["modify_pattern_start_time"] == start.isoformat()
assert result["times"]["sample_times"] == [start.isoformat()]
def test_mcp_online_simulation_does_not_copy_a_database(monkeypatch):
from app.mcp.scenario_service import calculate_scenario
+171
View File
@@ -0,0 +1,171 @@
import asyncio
from contextlib import asynccontextmanager
from types import SimpleNamespace
from unittest.mock import AsyncMock, Mock
from uuid import uuid4
import pytest
from fastapi import HTTPException
from app.mcp import query_service
from app.mcp.analysis_models import ScenarioInput
from app.mcp.catalog import TOOLS
from app.services import simulation
def parameters(**changes):
return ScenarioInput.model_validate({
"context_handle": "a" * 43, "start_time": "2026-07-16T08:00:00+08:00",
"duration": 60, "output_node_ids": ["J1"], **changes,
})
def context(monkeypatch):
permissions = {"simulation.run", "simulation.view", "scada.view", "webgis.view", "optimization.run"}
monkeypatch.setattr(query_service, "permissions_for_context", lambda _: permissions)
@asynccontextmanager
async def routing():
yield
scope = SimpleNamespace(
connection=SimpleNamespace(commit=AsyncMock()), simulation_routing=routing,
project=SimpleNamespace(project_code="owned", project_role="member"),
user=SimpleNamespace(username="actor"),
)
binding = SimpleNamespace(
tool_cap=["run_simulation"], permission_cap=sorted(permissions),
purpose="task_execution", task_id=uuid4(), task_scope_cap=[],
replay_context={"knowledge_cutoff": "2026-07-16T08:00:00+08:00"},
)
return scope, binding
def test_replay_coordinator_can_rebuild_online_baseline_without_live_scada(monkeypatch):
scope, binding = context(monkeypatch)
monkeypatch.setattr(simulation, "get_time", lambda _: {"HYDRAULIC TIMESTEP": "0:15:00"})
live_mappings = Mock(side_effect=AssertionError("Replay must not load live SCADA mappings"))
monkeypatch.setattr(simulation, "query_corresponding_element_id_and_query_id", live_mappings)
native = Mock(return_value={
"units": {"pressure": "MTR"},
"node_results": [{"node": "J1", "result": [{"pressure": 30}]}],
})
monkeypatch.setattr(simulation, "run_simulation", native)
result = asyncio.run(query_service.execute_query(scope, binding, TOOLS["run_simulation"], parameters()))
assert result.origin == "synthetic"
assert any("不读取实时SCADA" in gap for gap in result.data_gaps)
assert result.data["result"]["persisted_baseline"] is True
assert result.data["result"]["node_results"][0]["result"] == [{"pressure": 30}]
native.assert_called_once()
call = native.call_args.kwargs
assert call["name"] == "owned" and call["simulation_type"] == "realtime"
assert call["modify_pattern_start_time"] == "2026-07-16T08:00:00+08:00"
assert all(not values for values in vars(call["scada_mappings"]).values())
live_mappings.assert_not_called()
@pytest.mark.parametrize("problem", ["future_start", "reader", "permission", "tool_cap"])
def test_replay_online_baseline_preserves_existing_authorization(monkeypatch, problem):
scope, binding = context(monkeypatch)
args = parameters()
if problem == "future_start":
args = parameters(start_time="2026-07-16T08:15:00+08:00")
if problem == "reader":
binding.purpose = "task_reader"
if problem == "permission":
binding.permission_cap.remove("simulation.run")
if problem == "tool_cap":
binding.tool_cap.clear()
with pytest.raises(HTTPException) as caught:
asyncio.run(query_service.authorize_query(scope, binding, TOOLS["run_simulation"], args))
assert caught.value.status_code == (422 if problem == "future_start" else 403)
@pytest.mark.parametrize("name", ["analyze_valve_closure", "analyze_pressure_regulation"])
def test_replay_can_compare_hydraulic_alternatives_with_model_only_boundaries(monkeypatch, name):
from app.infra.db.postgresql.scada import ScadaElementMappings
from app.services import simulation_scenarios as scenarios
scope, binding = context(monkeypatch)
binding.tool_cap = [name]
captures = []
monkeypatch.setattr(scenarios, "get_option", lambda _: {"UNITS": "LPS"})
monkeypatch.setattr(scenarios, "set_option", lambda *_: None)
monkeypatch.setattr(simulation, "query_corresponding_element_id_and_query_id",
lambda _: ScadaElementMappings({}, {}, {"P1": "device"}, {}, {}, {}, {}))
monkeypatch.setattr(simulation, "run_simulation", lambda **kw: captures.append(kw) or {
"node_results": [{"node": "J1", "result": [{"pressure": 30}]}],
})
native_name = "valve_close_analysis" if name == "analyze_valve_closure" else "pressure_regulation"
native = getattr(scenarios, native_name).__wrapped__
monkeypatch.setattr(scenarios, native_name,
lambda **kw: native(**kw, _temporary_project="tjw_tmp_owned"))
args = TOOLS[name].model.model_validate({
**parameters().model_dump(),
**({"valves": ["V1"]} if name == "analyze_valve_closure" else {"pump_control": {"P1": [0]}}),
})
result = asyncio.run(query_service.execute_query(scope, binding, TOOLS[name], args))
assert result.origin == "synthetic"
assert result.data["result"]["persisted_baseline"] is False
assert len(captures) == 1
assert captures[0]["name"] == "tjw_tmp_owned" and captures[0]["persist_results"] is False
assert all(not values for values in vars(captures[0]["scada_mappings"]).values())
if name == "analyze_valve_closure":
assert captures[0]["modify_valve_opening"] == {"V1": 0.0}
else:
assert captures[0]["modify_fixed_pump_pattern"] == {"P1": [0]}
@pytest.mark.parametrize("name", [name for name, tool in TOOLS.items()
if tool.semantics == "asset" and not tool.mutating])
def test_replay_allows_authorized_asset_queries_for_input_discovery(monkeypatch, name):
scope, binding = context(monkeypatch)
binding.tool_cap = [name]
# Authorization needs only the schema kind; execution validates each tool's inputs.
asyncio.run(query_service.authorize_query(scope, binding, TOOLS[name], SimpleNamespace(kind="node")))
binding.permission_cap = []
with pytest.raises(HTTPException) as caught:
asyncio.run(query_service.authorize_query(scope, binding, TOOLS[name], SimpleNamespace(kind="node")))
assert caught.value.status_code == 403
@pytest.mark.parametrize("name,fields", [
("query_model_baseline_series", {"element_type": "junction", "element_ids": ["J1"],
"start": "2026-07-16T07:45:00+08:00", "end": "2026-07-16T08:00:00+08:00"}),
("get_model_baseline_at_time", {"element_type": "junction", "element_id": "J1", "time": "2026-07-16T08:00:00+08:00"}),
("query_model_baseline_field", {"element_type": "junction", "field": "pressure", "time": "2026-07-16T08:00:00+08:00"}),
])
def test_replay_can_verify_baseline_without_reading_future_values(monkeypatch, name, fields):
scope, binding = context(monkeypatch)
binding.tool_cap = [name]
args = TOOLS[name].model.model_validate({"context_handle": "a" * 43, **fields})
asyncio.run(query_service.authorize_query(scope, binding, TOOLS[name], args))
future = parameters(start_time="2026-07-16T08:15:00+08:00").start_time
with pytest.raises(HTTPException) as caught:
asyncio.run(query_service.authorize_query(scope, binding, TOOLS[name], args.model_copy(
update={"end" if "end" in fields else "time": future},
)))
assert caught.value.status_code == 422
@pytest.mark.parametrize("name", ["calculate_demand_allocation", "optimize_sensor_placement"])
def test_replay_can_choose_current_model_design_calculations(monkeypatch, name):
scope, binding = context(monkeypatch)
binding.tool_cap = [name]
asyncio.run(query_service.authorize_query(scope, binding, TOOLS[name], SimpleNamespace()))
binding.purpose = "task_reader"
with pytest.raises(HTTPException) as caught:
asyncio.run(query_service.authorize_query(scope, binding, TOOLS[name], SimpleNamespace()))
assert caught.value.status_code == 403
def test_replay_can_explore_contamination_without_requiring_map_generation(monkeypatch):
scope, binding = context(monkeypatch)
binding.tool_cap = ["run_contaminant_simulation"]
args = TOOLS["run_contaminant_simulation"].model.model_validate({
**parameters().model_dump(), "source": "J1", "concentration": 1,
"include_impact_coverage": False,
})
asyncio.run(query_service.authorize_query(scope, binding, TOOLS["run_contaminant_simulation"], args))