feat(workbench): add dynamic task approval lifecycle

2026-09-24 16:36:12 +08:00
parent db092b0f5e
commit ad813a7c16
35 changed files with 2956 additions and 564 deletions
@@ -11,6 +11,7 @@ from app.auth.permissions import (
    OPERATIONAL_ACTION_WORK_ORDER_APPROVE,
    OPERATIONAL_ACTION_WORK_ORDER_EXECUTE,
    OPERATIONAL_TASK_EXECUTE,
+    OPERATIONAL_TASK_ACCESS_APPROVE,
    OPERATIONAL_TASK_EXTERNAL_SYNC,
    OPERATIONAL_TASK_WRITE,
    SCADA_VIEW,
@@ -22,6 +23,11 @@ from app.domain.schemas.operational_task import (
    ActionExecuteCommand,
    ActionProposalCommand,
    ActionReconcileCommand,
+    WorkOrderDraftUpdateCommand,
+    WorkOrderConfirmDispatchCommand,
+    TaskAccessRequest,
+    TaskAccessRequestCreate,
+    TaskAccessRequestDecision,
    AgentArtifactsCommand,
    AgentArtifactsResult,
    AgentTaskContext,
@@ -35,6 +41,7 @@ from app.domain.schemas.operational_task import (
    ExecutionPlanSnapshot,
    ExternalFeedbackCommand,
    ExternalFeedbackResult,
+    ExternalSourceWithdrawalCommand,
    HumanApprovalAction,
    HumanEvidenceCommand,
    InvestigationReportRevision,
@@ -54,6 +61,7 @@ from app.domain.schemas.operational_task import (
    StoredEvidenceRecord,
 )
 from app.domain.schemas.operational_data import TaskDataSnapshot
+from app.domain.schemas.operational_data import TaskDataSource
 from app.services.content_digest import digest


@@ -124,6 +132,7 @@ class OperationalTaskService:
            can_execute_work_orders=OPERATIONAL_ACTION_WORK_ORDER_EXECUTE in permissions,
            can_decide_commands=OPERATIONAL_ACTION_COMMAND_APPROVE in permissions,
            can_execute_commands=OPERATIONAL_ACTION_COMMAND_EXECUTE in permissions,
+            can_decide_access=OPERATIONAL_TASK_ACCESS_APPROVE in permissions,
        )

    async def create(self, command: OperationalTaskCreate) -> OperationalTaskDetail:
@@ -210,6 +219,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 []
        active_run = await self._active_run(task)
        work_order_snapshot = None
        if task["task_type"] == "external_work_order" and task["source"].get("external_id"):
@@ -225,6 +235,7 @@ class OperationalTaskService:
            latest_report=report,
            evidence=evidence,
            actions=actions,
+            access_requests=access_requests,
            permissions=self._permission_view(),
            work_order_snapshot=work_order_snapshot,
        )
@@ -238,11 +249,130 @@ 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"])
        return AgentTaskContext(
            task=await self.detail(task_id),
-            approved_data_scopes=scopes,
+            approved_data_scopes=list(dict.fromkeys(scopes)),
        )

+    async def can_read_source(self, task_id: UUID, source: TaskDataSource) -> bool:
+        task = await self._task(task_id)
+        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
+        if not active_run.get("agent_session_id"):
+            return False
+        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))
+        )
+
+    async def request_access(self, task_id: UUID, command: TaskAccessRequestCreate) -> TaskAccessRequest:
+        body = _json(command)
+        scope = f"task:{task_id}:access:request"
+        async with self.repo.transaction():
+            replay = await self._request_replay(scope, command.request_id, body)
+            if replay:
+                return TaskAccessRequest.model_validate(replay["access_request"])
+            availability = getattr(self.repo, "access_requests_available", None)
+            require(
+                availability is None or await availability(),
+                "DEPENDENCY_UNAVAILABLE",
+                "Task data access requests are not enabled for this project database",
+                status=503,
+            )
+            await self.repo.lock(f"task:{task_id}")
+            task = await self._task(task_id)
+            active_run = await self._active_run(task)
+            require(active_run is not None and active_run.get("agent_session_id") == command.source_session_id,
+                    "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))
+                if item["status"] == "pending"
+            ]
+            require(
+                not pending,
+                "ACCESS_REQUEST_PENDING",
+                "Resolve the current task data access request before creating another",
+                refs=[item["access_request_id"] for item in pending],
+                status=409,
+            )
+            now = self._now()
+            request = TaskAccessRequest(
+                access_request_id=uuid4(), task_id=task_id, node_id=command.node_id,
+                scopes=list(dict.fromkeys(command.scopes)), reason=command.reason,
+                status="pending", approved_scopes=[], version=1,
+                requested_by=self._actor(), requested_at=now,
+            )
+            await self.repo.insert_access_request(str(task_id), _json(request))
+            task.update(revision=task["revision"] + 1, updated_at=now)
+            await self.repo.update_task(_task_json(OperationalTaskSummary.model_validate(task)))
+            await self._task_event(task, now)
+            await self._notify(task_id, "access_approval_pending", task["title"], command.reason, now,
+                               required_permission=OPERATIONAL_TASK_ACCESS_APPROVE,
+                               subject_type="access_request", subject_id=request.access_request_id)
+            await self._save_request(scope, command.request_id, body, {"access_request": _json(request)})
+            return request
+
+    async def decide_access(self, task_id: UUID, access_request_id: UUID,
+                            command: TaskAccessRequestDecision) -> TaskAccessRequest:
+        body = _json(command)
+        scope = f"task:{task_id}:access:{access_request_id}:decision"
+        async with self.repo.transaction():
+            replay = await self._request_replay(scope, command.request_id, body)
+            if replay:
+                return TaskAccessRequest.model_validate(replay["access_request"])
+            require(OPERATIONAL_TASK_ACCESS_APPROVE in self.permissions, "PERMISSION_DENIED",
+                    "Current user cannot approve task data access", status=403)
+            await self.repo.lock(f"access:{access_request_id}")
+            raw = await self.repo.get_access_request(str(task_id), str(access_request_id))
+            require(raw is not None, "ACCESS_REQUEST_NOT_FOUND", "Access request not found", status=404)
+            request = TaskAccessRequest.model_validate(raw)
+            require(request.version == command.expected_version, "VERSION_CONFLICT",
+                    "Access request version changed", status=409)
+            require(request.status == "pending", "INVALID_TRANSITION",
+                    "Access request has already been decided", status=409)
+            approved = set(command.approved_scopes)
+            require(approved <= set(request.scopes), "ACCESS_SCOPE_INVALID",
+                    "Approved scopes must be requested", status=422)
+            require((command.decision == "approved") == bool(approved), "ACCESS_SCOPE_INVALID",
+                    "Approval needs at least one scope; rejection grants none", status=422)
+            for source in approved:
+                permission = SCADA_VIEW if source.startswith("scada_") else SIMULATION_VIEW
+                require(permission in self.permissions, "PERMISSION_DENIED",
+                        "Approver cannot read a requested source", status=403)
+            now = self._now()
+            updated = request.model_copy(update={
+                "version": request.version + 1, "status": command.decision,
+                "approved_scopes": sorted(approved), "decided_by": self._actor(),
+                "decided_at": now, "comment": command.comment,
+            })
+            await self.repo.update_access_request(str(task_id), _json(updated), request.version)
+            await self._resolve_notifications(
+                task_id,
+                kinds=["access_approval_pending"],
+                subject_type="access_request",
+                subject_id=access_request_id,
+                now=now,
+                reason=f"access_request_{command.decision}",
+            )
+            task = await self._task(task_id)
+            await self._requeue_task(task, now)
+            await self.repo.update_task(_task_json(OperationalTaskSummary.model_validate(task)))
+            await self._task_event(task, now)
+            await self._save_request(scope, command.request_id, body, {"access_request": _json(updated)})
+            return updated
+
    async def list_runs(
        self, *, status: str | None, limit: int,
        cursor_created_at=None, cursor_run_id=None,
@@ -305,6 +435,14 @@ class OperationalTaskService:
            await self.repo.update_task(_task_json(OperationalTaskSummary.model_validate(task)))
            await self._run_event(run_model, now)
            await self._task_event(task, now)
+            await self._resolve_notifications(
+                task_id,
+                kinds=["run_delivery_degraded"],
+                subject_type="task",
+                subject_id=task_id,
+                now=now,
+                reason="run_claimed",
+            )
            await self._save_request(scope, command.request_id, body, {"run": _json(run_model)})
            return run_model

@@ -420,6 +558,14 @@ class OperationalTaskService:
            await self.repo.update_task(_task_json(OperationalTaskSummary.model_validate(task)))
            await self._run_event(run, now, queued=True)
            await self._task_event(task, now)
+            await self._resolve_notifications(
+                task_id,
+                kinds=["task_failed", "run_delivery_degraded"],
+                subject_type="task",
+                subject_id=task_id,
+                now=now,
+                reason="task_retried",
+            )
            await self._save_request(scope, command.request_id, body, {"task_id": str(task_id)})
            return await self.detail(task_id)

@@ -572,6 +718,35 @@ class OperationalTaskService:
                    known_evidence,
                )
                self._validate_plan_state(command.plan, known_evidence, actions)
