diff --git a/app/services/operational_tasks.py b/app/services/operational_tasks.py index 776cb07..935a463 100644 --- a/app/services/operational_tasks.py +++ b/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" diff --git a/docs/operational-task-workers.md b/docs/operational-task-workers.md index 7410bb8..272ff3d 100644 --- a/docs/operational-task-workers.md +++ b/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 不会因代码更新恢复。 diff --git a/tests/unit/test_operational_tasks.py b/tests/unit/test_operational_tasks.py index b352d17..4746170 100644 --- a/tests/unit/test_operational_tasks.py +++ b/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 diff --git a/tests/unit/test_peripheral_task_sync.py b/tests/unit/test_peripheral_task_sync.py index 3e9da45..f41eedd 100644 --- a/tests/unit/test_peripheral_task_sync.py +++ b/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: