F2: persist native job projections and expose task controls

2026-10-23 16:03:40 +08:00
parent bae5ec3c58
commit 33fde56edd
13 changed files with 550 additions and 5 deletions
@@ -0,0 +1,85 @@
+"""Persistent projections of native jobs; this is not another scheduler."""
+
+from dataclasses import asdict, dataclass, field
+from datetime import UTC, datetime
+from uuid import uuid4
+
+from sqlalchemy import select
+from pydantic.alias_generators import to_camel
+from db.models import BackgroundTask, BackgroundTaskEvent
+from db.session import SessionLocal
+
+TERMINAL = {"succeeded", "failed", "cancelled"}
+ACTIVE = {"queued", "submitted", "running", "waiting_input", "cancel_requested", "unknown"}
+
+
+@dataclass
+class TaskRecord:
+    id: str
+    assistant_id: str
+    conversation_id: str
+    tool_id: str
+    instruction: str
+    origin_key: str
+    connection: dict
+    secrets: dict
+    status: str = "queued"
+    summary: str = "任务已排队"
+    result: str = ""
+    remote_handle: dict = field(default_factory=dict)
+    input_request: dict = field(default_factory=dict)
+    job_id: str = ""
+    sequence: int = 0
+    created_at: str = field(default_factory=lambda: datetime.now(UTC).isoformat())
+    updated_at: str = field(default_factory=lambda: datetime.now(UTC).isoformat())
+
+    def public(self):
+        hidden = {"connection", "secrets", "origin_key", "remote_handle"}
+        return {to_camel(key): value for key, value in asdict(self).items() if key not in hidden}
+
+
+def from_row(row) -> TaskRecord:
+    return TaskRecord(**{key: getattr(row, key) for key in TaskRecord.__dataclass_fields__})
+
+
+class TaskRepository:
+    async def create(self, record: TaskRecord) -> TaskRecord:
+        async with SessionLocal() as db:
+            # All calls are serialized by the application entry point; unique
+            # origin_key also protects against duplicate tool delivery.
+            existing = (await db.execute(select(BackgroundTask).where(BackgroundTask.origin_key == record.origin_key))).scalar_one_or_none()
+            if existing:
+                return from_row(existing)
+            db.add(BackgroundTask(**asdict(record)))
+            await db.commit()
+        return record
+
+    async def get(self, task_id: str) -> TaskRecord | None:
+        async with SessionLocal() as db:
+            row = await db.get(BackgroundTask, task_id)
+            return from_row(row) if row else None
+
+    async def list(self, assistant_id: str | None = None, *, active=False) -> list[TaskRecord]:
+        statement = select(BackgroundTask).order_by(BackgroundTask.created_at.desc())
+        if not active:
+            statement = statement.limit(100)
+        if assistant_id:
+            statement = statement.where(BackgroundTask.assistant_id == assistant_id)
+        if active:
+            statement = statement.where(BackgroundTask.status.in_(ACTIVE))
+        async with SessionLocal() as db:
+            return [from_row(row) for row in (await db.execute(statement)).scalars().all()]
+
+    async def patch(self, task_id: str, **changes) -> TaskRecord:
+        async with SessionLocal() as db:
+            row = (await db.execute(select(BackgroundTask).where(BackgroundTask.id == task_id).with_for_update())).scalar_one()
+            if row.status in TERMINAL:
+                return from_row(row)
+            for key, value in changes.items():
+                setattr(row, key, value)
+            row.sequence += 1
+            row.updated_at = datetime.now(UTC).isoformat()
+            record = from_row(row)
+            db.add(BackgroundTaskEvent(id="te_" + uuid4().hex, task_id=task_id, sequence=row.sequence, data=record.public()))
+            await db.commit()
+            return record
@@ -0,0 +1,177 @@
+"""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()
+        for record in await self.repository.list(active=True):
+            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()
+
+    async def submit(self, *, assistant_id, conversation_id, tool, instruction, invocation_id):
+        if not self.runtime:
+            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 _save(self, task_id, **changes):
+        record = await self.repository.patch(task_id, **changes)
+        for listener in tuple(self._listeners.get(record.conversation_id, ())):
+            try:
+                await listener(record.public())
+            except Exception:
+                logger.warning("任务通知未送达，结果仍可查询: {}", task_id)
+        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)
+            async def emit(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._save(task_id, summary=record.summary)
+
+    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._save(task_id, summary=record.summary)
+
+    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="unknown", 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 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()