fix(skills): close recovered incidents without optional evidence gates

A completed field order and passed Server recovery evidence support the final closure report. Treat absent photos, valve records and field measurements as limits on confirmed details rather than new closure requirements.

Clarify already completed recovery state and keep any separate source-system administrative process conditional on its recorded requirements. Bump SCADA and burst handling skill versions.

app/services/operational_tasks.py

@@ -1513,11 +1513,13 @@ class OperationalTaskService:
         body = _json(command)
         scope = f"task:{task_id}:external-feedback"
         async with self.repo.transaction():
+            await self.repo.lock(f"task:{task_id}")
+            task = await self._task(task_id)
+            require(task.get("execution_scope") == self.execution_scope,
+                    "EXECUTION_SCOPE_MISMATCH", "当前环境不能同步其他执行范围的任务", status=403)
             replay = await self._request_replay(scope, command.request_id, body)
             if replay:
                 return ExternalFeedbackResult(task=await self.detail(task_id), **replay)
-            await self.repo.lock(f"task:{task_id}")
-            task = await self._task(task_id)
             if task["task_type"] == "event_investigation":
                 self._require_actionable_task(task)
                 own_source = task["source"].get("external_id")
@@ -1674,11 +1676,13 @@ class OperationalTaskService:
         body = _json(command)
         scope = f"task:{task_id}:withdraw-external-source"
         async with self.repo.transaction():
+            await self.repo.lock(f"task:{task_id}")
+            task = await self._task(task_id)
+            require(task.get("execution_scope") == self.execution_scope,
+                    "EXECUTION_SCOPE_MISMATCH", "当前环境不能同步其他执行范围的任务", status=403)
             replay = await self._request_replay(scope, command.request_id, body)
             if replay:
                 return await self.detail(task_id)
