fix(scada): align location capability and investigation guidance

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),
     (

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 = [
+        gaps = [result["notice"]] if result.get("notice") else []
-            "定位只使用任务知识截止前已固化的回放清洗输出；候选排序依赖漏量假设、模型和测点覆盖，仍需现场核实。"
-        ]
-        if used_demo:
-            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:

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:

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"]

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、在线模拟等未接入冻结输入的路径仍不可用；候选定位不以出厂流量命中或现场确诊为前提。
 

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字以内为目标，测点明细用表格、完整依据留在证据。候选、相似度和拓扑结果具备时，完成相应分析、展示与报告后再收尾；现场核验、审批和维修列为后续待办。现场工单与设备操作遵守服务端动作权限，本轮分析完成与现场事件闭环分别说明。

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)
 

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

tests/unit/test_scada_monitoring_calibration.py

@@ -7,7 +7,7 @@ from pydantic import ValidationError
 from app.domain.schemas.project_settings import ScadaMonitoringSettings, ScadaMonitoringWeights
 from app.services.scada_monitoring_calibration import MonitoringCalibrationBuilder, adaptive_score_threshold
 from app.services import scada_replay_monitoring as monitoring
-from test_scada_replay_monitoring import TARGET, PRESSURE_IDS, FLOW_IDS, _document
+from tests.unit.test_scada_replay_monitoring import TARGET, PRESSURE_IDS, FLOW_IDS, _document
 from app.domain.schemas.scada_replay import ScadaReplayCreate
 from app.services.scada_normal_profile import CALIBRATION_VERSION, DIAGNOSTIC_POLICY, normal_profile_bounds
 

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