diff --git a/app/infra/db/timescaledb/repositories/scada.py b/app/infra/db/timescaledb/repositories/scada.py index 2b42820..390bc97 100644 --- a/app/infra/db/timescaledb/repositories/scada.py +++ b/app/infra/db/timescaledb/repositories/scada.py @@ -1,5 +1,5 @@ from typing import List, Any -from datetime import datetime +from datetime import datetime, timedelta from collections import defaultdict from psycopg import AsyncConnection, Connection, sql from psycopg.rows import dict_row @@ -131,6 +131,29 @@ class ScadaRepository: ) return await cur.fetchall() + @staticmethod + async def get_recovery_scada_readings( + conn: AsyncConnection, device_ids: list[str], incident_at: datetime, + returned_at: datetime, through: datetime, *, baseline_limit: int = 100_001, + ) -> list[dict]: + """Read the causal reference and exactly the first stored frame after T.""" + if not device_ids or through <= returned_at: + return [] + async with conn.cursor() as cur: + 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) " + "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", + (device_ids, returned_at, through, device_ids, incident_at - timedelta(days=15), + incident_at, baseline_limit, device_ids), + ) + return [dict(row) for row in await cur.fetchall()] + @staticmethod async def get_latest_scada_readings( conn: AsyncConnection, diff --git a/app/services/operational_recovery.py b/app/services/operational_recovery.py index 7fb7751..27d6145 100644 --- a/app/services/operational_recovery.py +++ b/app/services/operational_recovery.py @@ -65,16 +65,12 @@ def accept_completion(task: dict, feedback: dict) -> bool: def evaluate_recovery(rows: list[dict], *, device_ids: list[str], returned_at: datetime, incident_at: datetime, interval_seconds: int) -> dict: - """Require two fresh complete samples inside the pre-event normal envelope. - - Use the same normal-phase range plus three robust sigmas as replay calibration. - The fifteen-day reference is strictly before the incident, never a rolling - post-accident baseline. Only finite, usable observations can prove recovery. - """ + """Verify the first SCADA sample strictly after the field feedback time T.""" base = {"status": "waiting", "device_ids": device_ids, "returned_at": returned_at.isoformat(), - "required_samples": 2, "interval_seconds": interval_seconds, + "required_samples": 1, "interval_seconds": interval_seconds, + "selection_method": "first_sample_after_feedback", "baseline_method": "normal_same_phase_range_plus_3_robust_sigma"} - if not device_ids or interval_seconds <= 0 or 86400 % interval_seconds: + if not device_ids: return {**base, "reason": "monitoring_configuration_missing"} if returned_at < incident_at: return {**base, "reason": "completion_precedes_incident"} @@ -86,6 +82,10 @@ def evaluate_recovery(rows: list[dict], *, device_ids: list[str], returned_at: d if row["device_id"] not in selected: continue at = row["time"] if isinstance(row["time"], datetime) else datetime.fromisoformat(row["time"]) + if at > returned_at: + # Keep the nearest frame even when it contains null/invalid values; + # never skip it in order to find a later normal-looking sample. + observations.setdefault(at, {}) if row.get("quality") not in {None, "valid", "corrected"}: continue value = row.get("cleaned_value") @@ -94,46 +94,33 @@ def evaluate_recovery(rows: list[dict], *, device_ids: list[str], returned_at: d if value is None or not isfinite(float(value)): continue value = float(value) - # Preserve the source sampling phase even for an off-grid event/receipt. phase = int(at.timestamp()) % 86400 if origin <= at < incident_at: baseline.setdefault((row["device_id"], phase), []).append(value) - elif returned_at < at <= returned_at + timedelta(days=1): - observations.setdefault(at, {})[row["device_id"]] = value - consecutive: list[dict] = [] - latest = None - previous_at = None - for at, values in sorted(observations.items()): - phase = int(at.timestamp()) % 86400 - checks = [] - for device in device_ids: - reference = baseline.get((device, phase), []) - if device not in values or len(reference) < 15: - checks.append({"device_id": device, "status": "missing_data"}) - continue - center = median(reference) - sigma = 1.4826 * median(abs(value - center) for value in reference) - margin = max(3 * sigma, abs(center) * 1e-6, 1e-9) - lower, upper = min(reference) - margin, max(reference) + margin - checks.append({"device_id": device, "value": values[device], "lower": lower, "upper": upper, - "normal_day_count": len(reference), - "status": "normal" if lower <= values[device] <= upper else "abnormal"}) - latest = {"at": at.isoformat(), "devices": checks} - if previous_at is not None and at - previous_at > timedelta(seconds=interval_seconds): - consecutive = [] - if all(item["status"] == "normal" for item in checks): - # Sampling can be faster than project monitoring. Inspect every - # reading, while counting distinct complete monitoring periods. - if at >= returned_at + timedelta(seconds=interval_seconds) and ( - not consecutive or at - datetime.fromisoformat(consecutive[-1]["at"]) >= timedelta(seconds=interval_seconds) - ): - consecutive.append(latest) - else: - consecutive = [] - previous_at = at - if len(consecutive) >= 2: - return {**base, "status": "passed", "verified_through": consecutive[-1]["at"], "samples": consecutive[-2:]} - return {**base, "reason": "post_completion_data_not_normal_or_incomplete", "latest_sample": latest} + elif at > returned_at: + observations[at][row["device_id"]] = value + if not observations: + return {**base, "reason": "post_completion_reading_missing"} + at = min(observations) + values = observations[at] + phase = int(at.timestamp()) % 86400 + checks = [] + for device in device_ids: + reference = baseline.get((device, phase), []) + if device not in values or len(reference) < 15: + checks.append({"device_id": device, "status": "missing_data"}) + continue + center = median(reference) + sigma = 1.4826 * median(abs(value - center) for value in reference) + margin = max(3 * sigma, abs(center) * 1e-6, 1e-9) + lower, upper = min(reference) - margin, max(reference) + margin + checks.append({"device_id": device, "value": values[device], "lower": lower, "upper": upper, + "normal_day_count": len(reference), + "status": "normal" if lower <= values[device] <= upper else "abnormal"}) + sample = {"at": at.isoformat(), "devices": checks} + if all(item["status"] == "normal" for item in checks): + return {**base, "status": "passed", "verified_through": at.isoformat(), "samples": [sample]} + return {**base, "reason": "nearest_post_completion_reading_not_normal_or_incomplete", "sample": sample} async def reconcile_recoveries(service: OperationalTaskService, timescale: AsyncConnection, @@ -155,7 +142,7 @@ async def reconcile_recoveries(service: OperationalTaskService, timescale: Async roles = settings.scada_monitoring devices = list(dict.fromkeys(roles.pressure_device_ids + roles.outlet_flow_device_ids)) if roles else [] source = "scada_readings" - through = min(service._now(), returned_at + timedelta(days=1)) + through = service._now() if context.get("device_ids"): devices = context["device_ids"] interval = context.get("interval_seconds", interval) @@ -173,8 +160,8 @@ async def reconcile_recoveries(service: OperationalTaskService, timescale: Async devices.extend(row["device_id"] for row in data if isinstance(row, dict) and isinstance(row.get("device_id"), str)) devices = list(dict.fromkeys(devices)) async with timescale.transaction(): - rows = await ScadaRepository.get_scada_by_ids_time_range( - timescale, devices, incident_at - timedelta(days=15), through, row_limit=100_001, + rows = await ScadaRepository.get_recovery_scada_readings( + timescale, devices, incident_at, returned_at, through, baseline_limit=100_001, ) if through > returned_at and devices else [] result = ({"status": "waiting", "reason": "reading_budget_exceeded"} if len(rows) > 100_000 else evaluate_recovery(rows, device_ids=devices, returned_at=returned_at, @@ -186,15 +173,15 @@ async def reconcile_recoveries(service: OperationalTaskService, timescale: Async if current.get("handling_context") != context or current["status"] in {"cancelled", "failed"}: continue now = service._now() - if devices and interval: - current["handling_context"].update(device_ids=devices, interval_seconds=interval) + if devices: + current["handling_context"].update(device_ids=devices, interval_seconds=interval or 0) current["handling_context"]["last_checked_at"] = now.isoformat() current["handling_context"]["last_check"] = result if result["status"] == "passed": evidence_id = "recovery:" + digest({"task_id": task["task_id"], "returned_at": context["returned_at"], "result": result}) evidence = StoredEvidenceRecord( evidence_id=evidence_id, kind="system_data", - summary="完工反馈后连续两个采样周期的 SCADA 恢复正常，等待 Agent 完成复核报告", + summary="完工反馈时间之后最近时刻的 SCADA 已恢复正常，等待 Agent 完成复核报告", source_label="恢复供水监测复核", source_ref=f"operational-task:{task['task_id']}", observed_at=datetime.fromisoformat(result["verified_through"]), recorded_at=now, recorded_by=service._actor(), has_snapshot=True, diff --git a/docs/burst-response-workflow.md b/docs/burst-response-workflow.md index ae4a087..44d8ef4 100644 --- a/docs/burst-response-workflow.md +++ b/docs/burst-response-workflow.md @@ -23,7 +23,7 @@ flowchart TD 3. **隔离方案**：用 analyze_valve_isolation 分析关联资产，保留必关阀、可选阀、直接隔离与供水失源节点；已知失效阀门用于替代方案分析，现场可操作性列为执行条件。 4. **影响说明**：先展示拓扑范围，再按可用SCADA、水力结果和用户关联资料解释事件当前影响、方案预计影响和保障需求。 5. **报告与处置**：保存阶段报告后提出核验、通知、保障、抢修和恢复等动作草稿；关联现场任务采用 general 工单。审批、派发和现场完成分别核对回执。 -6. **反馈复核**：已派发的关联工单保持 `waiting_work_order`，全部收到有效 `completed` 反馈后进入 `waiting_scada`。以现场反馈业务时间为起点，核验之后连续两个完整采样周期；正常范围使用事件前15天同采样相位的正常包络。数据未到或仍异常时保持等待，后台工作进程定期复核；通过后持久保存恢复证据并接续 Agent。Agent 最终报告必须引用该证据，运行完成后才能完成任务并解除对应告警。原事件由有权限人员在来源系统确认行政结案。 +6. **反馈复核**：已派发的关联工单保持 `waiting_work_order`，全部收到有效 `completed` 反馈后进入 `waiting_scada`。以现场反馈业务时间为 T，取严格晚于 T 的最近一个 SCADA 采样时刻进行复核；正常范围使用事件前15天同采样相位的正常包络。数据未到或仍异常时保持等待，后台工作进程定期复核；通过后持久保存恢复证据并接续 Agent。Agent 最终报告必须引用该证据，运行完成后才能完成任务并解除对应告警。原事件由有权限人员在来源系统确认行政结案。 SCADA作为事件影响的辅助证据，按目标测点和事件／对照窗口读取。具备水力输入且工具已授权时，使用 analyze_valve_closure 评估给定阀门方案；各步骤记录实际结果、适用范围或具体待补项。 @@ -51,9 +51,9 @@ SCADA作为事件影响的辅助证据，按目标测点和事件／对照窗口 业务阶段和工单、返回时间、复核回执保存在任务文档 `handling_context` JSON 中，不增加数据库列。外围样板更新工单时可提供带时区的 `occurred_at`，同时必须填写现场反馈；该时间写入活动记录，系统的 `updated_at` 仍是实际记录时间。同步器将现场反馈时间传给调查，避免历史演示误用电脑当前时间。 -预约时间 `scheduled_at` 可在派单前的工单草稿和已创建工单的编辑表单中调整。实际完工反馈时间与预约安排分别记录；Agent 以实际反馈时间为恢复复核起点。新的有效现场反馈修订完工时间后，旧复核证据保留为历史记录，当前复核起点更新并重新等待新时间之后的两个正常监测周期。只修改预约安排或其他工单信息，不改写已有现场反馈时间。 +预约时间 `scheduled_at` 可在派单前的工单草稿和已创建工单的编辑表单中调整。实际完工反馈时间与预约安排分别记录；Agent 以实际反馈时间为恢复复核起点。新的有效现场反馈修订完工时间后，旧复核证据保留为历史记录，当前复核起点更新，重新读取严格晚于新反馈时间的最近一个 SCADA 采样时刻。只修改预约安排或其他工单信息，不改写已有现场反馈时间。 -`POST /operational-tasks/recovery/reconcile` 由现有 operational worker 每轮调用，按项目及任务创建边界处理等待复核的任务；SCADA 读取失败不阻断其他调查派发。复核直接读取项目时序库中完工反馈时间之后的 SCADA，不依赖回放状态或推进回放时间。 +`POST /operational-tasks/recovery/reconcile` 由现有 operational worker 每轮调用，按项目及任务创建边界处理等待复核的任务；SCADA 读取失败不阻断其他调查派发。复核直接读取项目时序库中完工反馈时间之后最近时刻的 SCADA，不依赖回放状态或推进回放时间。 ## 验证 diff --git a/resources/work-order-skills/burst-event-handling/SKILL.md b/resources/work-order-skills/burst-event-handling/SKILL.md index 93b297f..557191f 100644 --- a/resources/work-order-skills/burst-event-handling/SKILL.md +++ b/resources/work-order-skills/burst-event-handling/SKILL.md @@ -2,7 +2,7 @@ name: burst-event-handling description: 根据爆管上报工单处理事件，完成资产关联、隔离方案、影响说明、处置建议与反馈复核。 metadata: - version: "1.1.0" + version: "1.1.1" --- # 爆管事件处理 @@ -48,6 +48,6 @@ metadata: 读取摘要中的 handling_context。waiting_work_order 表示已派单、等待有效完工反馈，创建回执的 completed_at 只证明创建动作完成。现场处理和恢复供水以关联工单的 completed 反馈为准；returned_at 是现场反馈的业务时间，演示时不得换成电脑当前时间。 -waiting_scada 表示恢复供水和监测复核阶段，查询 returned_at 之后的实际 SCADA，说明目标测点、时间、单位和正常对照。后台会按事件前15天的同期正常包络核验连续两个完整采样周期；缺测或仍异常时保持待复核，不宣称处置闭环。保存本轮阶段成果，等待后续数据接续，不反复派发同一工单。 +waiting_scada 表示恢复供水和监测复核阶段，以 returned_at 为反馈时刻 T，读取严格晚于 T 的最近一个 SCADA 采样时刻作为复核来源，说明测点、实际采样时间、读数、单位和正常范围；不等待额外周期，不使用 T 之前、T 当刻或更晚的正常数据替代。后台按事件前15天同期正常包络核验该时刻。最近时刻缺测或异常时保留实际问题，不宣称复核成功。保存本轮阶段成果，等待后续数据接续，不反复派发同一工单。 -verification_ready 时读取 verification_evidence_id 对应的证据与快照，核对工单反馈、返回时间、测点及两个正常采样周期，展示恢复曲线，发布 final/concluded 报告并引用该证据。运行完成后 Server 才完成任务并解除关联告警。原事件在来源系统的行政结案仍由有权限人员确认。 +verification_ready 时读取 verification_evidence_id 对应的证据与快照，核对工单反馈、T、最近采样时刻和测点，展示该时刻读数与正常范围；具备真实多时刻数据时再展示曲线，发布 final/concluded 报告并引用该证据。运行完成后 Server 才完成任务并解除关联告警。原事件在来源系统的行政结案仍由有权限人员确认。 diff --git a/tests/unit/test_operational_recovery.py b/tests/unit/test_operational_recovery.py index 4189d97..61ce9f9 100644 --- a/tests/unit/test_operational_recovery.py +++ b/tests/unit/test_operational_recovery.py @@ -27,11 +27,11 @@ def evaluate(rows): incident_at=INCIDENT, interval_seconds=900) -def test_recovery_uses_post_receipt_samples_and_pre_incident_same_phase_reference(): +def test_recovery_uses_nearest_post_receipt_sample_and_pre_incident_reference(): result = evaluate(observations()) assert result["status"] == "passed" - assert [item["at"] for item in result["samples"]] == [ - (RETURNED + timedelta(minutes=30)).isoformat(), (RETURNED + timedelta(minutes=45)).isoformat()] + assert result["required_samples"] == 1 + assert [item["at"] for item in result["samples"]] == [(RETURNED + timedelta(minutes=15)).isoformat()] assert all(check["normal_day_count"] == 15 for sample in result["samples"] for check in sample["devices"]) @@ -59,7 +59,7 @@ def test_samples_at_or_before_return_time_cannot_close_incident(): assert evaluate(rows)["status"] == "waiting" -def test_faster_sampling_counts_monitoring_periods_and_rejects_later_relapse(): +def test_nearest_sample_does_not_wait_for_monitoring_interval_or_use_later_data(): rows = [] for minute in range(5, 36, 5): at = RETURNED + timedelta(minutes=minute) @@ -67,9 +67,34 @@ def test_faster_sampling_counts_monitoring_periods_and_rejects_later_relapse(): rows.extend({"time": at - timedelta(days=day), "device_id": device, "monitored_value": 100.} for day in range(1, 16)) rows.append({"time": at, "device_id": device, "monitored_value": 100.}) - assert evaluate(rows)["status"] == "passed" + first = evaluate(rows) + assert first["status"] == "passed" + assert first["verified_through"] == (RETURNED + timedelta(minutes=5)).isoformat() next(row for row in rows if row["time"] == RETURNED + timedelta(minutes=25) and row["device_id"] == "P1")["monitored_value"] = 50. - assert evaluate(rows)["status"] == "waiting" + assert evaluate(rows) == first + + +@pytest.mark.parametrize("problem", ["abnormal", "invalid", "null", "missing_device"]) +def test_nearest_bad_sample_is_not_replaced_by_later_normal_samples(problem): + rows = observations() + nearest = RETURNED + timedelta(minutes=15) + row = next(row for row in rows if row["time"] == nearest and row["device_id"] == "P1") + if problem == "missing_device": + rows.remove(row) + elif problem == "invalid": + row["quality"] = "invalid" + else: + row["monitored_value"] = 50. if problem == "abnormal" else None + result = evaluate(rows) + assert result["status"] == "waiting" + assert result["sample"]["at"] == nearest.isoformat() + + +def test_off_grid_feedback_uses_first_later_sample_even_without_monitoring_interval(): + result = evaluate_recovery(observations(), device_ids=["P1", "F1"], + returned_at=RETURNED + timedelta(minutes=7), incident_at=INCIDENT, interval_seconds=0) + assert result["status"] == "passed" + assert result["verified_through"] == (RETURNED + timedelta(minutes=15)).isoformat() def test_dispatch_waits_for_all_orders_and_accepts_corrected_field_completion_time(): diff --git a/tests/unit/test_operational_tasks.py b/tests/unit/test_operational_tasks.py index 6400ae2..a1a0ac8 100644 --- a/tests/unit/test_operational_tasks.py +++ b/tests/unit/test_operational_tasks.py @@ -1128,8 +1128,8 @@ async def test_burst_waits_for_completion_then_scada_evidence_then_agent_report( at = returned + timedelta(minutes=15 * sample) rows.extend({"time": at - timedelta(days=day), "device_id": "P1", "monitored_value": 100.} for day in range(1, 16)) rows.append({"time": at, "device_id": "P1", "monitored_value": 100.}) - reader = AsyncMock(return_value=rows[:-1]) - monkeypatch.setattr("app.infra.db.timescaledb.repositories.scada.ScadaRepository.get_scada_by_ids_time_range", reader) + reader = AsyncMock(return_value=[row for row in rows if row["time"] < incident]) + monkeypatch.setattr("app.infra.db.timescaledb.repositories.scada.ScadaRepository.get_recovery_scada_readings", reader) assert await reconcile_recoveries(service, timescale, settings) == 0 assert repository.task["handling_context"]["phase"] == "waiting_scada" reader.return_value = rows @@ -1138,23 +1138,24 @@ async def test_burst_waits_for_completion_then_scada_evidence_then_agent_report( recovery = repository.task["handling_context"] evidence_id = recovery["verification_evidence_id"] assert repository.evidence[evidence_id]["data_snapshot"]["data"]["status"] == "passed" + assert repository.evidence[evidence_id]["data_snapshot"]["data"]["verified_through"] == (returned + timedelta(minutes=15)).isoformat() service._resolve_investigation_alerts.assert_not_awaited() # Correcting the appointment's field return time invalidates the old proof - # and requires two actual observations after the new business time. + # and selects the nearest actual observation after the new business time. corrected_time = returned + timedelta(minutes=15) await service.external_feedback(task["task_id"], ExternalFeedbackCommand( request_id=uuid4(), source_event_id=f"peripheral_sample:{child_id}", source_revision=3, status="completed", occurred_at=corrected_time, detail="修订现场实际完工时间")) assert repository.task["handling_context"]["phase"] == "waiting_scada" assert "verification_evidence_id" not in repository.task["handling_context"] + reader.return_value = [row for row in rows if row["time"] <= corrected_time] assert await reconcile_recoveries(service, timescale, settings) == 0 - at = returned + timedelta(minutes=45) - rows.extend({"time": at - timedelta(days=day), "device_id": "P1", "monitored_value": 100.} for day in range(1, 16)) - rows.append({"time": at, "device_id": "P1", "monitored_value": 100.}) + reader.return_value = rows assert await reconcile_recoveries(service, timescale, settings) == 1 revised_id = repository.task["handling_context"]["verification_evidence_id"] assert revised_id != evidence_id and evidence_id in repository.evidence + assert repository.evidence[revised_id]["data_snapshot"]["data"]["verified_through"] == (returned + timedelta(minutes=30)).isoformat() evidence_id = revised_id repository.task.update(status="running", agent_session_id="session-1") diff --git a/tests/unit/test_scada_repository.py b/tests/unit/test_scada_repository.py index 24a8119..55c4db9 100644 --- a/tests/unit/test_scada_repository.py +++ b/tests/unit/test_scada_repository.py @@ -1,5 +1,5 @@ import asyncio -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone import importlib.util from pathlib import Path @@ -172,3 +172,27 @@ def test_get_latest_scada_readings_skips_database_for_empty_page(): assert asyncio.run(ScadaRepository.get_latest_scada_readings(conn, [])) == {} assert conn.cursor_instance.calls == [] + + +def test_recovery_read_selects_first_raw_frame_strictly_after_feedback_and_bounded_reference(): + ScadaRepository = _load_scada_repository() + incident = datetime(2026, 7, 16, 0, 0, tzinfo=timezone.utc) + feedback = incident + timedelta(hours=5, minutes=7) + through = feedback + timedelta(hours=2) + rows = [{"time": incident - timedelta(days=1), "device_id": "P1", "monitored_value": 100., "cleaned_value": None}, + {"time": incident + timedelta(hours=5, minutes=15), "device_id": "P1", "monitored_value": 100., "cleaned_value": None}] + conn = _ReadingConnection(rows) + result = asyncio.run(ScadaRepository.get_recovery_scada_readings(conn, ["P1"], incident, feedback, through)) + 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 parameters == (["P1"], feedback, through, ["P1"], incident - timedelta(days=15), incident, 100_001, ["P1"]) + + +def test_recovery_read_does_not_query_when_feedback_is_not_in_the_past(): + ScadaRepository = _load_scada_repository() + at = datetime(2026, 7, 16, tzinfo=timezone.utc) + conn = _ReadingConnection([]) + assert asyncio.run(ScadaRepository.get_recovery_scada_readings(conn, ["P1"], at, at, at)) == [] + assert conn.cursor_instance.calls == []