-            await self.repo.lock(f"task:{task_id}")
-            task = await self._task(task_id)
             source = task["source"]
             require(
                 task["task_type"] == "external_work_order"

docs/burst-response-workflow.md

@@ -14,8 +14,8 @@ flowchart TD
     report --> approval[人工审批与现场执行]
     approval --> feedback[接收实际处置与恢复反馈]
     feedback --> verify[复核影响与结案条件]
-    verify --> close[来源系统确认事件闭环]
+    verify --> close[Server 完成任务并解除告警]
-    verify -->|新增事实或遗留事项| asset
+    verify -->|仍异常或有未完成处置| asset
 ```
 
 1. **接收核对**：保留来源工单ID、单调版本、发生时间、位置、描述、现场反馈和已有动作；后续轮次从新增事实接续。
@@ -23,7 +23,7 @@ flowchart TD
 3. **隔离方案**：用 analyze_valve_isolation 分析关联资产，保留必关阀、可选阀、直接隔离与供水失源节点；已知失效阀门用于替代方案分析，现场可操作性列为执行条件。
 4. **影响说明**：先展示拓扑范围，再按可用SCADA、水力结果和用户关联资料解释事件当前影响、方案预计影响和保障需求。
 5. **报告与处置**：保存阶段报告后提出核验、通知、保障、抢修和恢复等动作草稿；关联现场任务采用 general 工单。审批、派发和现场完成分别核对回执。
-6. **反馈复核**：已派发的关联工单保持 `waiting_work_order`，全部收到有效 `completed` 反馈后进入 `waiting_scada`。以现场反馈业务时间为 T，取严格晚于 T 的最近一个 SCADA 采样时刻进行复核；正常范围使用事件前15天同采样相位的正常包络。数据未到或仍异常时保持等待，后台工作进程定期复核；通过后持久保存恢复证据并接续 Agent。Agent 最终报告必须引用该证据，运行完成后才能完成任务并解除对应告警。原事件由有权限人员在来源系统确认行政结案。
+6. **反馈复核**：已派发的关联工单保持 `waiting_work_order`，全部收到有效 `completed` 反馈后进入 `waiting_scada`。以现场反馈业务时间为 T，取严格晚于 T 的最近一个 SCADA 采样时刻进行复核；正常范围使用事件前15天同采样相位的正常包络。数据未到或仍异常时保持等待，后台工作进程定期复核；通过后持久保存恢复证据并接续 Agent。Agent 最终报告必须引用该证据，运行完成后才能完成任务并解除对应告警。现场资料缺项仅限定未核实细节，不新增本任务的关闭门槛。来源系统另有明确行政结案流程时，按其实际状态另记。
 
 SCADA作为事件影响的辅助证据，按目标测点和事件／对照窗口读取。具备水力输入且工具已授权时，使用 analyze_valve_closure 评估给定阀门方案；各步骤记录实际结果、适用范围或具体待补项。
 

docs/operational-task-workers.md

@@ -210,7 +210,9 @@ SQL 同时增加必填范围约束，不设置数据库默认值。回滚版本
 Worker 应连接所属环境的 `OPERATIONAL_WORKER_SERVER_URL` 和
 `OPERATIONAL_WORKER_AGENT_URL`。Server 在运行列表、领取、租约回收与
 SCADA recovery 候选查询上限制执行范围；直接领取其他范围返回
-`403 EXECUTION_SCOPE_MISMATCH`。这属于配置隔离，不能替代服务凭据授权：
+`403 EXECUTION_SCOPE_MISMATCH`。外部工单反馈与来源撤回也必须匹配任务的执行范围，
+并在幂等重放前校验；开发环境同步器不能更新生产任务或重置其恢复监测起点。
+这属于配置隔离，不能替代服务凭据授权：
 不要让本地 Worker 通过线上 Server 领取任务。普通任务与历史记录仍共用存储。
 
 修改范围配置需要重启相应 Server。暂停的本地 Worker 不会因代码更新恢复。

resources/work-order-skills/burst-event-handling/SKILL.md

@@ -2,7 +2,7 @@
 name: burst-event-handling
 description: 根据爆管上报工单处理事件，完成资产关联、隔离方案、影响说明、处置建议与反馈复核。
 metadata:
-  version: "1.1.1"
+  version: "1.1.2"
 ---
 
 # 爆管事件处理
@@ -50,4 +50,4 @@ metadata:
 
 waiting_scada 表示恢复供水和监测复核阶段，以 returned_at 为反馈时刻 T，读取严格晚于 T 的最近一个 SCADA 采样时刻作为复核来源，说明测点、实际采样时间、读数、单位和正常范围；不等待额外周期，不使用 T 之前、T 当刻或更晚的正常数据替代。后台按事件前15天同期正常包络核验该时刻。最近时刻缺测或异常时保留实际问题，不宣称复核成功。保存本轮阶段成果，等待后续数据接续，不反复派发同一工单。
 
-verification_ready 时读取 verification_evidence_id 对应的证据与快照，核对工单反馈、T、最近采样时刻和测点，展示该时刻读数与正常范围；具备真实多时刻数据时再展示曲线，发布 final/concluded 报告并引用该证据。运行完成后 Server 才完成任务并解除关联告警。原事件在来源系统的行政结案仍由有权限人员确认。
+verification_ready 时读取 verification_evidence_id 对应的证据与快照，核对工单反馈、T、最近采样时刻和测点，展示该时刻读数与正常范围；具备真实多时刻数据时再展示曲线，工单 completed 且恢复证据 status=passed 时发布 final/concluded 恢复闭环报告并引用该证据。Server 完成运行后关闭任务并解除关联告警，phase=completed 表示已闭环。照片、阀位或复测记录未提供时仅记为资料缺项，不增设关闭前置条件。来源系统若另有明确行政结案流程，按其实际状态另记，不推断为当前任务尚未闭环。

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

@@ -2,7 +2,7 @@
 name: scada-anomaly
 description: 从 SCADA 指标异常开展数据核验、爆管诊断、候选定位与影响分析。
 metadata:
-  version: "1.0.13"
+  version: "1.0.14"
 ---
 
 # SCADA 数据异常调查
@@ -33,7 +33,7 @@ metadata:
 
 先查询并发布截至本轮knowledge_cutoff或异常时刻t的[t-3小时,t]当前趋势总览，覆盖本轮全部已标识SCADA设备，包括地图定位、等级标记及表格列出的目标设备。按指标和单位分图，同组各设备各占一条当前序列，系列名带device_id；每图最多8组，超过时分组发布，各组使用不同且稳定的chart_id。缺测填null，完全无读数的设备明确列为缺测并说明查询窗口，不能静默省略。压力和流量分别发布图表，包含正常与异常测点；某指标无设备或无有效读数时明确说明。每条当前序列保留三小时窗口内全部已取得采样点，不能用用于判断突变的[t-采样周期,t]相邻两点代替三小时趋势。15分钟数据正常应有13个时刻；缺测保留对应空槽并说明实际覆盖。收尾前逐项核对目标设备、查询覆盖、已发布图表和缺测清单，不用少量代表设备替代全量总览。图表“数据”视图沿用同一完整序列，报告按指标附测点、单位、窗口、有效点数和缺测说明表格。
 
-定位输入具备后继续推进定位。压力、流量均需发布按设备分图的七天历史复合对照，作为全量当前趋势总览之外的必要展示；不能因历史散点占用序列额度而遗漏其他设备的当前趋势。复合图采用 [t-3小时,t] 当前折线和前7天每天 t-30分钟、t-15分钟、t 的21个参考位置。chart_type=time_comparison表示同期复合对比，区别于普通line和bar。x_data只放UTC+8的HH:mm或HH:mm:ss共享时刻槽，不追加历史日期或历史槽位；15分钟数据用13槽，series中当前一个type=line、历史七个type=scatter，历史散点只填末3槽，其他为null。每组sample_times与data等长，每个有效值保留真实带时区日期和时刻，缺测值及其时间均填null；历史点按本地时刻放入对应槽，当前曲线使用同一窗口，每个历史点集使用此前某一天的对应窗口，跨午夜按窗口时间顺序排列。坐标轴注明单位。前7天按t转换后的当地日期逐日回退，不包含当天；按实际采样周期保留缺口，按需翻页并报告实际参考点数。基线未冻结、读取失败或缺测时明确说明并保留已有三小时趋势，不能静默退化为前一天或相邻两个时刻。
+定位输入具备后继续定位。压力、流量按设备另发七天同期复合图，不占用或替代全量当前总览。chart_type=time_comparison：x_data是UTC+8的HH:mm或HH:mm:ss共享槽，不拼接历史日期；当前[t-3小时,t]用series.type=line，前7天每天t-30分钟、t-15分钟、t用scatter。15分钟数据13槽，历史七组仅末3槽有值。每组sample_times与data等长，保留真实带时区日期；有效值与对应槽的本地时刻一致，缺测值和时间均null。当前曲线共用窗口，历史每组取此前某天对应窗口，跨午夜顺序排列；普通图用line/bar，坐标轴标单位。七天按t的本地日期回退，不含当天。按实际周期保留缺口，按需翻页并报告参考点数；基线未冻结、读取失败或缺测时说明并保留当前趋势，不退化为前一天或相邻两点。
 
 影响说明列出异常前参考、异常期间和已有恢复观测中的压力、流量或水质变化、持续时间、涉及测点及已核实资产。可比对象、单位、时段与运行条件下给出变化量，分别标明实际观测、模型预测和拓扑隔离。停水范围与漏量引用相应拓扑、水力、流量平衡或现场依据；恢复观测、用户关联或现场证据缺失时列为具体待核实项。
 
@@ -41,7 +41,7 @@ metadata:
 
 本轮时序分析保持任务冻结窗口，后续播放形成的观测由相应轮次处理。资产属性、空间查询和拓扑隔离使用项目当前模型，注明模型时态；水力工具按已开放的冻结输入能力调用。定位调用执行权限、设备与冻结观测校验；基线或覆盖缺口按实际返回记录。
 
-报告给出观测事实、规则信号、原因候选、定位与隔离结果、影响说明和建议。摘要以150字以内为目标，测点明细用表格、完整依据留在证据。候选、相似度和拓扑结果具备时，完成相应分析、展示与报告后再收尾；现场核验、审批和维修列为后续待办。现场工单与设备操作遵守服务端动作权限，本轮分析完成与现场事件闭环分别说明。
+报告给出观测事实、规则信号、原因候选、定位与隔离结果、影响说明和建议。摘要以150字以内为目标，测点明细用表格、完整依据留在证据。候选、相似度和拓扑结果具备时，完成相应分析、展示与报告后再收尾；现场核验、审批和维修按实际进度列为待办，已完工事项采用回执状态。现场工单与设备操作遵守服务端动作权限，恢复复核通过后按下节闭环。
 
 需要现场核验时，先发布本轮outcome=concluded的调查报告，再调用propose_task_action提出kind=create_work_order、work_order_type=general的关联预约工单草稿，引用候选与影响证据，写明核验对象、内容和回执要求。带replay_context的任务同样可提出草稿，描述中注明历史异常时刻及回放来源；草稿须由有权限的用户编辑、确认后发布，后续派发与反馈进入预约任务。草稿待确认时说明等待人工处理，不将调查报告完成写成现场处置完成。回放任务不支持设备命令。
 
@@ -49,4 +49,6 @@ metadata:
 
 收到关联工单的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越过原冻结上限重复查询，也不因该上限再建复测工单；确有独立现场工作时才提议新动作。完工时间修订后使用新返回的证据编号，旧证据仅作历史。正常读数不替代阀位、漏点照片等现场事实，报告分别说明监测恢复、现场待核验项与事件闭环。
+waiting_work_order等待完工；waiting_scada按last_check说明尚无后续采样、数据缺项或仍异常，等待监测复核。verification_ready时用get_task_evidence读取verification_evidence_id完整快照；工单completed且恢复证据status=passed时，发布final/concluded恢复闭环报告，同时引用原异常全量证据与当前恢复证据。Server完成运行后关闭任务并解除关联告警，phase=completed表示已闭环。阀位、漏点照片、现场复测数值等未提供时仅记为资料缺项，不新增现场核验、人工确认或更多采样的闭环条件。
+
+恢复观测由Server独立保存，原冻结上限不阻断此证据的使用；query_scada_readings仍遵守冻结窗口。完工时间修订后使用新证据编号，旧证据仅作历史。确有独立工作时再提议新动作。

tests/unit/test_operational_tasks.py

@@ -2487,3 +2487,59 @@ def test_task_scope_is_required_in_persisted_contract():
         OperationalTaskSummary.model_validate(task)
     assert any(item["loc"] == ("execution_scope",) and item["type"] == "missing"
                for item in error.value.errors())
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("operation", ["feedback", "withdrawal"])
+@pytest.mark.parametrize("task_scope,server_scope,allowed", [
+    (None, "production", False),
+    (None, "development", False),
+    ("production", "production", True),
+    ("production", "development", False),
+    ("development", "production", False),
+    ("development", "development", True),
+])
+async def test_external_sync_only_mutates_tasks_in_its_execution_scope(operation, task_scope, server_scope, allowed):
+    from copy import deepcopy
+
+    now = datetime(2026, 11, 8, tzinfo=timezone.utc)
+    context = _context()
+    task = _task_document(context, now, lease=None)
+    source_id = f"peripheral_sample:{uuid4()}"
+    task.update(task_type="external_work_order", source={
+        "kind": "external_system", "external_id": source_id,
+        "external_revision": 1, "event_kind": "general",
+    })
+    if task_scope is None:
+        task.pop("execution_scope")
+    else:
+        task["execution_scope"] = task_scope
+    repository = TaskRepository(task)
+    service = OperationalTaskService(repository, context, set(), clock=lambda: now, execution_scope=server_scope)
+    if operation == "feedback":
+        command = ExternalFeedbackCommand(request_id=uuid4(), source_event_id=source_id,
+            source_revision=2, status="completed", occurred_at=now, detail="已关阀修复管道")
+        method = "external_feedback"
+    else:
+        command = ExternalSourceWithdrawalCommand(request_id=uuid4(), source_event_id=source_id)
+        method = "withdraw_external_source"
+    before = deepcopy(repository.task)
+    if not allowed:
+        with pytest.raises(DomainError) as error:
+            await getattr(service, method)(task["task_id"], command)
+        assert error.value.code == "EXECUTION_SCOPE_MISMATCH" and error.value.status == 403
+        assert repository.task == before
+        assert repository.feedback == repository.evidence == repository.requests == {}
+        return
+
+    await getattr(service, method)(task["task_id"], command)
+    assert repository.task["status"] == ("completed" if operation == "feedback" else "cancelled")
+    # Check scope before idempotency replay as well as before the first write.
+    await getattr(service, method)(task["task_id"], command)
+    before = deepcopy(repository.task)
+    other = OperationalTaskService(repository, context, set(), clock=lambda: now,
+        execution_scope="development" if server_scope == "production" else "production")
+    with pytest.raises(DomainError) as error:
+        await getattr(other, method)(task["task_id"], command)
+    assert error.value.code == "EXECUTION_SCOPE_MISMATCH" and error.value.status == 403
+    assert repository.task == before

tests/unit/test_peripheral_task_sync.py

@@ -226,7 +226,7 @@ async def test_sample_sync_rejects_cross_project_data_before_writing():
 
 @pytest.mark.anyio
 @pytest.mark.parametrize("targeted", [False, True])
-@pytest.mark.parametrize("order_revision", [1, 2])
+@pytest.mark.parametrize("order_revision", [1, 4])
 @pytest.mark.parametrize(
     ("investigation_status", "expected_feedback_count", "routes_to_investigation"),
     [
@@ -254,9 +254,13 @@ async def test_created_work_order_receipt_preserves_projection_and_routes_active
                 "updated_at": "2026-09-23T09:00:00Z",
                 "title": "采样工单", "priority": "normal",
                 "work_order_type": "general", "created_at": "2026-09-23T08:00:00Z",
-                "scheduled_at": None,
+                "scheduled_at": "2026-09-23T08:30:00Z",
                 "activity": ([{"sequence": 2, "occurred_at": "2026-09-23T08:45:00Z", "actor": "现场人员",
-                               "kind": "feedback", "title": "现场反馈", "detail": "复测余氯0.2mg/L"}]
+                               "kind": "feedback", "title": "现场反馈", "detail": "复测余氯0.2mg/L"},
+                              {"sequence": 3, "occurred_at": "2026-09-23T08:50:00Z", "actor": "调度员",
+                               "kind": "updated", "title": "工单信息更新", "detail": "更新了状态。"},
+                              {"sequence": 4, "occurred_at": "2026-09-23T09:00:00Z", "actor": "调度员",
+                               "kind": "updated", "title": "工单信息更新", "detail": "更新了优先级。"}]
                              if order_revision > 1 else []),
             }
             if targeted: