diff --git a/backend/services/background_tasks/__init__.py b/backend/services/background_tasks/__init__.py new file mode 100644 index 0000000..f294ed1 --- /dev/null +++ b/backend/services/background_tasks/__init__.py @@ -0,0 +1 @@ +"""Prompt background tasks built on Pipecat's native Worker/Jobs.""" diff --git a/backend/services/background_tasks/native.py b/backend/services/background_tasks/native.py new file mode 100644 index 0000000..2fa1496 --- /dev/null +++ b/backend/services/background_tasks/native.py @@ -0,0 +1,83 @@ +"""A long-lived owner for native Pipecat jobs, independent of media calls.""" + +import asyncio +from collections.abc import Awaitable, Callable + +from pipecat.pipeline.job_context import JobParams, JobStatus +from pipecat.pipeline.job_decorator import job +from pipecat.workers.base_worker import BaseWorker +from pipecat.workers.runner import WorkerRunner + +JobCallback = Callable[[object], Awaitable[None]] +Execute = Callable[[object, BaseWorker], Awaitable[dict]] + + +class CoordinatorWorker(BaseWorker): + def __init__(self, on_update: JobCallback, on_result: JobCallback): + super().__init__("background-task-coordinator") + self._update = on_update + self._result = on_result + + async def on_job_update(self, message): + await self._update(message) + + async def on_job_response(self, message): + await self._result(message) + + +class RemoteAgentWorker(BaseWorker): + def __init__(self, execute: Execute, cancel: JobCallback): + super().__init__("background-task-executor") + self._execute = execute + self._cancel = cancel + + @job(name="execute") + async def execute(self, message): + try: + result = await self._execute(message, self) + except asyncio.CancelledError: + raise + except Exception: + # Error details belong in the repository; don't publish credentials. + await self.send_job_response(message.job_id, status=JobStatus.ERROR) + else: + await self.send_job_response(message.job_id, result) + + async def on_job_cancelled(self, message): + await self._cancel(message) + + +class NativeTaskRuntime: + """Only wiring: scheduling, deadlines and cancellation belong to Pipecat.""" + + def __init__(self, execute: Execute, cancel: JobCallback, + on_update: JobCallback, on_result: JobCallback): + self.coordinator = CoordinatorWorker(on_update, on_result) + self.executor = RemoteAgentWorker(execute, cancel) + self.runner: WorkerRunner | None = None + self._run: asyncio.Task | None = None + + async def start(self): + self.runner = WorkerRunner(handle_sigint=False, check_dangling_tasks=False) + ready = asyncio.Event() + async def on_ready(_runner): + ready.set() + self.runner.add_event_handler("on_ready", on_ready) + await self.runner.add_workers(self.coordinator, self.executor) + self._run = asyncio.create_task(self.runner.run(auto_end=False), name="background-jobs") + await asyncio.wait_for(ready.wait(), 10) + + async def submit(self, task_id: str, timeout: float) -> str: + return await self.coordinator.request_job( + self.executor.name, + params=JobParams(name="execute", payload={"task_id": task_id}, timeout=timeout), + ) + + async def cancel(self, job_id: str): + return await self.coordinator.request_cancel_job_group(job_id, reason="user requested stop") + + async def close(self): + if self.runner: + await self.runner.cancel(reason="application shutdown") + if self._run: + await self._run diff --git a/backend/tests/test_background_task_native.py b/backend/tests/test_background_task_native.py new file mode 100644 index 0000000..e7b7479 --- /dev/null +++ b/backend/tests/test_background_task_native.py @@ -0,0 +1,59 @@ +"""Real Pipecat scheduling with an explicitly simulated remote executor.""" +import asyncio +import json +import unittest +from pathlib import Path + +from pipecat.frames.frames import CancelFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.worker import PipelineWorker +from pipecat.processors.frame_processor import FrameProcessor +from services.background_tasks.native import NativeTaskRuntime + + +class Passthrough(FrameProcessor): + async def process_frame(self, frame, direction): + await super().process_frame(frame, direction) + await self.push_frame(frame, direction) + + +class NativeJobTest(unittest.IsolatedAsyncioTestCase): + async def test_progress_completion_and_media_disconnect(self): + progress, finished, release = asyncio.Event(), asyncio.Event(), asyncio.Event() + results = [] + + async def execute(message, worker): + # SIMULATED executor; no remote OpenCode or model is called here. + await worker.send_job_update(message.job_id, {"step": "working"}) + await release.wait() + return {"result": "done"} + + async def cancelled(message): + pass + + async def update(message): + progress.set() + + async def result(message): + results.append(message) + finished.set() + + runtime = NativeTaskRuntime(execute, cancelled, update, result) + await runtime.start() + try: + job_id = await runtime.submit("test-task", 5) + await asyncio.wait_for(progress.wait(), 2) + self.assertFalse(finished.is_set()) + media = PipelineWorker(Pipeline([Passthrough()]), enable_rtvi=False) + await runtime.runner.add_workers(media) + await media.queue_frame(CancelFrame(reason="client_disconnected")) + await asyncio.wait_for(media.wait(), 2) + self.assertIn(job_id, runtime.executor.active_jobs) + release.set() + await asyncio.wait_for(finished.wait(), 2) + self.assertEqual(results[0].status, "completed") + self.assertEqual(results[0].response, {"result": "done"}) + artifact = Path(__file__).resolve().parents[2] / "implement-stage/native-jobs-smoke.json" + artifact.write_text(json.dumps({"simulated_executor": True, "native_job": "completed", "survives_media_disconnect": True}) + "\n") + finally: + await runtime.close() diff --git a/implement-stage/ASSUMPTIONS.md b/implement-stage/ASSUMPTIONS.md new file mode 100644 index 0000000..51e9487 --- /dev/null +++ b/implement-stage/ASSUMPTIONS.md @@ -0,0 +1,17 @@ +# Assumption Ledger — Prompt background tasks + + +| ID | Under-determined by the request | Chosen | Class | Source | +|----|--------------------------------|--------|-------|--------| +| A-001 | 范围 | 只允许提示词 Pipeline 绑定任务工具;不开放 Workflow、Realtime、启动 Action | semantic | user | +| A-002 | 任务生命周期 | 应用级原生 Worker 拥有 Job;网页断线任务继续 | semantic | user | +| A-003 | 配置入口 | 新增 task 工具类型,复用 Tool CRUD/绑定;每个助手最多一个 | interface | default | +| A-004 | 任务作用域 | 平台生成 task_id,记录来源 conversation/assistant;管理接口沿用现有管理员鉴权 | interface | default | +| A-005 | 远端执行关联 | 一次任务一个 OpenCode session;提交前保存session及messageID,初始idle不算完成 | semantic | default | +| A-006 | 取消 | 原生Job取消后调用远端abort并核查;无法确认时保留待确认,不能宣称已停止 | semantic | default | +| A-007 | 通知 | 进度仅网页;完成/失败在线时触发前台总结;离线结果保留供查询 | semantic | default | +| A-008 | 恢复 | 单进程;重启恢复观察有远端句柄的任务,不自动重新发送未知提交 | semantic | default | +| A-009 | 远端配置 | 地址/工作目录/Basic鉴权由管理员工具配置,任务冻结连接信息;LLM不能改地址和目录 | interface | default | +| A-010 | 补充输入 | 同一运行任务可发送文本;权限/问题交互显式等待管理员处理,不自动批准 | semantic | default | +| A-011 | 验证边界 | 未提供真实远端服务,使用httpx MockTransport验证协议和真实Pipecat验证调度;真实联调待部署配置 | semantic | default | +| A-012 | 外部编码变更 | 仅明确调用任务工具时会启动远端执行,不因查看页面自动执行 | semantic | default | diff --git a/implement-stage/BUILD_NOTE.md b/implement-stage/BUILD_NOTE.md new file mode 100644 index 0000000..79d2fba --- /dev/null +++ b/implement-stage/BUILD_NOTE.md @@ -0,0 +1,16 @@ +# Build Note — Prompt background tasks + +| Rung | Feature | Acceptance check | Tier | Status | +|------|---------|------------------|------|--------| +| F0 | 原生Job主线、断线独立与取消 | backend/.venv/bin/python -m unittest discover -s backend/tests -p test_background_task_native.py | MUST | passed | +| F1 | OpenCode协议适配与状态关联 | cd backend && .venv/bin/python -m unittest tests.test_background_task_native tests.test_opencode_tasks | MUST | pending | +| F2 | 持久记录、管理API、应用宿主 | cd backend && .venv/bin/python -m unittest tests.test_background_tasks | MUST | pending | +| F3 | 提示词绑定、任务通知、网页 | cd frontend && npx tsc --noEmit | MUST | pending | +| F4 | 回归、文档与假设审阅 | make test | MUST | pending | + +## Run record +- F0: PYTHONPATH=backend backend/.venv/bin/python -m unittest discover -s backend/tests -p test_background_task_native.py → exit 0; native-jobs-smoke.json。 +## Deferred +- 真实远程 OpenCode + 模型联调:未提供服务地址/凭证。 +- 多进程分布式总线及通用插件发现:超出单进程 MVP。 +## Blockers diff --git a/implement-stage/SPEC.md b/implement-stage/SPEC.md new file mode 100644 index 0000000..e2fc5d8 --- /dev/null +++ b/implement-stage/SPEC.md @@ -0,0 +1,9 @@ +# 提示词 Pipeline 远程任务 MVP + +Base commit: d45be1e6864d1ea9ac73e9f15490873e2d6f606b + +目标:使用 Pipecat 1.13.0 原生 Worker/Jobs,为提示词 Pipeline 增加远程 OpenCode 后台任务。输入为管理员配置的任务工具、工作目录和用户指令;输出为任务回执、持久进度/结果、网页卡片及在线完成通知。 + +成功检查:backend/.venv/bin/python -m scripts.run_tests;frontend npm run lint、npm run test:contracts、npm run build。 + +范围:单应用进程,原生 AsyncQueueBus;任务所有者常驻应用;仅提示词 Pipeline。真实远端服务未提供,协议验证使用明确标注的本地模拟服务,不声称完成真实模型联调。 diff --git a/implement-stage/native-jobs-smoke.json b/implement-stage/native-jobs-smoke.json new file mode 100644 index 0000000..4df7c31 --- /dev/null +++ b/implement-stage/native-jobs-smoke.json @@ -0,0 +1 @@ +{"simulated_executor": true, "native_job": "completed", "survives_media_disconnect": true}