fix(scada): verify recovery after replay work-order completion

app/infra/db/postgresql/operational_tasks.py

@@ -717,7 +717,6 @@ class OperationalTaskRepository:
                WHERE project_id=%s AND task_type='event_investigation'
                  AND status NOT IN ('cancelled','failed')
                  AND document#>>'{handling_context,phase}'='waiting_scada'
-                 AND document#>>'{source,replay_context}' IS NULL
                  AND (%s::timestamptz IS NULL OR created_at>=%s)
                ORDER BY COALESCE((document#>>'{handling_context,last_checked_at}')::timestamptz,created_at),task_id LIMIT %s""",
             (self.project_id, created_from, created_from, limit),

app/services/operational_recovery.py

@@ -20,8 +20,7 @@ RECOVERY_WORKFLOWS = {"burst_report", "scada_anomaly"}
 
 
 def track_work_order(task: dict, action: dict) -> None:
-    if (task.get("workflow") not in RECOVERY_WORKFLOWS or action["kind"] != "create_work_order"
-            or task.get("source", {}).get("replay_context")):
+    if task.get("workflow") not in RECOVERY_WORKFLOWS or action["kind"] != "create_work_order":
         return
     receipt = action.get("receipt") or {}
     if action.get("execution_status") != "succeeded" or receipt.get("adapter") != "peripheral_sample":
@@ -160,6 +159,9 @@ 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():
+            # Field completion authorizes a separate recovery observation even
+            # when the incident was found in a frozen replay. Keep that replay's
+            # query window and batch boundary unchanged.
             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 []

docs/burst-response-workflow.md

@@ -55,6 +55,10 @@ SCADA作为事件影响的辅助证据，按目标测点和事件／对照窗口
 
 `POST /operational-tasks/recovery/reconcile` 由现有 operational worker 每轮调用，按项目及任务创建边界处理等待复核的任务；SCADA 读取失败不阻断其他调查派发。复核直接读取项目时序库中完工反馈时间之后最近时刻的 SCADA，不依赖回放状态或推进回放时间。
 
+这条恢复链路同样跟踪带 `replay_context` 的调查工单。原异常快照的 `knowledge_cutoff` 和 `batch_sequence` 保留，普通回放查询继续受其限制；工单完工反馈单独触发 Server 的 `ScadaRepository.get_recovery_scada_readings()`，以实际完工业务时间为 T，读取严格晚于 T 的最近一帧。例如 2026-07-16 13:00 完工，15 分钟采样时复核 13:15，而非原异常截止的 08:00。Agent 从 `handling_context.verification_evidence_id` 读取保存的恢复证据，并与原异常证据共同引用，不需放开回放查询或重建复测工单。
+
+已有调查在收到新的有效完工反馈时会补登记成功派发的关联工单。旧回报误用了电脑当前时间时，需在来源工单更正现场反馈业务时间并提交新版本；系统不猜测完工时刻，也不改写原报告或旧证据。
+
 ## 验证
 
 检查Skill名称、版本、摘要与正文digest一致、内联体积、角色权限、工单工具隔离和SCADA定位回归。Agent类型检查与会话工具策略测试共同覆盖运行时入口。发布后分别验收工单事件处理报告和SCADA候选定位，原有报告按既有修订保留。

docs/scada-peripheral-replay-design.md

@@ -18,6 +18,8 @@
 
 Agent 查询本轮已检测的固定批次快照，受任务的 replay_id、knowledge_cutoff、batch_sequence 和设备集合约束。新回放首次监测以只读一致快照冻结开始前 15 天历史，负序号静默基线按最多 2,000 行分块暂存为 `clean_prepared`；全部读取完成后，在一个 BizDB 短事务中把批次转为 `detected` 并原子发布基线与初始化标记，历史查询和定位对照才能读取。
 
+关联现场工单完工后的恢复复核采用独立证据链：Server 以现场反馈业务时间 T 读取项目时序库中严格晚于 T 的最近一帧及事件前对照，保存恢复监测证据后接续原调查。此路径不依赖回放推进，也不修改原异常快照的知识上限或批次；Agent 读取保存的恢复证据，不通过普通回放查询越界取数。详见[恢复复核流程](burst-response-workflow.md)。
+
 读取各页期间不持有 BizDB 事务，内存只保留当前页和设备短历史。固定准备预算为 1,000,000 行、512 MiB 序列化批次，覆盖 500 设备十五天每 15 分钟一条所需的 720,000 行并留有余量。超限不保存超限页，状态明确返回 `BASELINE_BUDGET_EXCEEDED` 及中文原因；未发布批次不进入观测、监测或待处理批次。失败重试只清理当前项目、本轮回放的未发布基线，从新的一致源快照重新准备，已发布基线保持不变。
 
 旧回放没有冻结基线时明确拒绝回放前查询，不回落到实时 TSDB。爆管定位支持受回放截止约束的冻结输入适配，其他未适配计算工具继续拒绝。[SCADA 调查 Skill](../resources/work-order-skills/scada-anomaly/SKILL.md) 要求据实说明证据限制。运行异常自动建立调查，数据质量异常仅建立告警；恢复时推进调查代次。真实模型推理与报告效果尚未联调。

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

@@ -2,7 +2,7 @@
 name: scada-anomaly
 description: 从 SCADA 指标异常开展数据核验、爆管诊断、候选定位与影响分析。
 metadata:
-  version: "1.0.12"
+  version: "1.0.13"
 ---
 
 # SCADA 数据异常调查
@@ -44,3 +44,9 @@ metadata:
 报告给出观测事实、规则信号、原因候选、定位与隔离结果、影响说明和建议。摘要以150字以内为目标，测点明细用表格、完整依据留在证据。候选、相似度和拓扑结果具备时，完成相应分析、展示与报告后再收尾；现场核验、审批和维修列为后续待办。现场工单与设备操作遵守服务端动作权限，本轮分析完成与现场事件闭环分别说明。
 
 需要现场核验时，先发布本轮outcome=concluded的调查报告，再调用propose_task_action提出kind=create_work_order、work_order_type=general的关联预约工单草稿，引用候选与影响证据，写明核验对象、内容和回执要求。带replay_context的任务同样可提出草稿，描述中注明历史异常时刻及回放来源；草稿须由有权限的用户编辑、确认后发布，后续派发与反馈进入预约任务。草稿待确认时说明等待人工处理，不将调查报告完成写成现场处置完成。回放任务不支持设备命令。
+
+## 完工后的恢复复核
+
+收到关联工单的completed反馈后，先读取get_task_context摘要中的handling_context，核对实际完工业务时间returned_at，不以工单创建、预约或电脑当前时间代替。Server通过现有恢复监测读取项目时序库中严格晚于returned_at的最近一帧及事件前15天同相位对照，独立保存恢复证据；例如13:00完工、15分钟采样时核对13:15。这条恢复链路适用于回放来源工单，不要求推进回放，原异常knowledge_cutoff与batch_sequence保持不变。
+
+phase=waiting_work_order时等待现场完工；phase=waiting_scada时查看last_check，分别说明尚无后续采样、基线或测点缺项、实际仍异常，等待监测复核，不将原回放查询越界解释为恢复数据不存在。phase=verification_ready时，用get_task_evidence读取verification_evidence_id的完整恢复证据，比较实际样本、正常范围与现场回报，最终报告同时引用原异常全量证据与当前恢复证据。恢复数据已由Server保存，不用query_scada_readings越过原冻结上限重复查询，也不因该上限再建复测工单；确有独立现场工作时才提议新动作。完工时间修订后使用新返回的证据编号，旧证据仅作历史。正常读数不替代阀位、漏点照片等现场事实，报告分别说明监测恢复、现场待核验项与事件闭环。

tests/unit/test_operational_recovery.py

@@ -97,8 +97,13 @@ def test_off_grid_feedback_uses_first_later_sample_even_without_monitoring_inter
     assert result["verified_through"] == (RETURNED + timedelta(minutes=15)).isoformat()
 
 
-def test_dispatch_waits_for_all_orders_and_accepts_corrected_field_completion_time():
+@pytest.mark.parametrize("replay", [False, True])
+def test_dispatch_waits_for_all_orders_and_accepts_corrected_field_completion_time(replay):
     task = {"workflow": "burst_report"}
+    if replay:
+        task["source"] = {"replay_context": {
+            "replay_id": str(uuid4()), "knowledge_cutoff": INCIDENT.isoformat(), "batch_sequence": 3,
+        }}
     for order_id in ("one", "two"):
         track_work_order(task, {"kind": "create_work_order", "action_id": str(uuid4()), "execution_status": "succeeded",
                                 "receipt": {"adapter": "peripheral_sample", "receipt_id": order_id}})
@@ -120,6 +125,9 @@ def test_dispatch_waits_for_all_orders_and_accepts_corrected_field_completion_ti
     assert task["handling_context"]["returned_at"] == corrected.isoformat()
     assert task["handling_context"]["phase"] == "waiting_scada"
     assert "verification_evidence_id" not in task["handling_context"] and "last_check" not in task["handling_context"]
+    if replay:
+        assert task["source"]["replay_context"]["knowledge_cutoff"] == INCIDENT.isoformat()
+        assert task["source"]["replay_context"]["batch_sequence"] == 3
 
 
 def test_field_feedback_time_requires_a_real_note_and_explicit_timezone():

tests/unit/test_operational_tasks.py

@@ -1128,21 +1128,39 @@ def _seed_concluded_report(repository, now):
 
 
 @pytest.mark.anyio
-async def test_burst_waits_for_completion_then_scada_evidence_then_agent_report(monkeypatch):
+@pytest.mark.parametrize("replay", [False, True])
+async def test_burst_waits_for_completion_then_scada_evidence_then_agent_report(monkeypatch, replay):
     from copy import deepcopy
     from app.domain.schemas.project_settings import ProjectSettings
     from app.services.operational_recovery import reconcile_recoveries, track_work_order
 
-    now = datetime(2026, 7, 16, 4, tzinfo=timezone.utc)
-    incident = now - timedelta(hours=4)
-    returned = incident + timedelta(hours=1)
+    now = datetime(2026, 7, 16, 6, tzinfo=timezone.utc)
+    incident = now - timedelta(hours=6)
+    returned = datetime.fromisoformat("2026-07-16T13:00:00+08:00")
     context = _context()
     task = _task_document(context, now, lease=now + timedelta(minutes=1))
     task.update(workflow="burst_report", trigger={"kind": "event", "occurred_at": incident.isoformat(), "detail": "爆管"})
+    if replay:
+        task["workflow"] = "scada_anomaly"
+        task["source"]["replay_context"] = {
+            "replay_id": str(uuid4()), "knowledge_cutoff": incident.isoformat(), "batch_sequence": 3,
+        }
     repository = RunTaskRepository(task, _active_run(task, now))
     repository.connection = object()
     service = OperationalTaskService(repository, context, set(), clock=lambda: now)
     monkeypatch.setattr(service, "_resolve_investigation_alerts", AsyncMock())
+    incident_evidence_id = None
+    if replay:
+        frozen = task["source"]["replay_context"]
+        evidence, _ = await service.record_data_snapshot(task["task_id"], TaskDataSnapshot(
+            source="query_scada_readings", source_ref="mcp:frozen-incident", queried_at=now,
+            origin="synthetic", truncated=False,
+            data={"data": {"devices": [{"device_id": "P1"}], "readings": [], "coverage": {
+                "mode": "all_scada", "query_complete": True,
+                "replay_id": frozen["replay_id"], "batch_sequence": 3,
+            }}, "window": {"start": (incident - timedelta(hours=3)).isoformat(), "end": incident.isoformat()}},
+        ))
+        incident_evidence_id = evidence.evidence_id
     _seed_concluded_report(repository, now)
     action = await service.propose_action(task["task_id"], ActionProposalCommand(
         request_id=uuid4(), kind="create_work_order", reason="隔离抢修与恢复供水",
@@ -1178,6 +1196,11 @@ async def test_burst_waits_for_completion_then_scada_evidence_then_agent_report(
     reader.return_value = rows
     assert await reconcile_recoveries(service, timescale, settings) == 1
     assert await reconcile_recoveries(service, timescale, settings) == 0
+    reader.assert_awaited_with(timescale, ["P1"], incident, returned, now, baseline_limit=100_001)
+    if replay:
+        assert repository.task["source"]["replay_context"]["replay_id"] == task["source"]["replay_context"]["replay_id"]
+        assert datetime.fromisoformat(repository.task["source"]["replay_context"]["knowledge_cutoff"]) == incident
+        assert repository.task["source"]["replay_context"]["batch_sequence"] == 3
     recovery = repository.task["handling_context"]
     evidence_id = recovery["verification_evidence_id"]
     assert repository.evidence[evidence_id]["data_snapshot"]["data"]["status"] == "passed"
@@ -1208,7 +1231,14 @@ async def test_burst_waits_for_completion_then_scada_evidence_then_agent_report(
         await service.complete_run(task["task_id"], repository.run["run_id"], AgentRunCommand(
             request_id=uuid4(), expected_revision=repository.run["revision"], source_session_id="session-1"))
     assert rejected.value.code == "RECOVERY_EVIDENCE_REQUIRED"
-    repository.reports[-1]["report"]["evidence_refs"] = [evidence_id]
+    if replay:
+        await service.publish_artifacts(task["task_id"], AgentArtifactsCommand(
+            request_id=uuid4(), source_session_id="session-1",
+            report={**_current_report(), "evidence_refs": [incident_evidence_id, evidence_id]},
+        ))
+        assert repository.reports[-1]["report"]["evidence_refs"] == [incident_evidence_id, evidence_id]
+    else:
+        repository.reports[-1]["report"]["evidence_refs"] = [evidence_id]
     result = await service.complete_run(task["task_id"], repository.run["run_id"], AgentRunCommand(
         request_id=uuid4(), expected_revision=repository.run["revision"], source_session_id="session-1"))
     assert result.status == "completed" and result.handling_context["phase"] == "completed"