From afcbd52a5a3e8de766c6433c740d8f431ed5f6a2 Mon Sep 17 00:00:00 2001 From: Huarch Date: Thu, 24 Sep 2026 17:11:21 +0800 Subject: [PATCH] refactor(tasks): centralize access request lookup --- app/services/operational_tasks.py | 52 +++++++++++++++---------------- 1 file changed, 25 insertions(+), 27 deletions(-) diff --git a/app/services/operational_tasks.py b/app/services/operational_tasks.py index 7db4692..62a975a 100644 --- a/app/services/operational_tasks.py +++ b/app/services/operational_tasks.py @@ -60,8 +60,7 @@ from app.domain.schemas.operational_task import ( RecoverExpiredLeasesResult, StoredEvidenceRecord, ) -from app.domain.schemas.operational_data import TaskDataSnapshot -from app.domain.schemas.operational_data import TaskDataSource +from app.domain.schemas.operational_data import TaskDataSnapshot, TaskDataSource from app.services.content_digest import digest @@ -122,6 +121,12 @@ class OperationalTaskService: append_event = getattr(self.repo, "append_event", None) return await append_event(**kwargs) if append_event is not None else None + async def _list_access_requests(self, task_id: UUID | str) -> list[dict]: + list_access_requests = getattr(self.repo, "list_access_requests", None) + if list_access_requests is None: + return [] + return await list_access_requests(str(task_id)) + def _permission_view(self) -> OperationalTaskPermissions: permissions = self.permissions return OperationalTaskPermissions( @@ -219,7 +224,7 @@ class OperationalTaskService: for document in await self.repo.list_evidence(str(task_id)) ] actions = await self.repo.list_actions(str(task_id)) - access_requests = await self.repo.list_access_requests(str(task_id)) if hasattr(self.repo, "list_access_requests") else [] + access_requests = await self._list_access_requests(task_id) active_run = await self._active_run(task) work_order_snapshot = None if task["task_type"] == "external_work_order" and task["source"].get("external_id"): @@ -249,10 +254,9 @@ class OperationalTaskService: ]) if SCADA_VIEW in self.permissions: scopes.extend(["scada_devices", "scada_readings"]) - if hasattr(self.repo, "list_access_requests"): - for request in await self.repo.list_access_requests(str(task_id)): - if request["status"] == "approved": - scopes.extend(request["approved_scopes"]) + for request in await self._list_access_requests(task_id): + if request["status"] == "approved": + scopes.extend(request["approved_scopes"]) return AgentTaskContext( task=await self.detail(task_id), approved_data_scopes=list(dict.fromkeys(scopes)), @@ -263,8 +267,6 @@ class OperationalTaskService: permission = SCADA_VIEW if source.startswith("scada_") else SIMULATION_VIEW if permission in self.permissions: return True - if not hasattr(self.repo, "list_access_requests"): - return False active_run = await self._active_run(task) if active_run is None or active_run.get("status") not in {"claimed", "running"}: return False @@ -273,7 +275,7 @@ class OperationalTaskService: self._require_active_run_lease(active_run, self._now()) return any( row["status"] == "approved" and source in row["approved_scopes"] - for row in await self.repo.list_access_requests(str(task_id)) + for row in await self._list_access_requests(task_id) ) async def request_access(self, task_id: UUID, command: TaskAccessRequestCreate) -> TaskAccessRequest: @@ -297,7 +299,7 @@ class OperationalTaskService: "SESSION_MISMATCH", "Current Agent session must own the task run", status=409) self._require_active_run_lease(active_run, self._now()) pending = [ - item for item in await self.repo.list_access_requests(str(task_id)) + item for item in await self._list_access_requests(task_id) if item["status"] == "pending" ] require( @@ -721,7 +723,7 @@ class OperationalTaskService: if command.plan.schema_name == "execution-plan@2": requests = { row["access_request_id"]: TaskAccessRequest.model_validate(row) - for row in await self.repo.list_access_requests(str(task_id)) + for row in await self._list_access_requests(task_id) } for node in command.plan.nodes: if node.kind == "gate": @@ -1357,16 +1359,15 @@ class OperationalTaskService: now=now, reason="external_source_withdrawn", ) - if hasattr(self.repo, "list_access_requests"): - for access_request in await self.repo.list_access_requests(str(task_id)): - await self._resolve_notifications( - task_id, - kinds=["access_approval_pending"], - subject_type="access_request", - subject_id=UUID(str(access_request["access_request_id"])), - now=now, - reason="external_source_withdrawn", - ) + for access_request in await self._list_access_requests(task_id): + await self._resolve_notifications( + task_id, + kinds=["access_approval_pending"], + subject_type="access_request", + subject_id=UUID(str(access_request["access_request_id"])), + now=now, + reason="external_source_withdrawn", + ) await self._resolve_notifications( task_id, kinds=["human_input_required", "run_delivery_degraded", "task_failed"], @@ -1667,7 +1668,7 @@ class OperationalTaskService: ] access_requests = [ TaskAccessRequest.model_validate(item) - for item in await self.repo.list_access_requests(str(task_id)) + for item in await self._list_access_requests(task_id) ] lifecycle = self._status_from_plan( plan.plan.nodes, actions, access_requests @@ -1712,10 +1713,7 @@ class OperationalTaskService: ) pending_access = any( request.get("status") == "pending" - for request in ( - await self.repo.list_access_requests(str(task_id)) - if hasattr(self.repo, "list_access_requests") else [] - ) + for request in await self._list_access_requests(task_id) ) if not pending_actions and not pending_access: await self._notify(