diff --git a/app/mcp/catalog.py b/app/mcp/catalog.py index 17c5f03..af1be5a 100644 --- a/app/mcp/catalog.py +++ b/app/mcp/catalog.py @@ -64,7 +64,7 @@ _specs = [ ("optimize_sensor_placement", "使用灵敏度或K-means算法推荐压力监测点；返回候选节点，正式部署另行保存并核验观测能力", SensorPlacementInput, "optimization.run", "analysis", False, True), ("run_leakage_identification", "基于SCADA压力与模型进行DMA漏损分配识别，需有依据的总漏损流量和时间对齐；返回待现场核验的区域估计。内部执行遗传算法与水力计算", LeakageIdentificationInput, "burst.run", "analysis", False, True), ("run_burst_detection", "爆管异常侦测：使用至少5个完整压力传感器此前15天数据；返回压力异常信号，候选管段由定位工具分析", BurstDetectionInput, "burst.run", "analysis", False, True), - ("run_burst_location", "爆管候选定位：对比异常与正常压力，可辅助流量；依据现有异常形成候选，水力算法使用有依据的burst_leakage作为输入。任务上下文location_demo.available=true时可省略漏量，Server直接返回候选管段与相似度，随后继续地图展示、隔离边界与影响分析。压力节点可省略，自动读取项目配置；monitoring正常窗省略时默认前一天同时间段，须说明假设；回放必须显式提供冻结正常窗。simulation须有已有分析运行UUID及同窗模型基线，结果标注模型来源", BurstLocationInput, "burst.run", "analysis", False, True), + ("run_burst_location", "爆管候选定位：对比异常与正常压力，可辅助流量；依据现有异常形成候选，水力算法使用有依据的burst_leakage作为输入。任务上下文location_capability.available=true时可省略漏量，Server直接返回候选管段与相似度，随后继续地图展示、隔离边界与影响分析。压力节点可省略，自动读取项目配置；monitoring正常窗省略时默认前一天同时间段，须说明假设；回放必须显式提供冻结正常窗。simulation须有已有分析运行UUID及同窗模型基线，结果标注模型来源", BurstLocationInput, "burst.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), ( diff --git a/app/mcp/query_service.py b/app/mcp/query_service.py index 217e160..624dcf7 100644 --- a/app/mcp/query_service.py +++ b/app/mcp/query_service.py @@ -181,7 +181,6 @@ 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" - gaps.append("历史数据回放证据，仅覆盖该轮调查已提交的批次，不表示现场实时观测。") elif name == "run_burst_location" and replay_context: from app.mcp.analysis_service import calculate_replay_burst_location from app.services.scada_replay_queries import replay_burst_location_inputs @@ -190,7 +189,6 @@ async def execute_query( conn, scope.project.project_id, replay_context, args, ) result = await location_demo_result(conn, scope.project.project_code, args) - used_demo = result is not None await conn.commit() if result is None: async with scope.simulation_routing(): @@ -204,11 +202,7 @@ async def execute_query( "result": result, } origin = "synthetic" - gaps = [ - "定位只使用任务知识截止前已固化的回放清洗输出；候选排序依赖漏量假设、模型和测点覆盖，仍需现场核实。" - ] - if used_demo: - gaps = [result["notice"]] if result.get("notice") else [] + gaps = [result["notice"]] if result.get("notice") else [] elif name == "run_burst_location" and ( demo_result := await location_demo_result(conn, scope.project.project_code, args) ) is not None: @@ -488,7 +482,7 @@ async def execute_query( and "run_burst_location" in binding.tool_cap): from app.services.burst_location_demo import location_demo_context - data["location_demo"] = location_demo_context( + data["location_capability"] = location_demo_context( getattr(scope.project, "project_code", None), replay_context, ) else: diff --git a/app/services/operational_tasks.py b/app/services/operational_tasks.py index f38c541..e76a05b 100644 --- a/app/services/operational_tasks.py +++ b/app/services/operational_tasks.py @@ -2153,7 +2153,7 @@ class OperationalTaskService: @staticmethod def _require_real_action(task: dict) -> None: require(not (task.get("source") or {}).get("replay_context"), "REPLAY_ACTION_DISABLED", - "SCADA 回放调查仅生成分析与报告，不执行真实工单或设备动作", status=409) + "当前任务的工单和设备动作未开放；请继续完成已授权的定位、地图展示、隔离影响分析与报告", status=409) @staticmethod def _require_actionable_task(task: dict) -> None: diff --git a/app/services/scada_replay.py b/app/services/scada_replay.py index 708e196..b6bcc57 100644 --- a/app/services/scada_replay.py +++ b/app/services/scada_replay.py @@ -532,7 +532,7 @@ class ScadaReplayService: f"爆管侦测得分 {score_text}；" f"压力下降测点 {len(pressure.get('hit_device_ids', []))} 个；" f"出厂边界流量增加测点 {len(flow.get('hit_device_ids', []))} 个。" - "任一指标异常即启动调查，加权分仅用于评估爆管嫌疑，不能据此提前确诊。" + "任一指标异常即启动调查，加权分用于评估爆管嫌疑；达到门槛后继续候选定位、隔离与影响分析。" ) if monitoring.get("anomaly_signals"): signals = monitoring["anomaly_signals"] diff --git a/docs/scada-peripheral-replay-design.md b/docs/scada-peripheral-replay-design.md index 37d839a..54d0001 100644 --- a/docs/scada-peripheral-replay-design.md +++ b/docs/scada-peripheral-replay-design.md @@ -573,7 +573,14 @@ next_cursor 绑定项目、设备、查询区间并固定本轮分页时间上 ### 自动播放与简短通知（2026-11-07） -调查任务仍按 `replay_context.knowledge_cutoff` 与 `batch_sequence` 截断知识，后续播放不自动扩大当前调查窗口。任务摘要只保留指标与异常数量，逐测点明细通过冻结 SCADA 证据查询。`scada-anomaly` 要求高嫌疑诊断继续候选定位；当前 Case3 的任务摘要和 skill 响应返回 `location_demo` 能力提示。提示可用时 `run_burst_location` 允许省略 `burst_leakage`，仍先核验冻结异常窗、正常窗及压力设备，再调用固定候选方法；水力算法仍要求漏量依据。候选输出不再含 mock 提示，MCP回放证据保留 `origin=synthetic`，不向 Agent 暴露事件真值。 +调查任务仍按 `replay_context.knowledge_cutoff` 与 `batch_sequence` 截断知识，后续播放不自动扩大当前调查窗口。任务摘要只保留指标与异常数量，逐测点明细通过冻结 SCADA 证据查询。`scada-anomaly` 要求高嫌疑诊断继续候选定位；当前 Case3 的任务摘要和 skill 响应返回 `location_capability` 能力提示。提示可用时 `run_burst_location` 允许省略 `burst_leakage`，仍先核验冻结异常窗、正常窗及压力设备，再调用固定候选方法；水力算法仍要求漏量依据。候选输出不再含 mock 提示，MCP回放证据保留 `origin=synthetic`，不向 Agent 暴露事件真值。 + +Agent可见能力字段统一为 `location_capability`，不再输出旧 `location_demo` 字段。 +公共指令与 SCADA Skill 明确候选及相似度返回后继续地图展示、隔离、节点覆盖、影响分析与报告， +来源字段本身不构成停止条件。有效 SCADA 查询不再将来源说明列为 `data_gaps`；实际缺测、 +窗口与设备校验仍按原规则处理。现场工单与设备动作维持服务端权限限制，动作拒绝文案明确 +允许继续已授权分析。旧工具回执和历史消息不改写；既有 OpenCode 进程的 system 配置副本 +需刷新后才使用新的公共指令。 回放调查开放项目资产属性、空间查询、拓扑隔离和节点覆盖，沿用原有业务权限。它们依据当前模型，不代表历史现场状态。实时 TSDB、在线模拟等未接入冻结输入的路径仍不可用；候选定位不以出厂流量命中或现场确诊为前提。 diff --git a/resources/work-order-skills/scada-anomaly/SKILL.md b/resources/work-order-skills/scada-anomaly/SKILL.md index bf89123..2ce7ca5 100644 --- a/resources/work-order-skills/scada-anomaly/SKILL.md +++ b/resources/work-order-skills/scada-anomaly/SKILL.md @@ -2,12 +2,12 @@ name: scada-anomaly description: 从 SCADA 指标异常开展数据核验、爆管诊断、候选定位与影响分析。 metadata: - version: "1.0.8" + version: "1.0.9" --- -# SCADA 回放异常调查 +# SCADA 数据异常调查 -读取任务与 replay_context，核对设备、单位、异常时刻、knowledge_cutoff 和 batch_sequence。来源标注为已存 SCADA 的演示回放；同一持续事件沿用已有调查。 +读取任务与 replay_context，核对设备、单位、异常时刻、knowledge_cutoff 和 batch_sequence。使用本轮已提交的 SCADA 观测开展调查，同一持续事件沿用已有调查。证据来源字段用于记录数据取得方式，不构成停止分析的条件。 报告与图表的阅读时区统一为Asia/Shanghai（UTC+8）。工具返回的+08:00时间按原值使用；Z时间换算后同步更新日期、历史系列名和横轴标签，例如2026-07-16T00:00:00Z对应7月16日08:00。sample_times记录相同真实时间点并使用+08:00偏移，查询与冻结边界按带时区的时间点比较。 @@ -25,7 +25,7 @@ metadata: 综合分达到嫌疑门槛，或压力与模型证据支持高嫌疑时，继续候选定位。流量变化和现场确认作为证据维度，定位输入按当前工具要求准备。 -检查摘要或 skill 响应顶层的 location_demo。available=true 时，用 run_burst_location 提交提示的异常窗、已核验的冻结正常对照窗及实际压力节点，burst_leakage 可省略。读取返回的候选管段与相似度，优先按 similarity_percent 展示百分比；只有 similarity 时，按0–1小数换算。取得候选后继续地图展示、隔离边界与影响分析。其他定位路径使用有依据的漏量估计或明确情景假设。缺少定位输入时记录待补项，继续独立的资产、空间与拓扑核查。 +检查摘要或 skill 响应顶层的 location_capability。available=true 时，用 run_burst_location 提交提示的异常窗、已核验的冻结正常对照窗及实际压力节点，burst_leakage 可省略。读取返回的候选管段与相似度，优先按 similarity_percent 展示百分比；只有 similarity 时，按0–1小数换算。取得候选后继续地图展示、隔离边界、节点覆盖与影响分析，完成报告后再收尾。其他定位路径使用有依据的漏量估计或明确情景假设。缺少定位输入时记录待补项，继续独立的资产、空间与拓扑核查。 取得候选后展示管段，再按授权用 analyze_valve_isolation 分析隔离边界，保留必关阀、可选阀、affected_nodes、supply_affected_nodes 及计数；用 generate_node_coverage 展示节点覆盖。候选与拓扑建议标注为待现场核验方案。最终报告前核对定位调用及返回结果，调用失败记录具体错误与输入缺口。 @@ -39,6 +39,6 @@ metadata: ## 范围与报告 -本轮时序分析保持任务冻结窗口，后续播放形成的观测由相应轮次处理。资产属性、空间查询和拓扑隔离使用项目当前模型，注明模型时态；水力工具按已开放的冻结输入能力调用。演示定位继续执行权限、设备与冻结观测校验；基线或覆盖缺口按实际返回记录。 +本轮时序分析保持任务冻结窗口，后续播放形成的观测由相应轮次处理。资产属性、空间查询和拓扑隔离使用项目当前模型，注明模型时态；水力工具按已开放的冻结输入能力调用。定位调用执行权限、设备与冻结观测校验；基线或覆盖缺口按实际返回记录。 -报告给出观测事实、规则信号、原因候选、定位与隔离结果、影响说明和建议。摘要以150字以内为目标，测点明细用表格、完整依据留在证据。演示分析只发布回放成果，本轮分析完成与现场事件闭环分别说明。 +报告给出观测事实、规则信号、原因候选、定位与隔离结果、影响说明和建议。摘要以150字以内为目标，测点明细用表格、完整依据留在证据。候选、相似度和拓扑结果具备时，完成相应分析、展示与报告后再收尾；现场核验、审批和维修列为后续待办。现场工单与设备操作遵守服务端动作权限，本轮分析完成与现场事件闭环分别说明。 diff --git a/tests/unit/test_burst_location_demo.py b/tests/unit/test_burst_location_demo.py index c0f5982..f5b4619 100644 --- a/tests/unit/test_burst_location_demo.py +++ b/tests/unit/test_burst_location_demo.py @@ -131,6 +131,8 @@ def test_location_capability_and_tool_contract_do_not_inject_mock_instructions() assert "mock" not in str(context) and "模拟" not in context["notice"] tool = TOOLS["run_burst_location"].protocol_tool() assert "相似度" in tool.description + assert "location_capability.available=true" in tool.description + assert "location_demo" not in tool.description assert "demo模拟候选" not in tool.description assert "明确标记的演示候选" not in str(tool.input_schema) diff --git a/tests/unit/test_mcp_work_order_skills.py b/tests/unit/test_mcp_work_order_skills.py index fc1defe..1bff77a 100644 --- a/tests/unit/test_mcp_work_order_skills.py +++ b/tests/unit/test_mcp_work_order_skills.py @@ -49,6 +49,16 @@ def test_work_order_handling_and_legacy_scada_replay_receive_separate_skills(): assert "run_burst_location" in replay["instructions"] +def test_scada_instructions_advance_candidate_analysis_independently_of_source_labels(): + skill = read_work_order_skill("scada_anomaly", include_content=True) + text = skill["instructions"] + assert "location_capability" in text + assert "证据来源字段用于记录数据取得方式，不构成停止分析的条件" in text + assert "隔离边界、节点覆盖与影响分析，完成报告后再收尾" in text + assert "现场工单与设备操作遵守服务端动作权限" in text + assert not any(word in text for word in ("演示", "mock", "location_demo", "只发布回放成果")) + + def test_task_skill_uses_persisted_workflow_and_preserves_task_scope(monkeypatch): task_id, project_id = uuid4(), uuid4() reader = AsyncMock(return_value={"task": {"workflow": "burst_report"}}) @@ -103,7 +113,8 @@ def test_replay_context_advertises_demo_location_without_revealing_truth(monkeyp for section in ("summary", "skill"): result = asyncio.run(query_service.execute_query(scope, binding, TOOLS["get_task_context"], TaskContextInput(context_handle=binding.context_handle, section=section))) - capability = result.data["location_demo"] + capability = result.data["location_capability"] + assert "location_demo" not in result.data assert capability["available"] is True assert capability["tool"] == "run_burst_location" assert capability["burst_leakage_required"] is False @@ -113,4 +124,4 @@ def test_replay_context_advertises_demo_location_without_revealing_truth(monkeyp binding.replay_context["knowledge_cutoff"] = "2026-07-15T23:45:00Z" result = asyncio.run(query_service.execute_query(scope, binding, TOOLS["get_task_context"], TaskContextInput(context_handle=binding.context_handle))) - assert result.data["location_demo"]["available"] is False + assert result.data["location_capability"]["available"] is False diff --git a/tests/unit/test_scada_replay_queries.py b/tests/unit/test_scada_replay_queries.py index 265cfdf..b5dfd0f 100644 --- a/tests/unit/test_scada_replay_queries.py +++ b/tests/unit/test_scada_replay_queries.py @@ -10,7 +10,7 @@ from fastapi import HTTPException from app.mcp.analysis_models import BurstLocationInput from app.mcp.catalog import TOOLS -from app.mcp.models import BoundContext +from app.mcp.models import BoundContext, ScadaInput from app.mcp import query_service from app.services.scada_replay_queries import replay_burst_location_inputs, replay_snapshot_rows @@ -18,6 +18,30 @@ from app.services.scada_replay_queries import replay_burst_location_inputs, repl START = datetime(2026, 10, 8, tzinfo=timezone.utc) +def test_replay_origin_is_preserved_without_a_provenance_only_data_gap(monkeypatch): + from app.services import scada_replay_queries + + monkeypatch.setattr(query_service, "authorize_query", AsyncMock()) + reading = {"time": START.isoformat(), "device_id": "P1", "cleaned_value": 20.0, + "monitored_value": 20.0, "quality": "valid"} + read = AsyncMock(return_value=[reading]) + monkeypatch.setattr(scada_replay_queries, "replay_snapshot_rows", read) + project_id = uuid4() + context = {"replay_id": str(uuid4()), "knowledge_cutoff": (START + timedelta(minutes=15)).isoformat(), + "batch_sequence": 9} + connection = object() + scope = SimpleNamespace(connection=connection, project=SimpleNamespace(project_id=project_id)) + binding = SimpleNamespace(replay_context=context) + args = ScadaInput(context_handle="a" * 43, device_ids=["P1"], start=START, + end=START + timedelta(minutes=15)) + result = asyncio.run(query_service.execute_query(scope, binding, TOOLS["query_scada_readings"], args)) + assert result.origin == "synthetic" + assert result.data[0]["device_id"] == "P1" + assert result.data_gaps == [] + read.assert_awaited_once_with(connection, project_id, context, ["P1"], args.start, args.end, + args.limit + 1, args.offset) + + class Cursor: def __init__(self, rows): self.rows = rows async def fetchone(self): return self.rows[0] if self.rows else None