fix(recovery): scope incident verification to relevant devices

Full-network browsing expanded persisted recovery gates to unrelated devices, while full fifteen-day reference reads exhausted the row budget. Rebuild legacy cached scopes from report devices and project monitoring roles, and query only the nearest post-feedback frame and its historical phase.

Regression tests fail before the fix for both new and cached scopes and verify automatic Agent requeue after recovery.

app/domain/schemas/operational_task.py

@@ -862,9 +862,18 @@ class TaskHighlightFeatures(StrictModel):
 
 class TaskLocateFeatures(StrictModel):
     action: Literal["locate_features"]
+    event_kind: Literal["burst", "pollution"] | None = Field(default=None,
+        description="定位事件语义：爆管候选传 burst，污染源传 pollution；普通资产定位省略")
     layer: TaskMapLayer
     feature_ids: Annotated[list[TaskFeatureId], Field(min_length=1, max_length=100)]
 
+    @model_serializer(mode="wrap")
+    def serialize_optional_kind(self, handler):
+        value = handler(self)
+        if self.event_kind is None:
+            value.pop("event_kind", None)
+        return value
+
 
 class TaskZoomToMap(StrictModel):
     action: Literal["zoom_to_map"]

app/infra/db/timescaledb/repositories/scada.py

@@ -143,9 +143,12 @@ class ScadaRepository:
             await cur.execute(
                 "WITH next_frame AS (SELECT min(time) AS time FROM scada.measurements "
                 "WHERE device_id=ANY(%s) AND time>%s AND time<=%s), "
-                "reference AS (SELECT time,device_id,monitored_value,cleaned_value "
-                "FROM scada.measurements WHERE device_id=ANY(%s) AND time>=%s AND time<%s "
-                "ORDER BY time,device_id LIMIT %s) "
+                "reference AS (SELECT m.time,m.device_id,m.monitored_value,m.cleaned_value "
+                "FROM scada.measurements m CROSS JOIN next_frame f "
+                "WHERE m.device_id=ANY(%s) AND m.time>=%s AND m.time<%s "
+                "AND mod(extract(epoch FROM m.time)::bigint,86400)="
+                "mod(extract(epoch FROM f.time)::bigint,86400) "
+                "ORDER BY m.time,m.device_id LIMIT %s) "
                 "SELECT time,device_id,monitored_value,cleaned_value FROM reference UNION ALL "
                 "SELECT m.time,m.device_id,m.monitored_value,m.cleaned_value FROM scada.measurements m "
                 "JOIN next_frame f ON m.time=f.time WHERE m.device_id=ANY(%s) ORDER BY time,device_id",

app/services/burst_impact.py

@@ -17,7 +17,7 @@ GRADES = (
 
 def _head_factor(unit: Any) -> float | None:
     # Standard water head; unsupported or absent units must not become zero loss.
-    return {"m": 1.0, "meters": 1.0, "metres": 1.0,
+    return {"m": 1.0, "mtr": 1.0, "meters": 1.0, "metres": 1.0,
             "kpa": 1 / 9.80665, "psi": 6894.757293168 / 9806.65,
             "bar": 100000 / 9806.65, "mpa": 1000000 / 9806.65,
             "ft": 0.3048}.get(str(unit).strip().lower())

app/services/operational_presentation.py

@@ -104,3 +104,17 @@ def resolve_overlay(snapshot: TaskDataSnapshot, mode: str) -> dict:
             normalized["properties"]["weight"] = weight
         features.append(normalized)
     return {"type": "FeatureCollection", "features": features}
+
+
+def validate_overlay_stage(data: dict, stage: str | None) -> None:
+    """An event-impact claim must be supported by the saved scenario geometry."""
+    if stage != "event_impact":
+        return
+    features = data.get("features", [])
+    require(bool(features) and all(
+        (feature.get("properties") or {}).get("impact_phase") in {
+            "after_burst_before_intervention", "after_contamination_before_intervention",
+        } for feature in features
+    ), "OVERLAY_IMPACT_PHASE_MISMATCH",
+        "事件影响阶段须引用处置前情景覆盖；关阀及供水失源覆盖请使用 isolation_plan，不能替代不可用的事件降压结果",
+        status=422)

app/services/operational_recovery.py

@@ -170,13 +170,18 @@ async def reconcile_recoveries(service: OperationalTaskService, timescale: Async
         devices = list(dict.fromkeys(roles.pressure_device_ids + roles.outlet_flow_device_ids)) if roles else []
         source = "scada_readings"
         through = service._now()
-        if context.get("device_ids"):
+        report_revision = task.get("current_report_revision")
+        if (context.get("device_ids") and context.get("device_selection_version") == 2
+                and context.get("source_report_revision") == report_revision):
             devices = context["device_ids"]
             interval = context.get("interval_seconds", interval)
         else:
-            # Retain the points the Agent actually inspected, including water
-            # quality or local incident sensors outside the project role lists.
-            for evidence in await service.repo.list_evidence(task["task_id"]):
+            # Full-network browsing is discovery, not a declaration that every
+            # device must recover for this incident. Prefer the report's scope.
+            report = await service.repo.get_latest_report(task["task_id"])
+            declared = ((report or {}).get("report") or {}).get("scada_device_ids") or []
+            devices.extend(declared)
+            for evidence in ([] if declared else await service.repo.list_evidence(task["task_id"])):
                 saved = evidence.get("data_snapshot") or {}
                 if saved.get("source") not in {"query_scada_readings", "scada_readings"}:
                     continue
@@ -200,11 +205,13 @@ async def reconcile_recoveries(service: OperationalTaskService, timescale: Async
             await service.repo.lock(f"task:{task['task_id']}")
             current = await service._task(UUID(task["task_id"]))
             # A new receipt, cancellation or another poll may have won the race.
-            if current.get("handling_context") != context or current["status"] in {"cancelled", "failed"}:
+            if (current.get("handling_context") != context or current["status"] in {"cancelled", "failed"}
+                    or current.get("current_report_revision") != report_revision):
                 continue
             now = service._now()
             if devices:
-                current["handling_context"].update(device_ids=devices, interval_seconds=interval or 0)
+                current["handling_context"].update(device_ids=devices, interval_seconds=interval or 0,
+                                                  device_selection_version=2, source_report_revision=report_revision)
             current["handling_context"]["last_checked_at"] = now.isoformat()
             current["handling_context"]["last_check"] = result
             if result["status"] == "passed":

app/services/operational_tasks.py

@@ -75,7 +75,7 @@ from app.domain.schemas.operational_task import (
 from app.domain.schemas.operational_data import TaskDataSnapshot, TaskDataSource
 from app.services.content_digest import digest
 from app.services.investigation_graph import adopt_legacy_investigation_graph, update_investigation_graph
-from app.services.operational_presentation import resolve_overlay
+from app.services.operational_presentation import resolve_overlay, validate_overlay_stage
 from app.services.operational_recovery import accept_completion, track_work_order
 from app.services.operational_scada_coverage import require_complete_scada_evidence
 from app.services import scada_presentation_coverage
@@ -1124,6 +1124,7 @@ class OperationalTaskService:
                     )
                     require(supplied == resolved, "OVERLAY_DATA_MISMATCH",
                             "Overlay data must match the immutable evidence snapshot", status=422)
+                validate_overlay_stage(resolved, command.narration.stage if command.narration else None)
                 payload["data"] = resolved
             metadata = {"task_id": str(task_id), "run_id": str(run["run_id"]),
                         "execution_generation": run["generation"], "command": payload}

contracts/manifest.json

@@ -3,7 +3,7 @@
   "contracts": {
     "server": {
       "file": "server-v1.openapi.json",
-      "sha256": "c31e9b4c71216951d1525acf1bcb8aeba4158b6cdc6b0b9dc01be02f9b6e6af3"
+      "sha256": "9801087d40ac0e1f67871f9a1db850cbcce47614cceec0a7e8771cdf58385f53"
     }
   }
 }

contracts/server-v1.openapi.json

@@ -9945,6 +9945,22 @@
             "title": "Action",
             "type": "string"
           },
+          "event_kind": {
+            "anyOf": [
+              {
+                "enum": [
+                  "burst",
+                  "pollution"
+                ],
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "description": "定位事件语义：爆管候选传 burst，污染源传 pollution；普通资产定位省略",
+            "title": "Event Kind"
+          },
           "feature_ids": {
             "items": {
               "maxLength": 128,

contracts/view-command.schema.json

@@ -7,6 +7,13 @@
           "type": "string",
           "const": "locate_features"
         },
+        "event_kind": {
+          "type": "string",
+          "enum": [
+            "burst",
+            "pollution"
+          ]
+        },
         "layer": {
           "type": "string",
           "enum": [

docs/burst-response-workflow.md

@@ -54,6 +54,8 @@ SCADA作为事件影响的辅助证据，按目标测点和事件／对照窗口
 
 业务阶段和工单、返回时间、复核回执保存在任务文档 `handling_context` JSON 中，不增加数据库列。外围样板更新工单时可提供带时区的 `occurred_at`，同时必须填写现场反馈；该时间写入活动记录，系统的 `updated_at` 仍是实际记录时间。同步器将现场反馈时间传给调查，避免历史演示误用电脑当前时间。
 
+恢复复核范围为项目配置的压力、出厂流量监测点与最新调查报告声明的 `scada_device_ids` 的并集；全网浏览查询不自动扩大本事件的复核范围。未声明测点的旧报告保留按已有查询证据补充测点的兼容路径。读取完工时间后最近一帧，并只取该帧时刻对应的事件前 15 天同期参考，不扫描完整 15 天曲线。旧等待任务保存的测点范围会按新选择版本自动重算；相关测点缺测或仍异常时继续等待，恢复通过后才唤醒 Agent。
+
 预约时间 `scheduled_at` 可在派单前的工单草稿和已创建工单的编辑表单中调整。实际完工反馈时间与预约安排分别记录；Agent 以实际反馈时间为恢复复核起点。新的有效现场反馈修订完工时间后，旧复核证据保留为历史记录，当前复核起点更新，重新读取严格晚于新反馈时间的最近一个 SCADA 采样时刻。只修改预约安排或其他工单信息，不改写已有现场反馈时间。
 
 `POST /operational-tasks/recovery/reconcile` 由现有 operational worker 每轮调用，按项目及任务创建边界处理等待复核的任务；SCADA 读取失败不阻断其他调查派发。复核直接读取项目时序库中完工反馈时间之后最近时刻的 SCADA，不依赖回放状态或推进回放时间。

resources/work-order-skills/burst-event-handling/SKILL.md

@@ -2,7 +2,7 @@
 name: burst-event-handling
 description: 根据爆管上报工单处理事件，完成资产关联、隔离方案、影响说明、处置建议与反馈复核。
 metadata:
-  version: "1.1.4"
+  version: "1.1.5"
 ---
 
 # 爆管事件处理
@@ -21,7 +21,7 @@ metadata:
 
 已有管段或节点编号时，用 get_element_properties 核对资产；已有坐标时直接使用，文字地址通过 geocode 解析后用 query_nearby_elements 获取附近资产。结合属性、道路、连接关系和现场记录确定处理对象，使用 get_node_connections 补充连接依据。
 
-在地图展示上报位置和关联资产，标明关联依据及现场确认状态。存在多个合理对象时保留候选及选择理由；关键位置需要人工补充时，先完成独立工作，再用 ask_task_user 提问。
+爆管候选管段用 locate_features 传 event_kind="burst" 定位。在地图展示上报位置和关联资产，标明关联依据及现场确认状态。存在多个合理对象时保留候选及选择理由；关键位置需要人工补充时，先完成独立工作，再用 ask_task_user 提问。
 
 定位到管道后立即保存 interim/concluded 阶段报告，明确“候选管段待现场核验”，调用 propose_task_action 提出 kind=create_work_order、work_order_type=general 的现场核查／抢修草稿，引用定位证据、地点和管段；沿用已有 action_id，不重复建单。人工审核、派发按 Server 权限与版本完成。审批或现场处理等待期间，继续独立的爆管影响模拟和隔离分析，不等模拟完成才发起工单；模拟结果补入同一调查与后续报告，不能把模型计算成功写成已派单。
 
@@ -31,13 +31,13 @@ metadata:
 
 工具在筛选展示节点前比较全网节点的起算同刻在线模拟压力，换算米水头，三档互斥：1≤下降<3、3≤下降<5、下降≥5。只用同刻有效值，不以代表节点、邻近时刻或缺失值代替；保留缺测、单位来源、模型时态、比较节点数及覆盖缺口。范围表示爆管后、关阀或抢修前的模型估计，单独于隔离失源范围。
 
-读取计算结果后按 data.presentation.command 的 evidence_id 用 publish_task_presentation 发布“爆管后、处置前的预计影响范围”，并发布三档节点计数、比较时刻和假设表；无地图几何时说明原因。证据、地图及表格进入正式任务事件，历史播放还原已保存结果，不重新查询在线基线。先保留此步骤，再展示隔离方案，不能用后续隔离覆盖替换本步。
+读取计算结果后按 data.presentation.command 的 evidence_id 用 publish_task_presentation 发布“爆管后、处置前的预计影响范围”，并发布三档节点计数、比较时刻和假设表；无地图几何时说明原因。证据、地图及表格进入正式任务事件，历史播放还原已保存结果，不重新查询在线基线。本步 narration.stage 用 event_impact。三级结果 unavailable 时单列原因和待补项；先保留此步骤，再展示隔离方案，不得用关阀覆盖替代。
 
 ## 4. 隔离方案
 
 对关联资产调用 analyze_valve_isolation，保留 isolatable、must_close_valves、optional_valves、affected_nodes、supply_affected_nodes 及计数。单个候选分别分析；联合事故按真实联合场景提交。已知失效阀门通过 disabled_valves 评估替代边界；现场可操作性列为执行前核验条件。
 
-展示方案阀门和直接隔离、供水失源两类节点；节点覆盖使用 generate_node_coverage，再通过 publish_task_presentation 发布地图请求。取得初步隔离方案后继续影响分析和处置建议。
+本步 narration.stage 用 isolation_plan，覆盖标题明确“关阀方案影响”。展示方案阀门和直接隔离、供水失源两类节点；节点覆盖使用 generate_node_coverage，再通过 publish_task_presentation 发布地图请求。取得初步隔离方案后继续影响分析和处置建议。
 
 ## 5. 影响说明
 

resources/work-order-skills/scada-anomaly/SKILL.md

@@ -2,7 +2,7 @@
 name: scada-anomaly
 description: 从 SCADA 指标异常开展数据核验、爆管诊断、候选定位与影响分析。
 metadata:
-  version: "1.0.19"
+  version: "1.0.20"
 ---
 
 # SCADA 数据异常调查
@@ -31,9 +31,9 @@ metadata:
 
 取得候选即展示管道，发 interim/concluded 报告及现场工单草稿；复用action_id，待审批继续模拟。
 
-run_burst_simulation设include_impact_coverage=true，用定位管道、异常时刻及有依据的破口面积cm²。全网同刻在线压力差按米水头分档1≤下降<3、3≤下降<5、下降≥5。按data.presentation发布地图及计数，缺口单列；先事件影响，再隔离。回放仅用当前模型、禁实时SCADA；在线参考非冻结观测，历史展示复用保存证据。
+run_burst_simulation设include_impact_coverage=true，用定位管道、异常时刻及有依据的破口面积cm²。全网同刻在线压力差按米水头分档1≤下降<3、3≤下降<5、下降≥5。按data.presentation发布地图及计数，narration.stage=event_impact；不可用时单列原因，禁用关阀范围替代。回放仅用当前模型、禁实时SCADA；在线参考非冻结观测，历史展示复用保存证据。
 
-再按授权用 analyze_valve_isolation 分析隔离边界，保留必关阀、可选阀、affected_nodes、supply_affected_nodes 及计数；用 generate_node_coverage 展示节点覆盖。候选与拓扑建议标注为待现场核验方案。最终报告前核对定位调用及返回结果，调用失败记录具体错误与输入缺口。
+再按授权用 analyze_valve_isolation 分析隔离边界，保留必关阀、可选阀、affected_nodes、supply_affected_nodes 及计数；用 generate_node_coverage 展示关阀覆盖，stage=isolation_plan。候选与拓扑建议标注为待现场核验方案。爆管候选用locate_features传event_kind=burst。报告前核对定位结果，失败列明错误与缺口。
 
 ## 对比展示与影响
 

tests/unit/test_burst_impact.py

@@ -195,3 +195,11 @@ async def test_order_notification_precedes_durable_impact_and_keeps_run_availabl
     result.data["result"]["map_payload"]["features"].clear()
     assert len(repository.events[-1]["metadata"]["command"]["data"]["features"]) == 3
     assert repository.run["status"] == "running"
+
+
+@pytest.mark.parametrize("unit", ["MTR", "mtr", " MTR "])
+def test_native_mtr_pressure_generates_all_three_grades(inputs, unit):
+    result = service.estimate_burst_impact("own", START, simulation([1, 2.999, 3, 4.999, 5], unit))
+    assert result["impact"]["status"] == "ready"
+    assert [g["node_ids"] for g in result["impact"]["grades"]] == [["J0", "J1"], ["J2", "J3"], ["J4"]]
+    assert len(result["map_payload"]["features"]) == 3

tests/unit/test_operational_presentation.py

@@ -7,7 +7,7 @@ from pydantic import ValidationError
 from app.domain.domain_errors import DomainError
 from app.domain.schemas.operational_data import TaskDataSnapshot
 from app.domain.schemas.operational_task import TaskPresentationEventCommand, TaskSafeChart
-from app.services.operational_presentation import resolve_overlay
+from app.services.operational_presentation import resolve_overlay, validate_overlay_stage
 
 
 def _snapshot(data):
@@ -58,6 +58,7 @@ def test_presentation_input_rejects_free_form_or_invalid_commands(command):
 
 @pytest.mark.parametrize("command", [
     {"action": "locate_features", "layer": "scada", "feature_ids": ["D-1"]},
+    {"action": "locate_features", "layer": "pipes", "feature_ids": ["P-1"], "event_kind": "burst"},
     {"action": "zoom_to_map", "center": [121.4, 31.1], "zoom": 24},
     {"action": "render_scada_analysis", "items": [{"device_id": "D-1", "level": "unrated"}]},
     {"action": "clear_scada_analysis"},
@@ -110,3 +111,14 @@ def test_safe_chart_preserves_sparse_scatter_and_real_sample_times():
 def test_safe_chart_rejects_invalid_sample_metadata_and_values(series):
     with pytest.raises(ValidationError):
         TaskSafeChart(title="X", chart_type="line", x_data=["t"], series=[series])
+
+
+@pytest.mark.parametrize("phase", ["after_burst_before_intervention", "after_contamination_before_intervention"])
+def test_event_impact_requires_pre_intervention_geometry(phase):
+    data = {"features": [{"properties": {"impact_phase": phase}}]}
+    validate_overlay_stage(data, "event_impact")
+    closure = {"features": [{"properties": {"coverage_group": "supply_affected_nodes"}}]}
+    validate_overlay_stage(closure, "isolation_plan")
+    with pytest.raises(DomainError) as error:
+        validate_overlay_stage(closure, "event_impact")
+    assert error.value.code == "OVERLAY_IMPACT_PHASE_MISMATCH"

tests/unit/test_operational_tasks.py

@@ -580,6 +580,15 @@ async def test_saved_mcp_coverage_envelope_publishes_groups_without_agent_geomet
         }}),
     )
     evidence, _ = await service.record_data_snapshot(task["task_id"], snapshot)
+    with pytest.raises(DomainError) as error:
+        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": "wrong-impact", "title": "事件影响",
+                     "mode": "geometry", "evidence_id": evidence.evidence_id},
+            narration={"stage": "event_impact", "summary": "预计影响范围，现场待核实。",
+                       "evidence_refs": [evidence.evidence_id]},
+        ))
+    assert error.value.code == "OVERLAY_IMPACT_PHASE_MISMATCH"
     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": "节点覆盖示意",

tests/unit/test_recovery_device_scope.py

@@ -0,0 +1,48 @@
+from copy import deepcopy
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+from app.domain.schemas.project_settings import ProjectSettings
+from app.services.operational_recovery import reconcile_recoveries
+from app.services.operational_tasks import OperationalTaskService
+from tests.support.operational_task_repository import TaskRepository, _active_run, _context, _task_document
+
+
+@pytest.fixture
+def anyio_backend():
+    return "asyncio"
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("cached", [False, True])
+async def test_recovery_uses_report_devices_instead_of_all_browsed_network_devices(monkeypatch, cached):
+    now = datetime(2026, 7, 16, 6, tzinfo=timezone.utc)
+    incident = now - timedelta(hours=6)
+    returned = now - timedelta(minutes=20)
+    task = _task_document(_context(), now, lease=now + timedelta(minutes=1))
+    task.update(workflow="scada_anomaly", status="waiting_human", lifecycle_status="waiting_human",
+                current_report_revision=1, trigger={"kind": "event", "occurred_at": incident.isoformat(), "detail": "异常"},
+                handling_context={"phase": "waiting_scada", "returned_at": returned.isoformat()})
+    if cached:
+        task["handling_context"].update(device_ids=["P1", "unrelated"], interval_seconds=900)
+    repo = TaskRepository(task, _active_run(task, now))
+    repo.get_latest_report = AsyncMock(return_value={"revision": 1, "report": {"scada_device_ids": ["P1"]}})
+    repo.list_evidence = AsyncMock(return_value=[{"data_snapshot": {"source": "query_scada_readings",
+        "data": {"data": [{"device_id": "P1"}, {"device_id": "unrelated"}]}}}])
+    repo.recovery_candidates = AsyncMock(side_effect=lambda *_: [deepcopy(repo.task)]
+        if repo.task["handling_context"]["phase"] == "waiting_scada" else [])
+    service = OperationalTaskService(repo, _context(), set(), clock=lambda: now)
+    at = returned + timedelta(minutes=5)
+    rows = [{"time": at - timedelta(days=day), "device_id": "P1", "monitored_value": 100.}
+            for day in range(1, 16)] + [{"time": at, "device_id": "P1", "monitored_value": 100.}]
+    reader = AsyncMock(side_effect=lambda _, devices, *args, **kwargs: rows if devices == ["P1"] else [{}] * 100_001)
+    monkeypatch.setattr("app.infra.db.timescaledb.repositories.scada.ScadaRepository.get_recovery_scada_readings", reader)
+    settings = ProjectSettings(scada_monitoring_interval_seconds=900, scada_monitoring={"pressure_device_ids": ["P1"]})
+    assert await reconcile_recoveries(service, SimpleNamespace(transaction=repo.transaction), settings) == 1
+    assert repo.task["handling_context"]["device_ids"] == ["P1"]
+    assert repo.task["handling_context"]["phase"] == "verification_ready"
+    assert repo.run["status"] == "queued"
+    assert await reconcile_recoveries(service, SimpleNamespace(transaction=repo.transaction), settings) == 0

tests/unit/test_scada_repository.py

@@ -186,7 +186,9 @@ def test_recovery_read_selects_first_raw_frame_strictly_after_feedback_and_bound
     assert result == rows
     query, parameters = conn.cursor_instance.calls[0]
     assert "min(time)" in query and "time>%s AND time<=%s" in query
-    assert "time>=%s AND time<%s" in query and "JOIN next_frame" in query
+    assert "m.time>=%s AND m.time<%s" in query and "JOIN next_frame" in query
+    assert "mod(extract(epoch FROM m.time)::bigint,86400)=" in query
+    assert "mod(extract(epoch FROM f.time)::bigint,86400)" in query
     assert parameters == (["P1"], feedback, through, ["P1"], incident - timedelta(days=15), incident, 100_001, ["P1"])