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/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 "
+                "reference AS (SELECT m.time,m.device_id,m.monitored_value,m.cleaned_value "
-                "FROM scada.measurements WHERE device_id=ANY(%s) AND time>=%s AND time<%s "
+                "FROM scada.measurements m CROSS JOIN next_frame f "
-                "ORDER BY time,device_id LIMIT %s) "
+                "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/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
+            # Full-network browsing is discovery, not a declaration that every
-            # quality or local incident sensors outside the project role lists.
+            # device must recover for this incident. Prefer the report's scope.
-            for evidence in await service.repo.list_evidence(task["task_id"]):
+            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":

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，不依赖回放状态或推进回放时间。

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