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}{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。