+                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 node in command.plan.nodes:
+                        if node.kind == "gate":
+                            gate_ref = str(node.gate_ref)
+                            access_request = requests.get(gate_ref)
+                            action = actions.get(gate_ref)
+                            require(
+                                access_request is not None or action is not None,
+                                "GATE_REFERENCE_NOT_FOUND",
+                                "Gate must reference a persisted request or action",
+                                refs=[node.id], status=409,
+                            )
+                            if node.status == "completed":
+                                require(
+                                    (
+                                        access_request is not None
+                                        and access_request.status != "pending"
+                                    ) or (
+                                        action is not None
+                                        and action.approval_status != "pending"
+                                    ),
+                                    "GATE_NOT_RESOLVED",
+                                    "A completed gate requires a completed human decision",
+                                    refs=[node.id], status=409,
+                                )
                plan_revision = (plan_revision or 0) + 1
                snapshot = ExecutionPlanSnapshot(
                    revision=plan_revision,
@@ -657,6 +832,7 @@ class OperationalTaskService:
                return HumanApprovalAction.model_validate(action)
            await self.repo.lock(f"task:{task_id}")
            task = await self._task(task_id)
+            self._require_actionable_task(task)
            self._validate_action_payload(command)
            if command.node_id:
                latest_plan = await self.repo.get_latest_plan(str(task_id))
@@ -699,12 +875,63 @@ class OperationalTaskService:
            await self._notify(
                task_id, "action_approval_pending", task["title"], command.reason, now,
                required_permission=approval_permission,
+                subject_type="action", subject_id=action.action_id,
            )
            await self._save_request(
                scope, command.request_id, body, {"action_id": str(action.action_id)}
            )
            return action

+    async def update_work_order_draft(
+        self, task_id: UUID, action_id: UUID, command: WorkOrderDraftUpdateCommand
+    ) -> HumanApprovalAction:
+        body = _json(command)
+        scope = f"task:{task_id}:action:{action_id}:draft"
+        async with self.repo.transaction():
+            replay = await self._request_replay(scope, command.request_id, body)
+            if replay:
+                return HumanApprovalAction.model_validate(replay["action"])
+            await self.repo.lock(f"task:{task_id}")
+            await self.repo.lock(f"action:{action_id}")
+            self._require_actionable_task(await self._task(task_id))
+            action = await self._action(task_id, action_id)
+            require(action.kind == "create_work_order", "INVALID_ACTION_KIND",
+                    "Only work-order drafts can be edited", status=422)
+            require(action.version == command.expected_version, "VERSION_CONFLICT",
+                    "Action version changed", status=409)
+            require(action.approval_status == "pending" and action.execution_status == "not_submitted",
+                    "INVALID_TRANSITION", "Submitted actions cannot be edited", status=409)
+            require(OPERATIONAL_ACTION_WORK_ORDER_APPROVE in self.permissions,
+                    "PERMISSION_DENIED", "Current user cannot edit this work-order draft", status=403)
+            now = self._now()
+            updated = action.model_copy(update={
+                "version": action.version + 1, "payload": command.payload,
+                "edited_by": self._actor(), "edited_at": now,
+            })
+            await self.repo.update_action(str(task_id), _json(updated))
+            await self._action_event(updated, now)
+            await self._save_request(scope, command.request_id, body, {"action": _json(updated)})
+            return updated
+
+    async def confirm_work_order_dispatch(
+        self, task_id: UUID, action_id: UUID, command: WorkOrderConfirmDispatchCommand,
+        dispatcher, credential: str | None,
+    ) -> HumanApprovalAction:
+        require(OPERATIONAL_ACTION_WORK_ORDER_APPROVE in self.permissions
+                and OPERATIONAL_ACTION_WORK_ORDER_EXECUTE in self.permissions,
+                "PERMISSION_DENIED", "Current user cannot confirm and dispatch work orders", status=403)
+        current = await self._action(task_id, action_id)
+        require(current.kind == "create_work_order", "INVALID_ACTION_KIND",
+                "Only work orders support one-step confirmation", status=422)
+        approval = await self.decide_action(task_id, action_id, ActionDecisionCommand(
+            request_id=uuid5(NAMESPACE_URL, f"confirm:{command.request_id}:approve"),
+            expected_version=command.expected_version, decision="approved",
+        ))
+        return await self.submit_action(task_id, action_id, ActionExecuteCommand(
+            request_id=uuid5(NAMESPACE_URL, f"confirm:{command.request_id}:dispatch"),
+            expected_version=approval.version,
+        ), dispatcher, credential)
+
    async def decide_action(
        self, task_id: UUID, action_id: UUID, command: ActionDecisionCommand
    ) -> HumanApprovalAction:
@@ -714,7 +941,9 @@ class OperationalTaskService:
            replay = await self._request_replay(scope, command.request_id, body)
            if replay:
                return HumanApprovalAction.model_validate(replay["action"])
+            await self.repo.lock(f"task:{task_id}")
            await self.repo.lock(f"action:{action_id}")
+            self._require_actionable_task(await self._task(task_id))
            action = await self._action(task_id, action_id)
            require(action.version == command.expected_version, "VERSION_CONFLICT",
                    "Action version changed", status=409)
@@ -730,6 +959,14 @@ class OperationalTaskService:
            })
            await self.repo.update_action(str(task_id), _json(action))
            await self._action_event(action, now)
