wehub-resource-sync
299 lines
9.4 KiB
Python
299 lines
9.4 KiB
Python
"""任务服务层 — CRUD + 状态机逻辑。
|
|
|
|
所有业务规则集中在此：
|
|
- 创建任务 → 事件写入 outbox 表（同一事务）
|
|
- 状态流转 → 校验合法性 + SELECT FOR UPDATE 防并发 + outbox 事件
|
|
- 查询、过滤、聚合
|
|
|
|
事件投递由 OutboxRelay worker 异步完成，保证 DB/Event 原子一致。
|
|
"""
|
|
|
|
import logging
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
from sqlalchemy import select, func, and_
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from ..models.outbox import OutboxEvent
|
|
from ..models.task import Task, TaskState, STATE_TRANSITIONS, TERMINAL_STATES
|
|
from .event_bus import (
|
|
TOPIC_TASK_CREATED,
|
|
TOPIC_TASK_STATUS,
|
|
TOPIC_TASK_COMPLETED,
|
|
TOPIC_TASK_DISPATCH,
|
|
)
|
|
|
|
log = logging.getLogger("edict.task_service")
|
|
|
|
|
|
class TaskService:
|
|
def __init__(self, db: AsyncSession, event_bus=None):
|
|
self.db = db
|
|
# event_bus 保留用于 request_dispatch 等直接发布场景
|
|
self.bus = event_bus
|
|
|
|
# ── 创建 ──
|
|
|
|
async def create_task(
|
|
self,
|
|
title: str,
|
|
description: str = "",
|
|
priority: str = "中",
|
|
assignee_org: str | None = None,
|
|
creator: str = "emperor",
|
|
tags: list[str] | None = None,
|
|
initial_state: TaskState = TaskState.Taizi,
|
|
meta: dict | None = None,
|
|
) -> Task:
|
|
"""创建任务，事件写入 outbox 表（同一事务原子提交）。"""
|
|
now = datetime.now(timezone.utc)
|
|
trace_id = str(uuid.uuid4())
|
|
target_org = Task.org_for_state(initial_state, assignee_org)
|
|
task_meta = meta or {}
|
|
|
|
task = Task(
|
|
trace_id=trace_id,
|
|
title=title,
|
|
description=description,
|
|
priority=priority,
|
|
state=initial_state,
|
|
assignee_org=assignee_org,
|
|
creator=creator,
|
|
tags=tags or [],
|
|
org=target_org,
|
|
official=creator,
|
|
now=description or "任务创建",
|
|
target_dept=assignee_org or "",
|
|
flow_log=[
|
|
{
|
|
"from": None,
|
|
"to": initial_state.value,
|
|
"agent": "system",
|
|
"reason": "任务创建",
|
|
"ts": now.isoformat(),
|
|
}
|
|
],
|
|
progress_log=[],
|
|
todos=[],
|
|
scheduler={},
|
|
meta=task_meta,
|
|
)
|
|
self.db.add(task)
|
|
await self.db.flush()
|
|
|
|
# 事件写入 outbox — 与 task 同一事务，原子提交
|
|
outbox = OutboxEvent(
|
|
topic=TOPIC_TASK_CREATED,
|
|
trace_id=trace_id,
|
|
event_type="task.created",
|
|
producer="task_service",
|
|
payload={
|
|
"task_id": str(task.task_id),
|
|
"title": title,
|
|
"state": initial_state.value,
|
|
"priority": priority,
|
|
"assignee_org": assignee_org,
|
|
},
|
|
)
|
|
self.db.add(outbox)
|
|
|
|
await self.db.commit()
|
|
log.info(f"Created task {task.task_id}: {title} [{initial_state.value}]")
|
|
return task
|
|
|
|
# ── 状态流转 ──
|
|
|
|
async def transition_state(
|
|
self,
|
|
task_id: uuid.UUID,
|
|
new_state: TaskState,
|
|
agent: str = "system",
|
|
reason: str = "",
|
|
) -> Task:
|
|
"""执行状态流转。SELECT FOR UPDATE 防止并发 flow_log 丢失。"""
|
|
# 行级排他锁 — 串行化同一任务的并发写入
|
|
stmt = select(Task).where(Task.task_id == task_id).with_for_update()
|
|
result = await self.db.execute(stmt)
|
|
task = result.scalar_one_or_none()
|
|
if task is None:
|
|
raise ValueError(f"Task not found: {task_id}")
|
|
|
|
old_state = task.state
|
|
|
|
# 校验合法流转
|
|
allowed = STATE_TRANSITIONS.get(old_state, set())
|
|
if new_state not in allowed:
|
|
raise ValueError(
|
|
f"Invalid transition: {old_state.value} → {new_state.value}. "
|
|
f"Allowed: {[s.value for s in allowed]}"
|
|
)
|
|
|
|
task.state = new_state
|
|
task.org = Task.org_for_state(new_state, task.assignee_org)
|
|
if reason:
|
|
task.now = reason
|
|
task.updated_at = datetime.now(timezone.utc)
|
|
|
|
# 在行锁保护下安全追加 flow_log
|
|
flow_entry = {
|
|
"from": old_state.value,
|
|
"to": new_state.value,
|
|
"agent": agent,
|
|
"reason": reason,
|
|
"ts": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
if task.flow_log is None:
|
|
task.flow_log = []
|
|
task.flow_log = [*task.flow_log, flow_entry]
|
|
|
|
# 事件写入 outbox（同一事务）
|
|
topic = TOPIC_TASK_COMPLETED if new_state in TERMINAL_STATES else TOPIC_TASK_STATUS
|
|
outbox = OutboxEvent(
|
|
topic=topic,
|
|
trace_id=str(task.trace_id),
|
|
event_type=f"task.state.{new_state.value}",
|
|
producer=agent,
|
|
payload={
|
|
"task_id": str(task_id),
|
|
"from": old_state.value,
|
|
"to": new_state.value,
|
|
"reason": reason,
|
|
"assignee_org": task.assignee_org,
|
|
},
|
|
)
|
|
self.db.add(outbox)
|
|
|
|
await self.db.commit()
|
|
log.info(f"Task {task_id} state: {old_state.value} → {new_state.value} by {agent}")
|
|
return task
|
|
|
|
# ── 派发请求 ──
|
|
|
|
async def request_dispatch(
|
|
self,
|
|
task_id: uuid.UUID,
|
|
target_agent: str,
|
|
message: str = "",
|
|
):
|
|
"""发布 task.dispatch 事件到 outbox，由 OutboxRelay 投递后 DispatchWorker 消费。"""
|
|
task = await self._get_task(task_id)
|
|
outbox = OutboxEvent(
|
|
topic=TOPIC_TASK_DISPATCH,
|
|
trace_id=str(task.trace_id),
|
|
event_type="task.dispatch.request",
|
|
producer="task_service",
|
|
payload={
|
|
"task_id": str(task_id),
|
|
"agent": target_agent,
|
|
"message": message,
|
|
"state": task.state.value,
|
|
},
|
|
)
|
|
self.db.add(outbox)
|
|
await self.db.commit()
|
|
log.info(f"Dispatch requested: task {task_id} → agent {target_agent}")
|
|
|
|
# ── 进度/备注更新 ──
|
|
|
|
async def add_progress(
|
|
self,
|
|
task_id: uuid.UUID,
|
|
agent: str,
|
|
content: str,
|
|
) -> Task:
|
|
task = await self._get_task(task_id)
|
|
entry = {
|
|
"agent": agent,
|
|
"content": content,
|
|
"ts": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
if task.progress_log is None:
|
|
task.progress_log = []
|
|
task.progress_log = [*task.progress_log, entry]
|
|
task.updated_at = datetime.now(timezone.utc)
|
|
await self.db.commit()
|
|
return task
|
|
|
|
async def update_todos(
|
|
self,
|
|
task_id: uuid.UUID,
|
|
todos: list[dict],
|
|
) -> Task:
|
|
task = await self._get_task(task_id)
|
|
task.todos = todos
|
|
task.updated_at = datetime.now(timezone.utc)
|
|
await self.db.commit()
|
|
return task
|
|
|
|
async def update_scheduler(
|
|
self,
|
|
task_id: uuid.UUID,
|
|
scheduler: dict,
|
|
) -> Task:
|
|
task = await self._get_task(task_id)
|
|
task.scheduler = scheduler
|
|
task.updated_at = datetime.now(timezone.utc)
|
|
await self.db.commit()
|
|
return task
|
|
|
|
# ── 查询 ──
|
|
|
|
async def get_task(self, task_id: uuid.UUID) -> Task:
|
|
return await self._get_task(task_id)
|
|
|
|
async def list_tasks(
|
|
self,
|
|
state: TaskState | None = None,
|
|
assignee_org: str | None = None,
|
|
priority: str | None = None,
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
) -> list[Task]:
|
|
stmt = select(Task)
|
|
conditions = []
|
|
if state is not None:
|
|
conditions.append(Task.state == state)
|
|
if assignee_org is not None:
|
|
conditions.append(Task.assignee_org == assignee_org)
|
|
if priority is not None:
|
|
conditions.append(Task.priority == priority)
|
|
if conditions:
|
|
stmt = stmt.where(and_(*conditions))
|
|
stmt = stmt.order_by(Task.created_at.desc()).limit(limit).offset(offset)
|
|
result = await self.db.execute(stmt)
|
|
return list(result.scalars().all())
|
|
|
|
async def get_live_status(self) -> dict[str, Any]:
|
|
"""生成兼容旧 live_status.json 格式的全局状态。"""
|
|
tasks = await self.list_tasks(limit=200)
|
|
active_tasks = {}
|
|
completed_tasks = {}
|
|
for t in tasks:
|
|
d = t.to_dict()
|
|
if t.state in TERMINAL_STATES:
|
|
completed_tasks[str(t.task_id)] = d
|
|
else:
|
|
active_tasks[str(t.task_id)] = d
|
|
return {
|
|
"tasks": active_tasks,
|
|
"completed_tasks": completed_tasks,
|
|
"last_updated": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
|
|
async def count_tasks(self, state: TaskState | None = None) -> int:
|
|
stmt = select(func.count(Task.task_id))
|
|
if state is not None:
|
|
stmt = stmt.where(Task.state == state)
|
|
result = await self.db.execute(stmt)
|
|
return result.scalar_one()
|
|
|
|
# ── 内部 ──
|
|
|
|
async def _get_task(self, task_id: uuid.UUID) -> Task:
|
|
task = await self.db.get(Task, task_id)
|
|
if task is None:
|
|
raise ValueError(f"Task not found: {task_id}")
|
|
return task
|