next-tjwater-server/app/services/operational_tasks.py

"""State machines for durable operational tasks and approved external actions."""
from __future__ import annotations

import json
from datetime import datetime, timedelta, timezone
from typing import Iterable
from uuid import NAMESPACE_URL, UUID, uuid4, uuid5

from app.auth.permissions import (
    OPERATIONAL_ACTION_COMMAND_APPROVE,
    OPERATIONAL_ACTION_COMMAND_EXECUTE,
    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,
    SIMULATION_VIEW,
)
from app.domain.domain_errors import DomainError, require
from app.domain.schemas.operational_task import (
    ActionDecisionCommand,
    ActionExecuteCommand,
    ActionProposalCommand,
    ActionReconcileCommand,
    WorkOrderDraftUpdateCommand,
    WorkOrderConfirmDispatchCommand,
    TaskAccessRequest,
    TaskAccessRequestCreate,
    TaskAccessRequestDecision,
    AgentArtifactsCommand,
    AgentArtifactsResult,
    AgentTaskContext,
    AgentRunCancelCommand,
    AgentRunCommand,
    AgentRunDispatchFailureCommand,
    AgentRunFailureCommand,
    CreateWorkOrderPayload,
    DispatchDueSchedulesResult,
    EvidenceRecord,
    ExecutionPlanSnapshot,
    ExternalFeedbackCommand,
    ExternalFeedbackResult,
    ExternalSourceWithdrawalCommand,
    HumanApprovalAction,
    HumanEvidenceCommand,
    InvestigationReportRevision,
    InvestigationReportReference,
    InvestigationProgressSnapshot,
    IssueCommandPayload,
    NotificationAcknowledgeCommand,
    OperationalTaskCreate,
    OperationalTaskDetail,
    OperationalTaskNotification,
    OperationalTaskPermissions,
    OperationalTaskSchedule,
    OperationalTaskScheduleCreate,
    OperationalTaskScheduleUpdate,
    OperationalTaskSummary,
    OperationalTaskRetryCommand,
    OperationalTaskRun,
    RecoverExpiredLeasesResult,
    StoredEvidenceRecord,
    TaskPresentationEventCommand,
    TaskPresentationEventResult,
    TaskPresentationCursor,
)
from app.domain.schemas.operational_data import TaskDataSnapshot, TaskDataSource
from app.services.content_digest import digest
from app.services.operational_presentation import resolve_overlay


LEASE_DURATION = timedelta(seconds=60)


def _json(model) -> dict:
    return model.model_dump(mode="json", by_alias=True)


def _task_json(model: OperationalTaskSummary) -> dict:
    return model.model_dump(mode="json", by_alias=True, exclude={"active_run", "latest_run"})