+            await self._resolve_notifications(
+                task_id,
+                kinds=["action_pending", "action_approval_pending"],
+                subject_type="action",
+                subject_id=action_id,
+                now=now,
+                reason=f"action_{command.decision}",
+            )
            if command.decision == "rejected":
                task = await self._task(task_id)
                now = self._now()
@@ -749,7 +986,9 @@ class OperationalTaskService:
            replay = await self._request_replay(scope, command.request_id, body)
            if replay:
                return HumanApprovalAction.model_validate(replay["action"])
+            await self.repo.lock(f"task:{task_id}")
            await self.repo.lock(f"action:{action_id}")
+            self._require_actionable_task(await self._task(task_id))
            action = await self._action(task_id, action_id)
            require(action.version == command.expected_version, "VERSION_CONFLICT",
                    "Action version changed", status=409)
@@ -771,6 +1010,14 @@ class OperationalTaskService:
            })
            await self.repo.update_action(str(task_id), _json(action))
            await self._action_event(action, now)
+            await self._resolve_notifications(
+                task_id,
+                kinds=["external_coordination_failed"],
+                subject_type="action",
+                subject_id=action_id,
+                now=now,
+                reason="external_coordination_retried",
+            )
            await self._save_request(
                scope, command.request_id, body, {"action": _json(action)}
            )
