148 lines
6.1 KiB
Python
148 lines
6.1 KiB
Python
"""AIOS Task Lifecycle.
|
|
|
|
Defines a governed task state machine for task creation, assignment, execution,
|
|
blocking, approval, retry, completion, notification and closure.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, dataclass, field
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
TERMINAL_STATES = {"completed", "cancelled", "failed", "closed"}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TaskTransition:
|
|
action: str
|
|
from_states: set[str]
|
|
to_state: str
|
|
required_role: str = "agent"
|
|
requires_reason: bool = False
|
|
notify: bool = False
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
data = asdict(self)
|
|
data["from_states"] = sorted(self.from_states)
|
|
return data
|
|
|
|
|
|
class TaskLifecycle:
|
|
"""Central task state machine."""
|
|
|
|
TRANSITIONS = {
|
|
"assign": TaskTransition("assign", {"new", "pending"}, "assigned", "dispatcher"),
|
|
"start": TaskTransition("start", {"assigned", "pending"}, "running", "executor"),
|
|
"update_progress": TaskTransition("update_progress", {"running"}, "running", "executor"),
|
|
"block": TaskTransition("block", {"running", "assigned", "pending"}, "blocked", "executor", requires_reason=True, notify=True),
|
|
"request_approval": TaskTransition("request_approval", {"running", "blocked"}, "waiting_approval", "executor", requires_reason=True, notify=True),
|
|
"approve": TaskTransition("approve", {"waiting_approval"}, "approved", "owner", notify=True),
|
|
"reject": TaskTransition("reject", {"waiting_approval"}, "rejected", "owner", requires_reason=True, notify=True),
|
|
"resume": TaskTransition("resume", {"blocked", "approved"}, "running", "executor"),
|
|
"retry": TaskTransition("retry", {"failed", "blocked"}, "running", "executor", notify=True),
|
|
"complete": TaskTransition("complete", {"running", "approved"}, "completed", "executor", notify=True),
|
|
"notify": TaskTransition("notify", {"completed"}, "completed", "notifier"),
|
|
"close": TaskTransition("close", {"completed", "cancelled", "failed"}, "closed", "owner"),
|
|
"cancel": TaskTransition("cancel", {"new", "pending", "assigned", "running", "blocked", "waiting_approval"}, "cancelled", "owner", requires_reason=True, notify=True),
|
|
"fail": TaskTransition("fail", {"running", "assigned"}, "failed", "executor", requires_reason=True, notify=True),
|
|
}
|
|
|
|
ROLE_ORDER = {
|
|
"user": 1,
|
|
"agent": 2,
|
|
"executor": 3,
|
|
"dispatcher": 4,
|
|
"notifier": 4,
|
|
"group_admin": 5,
|
|
"owner": 9,
|
|
}
|
|
|
|
def available_actions(self, task: dict[str, Any], role: str) -> list[dict[str, Any]]:
|
|
state = task.get("status", "new")
|
|
actions = []
|
|
for name, transition in self.TRANSITIONS.items():
|
|
allowed_state = state in transition.from_states
|
|
allowed_role = self._role_allows(role, transition.required_role)
|
|
actions.append({
|
|
"action": name,
|
|
"label": self._label(name),
|
|
"enabled": allowed_state and allowed_role,
|
|
"from_state": state,
|
|
"to_state": transition.to_state,
|
|
"required_role": transition.required_role,
|
|
"requires_reason": transition.requires_reason,
|
|
"notify": transition.notify,
|
|
"disabled_reason": None if allowed_state and allowed_role else self._disabled_reason(allowed_state, allowed_role, transition),
|
|
})
|
|
return actions
|
|
|
|
def transition(self, task: dict[str, Any], action: str, actor_role: str, reason: str | None = None) -> dict[str, Any]:
|
|
if action not in self.TRANSITIONS:
|
|
raise ValueError(f"unknown task action: {action}")
|
|
transition = self.TRANSITIONS[action]
|
|
state = task.get("status", "new")
|
|
if state not in transition.from_states:
|
|
raise ValueError(f"action {action} not allowed from state {state}")
|
|
if not self._role_allows(actor_role, transition.required_role):
|
|
raise PermissionError(f"role {actor_role} cannot execute {action}; requires {transition.required_role}")
|
|
if transition.requires_reason and not reason:
|
|
raise ValueError(f"action {action} requires reason")
|
|
updated = dict(task)
|
|
updated["status"] = transition.to_state
|
|
updated["updated_at"] = utc_now()
|
|
history = list(updated.get("history", []))
|
|
history.append({
|
|
"action": action,
|
|
"from": state,
|
|
"to": transition.to_state,
|
|
"actor_role": actor_role,
|
|
"reason": reason,
|
|
"ts": utc_now(),
|
|
})
|
|
updated["history"] = history
|
|
if action == "complete":
|
|
updated["completed_at"] = utc_now()
|
|
notification = dict(updated.get("notification", {}))
|
|
notification.setdefault("required", True)
|
|
notification.setdefault("sent", False)
|
|
updated["notification"] = notification
|
|
return updated
|
|
|
|
def _role_allows(self, role: str, required: str) -> bool:
|
|
return self.ROLE_ORDER.get(role, 0) >= self.ROLE_ORDER.get(required, 0)
|
|
|
|
def _label(self, action: str) -> str:
|
|
labels = {
|
|
"assign": "分配任务",
|
|
"start": "开始执行",
|
|
"update_progress": "更新进度",
|
|
"block": "标记阻塞",
|
|
"request_approval": "请求审批",
|
|
"approve": "批准",
|
|
"reject": "拒绝",
|
|
"resume": "继续执行",
|
|
"retry": "重试",
|
|
"complete": "标记完成",
|
|
"notify": "发送通知",
|
|
"close": "关闭任务",
|
|
"cancel": "取消任务",
|
|
"fail": "标记失败",
|
|
}
|
|
return labels.get(action, action)
|
|
|
|
def _disabled_reason(self, allowed_state: bool, allowed_role: bool, transition: TaskTransition) -> str:
|
|
if not allowed_state:
|
|
return "当前状态不允许该操作"
|
|
if not allowed_role:
|
|
return f"需要角色: {transition.required_role}"
|
|
return "不可用"
|
|
|
|
|
|
__all__ = ["TaskTransition", "TaskLifecycle", "TERMINAL_STATES"]
|