fix(tasks): isolate peripheral feedback writes by execution scope

An old development synchronizer could overwrite a production recovery timestamp in a shared project database even after production deployment. Check task scope before feedback or source-withdrawal writes and before idempotency replay.

Cover cross-scope rejection, same-scope success and replay, and preserve actual field feedback time when later order metadata changes.

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/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/scada-anomaly/SKILL.md

@@ -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的本地日期回退，不含当天。按实际周期保留缺口，按需翻页并报告参考点数；基线未冻结、读取失败或缺测时说明并保留当前趋势，不退化为前一天或相邻两点。
 
 影响说明列出异常前参考、异常期间和已有恢复观测中的压力、流量或水质变化、持续时间、涉及测点及已核实资产。可比对象、单位、时段与运行条件下给出变化量，分别标明实际观测、模型预测和拓扑隔离。停水范围与漏量引用相应拓扑、水力、流量平衡或现场依据；恢复观测、用户关联或现场证据缺失时列为具体待核实项。
 

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: