from contextlib import asynccontextmanager from datetime import datetime, timedelta, timezone from types import SimpleNamespace from unittest.mock import AsyncMock from uuid import uuid4 import pytest from pydantic import ValidationError from app.auth.permissions import ( OPERATIONAL_ACTION_WORK_ORDER_APPROVE, OPERATIONAL_ACTION_WORK_ORDER_EXECUTE, OPERATIONAL_TASK_VIEW, OPERATIONAL_TASK_ACCESS_APPROVE, SCADA_VIEW, ) from app.auth.project_dependencies import ProjectContext from app.domain.domain_errors import DomainError from app.domain.schemas.operational_task import ( ActionDecisionCommand, ActionExecuteCommand, ActionProposalCommand, AgentArtifactsCommand, AgentRunCommand, ExternalFeedbackCommand, ExternalSourceWithdrawalCommand, HumanApprovalAction, NotificationAcknowledgeCommand, OperationalTaskNotification, OperationalTaskRun, OperationalTaskSummary, ExecutionPlanInput, TaskAccessRequestCreate, TaskAccessRequestDecision, TaskAccessRequest, WorkOrderDraftUpdateCommand, WorkOrderConfirmDispatchCommand, ) from app.domain.schemas.operational_data import TaskDataSnapshot from app.infra.db.postgresql.operational_tasks import OperationalTaskRepository from app.infra.integrations.operational_actions import ActionDispatchOutcome from app.services.operational_tasks import OperationalTaskService @pytest.fixture def anyio_backend(): return "asyncio" class NotificationRepository: def __init__(self, notification): self.notification = notification self.reads = {} self.requests = {} self.events = [] @asynccontextmanager async def transaction(self): yield self async def lock(self, _key): return None async def get_request(self, scope, request_id): return self.requests.get((scope, request_id)) async def save_request(self, scope, request_id, request_digest, response): self.requests[(scope, request_id)] = { "request_digest": request_digest, "response": response, } async def get_notification(self, notification_id, user_id): if notification_id != str(self.notification.notification_id): return None read = self.reads.get(user_id) return self.notification.model_copy(update={ "status": "read" if read else "unread", "version": 2 if read else 1, "read_at": read, "read_by": f"principal:{user_id}" if read else None, }).model_dump(mode="json") async def mark_notification_read(self, notification_id, user_id, version, read_at): assert version == 2 self.reads[user_id] = read_at async def append_event(self, **event): self.events.append(event) return len(self.events) @pytest.mark.anyio async def test_access_request_read_keeps_existing_tasks_available_before_migration(): connection = SimpleNamespace(execute=AsyncMock()) connection.execute.return_value.fetchone = AsyncMock(return_value={"relation": None}) repository = OperationalTaskRepository(connection, uuid4()) assert await repository.list_access_requests(str(uuid4())) == [] assert connection.execute.await_count == 1 @pytest.mark.anyio async def test_access_request_read_uses_table_after_migration(): availability = SimpleNamespace(fetchone=AsyncMock(return_value={"relation": "workbench.operational_task_access_requests"})) records = [{"document": {"access_request_id": str(uuid4())}}] result = SimpleNamespace(fetchall=AsyncMock(return_value=records)) connection = SimpleNamespace(execute=AsyncMock(side_effect=[availability, result])) repository = OperationalTaskRepository(connection, uuid4()) assert await repository.list_access_requests(str(uuid4())) == [records[0]["document"]] assert connection.execute.await_count == 2 def _context(user_id=None, *, system_role="user"): return ProjectContext( project_id=uuid4(), project_code="test", user_id=user_id or uuid4(), project_role="member", system_role=system_role, ) @pytest.mark.anyio async def test_notification_read_state_is_isolated_per_user(): now = datetime(2026, 9, 23, tzinfo=timezone.utc) notification = OperationalTaskNotification( notification_id=uuid4(), task_id=uuid4(), kind="task_created", title="任务", description="已创建", status="unread", version=1, created_at=now, ) repository = NotificationRepository(notification) first = _context() second = _context() first_service = OperationalTaskService( repository, first, {OPERATIONAL_TASK_VIEW}, clock=lambda: now ) second_service = OperationalTaskService( repository, second, {OPERATIONAL_TASK_VIEW}, clock=lambda: now ) request_id = uuid4() read = await first_service.acknowledge_notification( notification.notification_id, NotificationAcknowledgeCommand(request_id=request_id, expected_version=1), ) assert read.status == "read" assert (await repository.get_notification( str(notification.notification_id), str(second.user_id) ))["status"] == "unread" second_read = await second_service.acknowledge_notification( notification.notification_id, NotificationAcknowledgeCommand(request_id=request_id, expected_version=1), ) assert second_read.status == "read" assert second_read.read_by == f"principal:{second.user_id}" assert [event["aggregate_type"] for event in repository.events] == [ "operational_notification_read", "operational_notification_read", ] assert repository.events[0]["aggregate_id"] != repository.events[1]["aggregate_id"] def test_running_plan_state_wins_over_future_human_waiting_node(): nodes = [ type("Node", (), {"status": "running", "owner": "agent"})(), type("Node", (), {"status": "waiting", "owner": "human"})(), ] assert OperationalTaskService._status_from_plan(nodes, []) == "running" def test_pending_approval_prevents_plan_from_marking_task_completed(): action = HumanApprovalAction( action_id=uuid4(), task_id=uuid4(), version=1, kind="create_work_order", reason="现场采样", payload={"title": "采样", "description": "核验", "work_order_type": "sampling"}, evidence_refs=[], required_approval_permission="work-order.approve", required_execution_permission="work-order.execute", approval_status="pending", execution_status="not_submitted", proposed_by="principal:test", proposed_at=datetime.now(timezone.utc), ) nodes = [type("Node", (), {"status": "completed", "owner": "agent"})()] assert OperationalTaskService._status_from_plan(nodes, [action]) == "waiting_human" def test_pending_access_request_prevents_plan_from_marking_task_completed(): request = TaskAccessRequest( access_request_id=uuid4(), task_id=uuid4(), node_id="read-scada", scopes=["scada_devices"], reason="核对监测点", status="pending", approved_scopes=[], version=1, requested_by="principal:agent", requested_at=datetime.now(timezone.utc), ) nodes = [type("Node", (), {"status": "completed", "owner": "agent"})()] assert OperationalTaskService._status_from_plan( nodes, [], [request] ) == "waiting_human" def test_explicit_agent_run_failure_marks_task_failed_even_with_waiting_work(): nodes = [ type("Node", (), {"id": "FIELD_SAMPLE", "status": "waiting", "owner": "human"})(), type("Node", (), {"id": "AGENT_RUN_FAILURE", "status": "failed", "owner": "agent"})(), ] assert OperationalTaskService._status_from_plan(nodes, []) == "failed" @pytest.mark.anyio async def test_targeted_notification_cannot_be_acknowledged_without_permission(): now = datetime(2026, 9, 23, tzinfo=timezone.utc) notification = OperationalTaskNotification( notification_id=uuid4(), task_id=uuid4(), kind="action_pending", title="待确认", description="命令", required_permission="command.approve", status="unread", version=1, created_at=now, ) service = OperationalTaskService( NotificationRepository(notification), _context(), {OPERATIONAL_TASK_VIEW}, clock=lambda: now, ) with pytest.raises(DomainError) as exc: await service.acknowledge_notification( notification.notification_id, NotificationAcknowledgeCommand(request_id=uuid4(), expected_version=1), ) assert exc.value.code == "PERMISSION_DENIED" def _task_document(context, now, *, lease=None): task_id = uuid4() return OperationalTaskSummary( project_id=context.project_id, task_id=task_id, task_type="event_investigation", workflow="water_quality_complaint", status="running", priority="normal", revision=2, execution_generation=1, execution_request_id=uuid4(), title="调查", summary="验证状态机", source={"kind": "operator", "event_kind": "water_quality_complaint"}, trigger={"kind": "manual", "occurred_at": now, "detail": "test"}, scheduled_for=None, agent_session_id="session-1", agent_lease_expires_at=lease, current_plan_revision=None, current_report_revision=None, created_at=now, updated_at=now, ).model_dump(mode="json") class TaskRepository: def __init__(self, task, *, action=None): self.task = task self.action = action self.requests = {} self.evidence = {} self.feedback = {} self.plans = [] self.reports = [] self.notifications = [] self.locks = [] self.access_requests = {} @asynccontextmanager async def transaction(self): yield self async def lock(self, key): self.locks.append(key) async def get_request(self, scope, request_id): return self.requests.get((scope, request_id)) async def save_request(self, scope, request_id, request_digest, response): self.requests[(scope, request_id)] = { "request_digest": request_digest, "response": response, } async def get_task(self, task_id): return self.task if task_id == self.task["task_id"] else None async def update_task(self, task): self.task = task async def get_latest_plan(self, _task_id): return self.plans[-1] if self.plans else None async def get_latest_report(self, _task_id): return self.reports[-1] if self.reports else None async def insert_plan(self, _task_id, snapshot): self.plans.append(snapshot) async def insert_report(self, _task_id, snapshot): self.reports.append(snapshot) async def list_evidence(self, _task_id): return list(self.evidence.values()) async def get_evidence(self, _task_id, evidence_id): return self.evidence.get(evidence_id) async def insert_evidence(self, _task_id, evidence): self.evidence[evidence["evidence_id"]] = evidence async def list_access_requests(self, _task_id): return list(self.access_requests.values()) async def get_access_request(self, _task_id, access_request_id): return self.access_requests.get(access_request_id) async def insert_access_request(self, _task_id, document): self.access_requests[document["access_request_id"]] = document async def update_access_request(self, _task_id, document, expected_version): assert document["version"] == expected_version + 1 self.access_requests[document["access_request_id"]] = document async def list_actions(self, _task_id): return [self.action] if self.action else [] async def get_action(self, _task_id, action_id): if self.action and self.action["action_id"] == action_id: return self.action return None async def update_action(self, _task_id, action): self.action = action async def insert_action(self, _task_id, action): self.action = action async def insert_notification(self, notification): self.notifications.append(notification) async def resolve_notifications( self, task_id, *, kinds, subject_type, subject_id, resolved_at, resolution_reason ): resolved = [] for notification in self.notifications: if ( notification["task_id"] == task_id and notification["kind"] in kinds and notification.get("lifecycle_status", "active") == "active" and notification.get("subject_type") == subject_type and notification.get("subject_id") == subject_id ): notification.update( lifecycle_status="resolved", resolved_at=resolved_at.isoformat(), resolution_reason=resolution_reason, ) resolved.append(notification.copy()) return resolved async def get_feedback(self, task_id, source_event_id): return self.feedback.get((task_id, source_event_id)) async def upsert_feedback(self, task_id, feedback, accepted_at): self.feedback[(task_id, feedback["source_event_id"])] = { "source_revision": feedback["source_revision"], "document": feedback, "accepted_at": accepted_at, } class RunTaskRepository(TaskRepository): def __init__(self, task, run): super().__init__(task) self.run = run self.events = [] async def get_active_run(self, task_id): if task_id == self.task["task_id"] and self.run["status"] in {"queued", "claimed", "running"}: return self.run return None async def get_run(self, run_id): return self.run if run_id == self.run["run_id"] else None async def update_run(self, run, expected_revision): assert run["revision"] == expected_revision + 1 self.run = run async def insert_run(self, run): self.run = run async def append_event(self, **event): self.events.append(event) return len(self.events) def _active_run(task, now): return OperationalTaskRun( run_id=uuid4(), task_id=task["task_id"], generation=task["execution_generation"], execution_request_id=task["execution_request_id"], status="running", revision=2, agent_session_id="session-1", lease_expires_at=now + timedelta(minutes=1), heartbeat_at=now, started_at=now, dispatch_attempts=0, next_dispatch_at=now, created_at=now, updated_at=now, ).model_dump(mode="json") @pytest.mark.anyio async def test_run_heartbeat_does_not_change_task_business_revision(): now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc) context = _context() task = _task_document(context, now, lease=now + timedelta(minutes=1)) repository = RunTaskRepository(task, _active_run(task, now)) service = OperationalTaskService(repository, context, set(), clock=lambda: now) updated = await service.heartbeat_run( task["task_id"], repository.run["run_id"], AgentRunCommand( request_id=uuid4(), expected_revision=2, source_session_id="session-1" ), ) assert updated.revision == 3 assert repository.task["revision"] == 2 assert repository.events == [] @pytest.mark.anyio async def test_complete_without_same_generation_artifacts_fails_run_explicitly(): now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc) context = _context() task = _task_document(context, now, lease=now + timedelta(minutes=1)) repository = RunTaskRepository(task, _active_run(task, now)) service = OperationalTaskService(repository, context, set(), clock=lambda: now) completed = await service.complete_run( task["task_id"], repository.run["run_id"], AgentRunCommand( request_id=uuid4(), expected_revision=2, source_session_id="session-1" ), ) assert repository.run["status"] == "failed" assert repository.run["failure_code"] == "ARTIFACT_INCOMPLETE" assert completed.lifecycle_status == "failed" assert completed.active_run is None assert [item["kind"] for item in repository.notifications] == ["task_failed"] @pytest.mark.anyio async def test_external_feedback_coalesces_into_an_already_queued_run(): now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc) context = _context() task = _task_document(context, now, lease=None) task.update( status="queued", lifecycle_status="open", execution_generation=2, execution_request_id=str(uuid4()), agent_session_id=None, ) run = _active_run(task, now) run.update( generation=2, execution_request_id=task["execution_request_id"], status="queued", revision=1, agent_session_id=None, lease_expires_at=None, heartbeat_at=None, started_at=None, ) repository = RunTaskRepository(task, run) service = OperationalTaskService(repository, context, set(), clock=lambda: now) result = await service.external_feedback( task["task_id"], ExternalFeedbackCommand( request_id=uuid4(), source_event_id="sample-result", source_revision=1, status="completed", occurred_at=now, detail="新的现场结果", ), ) assert result.task.execution_generation == 2 assert str(result.task.execution_request_id) == task["execution_request_id"] assert result.task.active_run is not None assert str(result.task.active_run.run_id) == run["run_id"] def _waiting_plan(action_id=None): node = { "id": "field-work", "title": "现场处置", "owner": "human", "status": "waiting", "detail": "等待审批与执行", "evidence_refs": [], "action_refs": [str(action_id)] if action_id else [], } return { "schema": "execution-plan@1", "title": "调查计划", "reason": "等待现场处置", "nodes": [node], "edges": [], } def _current_report(): return { "title": "本代调查报告", "current_conclusion": "等待现场处置", "conclusion_boundary": "尚无现场回执", "evidence_refs": [], "recommendations": ["完成现场处置"], "pending_items": ["现场处置"], } @pytest.mark.anyio async def test_plan_only_keeps_lease_until_same_generation_report_closes_run(): now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc) context = _context() task = _task_document(context, now, lease=now + timedelta(minutes=1)) repository = RunTaskRepository(task, _active_run(task, now)) service = OperationalTaskService(repository, context, set(), clock=lambda: now) planned = await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( request_id=uuid4(), source_session_id="session-1", plan=_waiting_plan(), )) assert planned.task.status == "running" assert planned.task.agent_lease_expires_at == now + timedelta(minutes=1) assert planned.task.latest_plan.execution_generation == 1 reported = await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( request_id=uuid4(), source_session_id="session-1", report=_current_report(), )) assert reported.task.status == "running" assert reported.task.latest_report.execution_generation == 1 completed = 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 completed.status == "waiting_human" assert completed.agent_lease_expires_at is None assert [item["kind"] for item in repository.notifications] == [ "human_input_required" ] @pytest.mark.anyio async def test_report_does_not_close_from_legacy_or_prior_generation_plan(): now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc) context = _context() task = _task_document(context, now, lease=now + timedelta(minutes=1)) task["execution_generation"] = 2 repository = TaskRepository(task) repository.plans.append({ "revision": 1, "execution_generation": None, "source_session_id": "session-1", "published_at": now.isoformat(), "published_by": "principal:test", "plan": _waiting_plan(), }) service = OperationalTaskService(repository, context, set(), clock=lambda: now) result = await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( request_id=uuid4(), source_session_id="session-1", report=_current_report(), )) assert result.task.status == "running" assert result.task.latest_report.execution_generation == 2 @pytest.mark.anyio async def test_action_proposal_keeps_running_lease_until_plan_and_report_close_run(): now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc) context = _context() task = _task_document(context, now, lease=now + timedelta(minutes=1)) repository = RunTaskRepository(task, _active_run(task, now)) service = OperationalTaskService(repository, context, set(), clock=lambda: now) await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( request_id=uuid4(), source_session_id="session-1", plan=_waiting_plan(), )) revision_before_proposal = repository.run["revision"] lease_before_proposal = repository.run["lease_expires_at"] action = await service.propose_action(task["task_id"], ActionProposalCommand( request_id=uuid4(), kind="create_work_order", reason="需要现场核验", node_id="field-work", evidence_refs=[], payload={"title": "现场核验", "description": "等待人工执行", "work_order_type": "sampling"}, )) assert repository.task["status"] == "running" assert repository.run["revision"] == revision_before_proposal assert repository.run["lease_expires_at"] == lease_before_proposal heartbeat = await service.heartbeat_run( task["task_id"], repository.run["run_id"], AgentRunCommand( request_id=uuid4(), expected_revision=revision_before_proposal, source_session_id="session-1", ), ) assert heartbeat.status == "running" published = await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( request_id=uuid4(), source_session_id="session-1", run_id=repository.run["run_id"], plan=_waiting_plan(action.action_id), report=_current_report(), )) closed = await service.complete_run( task["task_id"], repository.run["run_id"], AgentRunCommand( request_id=uuid4(), expected_revision=published.task.active_run.revision, source_session_id="session-1", ), ) assert closed.status == "waiting_human" assert closed.actions[0].action_id == action.action_id assert [item["kind"] for item in repository.notifications] == [ "action_approval_pending" ] @pytest.mark.anyio async def test_rejected_action_requeues_task_in_a_new_execution_generation(): now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc) context = _context() task = _task_document(context, now, lease=now + timedelta(minutes=1)) repository = RunTaskRepository(task, _active_run(task, now).copy()) service = OperationalTaskService( repository, context, {OPERATIONAL_ACTION_WORK_ORDER_APPROVE}, clock=lambda: now, ) action = await service.propose_action(task["task_id"], ActionProposalCommand( request_id=uuid4(), kind="create_work_order", reason="需要现场核验", evidence_refs=[], payload={"title": "现场核验", "description": "等待人工执行", "work_order_type": "sampling"}, )) original_request_id = repository.task["execution_request_id"] rejected = await service.decide_action( task["task_id"], action.action_id, ActionDecisionCommand( request_id=uuid4(), expected_version=action.version, decision="rejected", comment="当前方案不执行", ), ) assert rejected.approval_status == "rejected" assert rejected.version == action.version + 1 assert repository.task["status"] == "queued" assert repository.task["execution_generation"] == 2 assert repository.task["execution_request_id"] != original_request_id assert repository.task["agent_session_id"] is None assert repository.task["agent_lease_expires_at"] is None approval_notice = next( item for item in repository.notifications if item["kind"] == "action_approval_pending" ) assert approval_notice["lifecycle_status"] == "resolved" assert approval_notice["resolution_reason"] == "action_rejected" assert any( event["topic"] == "operational.notification.changed" and event["metadata"].get("change") == "resolved" for event in repository.events ) @pytest.mark.anyio async def test_action_decision_rejects_a_stale_expected_version(): now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc) context = _context() task = _task_document(context, now, lease=now + timedelta(minutes=1)) repository = RunTaskRepository(task, _active_run(task, now)) service = OperationalTaskService( repository, context, {OPERATIONAL_ACTION_WORK_ORDER_APPROVE}, clock=lambda: now, ) action = await service.propose_action(task["task_id"], ActionProposalCommand( request_id=uuid4(), kind="create_work_order", reason="需要现场核验", evidence_refs=[], payload={"title": "现场核验", "description": "等待人工执行", "work_order_type": "sampling"}, )) with pytest.raises(DomainError) as exc: await service.decide_action( task["task_id"], action.action_id, ActionDecisionCommand( request_id=uuid4(), expected_version=action.version + 1, decision="approved", ), ) assert exc.value.code == "VERSION_CONFLICT" assert repository.action["approval_status"] == "pending" assert repository.action["version"] == action.version def test_v2_plan_requires_explicit_decision_route_and_rejects_cycles(): plan = { "schema": "execution-plan@2", "title": "动态调查", "reason": "核对投诉与风险", "nodes": [ {"id": "risk", "title": "判断风险", "owner": "agent", "status": "completed", "detail": "按已核实事件选择", "kind": "decision", "selected_edge_id": "urgent", "evidence_refs": ["risk-evidence"]}, {"id": "urgent-check", "title": "紧急采样", "owner": "human", "status": "waiting", "detail": "等待现场人员", "kind": "activity"}, {"id": "routine-check", "title": "常规核查", "owner": "human", "status": "skipped", "detail": "紧急路线已选", "kind": "activity"}, ], "edges": [ {"id": "urgent", "from": "risk", "to": "urgent-check", "condition": "有紧急风险"}, {"id": "routine", "from": "risk", "to": "routine-check", "condition": "无紧急风险"}, ], } assert ExecutionPlanInput.model_validate(plan).nodes[0].selected_edge_id == "urgent" with pytest.raises(ValidationError, match="outgoing route"): ExecutionPlanInput.model_validate({**plan, "nodes": [{**plan["nodes"][0], "selected_edge_id": "missing"}, *plan["nodes"][1:]]}) with pytest.raises(ValidationError, match="acyclic"): ExecutionPlanInput.model_validate({**plan, "edges": [*plan["edges"], {"id": "loop", "from": "urgent-check", "to": "risk"}]}) with pytest.raises(ValidationError, match="unchosen route"): ExecutionPlanInput.model_validate({**plan, "nodes": [plan["nodes"][0], plan["nodes"][1], {**plan["nodes"][2], "status": "waiting"}]}) with pytest.raises(ValidationError, match="only decision or parallel split"): ExecutionPlanInput.model_validate({**plan, "nodes": [{**plan["nodes"][0], "kind": "activity", "selected_edge_id": None}, *plan["nodes"][1:]]}) @pytest.mark.anyio async def test_task_access_grant_is_project_scoped_reviewable_and_requeues(): now = datetime(2026, 9, 24, 8, tzinfo=timezone.utc) context = _context() task = _task_document(context, now, lease=now + timedelta(minutes=1)) initial_revision = task["revision"] repository = RunTaskRepository(task, _active_run(task, now)) agent = OperationalTaskService(repository, context, set(), clock=lambda: now) request = await agent.request_access(task["task_id"], TaskAccessRequestCreate( request_id=uuid4(), source_session_id="session-1", node_id="read-scada", scopes=["scada_devices"], reason="核对上游监测点", )) assert repository.task["revision"] == initial_revision + 1 assert any(event["topic"] == "operational.task.changed" for event in repository.events) assert not await agent.can_read_source(task["task_id"], "scada_devices") assert repository.notifications[0]["kind"] == "access_approval_pending" reviewer = OperationalTaskService(repository, context, {OPERATIONAL_TASK_ACCESS_APPROVE, SCADA_VIEW}, clock=lambda: now) approved = await reviewer.decide_access(task["task_id"], request.access_request_id, TaskAccessRequestDecision(request_id=uuid4(), expected_version=1, decision="approved", approved_scopes=["scada_devices"])) assert approved.version == 2 assert not await agent.can_read_source(task["task_id"], "scada_devices") repository.run.update(status="running", agent_session_id="session-2", lease_expires_at=now + timedelta(minutes=1)) assert await agent.can_read_source(task["task_id"], "scada_devices") assert not await agent.can_read_source(task["task_id"], "scada_readings") assert repository.task["status"] == "queued" access_notice = next( item for item in repository.notifications if item["kind"] == "access_approval_pending" ) assert access_notice["lifecycle_status"] == "resolved" assert access_notice["resolution_reason"] == "access_request_approved" @pytest.mark.anyio async def test_task_access_request_rejects_a_second_pending_request(): now = datetime(2026, 9, 24, 8, tzinfo=timezone.utc) context = _context() task = _task_document(context, now, lease=now + timedelta(minutes=1)) repository = RunTaskRepository(task, _active_run(task, now)) service = OperationalTaskService(repository, context, set(), clock=lambda: now) await service.request_access(task["task_id"], TaskAccessRequestCreate( request_id=uuid4(), source_session_id="session-1", node_id="read-scada", scopes=["scada_devices"], reason="核对上游监测点", )) with pytest.raises(DomainError) as exc: await service.request_access(task["task_id"], TaskAccessRequestCreate( request_id=uuid4(), source_session_id="session-1", node_id="read-scada", scopes=["scada_devices"], reason="重复申请", )) assert exc.value.code == "ACCESS_REQUEST_PENDING" assert len(repository.access_requests) == 1 assert len(repository.notifications) == 1 @pytest.mark.anyio async def test_task_access_request_fails_closed_before_project_migration(): now = datetime(2026, 9, 24, 8, tzinfo=timezone.utc) context = _context() task = _task_document(context, now, lease=now + timedelta(minutes=1)) repository = RunTaskRepository(task, _active_run(task, now)) repository.access_requests_available = AsyncMock(return_value=False) service = OperationalTaskService(repository, context, set(), clock=lambda: now) with pytest.raises(DomainError) as exc: await service.request_access(task["task_id"], TaskAccessRequestCreate( request_id=uuid4(), source_session_id="session-1", node_id="read-scada", scopes=["scada_devices"], reason="核对上游监测点", )) assert exc.value.code == "DEPENDENCY_UNAVAILABLE" assert exc.value.status == 503 @pytest.mark.anyio async def test_completed_gate_requires_access_request_decision(): now = datetime(2026, 9, 24, 8, tzinfo=timezone.utc) context = _context() task = _task_document(context, now, lease=now + timedelta(minutes=1)) repository = RunTaskRepository(task, _active_run(task, now)) service = OperationalTaskService(repository, context, set(), clock=lambda: now) request = await service.request_access(task["task_id"], TaskAccessRequestCreate( request_id=uuid4(), source_session_id="session-1", node_id="read-scada", scopes=["scada_devices"], reason="核对上游监测点", )) with pytest.raises(DomainError) as exc: await service.publish_artifacts(task["task_id"], AgentArtifactsCommand( request_id=uuid4(), source_session_id="session-1", plan={ "schema": "execution-plan@2", "title": "读取监测数据", "reason": "等待授权后读取", "nodes": [{ "id": "read-scada", "title": "数据访问审核", "owner": "human", "status": "completed", "detail": "尚未审核", "kind": "gate", "gate_ref": str(request.access_request_id), }], "edges": [], }, )) assert exc.value.code == "GATE_NOT_RESOLVED" @pytest.mark.anyio async def test_work_order_draft_edit_creates_new_review_version(): now = datetime(2026, 9, 24, 8, tzinfo=timezone.utc) context = _context() task = _task_document(context, now, lease=now + timedelta(minutes=1)) repository = RunTaskRepository(task, _active_run(task, now)) service = OperationalTaskService(repository, context, {OPERATIONAL_ACTION_WORK_ORDER_APPROVE}, clock=lambda: now) action = await service.propose_action(task["task_id"], ActionProposalCommand( request_id=uuid4(), kind="create_work_order", reason="现场采样", evidence_refs=[], payload={"title": "旧草稿", "description": "采样", "work_order_type": "general"}, )) updated = await service.update_work_order_draft(task["task_id"], action.action_id, WorkOrderDraftUpdateCommand(request_id=uuid4(), expected_version=1, payload={"title": "人工修订草稿", "description": "增加复测", "work_order_type": "general"})) assert updated.version == 2 assert updated.payload.title == "人工修订草稿" assert updated.approval_status == "pending" with pytest.raises(DomainError, match="Action version changed"): await service.update_work_order_draft(task["task_id"], action.action_id, WorkOrderDraftUpdateCommand(request_id=uuid4(), expected_version=1, payload={"title": "过期草稿", "description": "采样", "work_order_type": "general"})) assert repository.action["approval_status"] == "pending" assert repository.action["version"] == updated.version @pytest.mark.anyio async def test_one_click_work_order_confirmation_replays_without_duplicate_dispatch(): now = datetime(2026, 9, 24, 8, tzinfo=timezone.utc) context = _context() task = _task_document(context, now, lease=now + timedelta(minutes=1)) repository = RunTaskRepository(task, _active_run(task, now)) service = OperationalTaskService(repository, context, { OPERATIONAL_ACTION_WORK_ORDER_APPROVE, OPERATIONAL_ACTION_WORK_ORDER_EXECUTE, }, clock=lambda: now) action = await service.propose_action(task["task_id"], ActionProposalCommand( request_id=uuid4(), kind="create_work_order", reason="现场采样", evidence_refs=[], payload={"title": "采样", "description": "现场核验", "work_order_type": "general"}, )) class Dispatcher: calls = 0 async def dispatch(self, **_kwargs): self.calls += 1 return ActionDispatchOutcome("unsupported", "测试适配器未配置") dispatcher = Dispatcher() command = WorkOrderConfirmDispatchCommand(request_id=uuid4(), expected_version=action.version) first = await service.confirm_work_order_dispatch(task["task_id"], action.action_id, command, dispatcher, None) replay = await service.confirm_work_order_dispatch(task["task_id"], action.action_id, command, dispatcher, None) assert first.approval_status == "approved" assert first.execution_status == "unsupported" assert replay.version == first.version assert dispatcher.calls == 1 assert [item["kind"] for item in repository.notifications] == [ "action_approval_pending", "external_coordination_failed", ] @pytest.mark.anyio async def test_new_payload_creates_a_new_pending_action_without_inheriting_approval(): now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc) context = _context() task = _task_document(context, now, lease=now + timedelta(minutes=1)) repository = TaskRepository(task) service = OperationalTaskService( repository, context, {OPERATIONAL_ACTION_WORK_ORDER_APPROVE}, clock=lambda: now, ) original = await service.propose_action(task["task_id"], ActionProposalCommand( request_id=uuid4(), kind="create_work_order", reason="需要首次核验", evidence_refs=[], payload={"title": "首次核验", "description": "原始内容", "work_order_type": "sampling"}, )) approved = await service.decide_action( task["task_id"], original.action_id, ActionDecisionCommand( request_id=uuid4(), expected_version=original.version, decision="approved", ), ) revised = await service.propose_action(task["task_id"], ActionProposalCommand( request_id=uuid4(), kind="create_work_order", reason="内容已修改", evidence_refs=[], payload={"title": "修改后核验", "description": "新的内容", "work_order_type": "sampling"}, )) assert approved.approval_status == "approved" assert revised.action_id != approved.action_id assert revised.payload != approved.payload assert revised.approval_status == "pending" assert revised.execution_status == "not_submitted" assert revised.version == 1 assert revised.decided_by is None assert revised.decided_at is None def _work_order_snapshot(work_order_id, now): return { "work_order_id": str(work_order_id), "revision": 2, "title": "复测预约", "summary": "等待现场复测回令", "status": "awaiting_reply", "priority": "high", "assignee": "水质班组", "scheduled_at": (now + timedelta(hours=1)).isoformat(), "work_order_type": "water_quality_complaint", "location": "联调小区", "scenario_code": "WQ-SECONDARY-SUPPLY-001", "feedback_requirements": ["上传检测结果"], "activity": [{"sequence": 1, "occurred_at": now.isoformat(), "actor": "水质班组", "kind": "feedback", "title": "现场反馈", "detail": "上传复测结果"}], "origin": "synthetic", "updated_at": now.isoformat(), } @pytest.mark.anyio async def test_external_feedback_exposes_validated_work_order_snapshot_and_refreshes_summary(): now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc) context = _context() work_order_id = uuid4() task = _task_document(context, now, lease=None) task.update( task_type="external_work_order", workflow="water_quality_complaint", source={ "kind": "external_system", "external_id": f"peripheral_sample:{work_order_id}", "external_revision": 1, "event_kind": "water_quality_complaint", }, ) repository = TaskRepository(task) service = OperationalTaskService(repository, context, set(), clock=lambda: now) result = await service.external_feedback(task["task_id"], ExternalFeedbackCommand( request_id=uuid4(), source_event_id=f"peripheral_sample:{work_order_id}", source_revision=2, status="awaiting_feedback", occurred_at=now, detail="等待现场复测回令", work_order_snapshot=_work_order_snapshot(work_order_id, now), )) assert result.task.status == "waiting_human" assert result.task.title == "复测预约" assert result.task.summary == "等待现场复测回令" assert result.task.scheduled_for == now + timedelta(hours=1) assert result.task.priority == "urgent" assert result.task.work_order_snapshot is not None assert result.task.work_order_snapshot.activity[0].title == "现场反馈" assert result.task.work_order_snapshot.feedback_requirements == ["上传检测结果"] @pytest.mark.anyio async def test_withdraw_external_source_keeps_history_and_is_idempotent(): now = datetime(2026, 9, 24, 8, tzinfo=timezone.utc) context = _context() work_order_id = uuid4() source_id = f"peripheral_sample:{work_order_id}" task = _task_document(context, now, lease=None) task.update( task_type="external_work_order", source={"kind": "external_system", "external_id": source_id, "external_revision": 1, "event_kind": "general"}, ) repository = TaskRepository(task) service = OperationalTaskService(repository, context, set(), clock=lambda: now) command = ExternalSourceWithdrawalCommand( request_id=uuid4(), source_event_id=source_id ) withdrawn = await service.withdraw_external_source(task["task_id"], command) replayed = await service.withdraw_external_source(task["task_id"], command) assert withdrawn.source.deleted_at == now assert withdrawn.status == "cancelled" assert withdrawn.lifecycle_status == "cancelled" assert withdrawn.revision == 3 assert replayed.revision == withdrawn.revision assert repository.task["source"]["external_revision"] == 1 with pytest.raises(DomainError) as exc: await service.withdraw_external_source(task["task_id"], ExternalSourceWithdrawalCommand( request_id=uuid4(), source_event_id=f"peripheral_sample:{uuid4()}" )) assert exc.value.code == "EXTERNAL_SOURCE_MISMATCH" @pytest.mark.anyio async def test_withdrawn_external_task_rejects_action_changes_and_resolves_approval(): now = datetime(2026, 9, 24, 8, tzinfo=timezone.utc) context = _context() work_order_id = uuid4() source_id = f"peripheral_sample:{work_order_id}" task = _task_document(context, now, lease=None) task.update( task_type="external_work_order", status="scheduled", source={"kind": "external_system", "external_id": source_id, "external_revision": 1, "event_kind": "general"}, ) repository = TaskRepository(task) service = OperationalTaskService( repository, context, {OPERATIONAL_ACTION_WORK_ORDER_APPROVE}, clock=lambda: now, ) action = await service.propose_action(task["task_id"], ActionProposalCommand( request_id=uuid4(), kind="create_work_order", reason="需要现场核验", evidence_refs=[], payload={"title": "现场核验", "description": "等待人工执行", "work_order_type": "sampling"}, )) await service.withdraw_external_source( task["task_id"], ExternalSourceWithdrawalCommand(request_id=uuid4(), source_event_id=source_id), ) with pytest.raises(DomainError) as exc: await service.decide_action( task["task_id"], action.action_id, ActionDecisionCommand( request_id=uuid4(), expected_version=action.version, decision="approved", ), ) assert exc.value.code == "TASK_NOT_ACTIONABLE" assert repository.notifications[0]["lifecycle_status"] == "resolved" assert repository.notifications[0]["resolution_reason"] == "external_source_withdrawn" @pytest.mark.anyio async def test_event_investigation_rejects_work_order_snapshot(): now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc) context = _context() task = _task_document(context, now, lease=now + timedelta(minutes=1)) repository = TaskRepository(task) service = OperationalTaskService(repository, context, set(), clock=lambda: now) work_order_id = uuid4() with pytest.raises(DomainError) as exc: await service.external_feedback(task["task_id"], ExternalFeedbackCommand( request_id=uuid4(), source_event_id=str(work_order_id), source_revision=2, status="awaiting_feedback", occurred_at=now, detail="不应接受", work_order_snapshot=_work_order_snapshot(work_order_id, now), )) assert exc.value.code == "INVALID_FEEDBACK_SNAPSHOT" @pytest.mark.anyio async def test_expired_owned_lease_is_rejected_with_stable_code(): now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc) context = _context() task = _task_document(context, now, lease=now - timedelta(seconds=1)) repository = RunTaskRepository(task, _active_run(task, now)) repository.run["lease_expires_at"] = now - timedelta(seconds=1) service = OperationalTaskService(repository, context, set(), clock=lambda: now) with pytest.raises(DomainError) as exc: await service.heartbeat_run( task["task_id"], repository.run["run_id"], AgentRunCommand( request_id=uuid4(), expected_revision=2, source_session_id="session-1" ), ) assert exc.value.code == "LEASE_EXPIRED" assert repository.run["revision"] == 2 @pytest.mark.anyio async def test_interrupted_submit_replays_unknown_without_dispatching_twice(): now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc) context = _context() task = _task_document(context, now, lease=now + timedelta(minutes=1)) task.update(status="waiting_human", agent_lease_expires_at=None) action = HumanApprovalAction( action_id=uuid4(), task_id=task["task_id"], version=2, kind="create_work_order", reason="现场核验", payload={"title": "采样", "description": "核验", "work_order_type": "sampling"}, evidence_refs=[], required_approval_permission="work-order.approve", required_execution_permission=OPERATIONAL_ACTION_WORK_ORDER_EXECUTE, approval_status="approved", execution_status="not_submitted", proposed_by="principal:agent", proposed_at=now, decided_by="principal:approver", decided_at=now, ).model_dump(mode="json") repository = TaskRepository(task, action=action) service = OperationalTaskService( repository, context, {OPERATIONAL_ACTION_WORK_ORDER_EXECUTE}, clock=lambda: now, ) command = ActionExecuteCommand(request_id=uuid4(), expected_version=2) class LostResponseDispatcher: def __init__(self): self.calls = 0 async def dispatch(self, **_kwargs): self.calls += 1 raise RuntimeError("connection lost after submission") dispatcher = LostResponseDispatcher() with pytest.raises(RuntimeError): await service.submit_action( task["task_id"], action["action_id"], command, dispatcher, None ) replay = await service.submit_action( task["task_id"], action["action_id"], command, dispatcher, None ) assert dispatcher.calls == 1 assert replay.execution_status == "unknown" assert replay.execution_request_id is not None assert [item["kind"] for item in repository.notifications] == [ "external_coordination_failed" ] @pytest.mark.anyio async def test_snapshot_reuse_returns_the_original_provenance_and_rejects_collision(): now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc) context = _context() task = _task_document(context, now, lease=now + timedelta(minutes=1)) repository = TaskRepository(task) service = OperationalTaskService(repository, context, set(), clock=lambda: now) original = TaskDataSnapshot( source="peripheral_points", source_ref="peripheral_points:sha256:same", queried_at=now, origin="synthetic", data={"items": [{"id": "p-1"}]}, truncated=False, ) evidence, stored = await service.record_data_snapshot(task["task_id"], original) replayed = original.model_copy(update={"queried_at": now + timedelta(minutes=1)}) replay_evidence, replay_snapshot = await service.record_data_snapshot( task["task_id"], replayed ) assert replay_evidence == evidence assert replay_snapshot == stored == original collision = original.model_copy(update={"data": {"items": [{"id": "p-2"}]}}) with pytest.raises(DomainError) as exc: await service.record_data_snapshot(task["task_id"], collision) assert exc.value.code == "EVIDENCE_CONFLICT" @pytest.mark.anyio async def test_repository_task_update_rejects_a_stale_revision(): cursor = SimpleNamespace(rowcount=0) connection = SimpleNamespace(execute=AsyncMock(return_value=cursor)) repository = OperationalTaskRepository(connection, uuid4()) now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc) context = _context() task = _task_document(context, now, lease=now + timedelta(minutes=1)) task["revision"] = 3 with pytest.raises(DomainError) as exc: await repository.update_task(task) assert exc.value.code == "VERSION_CONFLICT"