diff --git a/app/infra/db/timescaledb/repositories/scada.py b/app/infra/db/timescaledb/repositories/scada.py index 390bc97..d0b008e 100644 --- a/app/infra/db/timescaledb/repositories/scada.py +++ b/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", diff --git a/app/services/operational_recovery.py b/app/services/operational_recovery.py index 759e801..d0fa075 100644 --- a/app/services/operational_recovery.py +++ b/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": diff --git a/docs/burst-response-workflow.md b/docs/burst-response-workflow.md index fdb7ea0..63da10c 100644 --- a/docs/burst-response-workflow.md +++ b/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，不依赖回放状态或推进回放时间。 diff --git a/tests/unit/test_recovery_device_scope.py b/tests/unit/test_recovery_device_scope.py new file mode 100644 index 0000000..a303cf5 --- /dev/null +++ b/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 diff --git a/tests/unit/test_scada_repository.py b/tests/unit/test_scada_repository.py index 55c4db9..5e6b824 100644 --- a/tests/unit/test_scada_repository.py +++ b/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"])