"""State machines for durable operational tasks and approved external actions.""" from __future__ import annotations 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_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, AgentArtifactsCommand, AgentArtifactsResult, AgentTaskContext, AgentRunCancelCommand, AgentRunCommand, AgentRunDispatchFailureCommand, AgentRunFailureCommand, CreateWorkOrderPayload, DispatchDueSchedulesResult, EvidenceRecord, ExecutionPlanSnapshot, ExternalFeedbackCommand, ExternalFeedbackResult, HumanApprovalAction, HumanEvidenceCommand, InvestigationReportRevision, IssueCommandPayload, NotificationAcknowledgeCommand, OperationalTaskCreate, OperationalTaskDetail, OperationalTaskNotification, OperationalTaskPermissions, OperationalTaskSchedule, OperationalTaskScheduleCreate, OperationalTaskScheduleUpdate, OperationalTaskSummary, OperationalTaskRetryCommand, OperationalTaskRun, RecoverExpiredLeasesResult, StoredEvidenceRecord, ) from app.domain.schemas.operational_data import TaskDataSnapshot from app.services.content_digest import digest 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"}) 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 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, ) 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 ) -> tuple[list[OperationalTaskSummary], int | None]: rows = await self.repo.list_tasks( task_type=task_type, status=status, limit=limit + 1, offset=offset ) 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)) 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)) active_run = await self._active_run(task) 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 != "active_run"}, active_run=active_run, latest_plan=plan, latest_report=report, evidence=evidence, actions=actions, permissions=self._permission_view(), work_order_snapshot=work_order_snapshot, ) async def agent_context(self, task_id: UUID) -> 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"]) return AgentTaskContext( task=await self.detail(task_id), approved_data_scopes=scopes, ) async def list_runs( self, *, status: str | None, limit: int, cursor_created_at=None, cursor_run_id=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, ) 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(): 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._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 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) 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._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: 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) -> RecoverExpiredLeasesResult: now = self._now() recovered: list[UUID] = [] async with self.repo.transaction(): for run in await self.repo.claim_expired_runs(limit): 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._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) 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)) } 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) 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) 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, ) 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"])}, ) 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 and current_plan is not None and current_plan.execution_generation == task["execution_generation"] and current_plan.source_session_id == command.source_session_id ): task["status"] = self._status_from_plan( current_plan.plan.nodes, actions.values() ) 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} await self._save_request(scope, command.request_id, body, response) return AgentArtifactsResult(task=await self.detail(task_id), **response) 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._validate_action_payload(command) if command.node_id: 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, ) await self._save_request( scope, command.request_id, body, {"action_id": str(action.action_id)} ) return action 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"action:{action_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) 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"action:{action_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._save_request( scope, command.request_id, body, {"action": _json(action)} ) 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, ) 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, ) 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" 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) outcome = await dispatcher.reconcile( context=self.context, task=await self.detail(task_id), 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, ) 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"] == "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" 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(prior) == digest(body), "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) 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": 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["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 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", aggregate_id=str(notification_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._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": 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 = ( plan is not None and report is not None and plan.execution_generation == run["generation"] and report.execution_generation == run["generation"] and plan.source_session_id == command.source_session_id and report.source_session_id == command.source_session_id ) if not same_generation: final_outcome = "failed" lifecycle = "failed" failure_code = "ARTIFACT_INCOMPLETE" failure_detail = ( "Run requires a final plan and report from the same generation" ) else: actions = [ HumanApprovalAction.model_validate(item) for item in await self.repo.list_actions(str(task_id)) ] lifecycle = self._status_from_plan(plan.plan.nodes, actions) 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) 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.repo.list_access_requests(str(task_id)) if hasattr(self.repo, "list_access_requests") else [] ) ) 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 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]) -> 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 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 _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": require(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"])) 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._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, ) -> 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, ) 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}, ) 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() }