@@ -795,6 +1042,7 @@ class OperationalTaskService:
                        "外部系统提交中断，执行结果未知，请核对回执",
                        self._now(),
                        required_permission=action.required_execution_permission,
+                        subject_type="action", subject_id=action_id,
                    )
            raise

@@ -822,6 +1070,16 @@ class OperationalTaskService:
                    outcome.detail,
                    outcome_recorded_at,
                    required_permission=action.required_execution_permission,
+                    subject_type="action", subject_id=action_id,
+                )
+            elif outcome.status == "succeeded":
+                await self._resolve_notifications(
+                    task_id,
+                    kinds=["external_coordination_failed"],
+                    subject_type="action",
+                    subject_id=action_id,
+                    now=outcome_recorded_at,
+                    reason="external_coordination_succeeded",
                )
            if outcome.status in {"succeeded", "failed", "unsupported"}:
                task_document = await self._task(task_id)
@@ -883,6 +1141,16 @@ class OperationalTaskService:
                    outcome.detail,
                    outcome_recorded_at,
                    required_permission=action.required_execution_permission,
+                    subject_type="action", subject_id=action_id,
+                )
+            elif outcome.status == "succeeded":
+                await self._resolve_notifications(
+                    task_id,
+                    kinds=["external_coordination_failed"],
+                    subject_type="action",
+                    subject_id=action_id,
+                    now=outcome_recorded_at,
+                    reason="external_coordination_succeeded",
                )
            if outcome.status in {"succeeded", "failed", "unsupported"}:
                task_document = await self._task(task_id)