class OperationalTaskService:
    def __init__(self, repository, context, permissions: Iterable[str], *, clock=None):
        self.repo = repository
        self.context = context
        self.permissions = frozenset(permissions)
        self.clock = clock or (lambda: datetime.now(timezone.utc))

    def _now(self) -> datetime:
        now = self.clock()
        require(now.tzinfo is not None, "SCHEMA_INVALID", "Clock must be timezone-aware")
        return now

    def _actor(self) -> str:
        return f"principal:{self.context.user_id}"

    @property
    def _supports_runs(self) -> bool:
        return hasattr(self.repo, "get_active_run") and hasattr(self.repo, "update_run")

    async def _active_run(self, task: dict) -> dict | None:
        if self._supports_runs:
            return await self.repo.get_active_run(str(task["task_id"]))
        if task.get("status") not in {"queued", "running"}:
            return None
        now = self._now()
        return _json(OperationalTaskRun(
            run_id=uuid5(NAMESPACE_URL, f"legacy-run:{task['task_id']}:{task['execution_generation']}"),
            task_id=task["task_id"],
            generation=task["execution_generation"],
            execution_request_id=task["execution_request_id"],
            status="running" if task["status"] == "running" else "queued",
            revision=task["revision"],
            agent_session_id=task.get("agent_session_id"),
            lease_expires_at=task.get("agent_lease_expires_at"),
            heartbeat_at=task.get("updated_at") if task["status"] == "running" else None,
            started_at=task.get("created_at") if task["status"] == "running" else None,
            dispatch_attempts=0,
            next_dispatch_at=task.get("created_at") or now,
            created_at=task.get("created_at") or now,
            updated_at=task.get("updated_at") or now,
        ))

    async def _append_event(self, **kwargs) -> int | None:
        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(
            can_publish_artifacts=OPERATIONAL_TASK_EXECUTE in permissions,
            can_propose_actions=OPERATIONAL_TASK_WRITE in permissions,
            can_sync_external_feedback=OPERATIONAL_TASK_EXTERNAL_SYNC in permissions,
            can_decide_work_orders=OPERATIONAL_ACTION_WORK_ORDER_APPROVE in permissions,
            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:
        body = _json(command)
        request_id = str(command.request_id)
        scope = "task:create"
        async with self.repo.transaction():
            await self.repo.lock(f"request:{scope}:{request_id}")
            prior = await self.repo.get_request(scope, request_id)
            if prior:
                self._require_same_request(prior, body)
                return await self.detail(UUID(prior["response"]["task_id"]))
            if command.source.external_id:
                await self.repo.lock(
                    f"external-source:{command.source.kind}:{command.source.external_id}"
                )
                existing = await self.repo.get_task_by_external_source(
                    command.source.kind, command.source.external_id
                )
                if existing:
                    await self.repo.save_request(
                        scope, request_id, digest(body), {"task_id": existing["task_id"]}
                    )
                    return await self.detail(UUID(existing["task_id"]))
            now = self._now()
            task_id = uuid4()
            task = OperationalTaskSummary(
                project_id=self.context.project_id,
                task_id=task_id,
                task_type=command.task_type,
                workflow=command.workflow,
                lifecycle_status="open",
                status="queued",
                priority=command.priority,
                revision=1,
                execution_generation=1,
                execution_request_id=self._execution_request_id(task_id, 1),
                title=command.title,
                summary=command.summary,
                source=command.source,
                trigger=command.trigger,
                scheduled_for=command.scheduled_for,
                agent_session_id=command.agent_session_id,
                agent_lease_expires_at=None,
                current_plan_revision=None,
                current_report_revision=None,
                created_at=now,
                updated_at=now,
            )
            task_document = _task_json(task)
            await self.repo.insert_task(task_document, request_id)
            if command.task_type != "external_work_order":
                run = self._new_run(task_id, 1, now)
                await self.repo.insert_run(_json(run))
                await self._run_event(run, now, queued=True)
            await self.repo.save_request(
                scope, request_id, digest(body), {"task_id": str(task_id)}
            )
            await self._task_event(task_document, now)
            return await self.detail(task_id)

    async def list(
        self, *, task_type: str | None, status: str | None, limit: int, offset: int,
        peripheral_work_order_id: UUID | None = None,
    ) -> tuple[list[OperationalTaskSummary], int | None]:
        rows = await self.repo.list_tasks(
            task_type=task_type, status=status, limit=limit + 1, offset=offset,
            **({"peripheral_work_order_id": str(peripheral_work_order_id)}
               if peripheral_work_order_id is not None else {}),
        )
        items: list[OperationalTaskSummary] = []
        for row in rows[:limit]:
            active_run = row.get("active_run") if "active_run" in row else await self._active_run(row)
            items.append(OperationalTaskSummary.model_validate({
                **row,
                "active_run": active_run,
            }))
        return items, offset + limit if len(rows) > limit else None

    async def detail(self, task_id: UUID) -> OperationalTaskDetail:
        task = await self.repo.get_task(str(task_id))
        require(task is not None, "TASK_NOT_FOUND", "Operational task not found", status=404)
        plan = await self.repo.get_latest_plan(str(task_id))
        report = await self.repo.get_latest_report(str(task_id))
        if report is not None:
            report = await self._report_with_progress_diagram(task_id, report)
        evidence = [
            self._public_evidence(document)
            for document in await self.repo.list_evidence(str(task_id))
        ]
        actions = await self.repo.list_actions(str(task_id))
        access_requests = await self._list_access_requests(task_id)
        active_run = await self._active_run(task)
        latest_run = (await self.repo.get_latest_run(str(task_id))
                      if hasattr(self.repo, "get_latest_run") else active_run)
        progress = await self.repo.get_latest_progress(str(task_id), task["execution_generation"])
        work_order_snapshot = None
        if task["task_type"] == "external_work_order" and task["source"].get("external_id"):
            feedback = await self.repo.get_feedback(
                str(task_id), task["source"]["external_id"]
            )
            if feedback:
                work_order_snapshot = feedback["document"].get("work_order_snapshot")
        return OperationalTaskDetail(
            **{key: value for key, value in task.items() if key not in {"active_run", "latest_run"}},
            active_run=active_run,
            latest_run=latest_run,
            latest_plan=plan,
            latest_report=report,
            latest_progress=progress,
            evidence=evidence,
            actions=actions,
            access_requests=access_requests,
            permissions=self._permission_view(),
            work_order_snapshot=work_order_snapshot,
        )

    async def _report_with_progress_diagram(
        self, task_id: UUID, value: dict | InvestigationReportRevision
    ) -> InvestigationReportRevision:
        """Archive an eligible recorded graph, without inventing completion states.

        Also projects older reports that predate graph archival. Read paths never
        rewrite their stored document or borrow a newer round's progress.
        """
        snapshot = InvestigationReportRevision.model_validate(value)
        if (snapshot.report.diagram is not None or snapshot.report.stage != "final"
                or snapshot.execution_generation is None):
            return snapshot
        stored = await self.repo.get_latest_progress(str(task_id), snapshot.execution_generation)
        if stored is None:
            return snapshot
        progress = InvestigationProgressSnapshot.model_validate(stored)
        diagram = progress.progress.diagram
        if (diagram is None or progress.execution_generation != snapshot.execution_generation
                or progress.source_session_id != snapshot.source_session_id
                or progress.published_at > snapshot.published_at
                or any(not set(node.evidence_refs) <= set(snapshot.report.evidence_refs)
                       for node in diagram.node_refs)):
            return snapshot
        return snapshot.model_copy(update={
            "report": snapshot.report.model_copy(update={"diagram": diagram}),
        })

    async def report(self, task_id: UUID, revision: int) -> InvestigationReportRevision:
        await self._task(task_id)
        value = await self.repo.get_report(str(task_id), revision)
        require(value is not None, "REPORT_NOT_FOUND", "Report revision not found", status=404)
        return await self._report_with_progress_diagram(task_id, value)

    async def list_reports(
        self, task_id: UUID, limit: int, offset: int
    ) -> tuple[list[InvestigationReportRevision], int | None]:
        await self._task(task_id)
        rows = await self.repo.list_reports(str(task_id), limit + 1, offset)
        return [await self._report_with_progress_diagram(task_id, row) for row in rows[:limit]], (
            offset + limit if len(rows) > limit else None
        )

    async def agent_context(self, task_id: UUID, report_revision: int | None = None) -> AgentTaskContext:
        scopes: list[str] = []
        if SIMULATION_VIEW in self.permissions:
            scopes.extend([
                "workbench_runtime", "peripheral_points", "peripheral_work_orders",
                "peripheral_observations", "peripheral_test_data",
            ])
        if SCADA_VIEW in self.permissions:
            scopes.extend(["scada_devices", "scada_readings"])
        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)),
            selected_report=await self.report(task_id, report_revision) if report_revision is not None else None,
        )

    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
        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._list_access_requests(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._list_access_requests(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,
        task_type: str | None = None, task_created_from: datetime | None = None,
    ) -> tuple[list[OperationalTaskRun], str | None]:
        rows = await self.repo.list_runs(
            status=status,
            limit=limit + 1,
            cursor_created_at=cursor_created_at,
            cursor_run_id=cursor_run_id,
            task_type=task_type,
            task_created_from=task_created_from,
        )
        items = [OperationalTaskRun.model_validate(row) for row in rows[:limit]]
        next_cursor = None
        if len(rows) > limit and items:
            last = items[-1]
            next_cursor = f"{last.created_at.isoformat()}|{last.run_id}"
        return items, next_cursor

    async def claim_run(
        self, task_id: UUID, run_id: UUID, command: AgentRunCommand
    ) -> OperationalTaskRun:
        body = _json(command)
        scope = f"run:{run_id}:claim"
        async with self.repo.transaction():
            await self.repo.require_execution_storage()
            replay = await self._request_replay(scope, command.request_id, body)
            if replay:
                return OperationalTaskRun.model_validate(replay["run"])
            await self.repo.lock(f"run:{run_id}")
            run = await self._run(task_id, run_id)
            require(
                str(command.request_id) == str(run["execution_request_id"]),
                "EXECUTION_REQUEST_MISMATCH",
                "Claim request does not match this run",
                status=409,
            )
            require(run["revision"] == command.expected_revision, "VERSION_CONFLICT",
                    "Run revision changed", status=409)
            require(run["status"] == "queued", "INVALID_TRANSITION",
                    "Only queued runs can be claimed", status=409)
            now = self._now()
            expected = run["revision"]
            run.update(
                status="running",
                revision=expected + 1,
                agent_session_id=command.source_session_id,
                lease_expires_at=now + LEASE_DURATION,
                heartbeat_at=now,
                started_at=run.get("started_at") or now,
                updated_at=now,
            )
            run_model = OperationalTaskRun.model_validate(run)
            await self.repo.update_run(_json(run_model), expected)
            task = await self._task(task_id)
            task.update(
                status="running",
                agent_session_id=command.source_session_id,
                agent_lease_expires_at=now + LEASE_DURATION,
                revision=task["revision"] + 1,
                updated_at=now,
            )
            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

    async def heartbeat_run(
        self, task_id: UUID, run_id: UUID, command: AgentRunCommand
    ) -> OperationalTaskRun:
        body = _json(command)
        scope = f"run:{run_id}:heartbeat"
        async with self.repo.transaction():
            replay = await self._request_replay(scope, command.request_id, body)
            if replay:
                return OperationalTaskRun.model_validate(replay["run"])
            await self.repo.lock(f"run:{run_id}")
            run = await self._run(task_id, run_id)
            require(run["revision"] == command.expected_revision, "VERSION_CONFLICT",
                    "Run revision changed", status=409)
            require(run["status"] in {"claimed", "running"}, "INVALID_TRANSITION",
                    "Only an active run has a lease", status=409)
            require(run.get("agent_session_id") == command.source_session_id,
                    "SESSION_MISMATCH", "Agent session does not own this run", status=409)
            now = self._now()
            self._require_active_run_lease(run, now)
            expected = run["revision"]
            run.update(
                status="running",
                revision=expected + 1,
                heartbeat_at=now,
                lease_expires_at=now + LEASE_DURATION,
                updated_at=now,
            )
            run_model = OperationalTaskRun.model_validate(run)
            await self.repo.update_run(_json(run_model), expected)
            # Deliberately no task update and no browser event: heartbeats are
            # lease maintenance, not business-state changes.
            await self._save_request(scope, command.request_id, body, {"run": _json(run_model)})
            return run_model

    async def complete_run(
        self, task_id: UUID, run_id: UUID, command: AgentRunCommand
    ) -> OperationalTaskDetail:
        return await self._finish_run(task_id, run_id, command, outcome="succeeded")

    async def task_runs(self, task_id: UUID, limit: int, offset: int) -> list[OperationalTaskRun]:
        await self._task(task_id)
        return [OperationalTaskRun.model_validate(row)
                for row in await self.repo.list_task_runs(str(task_id), limit, offset)]

    async def task_run(self, task_id: UUID, run_id: UUID) -> OperationalTaskRun:
        return OperationalTaskRun.model_validate(await self._run(task_id, run_id))

    async def _fail_run_messages(self, run_id, reason: str, now: datetime) -> None:
        if hasattr(self.repo, "fail_run_messages"):
            await self.repo.fail_run_messages(str(run_id), reason, now)

    async def fail_run(
        self, task_id: UUID, run_id: UUID, command: AgentRunFailureCommand
    ) -> OperationalTaskDetail:
        return await self._finish_run(
            task_id, run_id, command, outcome="failed",
            failure_code=command.failure_code, failure_detail=command.failure_detail,
        )

    async def cancel_run(
        self, task_id: UUID, run_id: UUID, command: AgentRunCancelCommand
    ) -> OperationalTaskDetail:
        body = _json(command)
        scope = f"run:{run_id}:cancel"
        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"run:{run_id}")
            run = await self._run(task_id, run_id)
            require(run["revision"] == command.expected_revision, "VERSION_CONFLICT",
                    "Run revision changed", status=409)
            require(run["status"] in {"queued", "claimed", "running"},
                    "INVALID_TRANSITION", "Run is already terminal", status=409)
            now = self._now()
            expected = run["revision"]
            run.update(
                status="cancelled", revision=expected + 1, finished_at=now,
                lease_expires_at=None, failure_code="CANCELLED",
                failure_detail=command.reason, updated_at=now,
            )
            run_model = OperationalTaskRun.model_validate(run)
            await self.repo.update_run(_json(run_model), expected)
            await self._fail_run_messages(run_id, "Run cancelled", now)
            task = await self._task(task_id)
            task.update(
                lifecycle_status="cancelled", status="cancelled",
                revision=task["revision"] + 1, agent_session_id=None,
                agent_lease_expires_at=None, updated_at=now,
            )
            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._save_request(scope, command.request_id, body, {"task_id": str(task_id)})
            return await self.detail(task_id)

    async def retry_task(
        self, task_id: UUID, command: OperationalTaskRetryCommand
    ) -> OperationalTaskDetail:
        body = _json(command)
        scope = f"task:{task_id}:retry"
        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)
            require(task["task_type"] != "external_work_order", "INVALID_TRANSITION",
                    "External work orders do not have Agent runs", status=409)
            require(await self.repo.get_active_run(str(task_id)) is None,
                    "ACTIVE_RUN_EXISTS", "Task already has an active run", status=409)
            now = self._now()
            generation = task["execution_generation"] + 1
            run = self._new_run(task_id, generation, now)
            await self.repo.insert_run(_json(run))
            task.update(
                lifecycle_status="open", status="queued",
                revision=task["revision"] + 1,
                execution_generation=generation,
                execution_request_id=str(run.execution_request_id),
                agent_session_id=None, agent_lease_expires_at=None, updated_at=now,
            )
            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)

    async def record_dispatch_failure(
        self, task_id: UUID, run_id: UUID, command: AgentRunDispatchFailureCommand
    ) -> OperationalTaskRun:
        body = _json(command)
        scope = f"run:{run_id}:dispatch-failure"
        async with self.repo.transaction():
            replay = await self._request_replay(scope, command.request_id, body)
            if replay:
                return OperationalTaskRun.model_validate(replay["run"])
            await self.repo.lock(f"run:{run_id}")
            run = await self._run(task_id, run_id)
            require(run["revision"] == command.expected_revision, "VERSION_CONFLICT",
                    "Run revision changed", status=409)
            require(run["status"] == "queued", "INVALID_TRANSITION",
                    "Only queued runs accept dispatch failures", status=409)
            now = self._now()
            expected = run["revision"]
            attempts = run["dispatch_attempts"] + 1
            retry_delay = min(300, 5 * (2 ** min(attempts - 1, 6)))
            run.update(
                revision=expected + 1,
                dispatch_attempts=attempts,
                failure_code=command.failure_code,
                failure_detail=command.failure_detail,
                next_dispatch_at=now + timedelta(seconds=retry_delay),
                updated_at=now,
            )
            if not command.retryable:
                run.update(status="failed", finished_at=now, next_dispatch_at=now)
            run_model = OperationalTaskRun.model_validate(run)
            await self.repo.update_run(_json(run_model), expected)
            await self._run_event(run_model, now, queued=command.retryable)
            if attempts == 10:
                task = await self._task(task_id)
                await self._notify(
                    task_id, "run_delivery_degraded", task["title"],
                    "Agent 执行投递连续失败 10 次，请检查服务连接", now,
                )
            if not command.retryable:
                await self._fail_run_messages(run_id, "Run dispatch failed", now)
                task = await self._task(task_id)
                task.update(
                    lifecycle_status="failed", status="failed",
                    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._save_request(scope, command.request_id, body, {"run": _json(run_model)})
            return run_model

    async def recover_expired_leases(
        self, limit: int, *, task_type: str | None = None, task_created_from: datetime | None = None,
    ) -> RecoverExpiredLeasesResult:
        now = self._now()
        recovered: list[UUID] = []
        async with self.repo.transaction():
            for run in await self.repo.claim_expired_runs(
                limit, task_type=task_type, task_created_from=task_created_from,
            ):
                # Lease recovery already holds a row lock. Try rather than wait
                # for the shared run lock to avoid reversing the lock order of
                # message/finish transactions; a busy run is checked next tick.
                if hasattr(self.repo, "try_lock") and not await self.repo.try_lock(f"run:{run['run_id']}"):
                    continue
                task_id = UUID(str(run["task_id"]))
                expected = run["revision"]
                run.update(
                    status="timed_out", revision=expected + 1, finished_at=now,
                    lease_expires_at=None, failure_code="LEASE_TIMEOUT",
                    failure_detail="Agent run lease expired", updated_at=now,
                )
                timed_out = OperationalTaskRun.model_validate(run)
                await self.repo.update_run(_json(timed_out), expected)
                await self._fail_run_messages(run["run_id"], "Run lease expired", now)
                await self._run_event(timed_out, now)
                task = await self._task(task_id)
                timeout_count = await self.repo.consecutive_timed_out_runs(str(task_id))
                task.update(
                    revision=task["revision"] + 1, agent_session_id=None,
                    agent_lease_expires_at=None, updated_at=now,
                )
                if timeout_count >= 3:
                    task.update(lifecycle_status="failed", status="failed")
                    await self._notify(
                        task_id, "task_failed", task["title"],
                        "Agent 连续三次执行超时，任务已停止自动重试", now,
                    )
                else:
                    generation = task["execution_generation"] + 1
                    next_run = self._new_run(task_id, generation, now)
                    await self.repo.insert_run(_json(next_run))
                    task.update(
                        lifecycle_status="open", status="queued",
                        execution_generation=generation,
                        execution_request_id=str(next_run.execution_request_id),
                    )
                    await self._run_event(next_run, now, queued=True)
                await self.repo.update_task(_task_json(OperationalTaskSummary.model_validate(task)))
                await self._task_event(task, now)
                recovered.append(task_id)
        return RecoverExpiredLeasesResult(task_ids=recovered, recovered_at=now)

    async def publish_artifacts(
        self, task_id: UUID, command: AgentArtifactsCommand
    ) -> AgentArtifactsResult:
        body = _json(command)
        scope = f"task:{task_id}:artifacts"
        async with self.repo.transaction():
            replay = await self._request_replay(scope, command.request_id, body)
            if replay:
                detail = await self.detail(task_id)
                return AgentArtifactsResult(task=detail, **replay)
            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, "RUN_NOT_FOUND",
                    "Task has no active Agent run", status=409)
            if command.run_id is not None:
                require(str(command.run_id) == str(active_run["run_id"]),
                        "EXECUTION_GENERATION_MISMATCH",
                        "Artifacts target a stale run", status=409)
            require(active_run.get("agent_session_id") == command.source_session_id,
                    "SESSION_MISMATCH", "Artifacts must come from the claimed Agent session",
                    status=409)
            require(active_run["status"] in {"claimed", "running"}, "INVALID_TRANSITION",
                    "Only a running run accepts Agent artifacts", status=409)
            now = self._now()
            self._require_active_run_lease(active_run, now)
            previous_report = await self.repo.get_latest_report(str(task_id))
            require(not previous_report or previous_report.get("execution_generation") != task["execution_generation"]
                    or previous_report["report"].get("outcome") != "insufficient_data",
                    "INVESTIGATION_ENDED", "Insufficient-data investigation requires a new run", status=409)
            known_evidence = {item["evidence_id"] for item in await self.repo.list_evidence(str(task_id))}
            for item in command.evidence:
                document = _json(EvidenceRecord(
                    **_json(item), recorded_at=now, recorded_by=self._actor()
                ))
                prior = await self.repo.get_evidence(str(task_id), item.evidence_id)
                if prior:
                    require(
                        digest(self._evidence_facts(prior)) == digest(_json(item)),
                        "EVIDENCE_CONFLICT",
                            "Evidence ids are immutable", status=409)
                else:
                    require(item.kind == "synthetic", "EVIDENCE_ORIGIN_INVALID",
                            "Agent may only add explicitly synthetic evidence; system, external, and human evidence must use their trusted ingestion endpoints",
                            status=422)
                    await self.repo.insert_evidence(str(task_id), document)
                known_evidence.add(item.evidence_id)
            plan_revision = task.get("current_plan_revision")
            report_revision = task.get("current_report_revision")
            current_plan = None
            actions = {
                item["action_id"]: HumanApprovalAction.model_validate(item)
                for item in await self.repo.list_actions(str(task_id))
            }
            progress_revision = None
            if command.progress is not None:
                require(command.run_id is not None, "RUN_REQUIRED",
                        "Progress must identify its current execution run", status=422)
                if command.progress.diagram:
                    for node in command.progress.diagram.node_refs:
                        self._require_evidence_refs(node.evidence_refs, known_evidence)
                        require(all(str(ref) in actions for ref in node.action_refs),
                                "ACTION_NOT_FOUND", "Progress references an unknown action", status=422)
                previous = await self.repo.get_latest_progress(str(task_id), task["execution_generation"])
                progress_revision = (previous["revision"] if previous else 0) + 1
                snapshot = InvestigationProgressSnapshot(
                    revision=progress_revision, run_id=active_run["run_id"],
                    execution_generation=task["execution_generation"],
                    source_session_id=command.source_session_id, published_at=now, progress=command.progress,
                )
                await self.repo.save_progress(str(task_id), _json(snapshot))
                await self._append_event(
                    topic="operational.progress.changed", aggregate_type="operational_task_progress",
                    aggregate_id=str(task_id), aggregate_revision=progress_revision, occurred_at=now,
                    metadata={"task_id": str(task_id), "run_id": str(active_run["run_id"]),
                              "execution_generation": task["execution_generation"]},
                )
            if command.plan:
                self._require_evidence_refs(
                    (ref for node in command.plan.nodes for ref in node.evidence_refs),
                    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._list_access_requests(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,
                    execution_generation=task["execution_generation"],
                    source_session_id=command.source_session_id,
                    published_at=now, published_by=self._actor(), plan=command.plan,
                )
                await self.repo.insert_plan(str(task_id), _json(snapshot))
                current_plan = snapshot
            if command.report:
                self._require_evidence_refs(command.report.evidence_refs, known_evidence)
                for gap in command.report.data_gaps:
                    require(set(gap.evidence_refs) <= set(command.report.evidence_refs),
                            "REPORT_GAP_EVIDENCE_INVALID",
                            "Gap evidence must be included in report evidence", status=422)
                if command.report.diagram:
                    for node in command.report.diagram.node_refs:
                        self._require_evidence_refs(node.evidence_refs, set(command.report.evidence_refs))
                        require(all(str(ref) in actions for ref in node.action_refs),
                                "ACTION_NOT_FOUND", "Diagram references an unknown action", status=422)
                if command.report.outcome == "insufficient_data":
                    require(not any(action.approval_status == "pending" or (
                        action.approval_status == "approved" and action.execution_status in {"not_submitted", "submitted", "unknown"}
                    ) for action in actions.values()), "REPORT_ACTION_PENDING",
                            "Resolve existing actions before ending for insufficient data", status=409)
                report_revision = (report_revision or 0) + 1
                report = InvestigationReportRevision(
                    revision=report_revision,
                    execution_generation=task["execution_generation"],
                    source_session_id=command.source_session_id,
                    published_at=now, published_by=self._actor(), report=command.report,
                )
                report = await self._report_with_progress_diagram(task_id, report)
                await self.repo.insert_report(str(task_id), _json(report))
                await self._append_event(
                    topic="operational.report.published",
                    aggregate_type="operational_report",
                    aggregate_id=str(task_id),
                    aggregate_revision=report_revision,
                    occurred_at=now,
                    metadata={"task_id": str(task_id), "run_id": str(active_run["run_id"])},
                )
                previous_result = previous_report["report"] if previous_report else None
                if command.report.stage == "final" and (
                    previous_report is None
                    or previous_report.get("execution_generation") != task["execution_generation"]
                    or previous_result.get("stage") != "final"
                    or any(previous_result.get(key, [] if key == "ui_blocks" else None) != _json(command.report).get(key)
                           for key in ("current_conclusion", "conclusion_boundary", "outcome", "stop_reason",
                                       "evidence_refs", "recommendations", "pending_items", "data_gaps", "ui_blocks"))
                ):
                    await self._resolve_notifications(
                        task_id, kinds=["report_updated"], subject_type="task", subject_id=task_id,
                        now=now, reason="report_superseded",
                    )
                    await self._notify(
                        task_id, "report_updated", task["title"], command.report.current_conclusion, now,
                        report_reference=InvestigationReportReference(
                            execution_generation=task["execution_generation"], revision=report_revision,
                        ),
                    )
                if current_plan is None:
                    stored_plan = await self.repo.get_latest_plan(str(task_id))
                    if stored_plan is not None:
                        current_plan = ExecutionPlanSnapshot.model_validate(stored_plan)
            task.update(
                revision=task["revision"] + 1,
                current_plan_revision=plan_revision,
                current_report_revision=report_revision,
                agent_lease_expires_at=now + LEASE_DURATION,
                updated_at=now,
            )
            await self.repo.update_task(_task_json(OperationalTaskSummary.model_validate(task)))
            run_expected = active_run["revision"]
            active_run.update(
                revision=run_expected + 1,
                heartbeat_at=now,
                lease_expires_at=now + LEASE_DURATION,
                updated_at=now,
            )
            if self._supports_runs:
                await self.repo.update_run(
                    _json(OperationalTaskRun.model_validate(active_run)), run_expected
                )
            else:
                # Compatibility for pre-015 repository adapters only.
                if command.report:
                    if task["task_type"] == "event_investigation":
                        task["status"] = self._investigation_status(actions.values()) if command.report.stage == "final" else "running"
                    else:
                        task["status"] = self._status_from_plan(current_plan.plan.nodes, actions.values()) if current_plan else "running"
                    if task["status"] in {"waiting_human", "completed", "failed", "cancelled"}:
                        task["lifecycle_status"] = task["status"]
                    task["agent_lease_expires_at"] = (
                        now + LEASE_DURATION if task["status"] == "running" else None
                    )
                    await self.repo.update_task(
                        _task_json(OperationalTaskSummary.model_validate(task))
                    )
            await self._task_event(task, now)
            response = {"plan_revision": plan_revision, "report_revision": report_revision}
            if progress_revision is not None:
                response["progress_revision"] = progress_revision
            await self._save_request(scope, command.request_id, body, response)
            return AgentArtifactsResult(task=await self.detail(task_id), **response)

    async def presentation_cursor(self, task_id: UUID) -> TaskPresentationCursor:
        # The shared event-order lock gives the run and cursor one committed boundary.
        async with self.repo.transaction():
            task = await self._task(task_id)
            run = await self._active_run(task)
            cursor = await self.repo.current_event_cursor()
            return TaskPresentationCursor(
                task_id=task_id, execution_generation=task["execution_generation"],
                active_run=({key: run[key] for key in ("run_id", "generation", "status")} if run else None),
                event_cursor=cursor,
            )

    async def publish_presentation_event(
        self, task_id: UUID, command: TaskPresentationEventCommand
    ) -> TaskPresentationEventResult:
        body = _json(command)
        scope = f"task:{task_id}:presentation"
        async with self.repo.transaction():
            replay = await self._request_replay(scope, command.request_id, body)
            if replay:
                return TaskPresentationEventResult.model_validate(replay)
            await self.repo.lock(f"task:{task_id}")
            await self.repo.lock(f"run:{command.run_id}")
            task = await self._task(task_id)
            self._require_actionable_task(task)
            run = await self._active_run(task)
            require(run is not None and str(run["run_id"]) == str(command.run_id)
                    and run["generation"] == task["execution_generation"],
                    "EXECUTION_GENERATION_MISMATCH", "Presentation targets a stale execution run", status=409)
            require(run["status"] in {"claimed", "running"}, "INVALID_TRANSITION",
                    "Only a running run can publish presentation", status=409)
            require(run.get("agent_session_id") == command.source_session_id,
                    "SESSION_MISMATCH", "Presentation must come from the claimed Agent session", status=409)
            now = self._now()
            self._require_active_run_lease(run, now)
            payload = _json(command.command)
            if command.command.action == "open_chart":
                known = {item["evidence_id"] for item in await self.repo.list_evidence(str(task_id))}
                self._require_evidence_refs(command.command.evidence_refs, known)
            elif command.command.action == "set_overlay":
                stored = await self.repo.get_evidence(str(task_id), command.command.evidence_id)
                require(stored is not None, "EVIDENCE_NOT_FOUND", "Overlay evidence does not belong to this task", status=422)
                evidence = StoredEvidenceRecord.model_validate(stored)
                require(evidence.data_snapshot is not None, "SNAPSHOT_NOT_FOUND",
                        "Overlay requires a Server-ingested immutable data snapshot", status=422)
                payload["data"] = resolve_overlay(evidence.data_snapshot, command.command.mode)
            metadata = {"task_id": str(task_id), "run_id": str(run["run_id"]),
                        "execution_generation": run["generation"], "command": payload}
            try:
                size = len(json.dumps(metadata, ensure_ascii=False, allow_nan=False).encode("utf-8"))
            except (ValueError, TypeError) as exc:
                raise DomainError("PRESENTATION_DATA_INVALID", "Presentation data must be finite JSON", status=422) from exc
            require(size <= 512 * 1024, "PRESENTATION_TOO_LARGE",
                    "Presentation event exceeds the 512 KiB limit", status=422)
            event_seq = await self._append_event(
                topic="operational.presentation.requested", aggregate_type="operational_task_presentation",
                aggregate_id=str(task_id), aggregate_revision=run["revision"], occurred_at=now,
                metadata=metadata,
            )
            require(event_seq is not None, "DEPENDENCY_UNAVAILABLE", "Presentation event storage is unavailable", status=503)
            result = TaskPresentationEventResult(event_seq=event_seq, run_id=run["run_id"],
                                                 execution_generation=run["generation"])
            await self._save_request(scope, command.request_id, body, _json(result))
            return result

    async def propose_action(
        self, task_id: UUID, command: ActionProposalCommand
    ) -> HumanApprovalAction:
        body = _json(command)
        scope = f"task:{task_id}:action:propose"
        async with self.repo.transaction():
            replay = await self._request_replay(scope, command.request_id, body)
            if replay:
                action = await self.repo.get_action(str(task_id), replay["action_id"])
                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 task["task_type"] == "event_investigation":
                report = await self.repo.get_latest_report(str(task_id))
                require(report is not None
                        and report.get("execution_generation") == task["execution_generation"]
                        and report["report"].get("outcome") == "concluded",
                        "ACTION_CONCLUSION_REQUIRED",
                        "Publish a current concluded report before proposing an action", status=409)
            if command.node_id:
                if task["task_type"] != "event_investigation":
                    latest_plan = await self.repo.get_latest_plan(str(task_id))
                    require(latest_plan is not None, "PLAN_NODE_NOT_FOUND",
                            "Action cannot be attached before a plan is published", status=409)
                    node_ids = {node["id"] for node in latest_plan["plan"]["nodes"]}
                    require(command.node_id in node_ids, "PLAN_NODE_NOT_FOUND",
                            "Action references an unknown plan node", refs=[command.node_id], status=409)
            known = {item["evidence_id"] for item in await self.repo.list_evidence(str(task_id))}
            self._require_evidence_refs(command.evidence_refs, known)
            now = self._now()
            if task["status"] == "running":
                active_run = await self._active_run(task)
                require(active_run is not None, "RUN_NOT_FOUND",
                        "Task has no active Agent run", status=409)
                self._require_active_run_lease(active_run, now)
            approval_permission, execution_permission = self._action_permissions(command.kind)
            action = HumanApprovalAction(
                action_id=uuid4(), task_id=task_id, version=1, kind=command.kind,
                reason=command.reason, payload=command.payload,
                evidence_refs=command.evidence_refs, node_id=command.node_id,
                required_approval_permission=approval_permission,
                required_execution_permission=execution_permission,
                approval_status="pending", execution_status="not_submitted",
                execution_request_id=None,
                proposed_by=self._actor(), proposed_at=now,
            )
            await self.repo.insert_action(str(task_id), _json(action))
            await self._action_event(action, now)
            if task["status"] != "running":
                task.update(
                    lifecycle_status="waiting_human", status="waiting_human",
                    revision=task["revision"] + 1,
                    agent_lease_expires_at=None, 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, "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)
        # A SELECT on the default non-autocommit connection starts an outer
        # transaction. End it explicitly, or approval/submit become savepoints
        # whose event lock survives into the peripheral HTTP call.
        async with self.repo.transaction():
            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:
        body = _json(command)
        scope = f"task:{task_id}:action:{action_id}:decision"
        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.version == command.expected_version, "VERSION_CONFLICT",
                    "Action version changed", status=409)
            require(action.approval_status == "pending", "INVALID_TRANSITION",
                    "Action has already been decided", status=409)
            require(action.required_approval_permission in self.permissions,
                    "PERMISSION_DENIED", "Current user cannot decide this action", status=403)
            now = self._now()
            action = action.model_copy(update={
                "version": action.version + 1, "approval_status": command.decision,
                "decided_by": self._actor(), "decided_at": now,
                "decision_comment": command.comment,
            })
            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()
                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, {"action": _json(action)})
            return action

    async def submit_action(
        self, task_id: UUID, action_id: UUID, command: ActionExecuteCommand,
        dispatcher, credential: str | None,
    ) -> HumanApprovalAction:
        body = _json(command)
        scope = f"task:{task_id}:action:{action_id}:submit"
        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.version == command.expected_version, "VERSION_CONFLICT",
                    "Action version changed", status=409)
            require(action.approval_status == "approved", "INVALID_TRANSITION",
                    "Only an approved action can be submitted", status=409)
            require(action.execution_status in {"not_submitted", "failed", "unsupported"},
                    "INVALID_TRANSITION", "Action cannot be submitted in its current state", status=409)
            require(action.required_execution_permission in self.permissions,
                    "PERMISSION_DENIED", "Current user cannot submit this action", status=403)
            now = self._now()
            action = action.model_copy(update={
                "version": action.version + 1, "execution_status": "unknown",
                "execution_request_id": uuid5(
                    NAMESPACE_URL,
                    f"operational-action:{self.context.project_id}:{task_id}:{action_id}:{action.version}",
                ),
                "executed_by": self._actor(), "executed_at": now,
                "execution_detail": "外部动作已开始提交，结果待回执确认",
            })
            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)}
            )

        async with self.repo.transaction():
            task = await self.detail(task_id)
        try:
            outcome = await dispatcher.dispatch(
                context=self.context, task=task, action=action, credential=credential
            )
        except Exception:
            async with self.repo.transaction():
                await self.repo.lock(f"action:{action_id}")
                current = await self._action(task_id, action_id)
                if (
                    current.version == action.version
                    and current.execution_status == "unknown"
                ):
                    await self._notify(
                        task_id,
                        "external_coordination_failed",
                        task.title,
                        "外部系统提交中断，执行结果未知，请核对回执",
                        self._now(),
                        required_permission=action.required_execution_permission,
                        subject_type="action", subject_id=action_id,
                    )
            raise

        async with self.repo.transaction():
            await self.repo.lock(f"request:{scope}:{command.request_id}")
            await self.repo.lock(f"action:{action_id}")
            current = await self._action(task_id, action_id)
            require(current.version == action.version and current.execution_status == "unknown",
                    "VERSION_CONFLICT", "Action changed while awaiting external receipt", status=409)
            self._validate_dispatch_outcome(outcome)
            current = current.model_copy(update={
                "version": current.version + 1,
                "execution_status": outcome.status,
                "receipt": outcome.receipt,
                "execution_detail": outcome.detail,
            })
            await self.repo.update_action(str(task_id), _json(current))
            outcome_recorded_at = self._now()
            await self._action_event(current, outcome_recorded_at)
            if outcome.status in {"unknown", "failed", "unsupported"}:
                await self._notify(
                    task_id,
                    "external_coordination_failed",
                    task.title,
                    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)
                await self._requeue_task(task_document, outcome_recorded_at)
                await self.repo.update_task(
                    _task_json(OperationalTaskSummary.model_validate(task_document))
                )
                await self._task_event(task_document, outcome_recorded_at)
            await self._save_request(scope, command.request_id, body, {"action": _json(current)})
            return current

    async def reconcile_action(
        self, task_id: UUID, action_id: UUID, command: ActionReconcileCommand,
        dispatcher, credential: str | None
    ) -> HumanApprovalAction:
        body = _json(command)
        scope = f"task:{task_id}:action:{action_id}:reconcile"
        async with self.repo.transaction():
            replay = await self.repo.get_request(scope, str(command.request_id))
            if replay:
                self._require_same_request(replay, body)
                return HumanApprovalAction.model_validate(replay["response"]["action"])
            action = await self._action(task_id, action_id)
            require(action.version == command.expected_version, "VERSION_CONFLICT",
                    "Action version changed", status=409)
            require(action.execution_status in {"submitted", "unknown"}, "INVALID_TRANSITION",
                    "Only submitted or unknown actions can be reconciled", status=409)
            require(action.required_execution_permission in self.permissions,
                    "PERMISSION_DENIED", "Current user cannot reconcile this action", status=403)
            task = await self.detail(task_id)
        outcome = await dispatcher.reconcile(
            context=self.context, task=task, action=action,
            credential=credential,
        )
        self._validate_dispatch_outcome(outcome)
        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"action:{action_id}")
            current = await self._action(task_id, action_id)
            require(current.version == action.version, "VERSION_CONFLICT",
                    "Action changed while reconciling", status=409)
            previous_execution_status = current.execution_status
            current = current.model_copy(update={
                "version": current.version + 1, "execution_status": outcome.status,
                "receipt": outcome.receipt, "execution_detail": outcome.detail,
            })
            await self.repo.update_action(str(task_id), _json(current))
            outcome_recorded_at = self._now()
            await self._action_event(current, outcome_recorded_at)
            if (
                outcome.status in {"unknown", "failed", "unsupported"}
                and outcome.status != previous_execution_status
            ):
                task = await self._task(task_id)
                await self._notify(
                    task_id,
                    "external_coordination_failed",
                    task["title"],
                    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)
                await self._requeue_task(task_document, outcome_recorded_at)
                await self.repo.update_task(
                    _task_json(OperationalTaskSummary.model_validate(task_document))
                )
                await self._task_event(task_document, outcome_recorded_at)
            await self._save_request(
                scope, command.request_id, body, {"action": _json(current)}
            )
            return current

    async def external_feedback(
        self, task_id: UUID, command: ExternalFeedbackCommand
    ) -> ExternalFeedbackResult:
        body = _json(command)
        scope = f"task:{task_id}:external-feedback"
        async with self.repo.transaction():
            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")
                matching_source = own_source is not None and command.source_event_id in {
                    own_source, f"peripheral_sample:{own_source}",
                }
                if task["status"] == "completed" and not matching_source:
                    actions = await self.repo.list_actions(str(task_id))
                    require(any(
                        item["kind"] == "create_work_order" and item.get("receipt")
                        and item["receipt"].get("adapter") == "peripheral_sample"
                        and command.source_event_id == f"peripheral_sample:{item['receipt']['receipt_id']}"
                        for item in actions
                    ), "EXTERNAL_SOURCE_MISMATCH", "Feedback does not match an associated child work order", status=409)
                    require(command.source_revision > 1, "INITIAL_CHILD_FEEDBACK",
                            "The child creation receipt is not new field feedback", status=409)
            if task["task_type"] == "external_work_order":
                require(
                    task["source"].get("external_id") == command.source_event_id,
                    "EXTERNAL_SOURCE_MISMATCH",
                    "External feedback does not match the task source",
                    status=409,
                )
            if command.work_order_snapshot is not None:
                require(
                    task["task_type"] == "external_work_order",
                    "INVALID_FEEDBACK_SNAPSHOT",
                    "Work order snapshots only apply to external work order tasks",
                    status=422,
                )
            existing = await self.repo.get_feedback(str(task_id), command.source_event_id)
            current_source_revision = (
                task.get("source", {}).get("external_revision")
                if task["task_type"] == "external_work_order" or (
                    task["task_type"] == "event_investigation" and matching_source
                )
                else None
            )
            if current_source_revision is not None:
                require(
                    command.source_revision >= current_source_revision,
                    "STALE_EXTERNAL_REVISION",
                    "External feedback revision is stale",
                    status=409,
                )
            if existing:
                prior = existing["document"]
                if command.source_revision == existing["source_revision"]:
                    require(digest({key: value for key, value in prior.items() if key != "request_id"})
                            == digest({key: value for key, value in body.items() if key != "request_id"}), "IDEMPOTENCY_CONFLICT",
                            "External feedback revision was reused with different content", status=409)
                    response = {
                        "source_event_id": command.source_event_id,
                        "source_revision": command.source_revision,
                        "accepted_at": existing["accepted_at"],
                    }
                    await self._save_request(scope, command.request_id, body, response)
                    return ExternalFeedbackResult(task=await self.detail(task_id), **response)
                require(command.source_revision > existing["source_revision"],
                        "STALE_EXTERNAL_REVISION", "External feedback revision is stale", status=409)
            if (task["task_type"] == "event_investigation" and matching_source
                    and current_source_revision is not None):
                require(command.source_revision > current_source_revision, "STALE_EXTERNAL_REVISION",
                        "Original event feedback must contain a new source version", status=409)
            now = self._now()
            await self.repo.upsert_feedback(str(task_id), body, now)
            if command.evidence:
                evidence = EvidenceRecord(
                    **_json(command.evidence),
                    recorded_at=now, recorded_by=self._actor(),
                )
                prior_evidence = await self.repo.get_evidence(
                    str(task_id), command.evidence.evidence_id
                )
                if prior_evidence:
                    require(
                        digest(self._evidence_facts(prior_evidence))
                        == digest(self._evidence_facts(_json(evidence))),
                            "EVIDENCE_CONFLICT", "Evidence ids are immutable", status=409)
                else:
                    await self.repo.insert_evidence(str(task_id), _json(evidence))
            self._apply_feedback_status(task, command.status)
            if task["task_type"] == "external_work_order":
                task["source"] = {
                    **task["source"],
                    "external_revision": command.source_revision,
                }
                if command.work_order_snapshot is not None:
                    task.update(
                        title=command.work_order_snapshot.title,
                        summary=command.work_order_snapshot.summary,
                        scheduled_for=command.work_order_snapshot.scheduled_at,
                        priority=(
                            "urgent"
                            if command.work_order_snapshot.priority in {"high", "urgent"}
                            else "normal"
                        ),
                    )
            if task["task_type"] == "event_investigation":
                if matching_source:
                    task["source"] = {**task["source"], "external_revision": command.source_revision}
                await self._requeue_task(task, now)
            else:
                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",
                    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"},
                )
            if task["status"] == "failed":
                await self._notify(
                    task_id, "task_failed",
                    task["title"], command.detail, now,
                )
            response = {
                "source_event_id": command.source_event_id,
                "source_revision": command.source_revision,
                "accepted_at": now,
            }
            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",
                        )
                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"],
                    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:
        body = _json(command)
        scope = (
            f"notification:{notification_id}:user:{self.context.user_id}:ack"
        )
        async with self.repo.transaction():
            replay = await self._request_replay(scope, command.request_id, body)
            if replay:
                return OperationalTaskNotification.model_validate(replay["notification"])
            await self.repo.lock(f"notification:{notification_id}")
            value = await self.repo.get_notification(
                str(notification_id), str(self.context.user_id)
            )
            require(value is not None, "NOTIFICATION_NOT_FOUND", "Notification not found", status=404)
            notification = OperationalTaskNotification.model_validate(value)
            require(
                notification.required_permission is None
                or notification.required_permission in self.permissions,
                "PERMISSION_DENIED",
                "Current user is not a recipient of this notification",
                status=403,
            )
            require(notification.version == command.expected_version, "VERSION_CONFLICT",
                    "Notification version changed", status=409)
            if notification.status == "unread":
                notification = notification.model_copy(update={
                    "status": "read", "version": notification.version + 1,
                    "read_at": self._now(), "read_by": self._actor(),
                })
                await self.repo.mark_notification_read(
                    str(notification_id), str(self.context.user_id),
                    notification.version, notification.read_at,
                )
                await self._append_event(
                    topic="operational.notification.changed",
                    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)},
                )
            await self._save_request(
                scope, command.request_id, body, {"notification": _json(notification)}
            )
            return notification

    async def record_data_snapshot(
        self, task_id: UUID, snapshot: TaskDataSnapshot
    ) -> tuple[EvidenceRecord, TaskDataSnapshot]:
        """Persist the bounded source response before an Agent may cite it."""
        await self._task(task_id)
        now = self._now()
        evidence = StoredEvidenceRecord(
            evidence_id=snapshot.source_ref,
            kind=snapshot.origin,
            summary=f"{snapshot.source} 数据快照",
            source_label=snapshot.source,
            source_ref=snapshot.source_ref,
            observed_at=snapshot.queried_at,
            recorded_at=now,
            recorded_by=self._actor(),
            has_snapshot=True,
            data_snapshot=snapshot,
        )
        async with self.repo.transaction():
            await self.repo.lock(f"task:{task_id}:evidence:{snapshot.source_ref}")
            prior = await self.repo.get_evidence(str(task_id), snapshot.source_ref)
            if prior:
                stored = StoredEvidenceRecord.model_validate(prior)
                require(
                    stored.data_snapshot is not None
                    and self._snapshot_facts(stored.data_snapshot)
                    == self._snapshot_facts(snapshot),
                    "EVIDENCE_CONFLICT",
                    "Evidence ids are immutable",
                    status=409,
                )
                return self._public_evidence(prior), stored.data_snapshot
            await self.repo.insert_evidence(str(task_id), _json(evidence))
        return self._public_evidence(_json(evidence)), snapshot

    async def add_human_evidence(
        self, task_id: UUID, command: HumanEvidenceCommand
    ) -> OperationalTaskDetail:
        body = _json(command)
        scope = f"task:{task_id}:human-evidence"
        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)
            require(command.evidence.kind == "human_input", "EVIDENCE_KIND_INVALID",
                    "This endpoint only accepts human_input evidence", status=422)
            now = self._now()
            record = EvidenceRecord(
                **_json(command.evidence), recorded_at=now, recorded_by=self._actor()
            )
            prior = await self.repo.get_evidence(str(task_id), command.evidence.evidence_id)
            if prior:
                require(
                    digest(self._evidence_facts(prior)) == digest(_json(command.evidence)),
                    "EVIDENCE_CONFLICT", "Evidence ids are immutable", status=409,
                )
            else:
                await self.repo.insert_evidence(str(task_id), _json(record))
            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)

    async def data_snapshot(self, task_id: UUID, evidence_id: str) -> TaskDataSnapshot:
        await self._task(task_id)
        value = await self.repo.get_evidence(str(task_id), evidence_id)
        require(value is not None, "EVIDENCE_NOT_FOUND", "Evidence not found", status=404)
        stored = StoredEvidenceRecord.model_validate(value)
        require(stored.data_snapshot is not None, "SNAPSHOT_NOT_FOUND",
                "Evidence has no stored data snapshot", status=404)
        return stored.data_snapshot

    async def create_schedule(
        self, command: OperationalTaskScheduleCreate
    ) -> OperationalTaskSchedule:
        body = _json(command)
        scope = "schedule:create"
        async with self.repo.transaction():
            await self.repo.lock(f"schedule-request:{command.request_id}")
            request = await self.repo.get_request(scope, str(command.request_id))
            if request:
                self._require_same_request(request, body)
                prior = await self.repo.get_schedule_by_request(str(command.request_id))
                require(prior is not None, "STATE_CORRUPT",
                        "Schedule idempotency record has no schedule", status=500)
                return OperationalTaskSchedule.model_validate(prior)
            prior = await self.repo.get_schedule_by_request(str(command.request_id))
            if prior:
                return OperationalTaskSchedule.model_validate(prior)
            now = self._now()
            schedule = OperationalTaskSchedule(
                schedule_id=uuid4(), schedule_key=command.schedule_key,
                title=command.title, summary=command.summary,
                interval_seconds=command.interval_seconds,
                next_run_at=command.next_run_at, priority=command.priority,
                enabled=True, version=1, created_at=now, updated_at=now,
            )
            await self.repo.insert_schedule(_json(schedule), str(command.request_id))
            await self._save_request(
                scope, command.request_id, body,
                {"schedule_id": str(schedule.schedule_id)},
            )
            return schedule

    async def list_schedules(self) -> list[OperationalTaskSchedule]:
        return [
            OperationalTaskSchedule.model_validate(item)
            for item in await self.repo.list_schedules()
        ]

    async def update_schedule(
        self, schedule_id: UUID, command: OperationalTaskScheduleUpdate
    ) -> OperationalTaskSchedule:
        body = _json(command)
        scope = f"schedule:{schedule_id}:update"
        async with self.repo.transaction():
            replay = await self._request_replay(scope, command.request_id, body)
            if replay:
                return OperationalTaskSchedule.model_validate(replay["schedule"])
            await self.repo.lock(f"schedule:{schedule_id}")
            value = await self.repo.get_schedule(str(schedule_id))
            require(value is not None, "SCHEDULE_NOT_FOUND", "Schedule not found", status=404)
            schedule = OperationalTaskSchedule.model_validate(value)
            require(schedule.version == command.expected_version, "VERSION_CONFLICT",
                    "Schedule version changed", status=409)
            schedule = schedule.model_copy(update={
                "enabled": command.enabled,
                "version": schedule.version + 1,
                "updated_at": self._now(),
            })
            await self.repo.update_schedule(_json(schedule))
            await self._save_request(
                scope, command.request_id, body, {"schedule": _json(schedule)}
            )
            return schedule

    async def list_notifications(
        self, *, unread_only: bool, limit: int, offset: int
    ) -> tuple[list[OperationalTaskNotification], int | None]:
        rows = await self.repo.list_notifications(
            user_id=str(self.context.user_id),
            allowed_permissions=sorted(self.permissions),
            unread_only=unread_only, limit=limit + 1, offset=offset
        )
        return (
            [OperationalTaskNotification.model_validate(item) for item in rows[:limit]],
            offset + limit if len(rows) > limit else None,
        )

    async def dispatch_due_schedules(self, limit: int) -> DispatchDueSchedulesResult:
        now = self._now()
        task_ids: list[UUID] = []
        async with self.repo.transaction():
            for schedule_document in await self.repo.claim_due_schedules(limit):
                schedule = OperationalTaskSchedule.model_validate(schedule_document)
                due_at = schedule.next_run_at
                task_id = uuid5(schedule.schedule_id, due_at.isoformat())
                request_id = uuid5(NAMESPACE_URL, f"schedule:{schedule.schedule_id}:{due_at.isoformat()}")
                task = self._scheduled_task(schedule, task_id, due_at, now)
                await self.repo.insert_task(_task_json(task), str(request_id))
                run = self._new_run(task_id, 1, now)
                await self.repo.insert_run(_json(run))
                await self.repo.insert_occurrence(
                    str(schedule.schedule_id), due_at, str(task_id), now
                )
                next_schedule = schedule.model_copy(update={
                    "next_run_at": due_at + timedelta(seconds=schedule.interval_seconds),
                    "version": schedule.version + 1, "updated_at": now,
                })
                await self.repo.advance_schedule(_json(next_schedule))
                await self._run_event(run, now, queued=True)
                await self._task_event(_task_json(task), now)
                task_ids.append(task_id)
        return DispatchDueSchedulesResult(task_ids=task_ids, dispatched_at=now)

    async def _finish_run(
        self, task_id: UUID, run_id: UUID, command: AgentRunCommand,
        *, outcome: str, failure_code: str | None = None,
        failure_detail: str | None = None,
    ) -> OperationalTaskDetail:
        body = _json(command)
        scope = f"run:{run_id}:{outcome}"
        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"run:{run_id}")
            run = await self._run(task_id, run_id)
            require(run["revision"] == command.expected_revision, "VERSION_CONFLICT",
                    "Run revision changed", status=409)
            require(run["status"] in {"claimed", "running"}, "INVALID_TRANSITION",
                    "Only an active run can finish", status=409)
            require(run.get("agent_session_id") == command.source_session_id,
                    "SESSION_MISMATCH", "Agent session does not own this run", status=409)
            now = self._now()
            self._require_active_run_lease(run, now)
            task = await self._task(task_id)
            final_outcome = outcome
            if outcome == "succeeded":
                if hasattr(self.repo, "pending_run_messages"):
                    pending_messages = await self.repo.pending_run_messages(str(run_id))
                    require(not pending_messages, "PENDING_AGENT_MESSAGES",
                            "Run has messages not yet entered into Agent context",
                            status=409, recovery="retry_same_request", refs=pending_messages)
                plan_document = await self.repo.get_latest_plan(str(task_id))
                report_document = await self.repo.get_latest_report(str(task_id))
                plan = (
                    ExecutionPlanSnapshot.model_validate(plan_document)
                    if plan_document is not None else None
                )
                report = (
                    InvestigationReportRevision.model_validate(report_document)
                    if report_document is not None else None
                )
                same_generation = (
                    report is not None
                    and report.execution_generation == run["generation"]
                    and report.source_session_id == command.source_session_id
                    and (task["task_type"] == "event_investigation" or (
                        plan is not None
                        and plan.execution_generation == run["generation"]
                        and plan.source_session_id == command.source_session_id
                    ))
                )
                if not same_generation or report.report.stage != "final":
                    final_outcome = "failed"
                    lifecycle = "failed"
                    failure_code = "ARTIFACT_INCOMPLETE"
                    failure_detail = (
                        "Run requires a final report from the same generation"
                    )
                else:
                    actions = [
                        HumanApprovalAction.model_validate(item)
                        for item in await self.repo.list_actions(str(task_id))
                    ]
                    access_requests = [
                        TaskAccessRequest.model_validate(item)
                        for item in await self._list_access_requests(task_id)
                    ]
                    if task["task_type"] == "event_investigation":
                        lifecycle = self._investigation_status(actions)
                    elif report.report.outcome == "insufficient_data":
                        terminal = all(node.status in {"completed", "skipped", "cancelled"}
                                       for node in plan.plan.nodes)
                        pending_action = any(action.approval_status == "pending" or (
                            action.approval_status == "approved" and action.execution_status in {"not_submitted", "submitted", "unknown"}
                        ) for action in actions)
                        lifecycle = "completed" if terminal and not pending_action else "running"
                    else:
                        lifecycle = self._status_from_plan(
                            plan.plan.nodes, actions, () if task["task_type"] == "event_investigation" else access_requests
                        )
                    if lifecycle == "running":
                        final_outcome = "failed"
                        lifecycle = "failed"
                        failure_code = "ARTIFACT_INCOMPLETE"
                        failure_detail = "Final plan still contains running work"
            else:
                lifecycle = "failed"
            expected = run["revision"]
            run.update(
                status=final_outcome, revision=expected + 1, finished_at=now,
                lease_expires_at=None, heartbeat_at=now,
                failure_code=failure_code, failure_detail=failure_detail, updated_at=now,
            )
            run_model = OperationalTaskRun.model_validate(run)
            await self.repo.update_run(_json(run_model), expected)
            if final_outcome != "succeeded":
                await self._fail_run_messages(run_id, failure_code or "Run failed", now)
            task.update(
                lifecycle_status=lifecycle,
                status=lifecycle,
                revision=task["revision"] + 1,
                agent_session_id=None,
                agent_lease_expires_at=None,
                updated_at=now,
            )
            await self.repo.update_task(_task_json(OperationalTaskSummary.model_validate(task)))
            await self._run_event(run_model, now)
            await self._task_event(task, now)
            if lifecycle == "failed":
                await self._notify(
                    task_id,
                    "task_failed",
                    task["title"],
                    failure_detail or "工况任务执行失败",
                    now,
                )
            elif lifecycle == "waiting_human":
                pending_actions = any(
                    action.approval_status == "pending" for action in actions
                )
                pending_access = any(
                    request.get("status") == "pending"
                    for request in await self._list_access_requests(task_id)
                )
                if not pending_actions and not pending_access:
                    await self._notify(
                        task_id,
                        "human_input_required",
                        task["title"],
                        "任务等待人工补充信息或业务确认",
                        now,
                    )
            await self._save_request(scope, command.request_id, body, {"task_id": str(task_id)})
            return await self.detail(task_id)

    async def _run(self, task_id: UUID, run_id: UUID) -> dict:
        run = await self.repo.get_run(str(run_id))
        require(
            run is not None and str(run["task_id"]) == str(task_id),
            "RUN_NOT_FOUND", "Operational task run not found", status=404,
        )
        return run

    async def _task(self, task_id: UUID) -> dict:
        task = await self.repo.get_task(str(task_id))
        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 task.get("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)
        return HumanApprovalAction.model_validate(value)

    async def _request_replay(self, scope: str, request_id: UUID, body: dict) -> dict | None:
        await self.repo.lock(f"request:{scope}:{request_id}")
        prior = await self.repo.get_request(scope, str(request_id))
        if prior:
            self._require_same_request(prior, body)
            return prior["response"]
        return None

    async def _save_request(self, scope: str, request_id: UUID, body: dict, response: dict) -> None:
        await self.repo.save_request(scope, str(request_id), digest(body), response)

    @staticmethod
    def _require_same_request(prior: dict, body: dict) -> None:
        require(prior["request_digest"] == digest(body), "IDEMPOTENCY_CONFLICT",
                "request_id was used with different content", status=409)

    @staticmethod
    def _execution_request_id(task_id: UUID, generation: int) -> UUID:
        return uuid5(NAMESPACE_URL, f"operational-task:{task_id}:generation:{generation}")

    @staticmethod
    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"
        if any(
            node.status == "failed" and node.id.startswith("AGENT_RUN_FAILURE")
            for node in nodes
        ):
            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"}:
            return "failed"
        if any(node.owner == "human" and node.status in {"pending", "waiting"} for node in nodes):
            return "waiting_human"
        return "running"

    @staticmethod
    def _investigation_status(actions: Iterable[HumanApprovalAction]) -> str:
        """Only recorded business actions can keep an investigation waiting."""
        return "waiting_human" if any(
            action.approval_status == "pending" or (
                action.approval_status == "approved"
                and action.execution_status in {"not_submitted", "submitted", "unknown"}
            ) for action in actions
        ) else "completed"

    @staticmethod
    def _validate_plan_state(plan, known_evidence: set[str], actions: dict[str, HumanApprovalAction]) -> None:
        nodes = {node.id: node for node in plan.nodes}
        predecessors: dict[str, set[str]] = {node.id: set() for node in plan.nodes}
        for edge in plan.edges:
            predecessors[edge.to_node].add(edge.from_node)
        for node in plan.nodes:
            if node.status in {"running", "completed"}:
                incomplete = [
                    predecessor for predecessor in predecessors[node.id]
                    if nodes[predecessor].status not in {"completed", "skipped"}
                ]
                require(not incomplete, "PLAN_DEPENDENCY_INCOMPLETE",
                        "A running or completed node has incomplete predecessors",
                        refs=incomplete, status=409)
            if node.status == "completed":
                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)
            for action_id in node.action_refs:
                action = actions.get(str(action_id))
                require(action is not None, "ACTION_NOT_FOUND",
                        "Plan node references an unknown action", refs=[str(action_id)], status=409)
                if node.status == "completed":
                    require(action.execution_status == "succeeded" and action.receipt is not None,
                            "ACTION_RECEIPT_REQUIRED",
                            "A completed action node requires a successful external receipt",
                            refs=[str(action_id)], status=409)

    @staticmethod
    def _evidence_facts(document: dict) -> dict:
        return {
            key: document.get(key)
            for key in (
                "evidence_id", "kind", "summary", "source_label", "source_ref", "observed_at"
            )
        }

    @staticmethod
    def _snapshot_facts(snapshot: TaskDataSnapshot) -> dict:
        return snapshot.model_dump(
            mode="json",
            exclude={"queried_at"},
        )

    @staticmethod
    def _require_active_run_lease(run: dict, now: datetime) -> None:
        lease = OperationalTaskRun.model_validate(run).lease_expires_at
        require(
            lease is not None and lease > now,
            "LEASE_EXPIRED",
            "Agent run lease has expired",
            status=409,
        )

    @staticmethod
    def _public_evidence(document: dict) -> EvidenceRecord:
        return EvidenceRecord.model_validate(
            {key: value for key, value in document.items() if key != "data_snapshot"}
        )

    @staticmethod
    def _require_evidence_refs(refs: Iterable[str], known: set[str]) -> None:
        missing = sorted(set(refs) - known)
        require(not missing, "EVIDENCE_NOT_FOUND", "Artifact references unknown evidence",
                refs=missing, status=409)

    @staticmethod
    def _validate_action_payload(command: ActionProposalCommand) -> None:
        model = CreateWorkOrderPayload if command.kind == "create_work_order" else IssueCommandPayload
        try:
            model.model_validate(command.payload)
        except Exception as exc:
            raise DomainError("ACTION_PAYLOAD_INVALID", "Action payload does not match its kind") from exc

    @staticmethod
    def _action_permissions(kind: str) -> tuple[str, str]:
        if kind == "create_work_order":
            return OPERATIONAL_ACTION_WORK_ORDER_APPROVE, OPERATIONAL_ACTION_WORK_ORDER_EXECUTE
        return OPERATIONAL_ACTION_COMMAND_APPROVE, OPERATIONAL_ACTION_COMMAND_EXECUTE

    @staticmethod
    def _validate_dispatch_outcome(outcome) -> None:
        if outcome.status in {"submitted", "succeeded"}:
            require(outcome.receipt is not None, "RECEIPT_REQUIRED",
                    "Submitted actions require an external receipt", status=502)
        if outcome.status == "succeeded":
            require(outcome.receipt.completed_at is not None, "RECEIPT_REQUIRED",
                    "Successful actions require a completion receipt", status=502)

    @staticmethod
    def _apply_feedback_status(task: dict, status: str) -> None:
        task["status"] = {
            "scheduled": "queued", "assigned": "running", "in_progress": "running",
            "awaiting_feedback": "waiting_human", "completed": "completed",
            "cancelled": "cancelled", "failed": "failed",
        }[status]
        task["lifecycle_status"] = {
            "scheduled": "open", "assigned": "open", "in_progress": "open",
            "awaiting_feedback": "waiting_human", "completed": "completed",
            "cancelled": "cancelled", "failed": "failed",
        }[status]
        task["agent_lease_expires_at"] = None

    async def _requeue_task(self, task: dict, now: datetime) -> None:
        if self._supports_runs:
            active_run = await self.repo.get_active_run(str(task["task_id"]))
            while active_run is not None:
                await self.repo.lock(f"run:{active_run['run_id']}")
                # Completion may have won the run lock after our first read.
                refreshed_run = await self.repo.get_active_run(str(task["task_id"]))
                if refreshed_run is None or str(refreshed_run["run_id"]) == str(active_run["run_id"]):
                    active_run = refreshed_run
                    break
                active_run = refreshed_run
            if active_run is not None and active_run["status"] == "queued":
                # Multiple inputs can arrive before the worker claims the next
                # generation. They are all visible when that queued run starts,
                # so coalesce them instead of violating the one-active-run rule.
                task.update(
                    lifecycle_status="open", status="queued",
                    revision=task["revision"] + 1,
                    execution_generation=active_run["generation"],
                    execution_request_id=str(active_run["execution_request_id"]),
                    agent_session_id=None, agent_lease_expires_at=None,
                    updated_at=now,
                )
                return
            if active_run is not None:
                expected = active_run["revision"]
                active_run.update(
                    status="cancelled", revision=expected + 1,
                    lease_expires_at=None, finished_at=now,
                    failure_code="SUPERSEDED_BY_REQUEUE",
                    failure_detail="New task input superseded the active run",
                    updated_at=now,
                )
                cancelled = OperationalTaskRun.model_validate(active_run)
                await self.repo.update_run(_json(cancelled), expected)
                await self._fail_run_messages(active_run["run_id"], "Run superseded by requeue", now)
                await self._run_event(cancelled, now)
        generation = task["execution_generation"] + 1
        task_id = UUID(task["task_id"])
        task.update(
            lifecycle_status="open", status="queued", revision=task["revision"] + 1,
            execution_generation=generation,
            execution_request_id=str(self._execution_request_id(task_id, generation)),
            agent_session_id=None, agent_lease_expires_at=None, updated_at=now,
        )
        run = self._new_run(task_id, generation, now)
        if self._supports_runs:
            await self.repo.insert_run(_json(run))
            await self._run_event(run, now, queued=True)

    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,
        report_reference: InvestigationReportReference | 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,
            report_reference=report_reference,
        )
        await self.repo.insert_notification(_json(notification))
        await self._append_event(
            topic="operational.notification.changed",
            aggregate_type="operational_notification",
            aggregate_id=str(notification.notification_id),
            aggregate_revision=notification.version,
            occurred_at=now,
            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:
        return OperationalTaskRun(
            run_id=uuid4(),
            task_id=task_id,
            generation=generation,
            execution_request_id=self._execution_request_id(task_id, generation),
            status="queued",
            revision=1,
            dispatch_attempts=0,
            next_dispatch_at=now,
            created_at=now,
            updated_at=now,
        )

    async def _task_event(self, task: dict, now: datetime) -> None:
        await self._append_event(
            topic="operational.task.changed",
            aggregate_type="operational_task",
            aggregate_id=str(task["task_id"]),
            aggregate_revision=int(task["revision"]),
            occurred_at=now,
            metadata={
                "task_id": str(task["task_id"]),
                "lifecycle_status": task.get("lifecycle_status", "open"),
            },
        )

    async def _run_event(
        self, run: OperationalTaskRun, now: datetime, *, queued: bool = False
    ) -> None:
        await self._append_event(
            topic="operational.run.queued" if queued else "operational.run.changed",
            aggregate_type="operational_task_run",
            aggregate_id=str(run.run_id),
            aggregate_revision=run.revision,
            occurred_at=now,
            metadata={
                "task_id": str(run.task_id),
                "run_id": str(run.run_id),
                "status": run.status,
            },
        )

    async def _action_event(
        self, action: HumanApprovalAction, now: datetime
    ) -> None:
        await self._append_event(
            topic="operational.action.changed",
            aggregate_type="operational_action",
            aggregate_id=str(action.action_id),
            aggregate_revision=action.version,
            occurred_at=now,
            metadata={"task_id": str(action.task_id)},
        )

    def _scheduled_task(
        self, schedule: OperationalTaskSchedule, task_id: UUID,
        due_at: datetime, now: datetime,
    ) -> OperationalTaskSummary:
        return OperationalTaskSummary(
            project_id=self.context.project_id, task_id=task_id,
            task_type="scheduled_flow", workflow=schedule.schedule_key,
            lifecycle_status="open", status="queued", priority=schedule.priority, revision=1,
            execution_generation=1,
            execution_request_id=self._execution_request_id(task_id, 1),
            title=schedule.title, summary=schedule.summary,
            source={
                "kind": "schedule",
                "external_id": f"{schedule.schedule_id}:{due_at.isoformat()}",
            },
            trigger={"kind": "schedule", "occurred_at": due_at,
                     "detail": f"定时流程 {schedule.schedule_key} 到期"},
            scheduled_for=due_at, agent_session_id=None,
            agent_lease_expires_at=None, current_plan_revision=None,
            current_report_revision=None, created_at=now, updated_at=now,
        )


def _json_dict(value: dict) -> dict:
    """Convert datetime values without allowing arbitrary object serialization."""
    return {
        key: item.isoformat().replace("+00:00", "Z") if isinstance(item, datetime) else item
        for key, item in value.items()
    }