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 "
-                "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/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":