@@ -988,6 +1256,24 @@ class OperationalTaskService:
                task.update(revision=task["revision"] + 1, updated_at=now)
            await self.repo.update_task(_task_json(OperationalTaskSummary.model_validate(task)))
            await self._task_event(task, now)
+            if task["status"] != "failed":
+                await self._resolve_notifications(
+                    task_id,
+                    kinds=["task_failed"],
+                    subject_type="task",
+                    subject_id=task_id,
+                    now=now,
+                    reason="external_status_recovered",
+                )
+            if task["status"] != "waiting_human":
+                await self._resolve_notifications(
+                    task_id,
+                    kinds=["human_input_required"],
+                    subject_type="task",
+                    subject_id=task_id,
+                    now=now,
+                    reason="external_status_progressed",
+                )
            if task["task_type"] == "external_work_order":
                await self._append_event(
                    topic="workbench.work_order.changed",
@@ -1010,6 +1296,88 @@ class OperationalTaskService:
            await self._save_request(scope, command.request_id, body, _json_dict(response))
            return ExternalFeedbackResult(task=await self.detail(task_id), **response)

+    async def withdraw_external_source(
+        self, task_id: UUID, command: ExternalSourceWithdrawalCommand
+    ) -> OperationalTaskDetail:
+        body = _json(command)
+        scope = f"task:{task_id}:withdraw-external-source"
+        async with self.repo.transaction():
+            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"
+                and source.get("kind") == "external_system"
+                and source.get("external_id") == command.source_event_id
+                and command.source_event_id.startswith("peripheral_sample:"),
+                "EXTERNAL_SOURCE_MISMATCH",
+                "Only matching peripheral work orders can be withdrawn",
+                status=409,
+            )
+            if source.get("deleted_at") is None:
+                now = self._now()
+                task["source"] = {**source, "deleted_at": now}
+                task.update(
+                    status="cancelled",
+                    lifecycle_status="cancelled",
+                    revision=task["revision"] + 1,
+                    updated_at=now,
+                )
+                await self.repo.update_task(
+                    _task_json(OperationalTaskSummary.model_validate(task))
+                )
+                await self._task_event(task, now)
+                await self._append_event(
+                    topic="workbench.work_order.changed",
+                    aggregate_type="operational_task",
+                    aggregate_id=str(task_id),
+                    aggregate_revision=task["revision"],
+                    occurred_at=now,
+                    metadata={"task_id": str(task_id), "namespace": "operational-task"},
+                )
+                for action in await self.repo.list_actions(str(task_id)):
+                    action_id = UUID(str(action["action_id"]))
+                    await self._resolve_notifications(
+                        task_id,
+                        kinds=["action_pending", "action_approval_pending"],
+                        subject_type="action",
+                        subject_id=action_id,
+                        now=now,
+                        reason="external_source_withdrawn",
+                    )
+                    if action["execution_status"] == "not_submitted":
+                        await self._resolve_notifications(
+                            task_id,
+                            kinds=["external_coordination_failed"],
+                            subject_type="action",
+                            subject_id=action_id,
+                            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",
+                        )
+                await self._resolve_notifications(
+                    task_id,
+                    kinds=["human_input_required", "run_delivery_degraded", "task_failed"],
+                    subject_type="task",
+                    subject_id=task_id,
+                    now=now,
+                    reason="external_source_withdrawn",
+                )
+            await self._save_request(scope, command.request_id, body, {"task_id": str(task_id)})
+            return await self.detail(task_id)
+
    async def acknowledge_notification(
        self, notification_id: UUID, command: NotificationAcknowledgeCommand
    ) -> OperationalTaskNotification:
