264 lines
13 KiB
Python
264 lines
13 KiB
Python
"""Task records and backend selection around application-owned Pipecat Jobs."""
|
|
|
|
import asyncio
|
|
from collections import defaultdict
|
|
from copy import deepcopy
|
|
from datetime import UTC, datetime, timedelta
|
|
from uuid import uuid4
|
|
|
|
from loguru import logger
|
|
from services.background_tasks.native import NativeTaskRuntime
|
|
from services.background_tasks.opencode import OpenCodeExecutor
|
|
from services.background_tasks.records import TaskRecord, TaskRepository, TERMINAL
|
|
|
|
|
|
class TaskService:
|
|
def __init__(self, repository=None, executor_factory=None):
|
|
self.repository = repository or TaskRepository()
|
|
self.executor_factory = executor_factory
|
|
self.runtime = None
|
|
self._listeners = defaultdict(set)
|
|
self._jobs = {}
|
|
self._executors = {}
|
|
self._submit_lock = asyncio.Lock()
|
|
self._closing = False
|
|
|
|
async def start(self):
|
|
self._closing = False
|
|
self.runtime = NativeTaskRuntime(self._execute, self._cancel_backend, self._on_update, self._on_result)
|
|
await self.runtime.start()
|
|
try:
|
|
records = await self.repository.list(active=True)
|
|
except Exception:
|
|
await self.close()
|
|
raise
|
|
for record in records:
|
|
if record.backend_type == "reasoner":
|
|
await self.repository.patch(record.id, status="failed", summary="服务重启，内置任务已中断；未自动重放工具，请核查已执行操作")
|
|
elif record.remote_handle:
|
|
await self._dispatch(record)
|
|
else:
|
|
await self.repository.patch(record.id, status="unknown", summary="服务曾中断，未自动重新提交；请核查执行后端后再创建任务")
|
|
|
|
async def close(self):
|
|
self._closing = True
|
|
if self.runtime:
|
|
await self.runtime.close()
|
|
self.runtime = None
|
|
self._listeners.clear()
|
|
self._jobs.clear()
|
|
self._executors.clear()
|
|
|
|
async def submit(self, *, assistant_id, conversation_id, backend, instruction, invocation_id, variable_state=None):
|
|
if not self.runtime:
|
|
raise ValueError("后台任务服务尚未启动")
|
|
if not assistant_id or not conversation_id:
|
|
raise ValueError("后台任务需要已保存的助手和会话标识")
|
|
text = instruction.strip()
|
|
if not text or len(text) > 12000:
|
|
raise ValueError("任务指令必须为 1-12000 个字符")
|
|
if backend.type not in {"reasoner", "opencode"}:
|
|
raise ValueError("不支持该任务后端")
|
|
snapshot = backend.model_dump(mode="json") if hasattr(backend, "model_dump") else {
|
|
"id": backend.id, "name": getattr(backend, "name", ""), "type": backend.type,
|
|
"config": deepcopy(backend.config), "secrets": deepcopy(backend.secrets),
|
|
}
|
|
connection = deepcopy(backend.config)
|
|
connection["_backend_type"] = backend.type
|
|
connection["_backend"] = snapshot
|
|
state = variable_state or {}
|
|
connection["_variables"] = deepcopy(state.get("values", {}))
|
|
connection["_variable_definitions"] = deepcopy(state.get("definitions", {}))
|
|
connection["_secret_variables"] = deepcopy(state.get("secrets", {}))
|
|
timeout = int(connection.get("timeout_seconds", 600))
|
|
deadline = (datetime.now(UTC) + timedelta(seconds=timeout)).isoformat()
|
|
async with self._submit_lock:
|
|
record = await self.repository.create(TaskRecord(
|
|
id="task_" + uuid4().hex,
|
|
assistant_id=assistant_id, conversation_id=conversation_id, tool_id="",
|
|
backend_id=backend.id, backend_type=backend.type, deadline_at=deadline,
|
|
instruction=text, origin_key=f"{conversation_id}:{invocation_id}"[:256],
|
|
connection=connection, secrets=deepcopy(backend.secrets),
|
|
))
|
|
if not record.job_id and record.status == "queued":
|
|
await self._dispatch(record)
|
|
return (await self.get(record.id)).public()
|
|
|
|
async def _dispatch(self, record):
|
|
timeout = float(record.connection.get("timeout_seconds", 600))
|
|
if record.deadline_at:
|
|
timeout = (datetime.fromisoformat(record.deadline_at) - datetime.now(UTC)).total_seconds()
|
|
if timeout <= 0:
|
|
await self._save(record.id, status="cancel_requested", summary="任务已超过等待时限，正在核查停止状态")
|
|
await self._stop_record(await self.get(record.id))
|
|
return
|
|
job_id = await self.runtime.submit(record.id, timeout)
|
|
self._jobs[job_id] = record.id
|
|
await self.repository.patch(record.id, job_id=job_id)
|
|
|
|
async def get(self, task_id, assistant_id=None):
|
|
record = await self.repository.get(task_id)
|
|
if not record or (assistant_id and record.assistant_id != assistant_id):
|
|
raise ValueError("任务不存在或不属于当前助手")
|
|
return record
|
|
|
|
def subscribe(self, conversation_id, listener):
|
|
self._listeners[conversation_id].add(listener)
|
|
def detach():
|
|
self._listeners[conversation_id].discard(listener)
|
|
if not self._listeners[conversation_id]:
|
|
self._listeners.pop(conversation_id, None)
|
|
return detach
|
|
|
|
async def _publish(self, record):
|
|
for listener in tuple(self._listeners.get(record.conversation_id, ())):
|
|
try:
|
|
await listener(record.public())
|
|
except Exception:
|
|
logger.warning("任务通知未送达，结果仍可查询: {}", record.id)
|
|
|
|
async def _save(self, task_id, **changes):
|
|
record = await self.repository.patch(task_id, **changes)
|
|
await self._publish(record)
|
|
return record
|
|
|
|
def _executor(self, record):
|
|
if record.id in self._executors:
|
|
return self._executors[record.id]
|
|
config = {**record.connection, "_task_id": record.id}
|
|
if self.executor_factory:
|
|
return self.executor_factory(config, record.secrets)
|
|
if record.backend_type == "reasoner":
|
|
from services.background_tasks.reasoner import ReasonerExecutor
|
|
return ReasonerExecutor(config, record.secrets, runtime=self.runtime)
|
|
return OpenCodeExecutor(config, record.secrets)
|
|
|
|
async def _execute(self, message, worker):
|
|
task_id = message.payload["task_id"]
|
|
self._jobs[message.job_id] = task_id
|
|
record = await self.get(task_id)
|
|
if record.status in TERMINAL:
|
|
return {"task_id": task_id}
|
|
if record.status == "cancel_requested":
|
|
await self._stop_record(record)
|
|
return {"task_id": task_id}
|
|
executor = self._executor(record)
|
|
self._executors[task_id] = executor
|
|
resume = bool(record.remote_handle)
|
|
try:
|
|
if not resume:
|
|
outstanding = await self.repository.list(active=True)
|
|
if any(other.id != task_id and other.backend_type == "opencode" and other.remote_handle for other in outstanding):
|
|
await self.repository.patch(task_id, status="failed", summary="另一远端任务尚未结束，请先核查或停止该任务；本任务未提交到后端")
|
|
return {"task_id": task_id}
|
|
record = await self.repository.patch(task_id, remote_handle=await executor.prepare())
|
|
last_event = None
|
|
async def emit(event):
|
|
nonlocal last_event
|
|
if event == last_event:
|
|
return
|
|
last_event = deepcopy(event)
|
|
await self.repository.patch(task_id, **event)
|
|
await worker.send_job_update(message.job_id, {"task_id": task_id})
|
|
# Only OpenCode can reconnect observations without replaying work.
|
|
attempts = 1 if record.backend_type == "reasoner" else 5
|
|
for attempt in range(attempts):
|
|
try:
|
|
result = await executor.run(record.remote_handle, record.instruction, emit, resume=resume)
|
|
await self.repository.patch(task_id, **result, input_request={})
|
|
return {"task_id": task_id}
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
if attempt == attempts - 1:
|
|
raise
|
|
resume = True
|
|
await emit({"summary": "远端连接中断，正在核查任务状态"})
|
|
await asyncio.sleep(min(2 ** attempt, 10))
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
if record.backend_type == "reasoner":
|
|
await self.repository.patch(task_id, status="failed", summary="内置推理任务中断；未自动重复执行，请核查任务配置与已执行操作")
|
|
else:
|
|
await self.repository.patch(task_id, status="unknown", summary="无法确认远端执行状态，请检查远端服务；不会自动重复提交")
|
|
return {"task_id": task_id}
|
|
|
|
async def _on_update(self, message):
|
|
task_id = (message.update or {}).get("task_id")
|
|
if task_id:
|
|
await self._publish(await self.get(task_id))
|
|
|
|
async def _on_result(self, message):
|
|
task_id = self._jobs.pop(message.job_id, None)
|
|
if not task_id:
|
|
return
|
|
self._executors.pop(task_id, None)
|
|
if self._closing:
|
|
return
|
|
record = await self.get(task_id)
|
|
if record.status not in TERMINAL and str(message.status) in {"error", "failed"}:
|
|
await self._save(task_id, status="unknown", summary="任务运行中断，请核查执行后端状态")
|
|
else:
|
|
await self._publish(record)
|
|
|
|
async def cancel(self, task_id, assistant_id=None):
|
|
record = await self.get(task_id, assistant_id)
|
|
if record.status in TERMINAL:
|
|
return record.public()
|
|
await self._save(task_id, status="cancel_requested", summary="正在请求停止任务")
|
|
accepted = bool(self.runtime and record.job_id and await self.runtime.cancel(record.job_id))
|
|
if not accepted:
|
|
await self._stop_record(record)
|
|
return (await self.get(task_id)).public()
|
|
|
|
async def _cancel_backend(self, message):
|
|
task_id = self._jobs.get(message.job_id)
|
|
if not task_id or self._closing:
|
|
return
|
|
await self._stop_record(await self.get(task_id))
|
|
|
|
async def _stop_record(self, record):
|
|
if record.status in TERMINAL:
|
|
return
|
|
if not record.remote_handle:
|
|
await self._save(record.id, status="cancelled", summary="已取消，尚未向任务后端提交指令")
|
|
return
|
|
try:
|
|
stopped = await self._executor(record).cancel(record.remote_handle)
|
|
except Exception:
|
|
stopped = False
|
|
summary = "已停止内置推理；已发出的工具请求结果请核查" if record.backend_type == "reasoner" else "远端已停止"
|
|
await self._save(record.id, status="cancelled" if stopped else "cancel_requested", summary=summary if stopped else "停止尚未确认，请检查执行后端")
|
|
|
|
async def reconcile(self, task_id):
|
|
record = await self.get(task_id)
|
|
if not record.remote_handle or record.status in TERMINAL:
|
|
return record.public()
|
|
if record.status == "cancel_requested":
|
|
await self._stop_record(record)
|
|
else:
|
|
snapshot = await self._executor(record).snapshot(record.remote_handle)
|
|
if snapshot["status"] == "running" and record.input_request:
|
|
snapshot.update(status="waiting_input", summary="任务后端等待补充信息")
|
|
await self._save(task_id, **snapshot)
|
|
if snapshot["status"] not in TERMINAL and self.runtime and record.job_id not in self.runtime.coordinator.job_groups:
|
|
if record.backend_type == "reasoner":
|
|
await self._save(task_id, status="failed", summary="内置任务已中断，不自动重放工具")
|
|
else:
|
|
await self._dispatch(await self.get(task_id))
|
|
return (await self.get(task_id)).public()
|
|
|
|
async def send_input(self, task_id, value, assistant_id=None):
|
|
record = await self.get(task_id, assistant_id)
|
|
if record.status != "waiting_input" or not record.input_request:
|
|
raise ValueError("任务当前没有待处理的确认或问题")
|
|
await self._executor(record).send_input(record.input_request, value)
|
|
latest = await self.get(task_id)
|
|
if latest.input_request.get("id") == record.input_request.get("id"):
|
|
await self._save(task_id, status="running", summary="已发送回复，等待任务后端继续", input_request={})
|
|
return (await self.get(task_id)).public()
|
|
|
|
|
|
task_service = TaskService()
|