"""Business entry point backed by native Pipecat Jobs.""" import asyncio from collections import defaultdict from copy import deepcopy 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=OpenCodeExecutor): self.repository = repository or TaskRepository() self.executor_factory = executor_factory self.runtime = None self._listeners = defaultdict(set) self._jobs = {} self._submit_lock = asyncio.Lock() self._closing = False async def start(self): self._closing = False self.runtime = NativeTaskRuntime(self._execute, self._cancel_remote, 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.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() async def submit(self, *, assistant_id, conversation_id, tool, instruction, invocation_id): 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 tool.type != "task": raise ValueError("请选择后台任务工具") 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=tool.id, instruction=text, origin_key=f"{conversation_id}:{invocation_id}"[:256], connection=deepcopy(tool.definition["config"]), secrets=deepcopy(tool.secrets), )) if not record.job_id and record.status == "queued": await self._dispatch(record) return record.public() async def _dispatch(self, record): job_id = await self.runtime.submit(record.id, record.connection.get("timeout_seconds", 600)) 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 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) executor = self.executor_factory(record.connection, record.secrets) resume = bool(record.remote_handle) try: if not resume: handle = await executor.prepare() record = await self.repository.patch(task_id, remote_handle=handle) 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}) # Reconnect only observations; never repeat a mutating prompt. for attempt in range(5): 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: resume = True await emit({"summary": "远端连接中断，正在核查任务状态"}) if attempt == 4: raise await asyncio.sleep(min(2 ** attempt, 10)) except asyncio.CancelledError: raise except Exception: 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: record = await self.get(task_id) await self._publish(record) async def _on_result(self, message): if self._closing: return task_id = self._jobs.pop(message.job_id, None) if not task_id: 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_remote(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_factory(record.connection, record.secrets).cancel(record.remote_handle) except Exception: stopped = False await self._save(record.id, status="cancelled" if stopped else "cancel_requested", 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() executor = self.executor_factory(record.connection, record.secrets) if record.status == "cancel_requested": await self._stop_record(record) else: snapshot = await executor.snapshot(record.remote_handle) # Session status does not describe pending questions/permissions. # Keep the saved request visible until a reply or terminal result. 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: 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_factory(record.connection, record.secrets).send_input(record.input_request, value) await self._save(task_id, status="running", summary="已发送回复，等待远端继续", input_request={}) return (await self.get(task_id)).public() task_service = TaskService()