@@ -1047,8 +1415,8 @@ class OperationalTaskService:
                )
                await self._append_event(
                    topic="operational.notification.changed",
-                    aggregate_type="operational_notification",
-                    aggregate_id=str(notification_id),
+                    aggregate_type="operational_notification_read",
+                    aggregate_id=f"{notification_id}:{self.context.user_id}",
                    aggregate_revision=notification.version,
                    occurred_at=notification.read_at,
                    metadata={"task_id": str(notification.task_id)},
@@ -1121,6 +1489,14 @@ class OperationalTaskService:
            await self._requeue_task(task, now)
            await self.repo.update_task(_task_json(OperationalTaskSummary.model_validate(task)))
            await self._task_event(task, now)
+            await self._resolve_notifications(
+                task_id,
+                kinds=["human_input_required"],
+                subject_type="task",
+                subject_id=task_id,
+                now=now,
+                reason="human_evidence_added",
+            )
            await self._save_request(scope, command.request_id, body, {"task_id": str(task_id)})
            return await self.detail(task_id)

@@ -1289,7 +1665,13 @@ class OperationalTaskService:
                        HumanApprovalAction.model_validate(item)
                        for item in await self.repo.list_actions(str(task_id))
                    ]
-                    lifecycle = self._status_from_plan(plan.plan.nodes, actions)
+                    access_requests = [
+                        TaskAccessRequest.model_validate(item)
+                        for item in await self.repo.list_access_requests(str(task_id))
+                    ]
+                    lifecycle = self._status_from_plan(
+                        plan.plan.nodes, actions, access_requests
+                    )
                    if lifecycle == "running":
                        final_outcome = "failed"
                        lifecycle = "failed"
@@ -1359,6 +1741,16 @@ class OperationalTaskService:
        require(task is not None, "TASK_NOT_FOUND", "Operational task not found", status=404)
        return task

