From 8ac75a187380ffa463aa67de07673f41bbaae752 Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Fri, 23 Oct 2026 16:27:00 +0800 Subject: [PATCH] F4: verify task recovery and document Prompt background jobs --- backend/README.md | 2 + backend/contracts/openapi.json | 404 +++++++++++++++++- backend/routes/background_tasks.py | 10 + backend/services/background_tasks/native.py | 2 +- backend/services/background_tasks/service.py | 48 ++- backend/services/pipecat/pipeline.py | 61 +-- backend/tests/test_background_task_native.py | 20 + backend/tests/test_background_tasks.py | 84 +++- docs-developer/prompt-background-tasks.mdx | 50 +++ .../background-task-panel.tsx | 1 + .../components/tools/tool-binding-editor.tsx | 5 +- frontend/src/lib/api.ts | 1 + implement-stage/ASSUMPTIONS.md | 6 +- implement-stage/BUILD_NOTE.md | 6 +- 14 files changed, 654 insertions(+), 46 deletions(-) create mode 100644 docs-developer/prompt-background-tasks.mdx diff --git a/backend/README.md b/backend/README.md index cf0daf8..131cea9 100644 --- a/backend/README.md +++ b/backend/README.md @@ -61,3 +61,5 @@ make contract-check ``` 开发计划中的生产可观测性、并发保护和可靠性工作尚未全部完成，见[单租户生产化计划](../docs-developer/development-plan.mdx)。 + +提示词 Pipeline 的远程编程任务接入与迁移步骤见 [后台任务说明](../docs-developer/prompt-background-tasks.mdx)。使用 Pipecat 原生 Jobs，需先迁移数据库并配置已有的远程 OpenCode 服务。 diff --git a/backend/contracts/openapi.json b/backend/contracts/openapi.json index 7866f8e..eb0d241 100644 --- a/backend/contracts/openapi.json +++ b/backend/contracts/openapi.json @@ -5082,6 +5082,148 @@ "title": "StateUpdateRule", "type": "object" }, + "SubmitTask": { + "properties": { + "assistantId": { + "title": "Assistantid", + "type": "string" + }, + "instruction": { + "maxLength": 12000, + "minLength": 1, + "title": "Instruction", + "type": "string" + }, + "requestId": { + "maxLength": 80, + "minLength": 1, + "title": "Requestid", + "type": "string" + }, + "toolId": { + "title": "Toolid", + "type": "string" + } + }, + "required": [ + "assistantId", + "toolId", + "instruction", + "requestId" + ], + "title": "SubmitTask", + "type": "object" + }, + "TaskInput": { + "properties": { + "answers": { + "anyOf": [ + { + "items": { + "items": { + "type": "string" + }, + "type": "array" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "title": "Answers" + }, + "reply": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Reply" + } + }, + "title": "TaskInput", + "type": "object" + }, + "TaskToolConfig": { + "properties": { + "allowInterruptions": { + "const": true, + "default": true, + "title": "Allowinterruptions", + "type": "boolean" + }, + "directory": { + "maxLength": 2048, + "minLength": 1, + "title": "Directory", + "type": "string" + }, + "executionMode": { + "const": "immediate", + "default": "immediate", + "title": "Executionmode", + "type": "string" + }, + "provider": { + "const": "opencode", + "default": "opencode", + "title": "Provider", + "type": "string" + }, + "timeoutSeconds": { + "default": 600, + "maximum": 7200.0, + "minimum": 30.0, + "title": "Timeoutseconds", + "type": "integer" + }, + "url": { + "maxLength": 2048, + "minLength": 1, + "title": "Url", + "type": "string" + }, + "username": { + "default": "opencode", + "maxLength": 128, + "title": "Username", + "type": "string" + } + }, + "required": [ + "url", + "directory" + ], + "title": "TaskToolConfig", + "type": "object" + }, + "TaskToolDefinition": { + "properties": { + "config": { + "$ref": "#/components/schemas/TaskToolConfig" + }, + "schemaVersion": { + "default": 2, + "title": "Schemaversion", + "type": "integer" + }, + "type": { + "const": "task", + "default": "task", + "title": "Type", + "type": "string" + } + }, + "required": [ + "config" + ], + "title": "TaskToolDefinition", + "type": "object" + }, "TemplateValueSource": { "properties": { "source": { @@ -5648,7 +5790,8 @@ "mapping": { "client": "#/components/schemas/ClientToolDefinition", "http": "#/components/schemas/HttpToolDefinition", - "mcp": "#/components/schemas/McpToolDefinition" + "mcp": "#/components/schemas/McpToolDefinition", + "task": "#/components/schemas/TaskToolDefinition" }, "propertyName": "type" }, @@ -5661,6 +5804,9 @@ }, { "$ref": "#/components/schemas/McpToolDefinition" + }, + { + "$ref": "#/components/schemas/TaskToolDefinition" } ], "title": "Definition" @@ -5727,7 +5873,8 @@ "enum": [ "http", "mcp", - "client" + "client", + "task" ], "title": "Type", "type": "string" @@ -5927,7 +6074,8 @@ "mapping": { "client": "#/components/schemas/ClientToolDefinition", "http": "#/components/schemas/HttpToolDefinition", - "mcp": "#/components/schemas/McpToolDefinition" + "mcp": "#/components/schemas/McpToolDefinition", + "task": "#/components/schemas/TaskToolDefinition" }, "propertyName": "type" }, @@ -5940,6 +6088,9 @@ }, { "$ref": "#/components/schemas/McpToolDefinition" + }, + { + "$ref": "#/components/schemas/TaskToolDefinition" } ], "title": "Definition" @@ -6541,6 +6692,253 @@ ] } }, + "/api/background-tasks": { + "get": { + "operationId": "list_tasks_api_background_tasks_get", + "parameters": [ + { + "in": "query", + "name": "assistant_id", + "required": true, + "schema": { + "title": "Assistant Id", + "type": "string" + } + } + ], + "responses": { + "200": { + "content": { + "application/json": { + "schema": {} + } + }, + "description": "Successful Response" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + "description": "Validation Error" + } + }, + "summary": "List Tasks", + "tags": [ + "background-tasks" + ] + }, + "post": { + "operationId": "submit_task_api_background_tasks_post", + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SubmitTask" + } + } + }, + "required": true + }, + "responses": { + "200": { + "content": { + "application/json": { + "schema": {} + } + }, + "description": "Successful Response" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + "description": "Validation Error" + } + }, + "summary": "Submit Task", + "tags": [ + "background-tasks" + ] + } + }, + "/api/background-tasks/{task_id}": { + "get": { + "operationId": "get_task_api_background_tasks__task_id__get", + "parameters": [ + { + "in": "path", + "name": "task_id", + "required": true, + "schema": { + "title": "Task Id", + "type": "string" + } + } + ], + "responses": { + "200": { + "content": { + "application/json": { + "schema": {} + } + }, + "description": "Successful Response" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + "description": "Validation Error" + } + }, + "summary": "Get Task", + "tags": [ + "background-tasks" + ] + } + }, + "/api/background-tasks/{task_id}/cancel": { + "post": { + "operationId": "cancel_task_api_background_tasks__task_id__cancel_post", + "parameters": [ + { + "in": "path", + "name": "task_id", + "required": true, + "schema": { + "title": "Task Id", + "type": "string" + } + } + ], + "responses": { + "200": { + "content": { + "application/json": { + "schema": {} + } + }, + "description": "Successful Response" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + "description": "Validation Error" + } + }, + "summary": "Cancel Task", + "tags": [ + "background-tasks" + ] + } + }, + "/api/background-tasks/{task_id}/input": { + "post": { + "operationId": "send_input_api_background_tasks__task_id__input_post", + "parameters": [ + { + "in": "path", + "name": "task_id", + "required": true, + "schema": { + "title": "Task Id", + "type": "string" + } + } + ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/TaskInput" + } + } + }, + "required": true + }, + "responses": { + "200": { + "content": { + "application/json": { + "schema": {} + } + }, + "description": "Successful Response" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + "description": "Validation Error" + } + }, + "summary": "Send Input", + "tags": [ + "background-tasks" + ] + } + }, + "/api/background-tasks/{task_id}/refresh": { + "post": { + "operationId": "refresh_task_api_background_tasks__task_id__refresh_post", + "parameters": [ + { + "in": "path", + "name": "task_id", + "required": true, + "schema": { + "title": "Task Id", + "type": "string" + } + } + ], + "responses": { + "200": { + "content": { + "application/json": { + "schema": {} + } + }, + "description": "Successful Response" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + "description": "Validation Error" + } + }, + "summary": "Refresh Task", + "tags": [ + "background-tasks" + ] + } + }, "/api/conversations": { "get": { "operationId": "list_conversations_api_conversations_get", diff --git a/backend/routes/background_tasks.py b/backend/routes/background_tasks.py index 0b50a31..d2837b8 100644 --- a/backend/routes/background_tasks.py +++ b/backend/routes/background_tasks.py @@ -66,3 +66,13 @@ async def send_input(task_id: str, body: TaskInput): raise HTTPException(400, str(exc)) from exc except Exception as exc: raise HTTPException(502, "远端暂未接受回复，请稍后重试") from exc + + +@router.post("/{task_id}/refresh") +async def refresh_task(task_id: str): + try: + return await task_service.reconcile(task_id) + except ValueError as exc: + raise HTTPException(404, str(exc)) from exc + except Exception as exc: + raise HTTPException(502, "暂时无法核查远端状态") from exc diff --git a/backend/services/background_tasks/native.py b/backend/services/background_tasks/native.py index 2fa1496..6d1de09 100644 --- a/backend/services/background_tasks/native.py +++ b/backend/services/background_tasks/native.py @@ -31,7 +31,7 @@ class RemoteAgentWorker(BaseWorker): self._execute = execute self._cancel = cancel - @job(name="execute") + @job(name="execute", sequential=True) async def execute(self, message): try: result = await self._execute(message, self) diff --git a/backend/services/background_tasks/service.py b/backend/services/background_tasks/service.py index f212dd4..ebcac41 100644 --- a/backend/services/background_tasks/service.py +++ b/backend/services/background_tasks/service.py @@ -25,7 +25,12 @@ class TaskService: 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): + 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: @@ -37,10 +42,13 @@ class TaskService: 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 个字符") @@ -77,13 +85,16 @@ class TaskService: self._listeners.pop(conversation_id, None) return detach - async def _save(self, task_id, **changes): - record = await self.repository.patch(task_id, **changes) + async def _publish(self, record): for listener in tuple(self._listeners.get(record.conversation_id, ())): try: await listener(record.public()) except Exception: - logger.warning("任务通知未送达，结果仍可查询: {}", task_id) + 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): @@ -96,7 +107,12 @@ class TaskService: 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. @@ -123,7 +139,7 @@ class TaskService: 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) + await self._publish(record) async def _on_result(self, message): if self._closing: @@ -135,7 +151,7 @@ class TaskService: 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) + await self._publish(record) async def cancel(self, task_id, assistant_id=None): record = await self.get(task_id, assistant_id) @@ -157,7 +173,7 @@ class TaskService: if record.status in TERMINAL: return if not record.remote_handle: - await self._save(record.id, status="unknown", summary="尚无远端句柄，无法确认是否已停止") + await self._save(record.id, status="cancelled", summary="已取消，尚未向远端提交指令") return try: stopped = await self.executor_factory(record.connection, record.secrets).cancel(record.remote_handle) @@ -165,6 +181,24 @@ class TaskService: 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: diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index b88a789..26edc26 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -847,40 +847,43 @@ async def run_pipeline( messages.insert(0, {"role": "system", "content": content}) set_visible_tools([]) - await brain.setup( - cfg, - BrainRuntime( - context=context, - llm=llm, - queue_frame=worker.queue_frame, - set_system_prompt=set_system_prompt, - set_tools=set_visible_tools, - call_end=call_end, - session_id=cfg.conversation_id, - client_tools=client_tools, - worker=worker, - context_aggregator=WorkflowAggregatorPair( - user_aggregator, - assistant_aggregator, - ), - transport=transport, - switch_services=service_controller.switch, - set_knowledge_scope=knowledge_retrieval.set_scope, - set_vision_scope=lambda scope: workflow_vision_scope.update(scope), - vision_function=workflow_vision_function, - set_input_enabled=lambda enabled: input_state.__setitem__("enabled", enabled), - apply_turn_config=apply_workflow_turn_config, - flow_global_functions=flow_global_functions, - ), - ) try: + await brain.setup( + cfg, + BrainRuntime( + context=context, + llm=llm, + queue_frame=worker.queue_frame, + set_system_prompt=set_system_prompt, + set_tools=set_visible_tools, + call_end=call_end, + session_id=cfg.conversation_id, + client_tools=client_tools, + worker=worker, + context_aggregator=WorkflowAggregatorPair( + user_aggregator, + assistant_aggregator, + ), + transport=transport, + switch_services=service_controller.switch, + set_knowledge_scope=knowledge_retrieval.set_scope, + set_vision_scope=lambda scope: workflow_vision_scope.update(scope), + vision_function=workflow_vision_function, + set_input_enabled=lambda enabled: input_state.__setitem__("enabled", enabled), + apply_turn_config=apply_workflow_turn_config, + flow_global_functions=flow_global_functions, + ), + ) await brain.run_preflight() - except Exception: + # Preflight tools may assign variables used by the opening speech. + greeting = await brain.greeting(cfg) + except BaseException: + close = getattr(brain, "close", None) + if close: + await close() if recorder: await recorder.finish(status="failed") raise - # Preflight tools may assign variables used by the opening speech. - greeting = await brain.greeting(cfg) async def submit_user_input(value: UserInput) -> None: if not value.has_image: diff --git a/backend/tests/test_background_task_native.py b/backend/tests/test_background_task_native.py index e7b7479..b6fa967 100644 --- a/backend/tests/test_background_task_native.py +++ b/backend/tests/test_background_task_native.py @@ -18,6 +18,26 @@ class Passthrough(FrameProcessor): class NativeJobTest(unittest.IsolatedAsyncioTestCase): + async def test_native_timeout_calls_remote_cleanup(self): + cancelled = asyncio.Event() + + async def execute(message, worker): + await asyncio.Event().wait() + + async def cleanup(message): + cancelled.set() + + async def ignore(message): + pass + + runtime = NativeTaskRuntime(execute, cleanup, ignore, ignore) + await runtime.start() + try: + await runtime.submit("timeout-test", 0.05) + await asyncio.wait_for(cancelled.wait(), 2) + finally: + await runtime.close() + async def test_progress_completion_and_media_disconnect(self): progress, finished, release = asyncio.Event(), asyncio.Event(), asyncio.Event() results = [] diff --git a/backend/tests/test_background_tasks.py b/backend/tests/test_background_tasks.py index af98b1b..0a7bc2b 100644 --- a/backend/tests/test_background_tasks.py +++ b/backend/tests/test_background_tasks.py @@ -35,6 +35,7 @@ class MemoryRepository: class SimulatedExecutor: release = None calls = 0 + confirms_stop = True def __init__(self, config, secrets): pass async def prepare(self): @@ -45,13 +46,16 @@ class SimulatedExecutor: await type(self).release.wait() return {"status": "succeeded", "summary": "完成", "result": "真实Job，模拟远端结果"} async def cancel(self, handle): - return True + return type(self).confirms_stop + async def snapshot(self, handle): + return {"status": "running", "summary": "远端仍在执行", "result": ""} class TaskServiceTest(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self): SimulatedExecutor.release = asyncio.Event() SimulatedExecutor.calls = 0 + SimulatedExecutor.confirms_stop = True self.service = TaskService(MemoryRepository(), SimulatedExecutor) await self.service.start() self.tool = SimpleNamespace(type="task", id="tool_task", definition={"config": {"url": "https://remote.test", "directory": "/demo", "timeout_seconds": 30}}, secrets={"password": "never-public"}) @@ -84,7 +88,85 @@ class TaskServiceTest(unittest.IsolatedAsyncioTestCase): await self.wait_status(task["id"], "running") await self.service.close() self.assertEqual((await self.service.get(task["id"])).status, "running") + async def test_unconfirmed_remote_stop_is_not_reported_cancelled(self): + SimulatedExecutor.confirms_stop = False + task = await self.submit() + await self.wait_status(task["id"], "running") + await self.service.cancel(task["id"]) + async with asyncio.timeout(2): + while self.service.runtime.executor.active_jobs: + await asyncio.sleep(.005) + self.assertEqual((await self.service.get(task["id"])).status, "cancel_requested") + + async def test_queued_task_can_be_cancelled_without_remote_submission(self): + first = await self.submit("call_first") + await self.wait_status(first["id"], "running") + second = await self.submit("call_second") + await self.service.cancel(second["id"]) + await self.wait_status(second["id"], "cancelled") + self.assertEqual(SimulatedExecutor.calls, 1) + + async def test_restart_observes_existing_remote_handle(self): + task = await self.submit() + await self.wait_status(task["id"], "running") + handle = (await self.service.get(task["id"])).remote_handle + await self.service.close() + await self.service.start() + async with asyncio.timeout(2): + while SimulatedExecutor.calls < 2: + await asyncio.sleep(.005) + self.assertEqual((await self.service.get(task["id"])).remote_handle, handle) + SimulatedExecutor.release.set() + await self.wait_status(task["id"], "succeeded") + + async def test_reconcile_preserves_a_pending_question(self): + task = await self.submit() + await self.wait_status(task["id"], "running") + question = {"kind": "question", "id": "que_1", "questions": [{"question": "使用哪个分支？"}]} + await self.service.repository.patch(task["id"], status="waiting_input", input_request=question) + result = await self.service.reconcile(task["id"]) + self.assertEqual(result["status"], "waiting_input") + self.assertEqual(result["inputRequest"], question) + def test_task_tool_preserves_masked_password(self): body = ToolUpsert(name="后台编程", functionName="coding_task", definition={"type": "task", "config": {"url": "https://remote.test", "directory": "/demo"}}, secrets={"password": "****"}) data = _payload(body, {"password": "old-password"}) self.assertEqual(data["secrets"]["password"], "old-password") + + +class TaskApiTest(unittest.IsolatedAsyncioTestCase): + async def test_admin_auth_and_real_native_job_through_http_api(self): + import httpx + from fastapi import FastAPI + from unittest.mock import AsyncMock, patch + from routes import background_tasks as routes + from services.auth import require_admin + from db.session import get_session + SimulatedExecutor.release = asyncio.Event() + service = TaskService(MemoryRepository(), SimulatedExecutor) + await service.start() + app = FastAPI() + app.include_router(routes.router) + async def database(): + yield None + app.dependency_overrides[get_session] = database + cfg = SimpleNamespace(type="prompt", runtimeMode="pipeline", tool_bindings=[SimpleNamespace(tool_id="tool_task")], tools=[SimpleNamespace(type="task", id="tool_task", definition={"config": {"url": "https://remote.test", "directory": "/demo", "timeout_seconds": 30}}, secrets={})]) + try: + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="https://platform.test") as client: + unauthenticated = await client.get("/api/background-tasks?assistant_id=asst_1") + self.assertEqual(unauthenticated.status_code, 401) + app.dependency_overrides[require_admin] = lambda: {"username": "test"} + with patch.object(routes, "task_service", service), patch.object(routes, "resolve_runtime_config", AsyncMock(return_value=cfg)): + response = await client.post("/api/background-tasks", json={"assistantId": "asst_1", "toolId": "tool_task", "instruction": "测试请求", "requestId": "request_1"}) + self.assertEqual(response.status_code, 200) + task_id = response.json()["id"] + self.assertEqual(response.json()["assistantId"], "asst_1") + SimulatedExecutor.release.set() + async with asyncio.timeout(2): + while (await service.get(task_id)).status != "succeeded": + await asyncio.sleep(.005) + task = await client.get(f"/api/background-tasks/{task_id}") + self.assertEqual(task.json()["status"], "succeeded") + self.assertNotIn("secrets", task.json()) + finally: + await service.close() diff --git a/docs-developer/prompt-background-tasks.mdx b/docs-developer/prompt-background-tasks.mdx new file mode 100644 index 0000000..c4b67b6 --- /dev/null +++ b/docs-developer/prompt-background-tasks.mdx @@ -0,0 +1,50 @@ +--- +title: 提示词后台任务 +description: 使用 Pipecat 原生 Jobs 委派远程 OpenCode 任务。 +--- + +# 提示词后台任务 MVP + +提示词 Pipeline 可以绑定一个「后台任务」工具。对话模型提交任务后立即获得任务编号，继续对话；应用级 Pipecat Worker 拥有长任务，语音断线不取消远端执行。OpenCode 是当前唯一实现的执行器。 + +## 使用 + +1. 更新后端依赖，并执行数据库迁移：`cd backend && .venv/bin/python -m alembic upgrade head`。迁移新增任务记录/事件表，并允许 task 工具类型。 +2. 重启后端。第一版要求单个应用进程（一个 Uvicorn worker），使用原生 AsyncQueueBus。 +3. 在远端准备好仓库、OpenCode 模型与服务。此功能连接已有 `opencode serve`，不会在远端安装软件或启动进程。服务应通过后端可访问的地址提供 HTTP(S)，如开启 Basic 鉴权，请配置 `OPENCODE_SERVER_PASSWORD`。 +4. 在「组件库 / 工具」新建「后台任务 · OpenCode」，填写服务地址、远端工作目录、用户名、服务密码和任务时限。服务地址/目录固定于配置，不由 LLM 参数决定。 +5. 在提示词 Pipeline 助手的「外部工具」中选择该工具。默认 instruction 参数由模型提供，可配置固定值或助手变量。任务工具不支持最终结果映射到会话变量。 +6. 保存助手，在手动语音调试中提出任务。展开「后台任务」查看进展、结果，处理远端问题/权限，或停止任务。网页刷新后仍可按助手查询任务。 + +前台新增三个控制函数：`get_background_task`、`cancel_background_task`、`send_background_task_message`。主提交函数使用工具资源的函数名。控制函数名由平台保留。单个远端问题可由对话工具回答；权限和多个问题请在网页卡片处理。不会自动批准权限。 + +## 运行与恢复 + +- NativeTaskRuntime 只组装 `WorkerRunner`、常驻 CoordinatorWorker 和 RemoteAgentWorker。派发、超时、取消和 FIFO 串行执行由 Pipecat 原生 Job 系统负责。 +- 任务时限包含原生队列等待时间；本版所有 coding 任务串行执行，避免同时修改同一仓库。 +- 提交回执表示平台接受任务。OpenCode 的 204 只表示远端接收；必须取得关联到本次 messageID 的完成消息才能显示成功。 +- OpenCodeAdapter 负责创建远端 session、订阅 SSE、异步提交、结果查询和 abort。每个任务使用独立远端 session，内部步骤仍由 OpenCode 完成。 +- 原生 Job 取消不代表远端已停止。适配器调用 abort 并查询远端状态；未确认时保留「停止待确认」。 +- 连接中断会尝试重新观察，不重新发送原指令。无法核查时显示「状态待核查」，可显式核查或请求停止。 +- 应用退出仅停止本地监听，不主动 abort 远端任务；重启后使用已保存的远端句柄重新观察。没有远端句柄的中断任务不会自动重新提交。 +- 任务和远端连接配置保存在数据库；公开 API 不返回密码或连接快照。资源后续编辑不会改变已运行任务的地址/目录。 +- 在线进度通过现有自定义 app message 传送，网页任务卡片另按助手轮询快照以支持刷新。完成/失败/等待输入通过 Pipecat 原生上下文消息触发前台模型回复；用户/助手正在说话时由原生聚合器安排。离线结果只保留在任务卡片，不自动在新语音连接中播报旧结果。 + +## 第一版边界 + +仅提示词 Pipeline 支持任务工具。Workflow、Realtime、启动 Action 不支持。自动测试/Responses 测试的模拟工具执行路径不会启动真实任务；请使用手动语音调试验收。运行中的任意实时 steer 尚未支持；需要修改目标时，先停止，再提交新的任务。 + +后端使用现有管理员鉴权；第一版沿用整个管理台的单管理员访问范围，尚无多租户隔离。持久化记录是业务历史和恢复依据，不是另一套任务调度器。 + +## 管理接口 + +- `GET /api/background-tasks?assistant_id=...`：最近100项任务。 +- `GET /api/background-tasks/{id}`：任务快照。 +- `POST /api/background-tasks`：提交已绑定工具的任务，携带 assistantId/toolId/instruction/requestId；重复 requestId 返回原任务。 +- `POST /api/background-tasks/{id}/cancel`：请求停止。 +- `POST /api/background-tasks/{id}/refresh`：核查远端并重新观察，不重发原任务。 +- `POST /api/background-tasks/{id}/input`：回复待处理的 permission/question。 + +## 验证边界 + +测试覆盖真实 Pipecat Jobs 的派发/进度/断线独立/取消，以及显式模拟的 OpenCode HTTP/SSE 协议。不把模拟执行结果当作真实代码修改结果。当前开发环境没有可连接的数据库或远端 OpenCode 服务，因此数据库迁移生成检查与真实服务联调应在部署环境执行。 diff --git a/frontend/src/components/assistant-editor/background-task-panel.tsx b/frontend/src/components/assistant-editor/background-task-panel.tsx index 461f0af..f9cbfe9 100644 --- a/frontend/src/components/assistant-editor/background-task-panel.tsx +++ b/frontend/src/components/assistant-editor/background-task-panel.tsx @@ -45,6 +45,7 @@ function TaskCard({ task, onChanged }: { task: BackgroundTask; onChanged: () => } {task.result &&
查看结果
{task.result}
} + {task.status === "unknown" && } {!terminal.has(task.status) && } {error &&

{error}

} ; diff --git a/frontend/src/components/tools/tool-binding-editor.tsx b/frontend/src/components/tools/tool-binding-editor.tsx index 0625f83..29aa59b 100644 --- a/frontend/src/components/tools/tool-binding-editor.tsx +++ b/frontend/src/components/tools/tool-binding-editor.tsx @@ -556,9 +556,10 @@ export function ToolBindingEditor({ () => tools.filter((tool) => tool.status === "active"), [tools], ); + const selectedTaskId = value.find(binding => tools.find(tool => tool.id === binding.toolId)?.type === "task")?.toolId; const pickerTools = useMemo( - () => (selectableTools ?? tools).filter((tool) => tool.status === "active" && (allowTask || tool.type !== "task")), - [selectableTools, tools, allowTask], + () => (selectableTools ?? tools).filter((tool) => tool.status === "active" && (allowTask || tool.type !== "task") && (tool.type !== "task" || !selectedTaskId || tool.id === selectedTaskId)), + [selectableTools, tools, allowTask, selectedTaskId], ); const selectedIds = new Set(value.map((binding) => binding.toolId)); const variableNames = [...Object.keys(variables), ...SYSTEM_VARIABLE_NAMES]; diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index ef89f7b..403bd29 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -1004,6 +1004,7 @@ export type BackgroundTask = { }; }; export const backgroundTasksApi = { + refresh: (id: string) => request(`/api/background-tasks/${encodeURIComponent(id)}/refresh`, { method: "POST" }), list: (assistantId: string) => request(`/api/background-tasks?assistant_id=${encodeURIComponent(assistantId)}`), cancel: (id: string) => request(`/api/background-tasks/${encodeURIComponent(id)}/cancel`, { method: "POST" }), input: (id: string, value: { reply?: string; answers?: string[][] }) => request(`/api/background-tasks/${encodeURIComponent(id)}/input`, { method: "POST", body: JSON.stringify(value) }), diff --git a/implement-stage/ASSUMPTIONS.md b/implement-stage/ASSUMPTIONS.md index 5b6c142..faaf587 100644 --- a/implement-stage/ASSUMPTIONS.md +++ b/implement-stage/ASSUMPTIONS.md @@ -12,9 +12,11 @@ | A-007 | 通知 | 进度仅网页；完成/失败在线时触发前台总结；离线结果保留供查询 | semantic | default | | A-008 | 恢复 | 单进程；重启恢复观察有远端句柄的任务，不自动重新发送未知提交 | semantic | default | | A-009 | 远端配置 | 地址/工作目录/Basic鉴权由管理员工具配置，任务冻结连接信息；LLM不能改地址和目录 | interface | default | -| A-010 | 补充输入 | 同一运行任务可发送文本；权限/问题交互显式等待管理员处理，不自动批准 | semantic | default | +| A-010 | 补充输入 | 本版仅支持回答远端问题及权限确认；任意实时修改要求需先停止再提交，不自动批准权限 | semantic | default | | A-011 | 验证边界 | 未提供真实远端服务，使用httpx MockTransport验证协议和真实Pipecat验证调度；真实联调待部署配置 | semantic | default | | A-012 | 外部编码变更 | 仅明确调用任务工具时会启动远端执行，不因查看页面自动执行 | semantic | default | - | A-013 | 任意运行中修改任务 | 本版支持远端待回答的问题/权限；任意实时 steer 延后，新增要求通过停止后重新提交 | semantic | default | | A-014 | 任务API字段 | 与现有API一致使用camelCase；执行器内部保持Python snake_case | interface | default | +| A-015 | 同一工作区并发修改 | 本版原生execute Job采用全局FIFO串行；排队也计入任务时限，避免同时改同一仓库 | semantic | default | +| A-016 | 状态未知后恢复 | 网页提供显式核查/重新观察；只重连已有远端句柄，绝不重复发送原任务 | semantic | default | +| A-017 | 自动测试边界 | task工具不出现在普通工具模拟列表；测试运行器中调用task只返回不支持，真实执行需手动语音调试 | semantic | default | diff --git a/implement-stage/BUILD_NOTE.md b/implement-stage/BUILD_NOTE.md index 54158b7..ed867c5 100644 --- a/implement-stage/BUILD_NOTE.md +++ b/implement-stage/BUILD_NOTE.md @@ -6,13 +6,17 @@ | F1 | OpenCode协议适配与状态关联 | cd backend && .venv/bin/python -m unittest tests.test_background_task_native tests.test_opencode_tasks | MUST | passed | | F2 | 持久记录、管理API、应用宿主 | cd backend && .venv/bin/python -m unittest tests.test_background_tasks | MUST | passed | | F3 | 提示词绑定、任务通知、网页 | cd frontend && npx tsc --noEmit | MUST | passed | -| F4 | 回归、文档与假设审阅 | make test | MUST | pending | +| F4 | 回归、文档与假设审阅 | make test | MUST | checks passed; sweep 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。 - F1: 原生 Job + httpx 模拟远端协议共 3 项测试通过；未调用真实模型。 - F2: 累计 7 项原生Job、协议、服务生命周期与密钥保留测试通过。 - F3: 累计9项后端任务测试通过，TypeScript检查通过，前端lint无错误（BatchTestPage已有1条warning）。 +- F4: `make test` → exit 0；360 项后端测试、4 项前端契约测试、5 项公开契约检查与生产构建通过。前端已有 BatchTestPage useEffect warning 1 条，无错误。 +- F4: 初次回归发现公开契约快照需更新；随后构建受沙箱 Google Fonts 网络限制，联网重试通过；未修改字体设计。 +- F4: 补充待回答问题的状态核查保留与原生超时清理检查；任务生命周期专项测试通过。假设审阅待执行。 +- 新增迁移区间 SQL：`cd backend && .venv/bin/python -m alembic upgrade 20260907_0025:head --sql` → exit 0。旧历史迁移含数据库读取，未声称全历史离线生成通过。 ## Deferred - 真实远程 OpenCode + 模型联调：未提供服务地址/凭证。 - 多进程分布式总线及通用插件发现：超出单进程 MVP。