+    @staticmethod
+    def _require_actionable_task(task: dict) -> None:
+        source = task.get("source") or {}
+        require(
+            task.get("lifecycle_status") != "cancelled" and source.get("deleted_at") is None,
+            "TASK_NOT_ACTIONABLE",
+            "Cancelled or withdrawn tasks do not accept action changes",
+            status=409,
+        )
+
    async def _action(self, task_id: UUID, action_id: UUID) -> HumanApprovalAction:
        value = await self.repo.get_action(str(task_id), str(action_id))
        require(value is not None, "ACTION_NOT_FOUND", "Action not found", status=404)
@@ -1385,7 +1777,11 @@ class OperationalTaskService:
        return uuid5(NAMESPACE_URL, f"operational-task:{task_id}:generation:{generation}")

    @staticmethod
-    def _status_from_plan(nodes, actions: Iterable[HumanApprovalAction]) -> str:
+    def _status_from_plan(
+        nodes,
+        actions: Iterable[HumanApprovalAction],
+        access_requests: Iterable[TaskAccessRequest] = (),
+    ) -> str:
        statuses = {node.status for node in nodes}
        if "running" in statuses:
            return "running"
@@ -1396,6 +1792,8 @@ class OperationalTaskService:
            return "failed"
        if any(action.approval_status == "pending" for action in actions):
            return "waiting_human"
+        if any(request.status == "pending" for request in access_requests):
+            return "waiting_human"
        if statuses <= {"completed", "skipped", "cancelled"}:
            return "completed"
        if "failed" in statuses and statuses <= {"completed", "failed", "skipped", "cancelled"}:
@@ -1420,7 +1818,10 @@ class OperationalTaskService:
                        "A running or completed node has incomplete predecessors",
                        refs=incomplete, status=409)
            if node.status == "completed":
-                require(bool(node.evidence_refs or node.action_refs),
+                structural = plan.schema_name == "execution-plan@2" and node.kind in {
+                    "parallel_split", "parallel_join", "gate"
+                }
+                require(structural or bool(node.evidence_refs or node.action_refs),
                        "PLAN_NODE_UNSUPPORTED",
                        "Completed nodes require evidence or an executed action receipt",
                        refs=[node.id], status=409)
@@ -1553,11 +1954,14 @@ class OperationalTaskService:
    async def _notify(
        self, task_id: UUID, kind: str, title: str, description: str, now: datetime,
        *, required_permission: str | None = None,
+        subject_type: str = "task", subject_id: UUID | None = None,
    ) -> None:
        notification = OperationalTaskNotification(
            notification_id=uuid4(), task_id=task_id, kind=kind,
            title=title, description=description, status="unread", version=1,
            created_at=now, required_permission=required_permission,
+            lifecycle_status="active", subject_type=subject_type,
+            subject_id=subject_id or task_id,
        )
        await self.repo.insert_notification(_json(notification))
        await self._append_event(
@@ -1569,6 +1973,42 @@ class OperationalTaskService:
            metadata={"task_id": str(task_id), "kind": kind},
        )

+    async def _resolve_notifications(
+        self,
+        task_id: UUID,
+        *,
+        kinds: list[str],
+        subject_type: str,
+        subject_id: UUID,
+        now: datetime,
+        reason: str,
+    ) -> None:
+        resolve = getattr(self.repo, "resolve_notifications", None)
+        if resolve is None:
+            return
+        notifications = await resolve(
+            str(task_id),
+            kinds=kinds,
+            subject_type=subject_type,
+            subject_id=str(subject_id),
+            resolved_at=now,
+            resolution_reason=reason,
+        )
+        for notification in notifications:
+            await self._append_event(
+                topic="operational.notification.changed",
+                aggregate_type="operational_notification",
+                aggregate_id=notification["notification_id"],
+                aggregate_revision=2,
+                occurred_at=now,
+                metadata={
+                    "task_id": str(task_id),
+                    "kind": notification["kind"],
+                    "change": "resolved",
+                    "reason": reason,
+                },
+            )
+
    def _new_run(
        self, task_id: UUID, generation: int, now: datetime
    ) -> OperationalTaskRun: