diff --git a/backend/routes/background_tasks.py b/backend/routes/background_tasks.py index d2837b8..e2228ab 100644 --- a/backend/routes/background_tasks.py +++ b/backend/routes/background_tasks.py @@ -12,7 +12,6 @@ router = APIRouter(prefix="/api/background-tasks", tags=["background-tasks"], de class SubmitTask(CamelModel): assistant_id: str - tool_id: str instruction: str = Field(min_length=1, max_length=12000) request_id: str = Field(min_length=1, max_length=80) @@ -32,12 +31,15 @@ async def submit_task(body: SubmitTask, db=Depends(get_session)): cfg = await resolve_runtime_config(db, body.assistant_id) if cfg.type != "prompt" or cfg.runtimeMode != "pipeline": raise HTTPException(400, "后台任务仅支持提示词 Pipeline") - bound = {binding.tool_id for binding in cfg.tool_bindings} - tool = next((tool for tool in cfg.tools if tool.id == body.tool_id and tool.id in bound and tool.type == "task"), None) - if not tool: - raise HTTPException(400, "助手未绑定该任务工具") + if cfg.task_backend is None: + raise HTTPException(400, "助手未绑定任务后端") try: - return await task_service.submit(assistant_id=body.assistant_id, conversation_id="web_" + body.assistant_id, tool=tool, instruction=body.instruction, invocation_id=body.request_id) + return await task_service.submit( + assistant_id=body.assistant_id, conversation_id="web_" + body.assistant_id, + backend=cfg.task_backend, instruction=body.instruction, invocation_id=body.request_id, + variable_state={"values": cfg.dynamic_variables, "definitions": cfg.dynamic_variable_definitions, + "secrets": cfg.secret_dynamic_variables}, + ) except ValueError as exc: raise HTTPException(400, str(exc)) from exc diff --git a/backend/services/background_tasks/opencode.py b/backend/services/background_tasks/opencode.py index ee9be5d..1a2028b 100644 --- a/backend/services/background_tasks/opencode.py +++ b/backend/services/background_tasks/opencode.py @@ -76,9 +76,21 @@ class OpenCodeExecutor: async with client.stream("GET", "event") as stream: stream.raise_for_status() if not resume: + payload = {"messageID": handle["message_id"], "parts": [{"type": "text", "text": instruction}]} + if self.config.get("agent"): + payload["agent"] = self.config["agent"] + if self.config.get("model_provider_id") and self.config.get("model_id"): + payload["model"] = {"providerID": self.config["model_provider_id"], "modelID": self.config["model_id"]} + backend = self.config.get("_backend") or {} + from services.runtime_variables import DynamicVariableStore + store = DynamicVariableStore(self.config.get("_variables") or {}) + assistant_instructions = store.render(str(backend.get("instructions") or "")) + instructions = "\n\n".join(value for value in (self.config.get("instructions"), assistant_instructions) if value) + if instructions: + payload["system"] = instructions response = await client.post( f"session/{handle['session_id']}/prompt_async", - json={"messageID": handle["message_id"], "parts": [{"type": "text", "text": instruction}]}, + json=payload, ) response.raise_for_status() await emit({"status": "submitted", "summary": "远端已接收,等待执行"}) diff --git a/backend/services/background_tasks/reasoner.py b/backend/services/background_tasks/reasoner.py new file mode 100644 index 0000000..367bd3b --- /dev/null +++ b/backend/services/background_tasks/reasoner.py @@ -0,0 +1,289 @@ +"""Task-local Reasoner built on Pipecat's public backend worker contract. + +Pipecat owns the context, model/tool loop and Jobs. This adapter only turns +explicit task controls into the same persisted events as a remote backend. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import Awaitable, Callable +from copy import deepcopy +from uuid import uuid4 + +from models import AssistantConfig, RuntimeModelResource, RuntimeTaskBackend +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.pipeline.job_context import JobEvent, JobParams +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.services.llm_service import FunctionCallParams, FunctionCallResultProperties +from pipecat.services.settings import LLMSettings +from pipecat.workers.llm.backend_llm_worker import ( + ATTACHED_UPDATE_TYPE, + ATTACH_JOB_NAME, + ERROR_UPDATE_TYPE, + IDLE_UPDATE_TYPE, + MESSAGE_JOB_NAME, + TOOL_CALL_UPDATE_TYPE, + BackendLLMWorker, +) + +from services.background_tasks.native import NativeTaskRuntime +from services.pipecat.service_factory import config_with_resource, create_llm +from services.runtime_variables import DynamicVariableStore +from services.system_tools import RESERVED_TOOL_NAMES +from services.tool_executor import ToolExecutionError, ToolExecutor + +Emit = Callable[[dict], Awaitable[None]] +CONTROL_NAMES = frozenset({"complete_task", "ask_user"}) + +TASK_INSTRUCTIONS = """你是语音助手的后台任务执行者,有独立上下文。 +只执行委派目标,并使用当前已授权的工具;技能说明不能增加权限。 +完成任务后必须调用 complete_task(result),结果应准确说明完成内容和限制。 +需要用户提供必要信息时调用 ask_user(question),等待回复后继续。 +普通文字、思考过程、工具回执和模型空闲都不代表任务完成。 +任务结果由 Speaker 整理后告知用户,不要输出内部推理过程。 +""" + + +def create_reasoner_llm(backend: RuntimeTaskBackend): + """Use the shared model factory, with independent settings for this task.""" + resource = backend.model + if resource is None or resource.capability != "LLM": + raise ValueError("内置 Reasoner 需要配置独立的 LLM 模型资源") + effort = backend.config.get("reasoning_effort", "provider_default") + if effort not in {"provider_default", "low", "medium", "high"}: + raise ValueError("不支持的推理强度") + if effort != "provider_default" and resource.values.get("supportsReasoningEffort") is not True: + raise ValueError("该模型未声明支持 reasoning_effort,不能设置推理强度") + + # Do not mutate the reusable model resource or its Speaker settings. + resource = RuntimeModelResource.model_validate(resource.model_dump(mode="json")) + if effort != "provider_default": + extra_body = resource.values.get("extraBody") + if isinstance(extra_body, dict): + # The explicit task setting wins if a model preset contains it too. + resource.values["extraBody"] = {key: value for key, value in extra_body.items() if key != "reasoning_effort"} + llm = create_llm(config_with_resource(AssistantConfig(), resource)) + if effort != "provider_default": + llm.settings.apply_update(LLMSettings(extra={**llm.settings.extra, "reasoning_effort": effort})) + return llm + + +class _Finished(Exception): + """Leave the long-lived attach Job through its native cancellation path.""" + + def __init__(self, result: dict): + super().__init__("task finished") + self.result = result + + +class ReasonerExecutor: + def __init__(self, config: dict, secrets: dict, *, runtime: NativeTaskRuntime): + self.config = config + self.runtime = runtime + self.backend = RuntimeTaskBackend.model_validate(config["_backend"]) + self._worker: BackendLLMWorker | None = None + self._emit: Emit | None = None + self._pending: dict = {} + self._final: dict | None = None + self._confirmed: dict | None = None + self._registered: list[str] = [] + self._stopped = False + self._sending_input = False + self._cleanup_lock = asyncio.Lock() + self._tools = ToolExecutor(DynamicVariableStore.from_config(AssistantConfig( + dynamic_variables=deepcopy(config.get("_variables") or {}), + dynamic_variable_definitions=deepcopy(config.get("_variable_definitions") or {}), + secret_dynamic_variables=deepcopy(config.get("_secret_variables") or {}), + ))) + + async def prepare(self) -> dict: + return {"kind": "reasoner", "task_id": self.config["_task_id"]} + + def _system_prompt(self) -> str: + parts = [ + TASK_INSTRUCTIONS, + str(self.backend.config.get("instructions") or "").strip(), + self._tools.store.render(self.backend.instructions).strip(), + ] + for skill in self.backend.config.get("skills") or []: + parts.append(f"技能:{skill['name']}\n{skill['content']}") + return "\n\n".join(part for part in parts if part) + + def _build_worker(self) -> BackendLLMWorker: + bindings = {binding.tool_id: binding for binding in self.backend.tool_bindings} + names = set(CONTROL_NAMES | RESERVED_TOOL_NAMES | {"delegate_task"}) + schemas = [] + callbacks = [] + for tool in self.backend.tools: + if tool.type not in {"http", "mcp"}: + raise ValueError("Reasoner 仅允许 HTTP/MCP 服务端工具") + if tool.function_name in names: + raise ValueError("Reasoner 工具函数名重复或与任务控制函数冲突") + names.add(tool.function_name) + binding = bindings.get(tool.id) + if binding is None: + raise ValueError("Reasoner 工具必须显式绑定才可执行") + properties, required = self._tools.schema_parts(tool, binding) + schemas.append(FunctionSchema(name=tool.function_name, description=tool.description or tool.name, + properties=properties, required=required)) + + async def execute(params, current_tool=tool, current_binding=binding): + if self._pending or self._final or self._stopped: + await params.result_callback({"status": "error", "message": "任务已暂停或结束,不执行新工具"}, + properties=FunctionCallResultProperties(run_llm=False)) + return + try: + result = await self._tools.execute(current_tool, dict(params.arguments or {}), binding=current_binding) + except (ToolExecutionError, ValueError): + # Tool errors can carry URLs or sensitive payloads. Let the + # model report a useful failure without exposing those. + result = {"status": "error", "message": "服务端工具执行失败,请检查工具配置或稍后重试"} + await params.result_callback(result, properties=FunctionCallResultProperties( + run_llm=not (self._pending or self._final or self._stopped), + )) + + callbacks.append((tool.function_name, execute)) + + schemas.extend([ + FunctionSchema(name="complete_task", description="明确完成任务并提交最终结果。", + properties={"result": {"type": "string", "description": "完成结果、依据和未完成的限制。"}}, required=["result"]), + FunctionSchema(name="ask_user", description="需要必要信息时暂停任务,向用户提出一个问题。", + properties={"question": {"type": "string", "description": "需要用户回答的问题。"}}, required=["question"]), + ]) + llm = create_reasoner_llm(self.backend) + for name, callback in callbacks + [("complete_task", self._complete_task), ("ask_user", self._ask_user)]: + # Synchronous native tools drive BackendLLMWorker's existing loop. + llm.register_function(name, callback, cancel_on_interruption=True) + self._registered.append(name) + return BackendLLMWorker( + name="reasoner-" + uuid4().hex, + llm=llm, + context=LLMContext([{"role": "system", "content": self._system_prompt()}], tools=schemas), + pairing_instruction="任务由 Speaker 发送给你。使用任务控制函数提交结果或提问;普通输出不作为给用户的通知。", + ) + + async def _complete_task(self, params: FunctionCallParams): + result = str(params.arguments.get("result") or "").strip() + if not result or self._pending or self._stopped: + await params.result_callback( + {"status": "error", "message": "需要非空结果,且任务不能处于等待输入或停止状态"}, + properties=FunctionCallResultProperties(run_llm=not (self._pending or self._stopped)), + ) + return + self._final = {"status": "succeeded", "summary": "任务已完成", "result": result} + await params.result_callback({"status": "accepted"}, properties=FunctionCallResultProperties(run_llm=False)) + + async def _ask_user(self, params: FunctionCallParams): + question = str(params.arguments.get("question") or "").strip() + if not question or self._pending or self._final or self._stopped: + await params.result_callback( + {"status": "error", "message": "请在任务运行时提出一个非空问题"}, + properties=FunctionCallResultProperties(run_llm=not (self._pending or self._final or self._stopped)), + ) + return + self._pending = {"id": "question_" + uuid4().hex, "kind": "question", + "questions": [{"header": "补充信息", "question": question, "options": [], "custom": True}]} + await self._emit({"status": "waiting_input", "summary": "任务需要补充信息", "input_request": deepcopy(self._pending)}) + await params.result_callback({"status": "waiting_input"}, properties=FunctionCallResultProperties(run_llm=False)) + + async def _message(self, text: str): + if self._worker is None or self._stopped: + raise ValueError("内置 Reasoner 已停止,不能继续当前任务") + async with self.runtime.coordinator.job(self._worker.name, params=JobParams( + name=MESSAGE_JOB_NAME, payload={"request": text}, timeout=15, + )): + pass + + async def run(self, handle: dict, instruction: str, emit: Emit, *, resume=False) -> dict: + if resume: + return {"status": "failed", "summary": "内置任务执行曾中断,不会自动重放工具", "result": "请核查已有操作后重新创建任务。"} + if self.runtime.runner is None: + raise ValueError("后台任务运行器尚未启动") + self._emit = emit + self._worker = self._build_worker() + try: + await self.runtime.runner.add_workers(self._worker) + async with self.runtime.coordinator.job(self._worker.name, params=JobParams(name=ATTACH_JOB_NAME)) as attached: + async for event in attached: + if event.type != JobEvent.UPDATE: + continue + data = event.data or {} + kind = data.get("type") + if kind == ATTACHED_UPDATE_TYPE: + await emit({"status": "running", "summary": "内置 Reasoner 正在执行任务"}) + await self._message(instruction) + elif kind == TOOL_CALL_UPDATE_TYPE and data.get("function_name") not in CONTROL_NAMES: + if not self._pending: + await emit({"status": "running", "summary": f"服务端工具 · {data.get('function_name', '')}"[:300]}) + elif kind == ERROR_UPDATE_TYPE: + raise _Finished({"status": "failed", "summary": "内置推理服务执行失败", "result": "请检查模型连接和工具配置。"}) + elif kind == IDLE_UPDATE_TYPE: + # An old idle update can already be on the bus when a + # user reply starts the next native message Job. + if self._sending_input or self._worker.working: + continue + if self._final: + raise _Finished(self._final) + if not self._pending: + raise _Finished({"status": "failed", "summary": "模型没有明确完成任务", "result": "模型已停止生成,但没有调用 complete_task 或 ask_user;请调整任务指令或模型后重试。"}) + return {"status": "failed", "summary": "内置任务连接已中断", "result": "不会自动重放工具调用。"} + except _Finished as finished: + self._confirmed = finished.result + return finished.result + finally: + await self._cleanup() + + async def _cleanup(self): + async with self._cleanup_lock: + worker = self._worker + self._stopped = True + if worker is None: + return + # This is PipelineWorker.cancel(), which cancels only this pipeline. + # BaseWorker.cancel()/runner.cancel() would stop unrelated tasks too. + await worker.cancel(reason="task finished or cancelled") + await asyncio.wait_for(worker.wait(), 10) + # WorkerRunner retains ended worker registrations. Release transcript + # and handler captures through public APIs instead of editing Pipecat. + worker.context.set_messages([]) + worker.context.set_tools([]) + for name in self._registered: + worker.llm.unregister_function(name) + self._registered.clear() + self._worker = None + + async def cancel(self, handle: dict) -> bool: + if self._worker is not None: + await self._cleanup() + return True + + async def snapshot(self, handle: dict) -> dict: + if self._confirmed and self._worker is None: + return dict(self._confirmed) + if self._pending: + return {"status": "waiting_input", "summary": "任务需要补充信息", "input_request": deepcopy(self._pending)} + if self._stopped: + return {"status": "failed", "summary": "内置任务已中断", "result": "不会自动重放工具调用。"} + return {"status": "running", "summary": "内置 Reasoner 正在执行任务"} + + async def send_input(self, request: dict, value: dict): + if not self._pending or request.get("id") != self._pending["id"]: + raise ValueError("该问题已处理或不属于当前任务") + answers = value.get("answers") + if not isinstance(answers, list) or len(answers) != 1 or not isinstance(answers[0], list): + raise ValueError("请回答当前任务的问题") + text = "\n".join(str(answer).strip() for answer in answers[0]).strip() + if not text: + raise ValueError("回答不能为空") + previous = self._pending + self._pending = {} + self._sending_input = True + try: + await self._message("用户补充信息:\n" + text) + except Exception: + if not self._pending and not self._stopped: + self._pending = previous + raise + finally: + self._sending_input = False diff --git a/backend/services/background_tasks/records.py b/backend/services/background_tasks/records.py index bdfe52e..f967781 100644 --- a/backend/services/background_tasks/records.py +++ b/backend/services/background_tasks/records.py @@ -23,6 +23,9 @@ class TaskRecord: origin_key: str connection: dict secrets: dict + backend_id: str = "" + backend_type: str = "opencode" + deadline_at: str | None = None status: str = "queued" summary: str = "任务已排队" result: str = "" diff --git a/backend/services/background_tasks/service.py b/backend/services/background_tasks/service.py index 0615efe..b827989 100644 --- a/backend/services/background_tasks/service.py +++ b/backend/services/background_tasks/service.py @@ -1,8 +1,9 @@ -"""Business entry point backed by native Pipecat Jobs.""" +"""Task records and backend selection around application-owned Pipecat Jobs.""" import asyncio from collections import defaultdict from copy import deepcopy +from datetime import UTC, datetime, timedelta from uuid import uuid4 from loguru import logger @@ -12,18 +13,19 @@ from services.background_tasks.records import TaskRecord, TaskRepository, TERMIN class TaskService: - def __init__(self, repository=None, executor_factory=OpenCodeExecutor): + def __init__(self, repository=None, executor_factory=None): self.repository = repository or TaskRepository() self.executor_factory = executor_factory self.runtime = None self._listeners = defaultdict(set) self._jobs = {} + self._executors = {} 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) + self.runtime = NativeTaskRuntime(self._execute, self._cancel_backend, self._on_update, self._on_result) await self.runtime.start() try: records = await self.repository.list(active=True) @@ -31,10 +33,12 @@ class TaskService: await self.close() raise for record in records: - if record.remote_handle: + if record.backend_type == "reasoner": + await self.repository.patch(record.id, status="failed", summary="服务重启,内置任务已中断;未自动重放工具,请核查已执行操作") + elif record.remote_handle: await self._dispatch(record) else: - await self.repository.patch(record.id, status="unknown", summary="服务曾中断,未自动重新提交;请核查远端后再创建任务") + await self.repository.patch(record.id, status="unknown", summary="服务曾中断,未自动重新提交;请核查执行后端后再创建任务") async def close(self): self._closing = True @@ -43,8 +47,9 @@ class TaskService: self.runtime = None self._listeners.clear() self._jobs.clear() + self._executors.clear() - async def submit(self, *, assistant_id, conversation_id, instruction, invocation_id, backend=None, tool=None): + async def submit(self, *, assistant_id, conversation_id, backend, instruction, invocation_id, variable_state=None): if not self.runtime: raise ValueError("后台任务服务尚未启动") if not assistant_id or not conversation_id: @@ -52,32 +57,42 @@ class TaskService: text = instruction.strip() if not text or len(text) > 12000: raise ValueError("任务指令必须为 1-12000 个字符") - if backend is not None: - if backend.type != "opencode": - raise ValueError("该任务后端尚未接通") - backend_id = backend.id - connection = deepcopy(backend.config) - connection["_backend_type"] = backend.type - secrets = deepcopy(backend.secrets) - else: - # F0 transition: retired when saved Tool bindings migrate in F1. - if tool is None or tool.type != "task": - raise ValueError("请选择任务后端") - backend_id, connection, secrets = tool.id, deepcopy(tool.definition["config"]), deepcopy(tool.secrets) + if backend.type not in {"reasoner", "opencode"}: + raise ValueError("不支持该任务后端") + snapshot = backend.model_dump(mode="json") if hasattr(backend, "model_dump") else { + "id": backend.id, "name": getattr(backend, "name", ""), "type": backend.type, + "config": deepcopy(backend.config), "secrets": deepcopy(backend.secrets), + } + connection = deepcopy(backend.config) + connection["_backend_type"] = backend.type + connection["_backend"] = snapshot + state = variable_state or {} + connection["_variables"] = deepcopy(state.get("values", {})) + connection["_variable_definitions"] = deepcopy(state.get("definitions", {})) + connection["_secret_variables"] = deepcopy(state.get("secrets", {})) + timeout = int(connection.get("timeout_seconds", 600)) + deadline = (datetime.now(UTC) + timedelta(seconds=timeout)).isoformat() 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=backend_id, instruction=text, - origin_key=f"{conversation_id}:{invocation_id}"[:256], - connection=connection, secrets=secrets, + assistant_id=assistant_id, conversation_id=conversation_id, tool_id="", + backend_id=backend.id, backend_type=backend.type, deadline_at=deadline, + instruction=text, origin_key=f"{conversation_id}:{invocation_id}"[:256], + connection=connection, secrets=deepcopy(backend.secrets), )) if not record.job_id and record.status == "queued": await self._dispatch(record) - return record.public() + return (await self.get(record.id)).public() async def _dispatch(self, record): - job_id = await self.runtime.submit(record.id, record.connection.get("timeout_seconds", 600)) + timeout = float(record.connection.get("timeout_seconds", 600)) + if record.deadline_at: + timeout = (datetime.fromisoformat(record.deadline_at) - datetime.now(UTC)).total_seconds() + if timeout <= 0: + await self._save(record.id, status="cancel_requested", summary="任务已超过等待时限,正在核查停止状态") + await self._stop_record(await self.get(record.id)) + return + job_id = await self.runtime.submit(record.id, timeout) self._jobs[job_id] = record.id await self.repository.patch(record.id, job_id=job_id) @@ -107,23 +122,36 @@ class TaskService: await self._publish(record) return record + def _executor(self, record): + if record.id in self._executors: + return self._executors[record.id] + config = {**record.connection, "_task_id": record.id} + if self.executor_factory: + return self.executor_factory(config, record.secrets) + if record.backend_type == "reasoner": + from services.background_tasks.reasoner import ReasonerExecutor + return ReasonerExecutor(config, record.secrets, runtime=self.runtime) + return OpenCodeExecutor(config, record.secrets) + 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) + if record.status in TERMINAL: + return {"task_id": task_id} if record.status == "cancel_requested": await self._stop_record(record) return {"task_id": task_id} - executor = self.executor_factory(record.connection, record.secrets) + executor = self._executor(record) + self._executors[task_id] = executor resume = bool(record.remote_handle) try: if not resume: outstanding = await self.repository.list(active=True) - if any(other.id != task_id and other.remote_handle for other in outstanding): - await self.repository.patch(task_id, status="failed", summary="另一远端任务尚未结束,请先核查或停止该任务;本任务未提交到远端") + if any(other.id != task_id and other.backend_type == "opencode" and other.remote_handle for other in outstanding): + await self.repository.patch(task_id, status="failed", summary="另一远端任务尚未结束,请先核查或停止该任务;本任务未提交到后端") return {"task_id": task_id} - handle = await executor.prepare() - record = await self.repository.patch(task_id, remote_handle=handle) + record = await self.repository.patch(task_id, remote_handle=await executor.prepare()) last_event = None async def emit(event): nonlocal last_event @@ -132,8 +160,9 @@ class TaskService: 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): + # Only OpenCode can reconnect observations without replaying work. + attempts = 1 if record.backend_type == "reasoner" else 5 + for attempt in range(attempts): try: result = await executor.run(record.remote_handle, record.instruction, emit, resume=resume) await self.repository.patch(task_id, **result, input_request={}) @@ -141,32 +170,35 @@ class TaskService: except asyncio.CancelledError: raise except Exception: + if attempt == attempts - 1: + raise 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="无法确认远端执行状态,请检查远端服务;不会自动重复提交") + if record.backend_type == "reasoner": + await self.repository.patch(task_id, status="failed", summary="内置推理任务中断;未自动重复执行,请核查任务配置与已执行操作") + else: + 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) + await self._publish(await self.get(task_id)) async def _on_result(self, message): - if self._closing: - return task_id = self._jobs.pop(message.job_id, None) if not task_id: return + self._executors.pop(task_id, None) + if self._closing: + 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="任务运行中断,请核查远端状态") + await self._save(task_id, status="unknown", summary="任务运行中断,请核查执行后端状态") else: await self._publish(record) @@ -174,13 +206,13 @@ class TaskService: 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="正在请求远端停止") + 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): + async def _cancel_backend(self, message): task_id = self._jobs.get(message.job_id) if not task_id or self._closing: return @@ -190,38 +222,41 @@ class TaskService: if record.status in TERMINAL: return if not record.remote_handle: - await self._save(record.id, status="cancelled", summary="已取消,尚未向远端提交指令") + await self._save(record.id, status="cancelled", summary="已取消,尚未向任务后端提交指令") return try: - stopped = await self.executor_factory(record.connection, record.secrets).cancel(record.remote_handle) + stopped = await self._executor(record).cancel(record.remote_handle) except Exception: stopped = False - await self._save(record.id, status="cancelled" if stopped else "cancel_requested", summary="远端已停止" if stopped else "停止尚未确认,请检查远端服务") + summary = "已停止内置推理;已发出的工具请求结果请核查" if record.backend_type == "reasoner" else "远端已停止" + await self._save(record.id, status="cancelled" if stopped else "cancel_requested", summary=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. + snapshot = await self._executor(record).snapshot(record.remote_handle) if snapshot["status"] == "running" and record.input_request: - snapshot.update(status="waiting_input", summary="远端等待确认或补充信息") + 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)) + if record.backend_type == "reasoner": + await self._save(task_id, status="failed", summary="内置任务已中断,不自动重放工具") + else: + 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={}) + await self._executor(record).send_input(record.input_request, value) + latest = await self.get(task_id) + if latest.input_request.get("id") == record.input_request.get("id"): + await self._save(task_id, status="running", summary="已发送回复,等待任务后端继续", input_request={}) return (await self.get(task_id)).public() diff --git a/backend/services/brains/prompt_brain.py b/backend/services/brains/prompt_brain.py index c94cc7c..7ee0746 100644 --- a/backend/services/brains/prompt_brain.py +++ b/backend/services/brains/prompt_brain.py @@ -102,7 +102,7 @@ class PromptBrain(BaseBrain): def system_prompt(self, cfg: AssistantConfig) -> str: prompt = self._store.render(cfg.prompt) if self._dynamic_enabled else cfg.prompt - if cfg.runtimeMode == "pipeline" and any(tool.type == "task" for tool in cfg.tools): + if cfg.runtimeMode == "pipeline" and cfg.task_backend is not None: from services.brains.prompt_tasks import TASK_HINT prompt += "\n" + TASK_HINT return prompt @@ -158,20 +158,6 @@ class PromptBrain(BaseBrain): ): logger.warning(f"跳过客户端未上报的工具: {tool.function_name}") continue - if tool.type == "task": - if cfg.runtimeMode != "pipeline": - raise ValueError("后台任务工具仅支持提示词 Pipeline") - from services.brains.prompt_tasks import PromptTaskBridge - if self._task_bridge is not None: - raise ValueError("最多绑定一个后台任务工具") - self._task_bridge = PromptTaskBridge(cfg, runtime, self._tools) - for schema, handler in self._task_bridge.functions(tool, binding_by_id[tool.id]): - if schema.name in registered_names: - raise ValueError("后台任务函数名与已有工具冲突") - schemas.append(schema) - registered_names.add(schema.name) - runtime.llm.register_function(schema.name, handler, cancel_on_interruption=False) - continue if tool.type == "system": schema, handler = self._make_system_tool(tool, runtime) elif tool.type in {"http", "mcp", "client"}: @@ -195,6 +181,17 @@ class PromptBrain(BaseBrain): handler, cancel_on_interruption=policy.cancel_on_interruption, ) + if cfg.task_backend is not None: + if cfg.runtimeMode != "pipeline": + raise ValueError("任务后端仅支持提示词 Pipeline") + from services.brains.prompt_tasks import PromptTaskBridge + self._task_bridge = PromptTaskBridge(cfg, runtime, self._tools) + for schema, handler in self._task_bridge.functions(): + if schema.name in registered_names: + raise ValueError("任务委派函数名与已有工具冲突") + schemas.append(schema) + registered_names.add(schema.name) + runtime.llm.register_function(schema.name, handler, cancel_on_interruption=False) runtime.set_tools(schemas) async def close(self): diff --git a/backend/services/brains/prompt_tasks.py b/backend/services/brains/prompt_tasks.py index e031b52..b22d6d4 100644 --- a/backend/services/brains/prompt_tasks.py +++ b/backend/services/brains/prompt_tasks.py @@ -7,7 +7,8 @@ from services.background_tasks.records import TERMINAL CONTROL_NAMES = {"get_background_task", "cancel_background_task", "send_background_task_message"} TASK_HINT = """ -你可以把需要远程 coding agent 执行的工作交给后台任务工具。提交回执仅表示已接受, +你负责及时交流、澄清需求和播报进度;复杂推理或执行任务可调用 delegate_task 委派给任务后端。 +任务后端拥有独立上下文,请在任务要求中提供完整目标和必要背景。提交回执仅表示已接受, 不能据此宣称代码已经修改或测试通过。后台执行时继续回答用户;查询进度用任务查询工具。 用户明确要求停止时调用取消工具,普通插话不会停止任务。收到后台任务结果后用简洁口语 告知用户一次,依据实际结果描述成功或失败。任务状态未知时如实告知,不要重复提交。 @@ -24,19 +25,40 @@ class PromptTaskBridge: def close(self): self._detach() - def functions(self, tool, binding): - properties, required = self.executor.schema_parts(tool, binding) + def functions(self): + backend = self.cfg.task_backend + if backend is None: + raise ValueError("助手未绑定任务后端") + source = backend.instruction_source or {"source": "model"} + model_supplied = source.get("source") == "model" + properties = {"instruction": {"type": "string", "description": "完整任务目标、必要背景和验收要求"}} if model_supplied else {} + required = ["instruction"] if model_supplied else [] async def submit(params): if self.runtime.tool_executor_factory is not None: await params.result_callback({"status": "error", "message": "自动测试不启动真实后台任务,请使用语音调试验证"}) return try: - arguments = self.executor.resolve_arguments(tool, binding, params.arguments or {}) + kind = source.get("source") + if kind == "model": + instruction = (params.arguments or {}).get("instruction") + elif kind == "literal": + instruction = source.get("value") + elif kind == "variable": + instruction = self.executor.store.value(str(source.get("variable_name") or source.get("variableName") or "")) + elif kind == "template": + instruction = self.executor.store.render_data(source.get("value")) + else: + raise ValueError("不支持的任务指令来源") + if not isinstance(instruction, str): + raise ValueError("任务指令必须是字符串") result = await self.service.submit( assistant_id=self.cfg.dynamic_variables.get("system__agent_id", ""), conversation_id=self.cfg.conversation_id, - tool=tool, instruction=arguments["instruction"], invocation_id=params.tool_call_id, + backend=backend, instruction=instruction, invocation_id=params.tool_call_id, + variable_state={"values": self.executor.store.values, + "definitions": self.cfg.dynamic_variable_definitions, + "secrets": self.executor.store.secrets}, ) await params.result_callback({"status": "accepted", "task": result}) except ValueError as exc: @@ -64,10 +86,10 @@ class PromptTaskBridge: except ValueError as exc: await params.result_callback({"status": "error", "message": str(exc)}) - schemas = [(FunctionSchema(name=tool.function_name, description=tool.description or "把编程工作交给远程后台任务执行,立即返回任务编号", properties=properties, required=required), submit)] + schemas = [(FunctionSchema(name="delegate_task", description="委派复杂推理或执行任务给已配置的任务后端,立即返回任务编号", properties=properties, required=required), submit)] for name, description in ( ("get_background_task", "查询当前助手后台任务的真实进度和结果"), - ("cancel_background_task", "用户明确要求停止时,请求停止远程后台任务"), + ("cancel_background_task", "用户明确要求停止时,请求停止后台任务"), ("send_background_task_message", "回答后台任务当前提出的一个问题;权限确认请用户在网页操作"), ): props = {"task_id": {"type": "string", "description": "提交回执中的任务id"}} diff --git a/backend/services/tool_executor.py b/backend/services/tool_executor.py index cce80cd..02fa48a 100644 --- a/backend/services/tool_executor.py +++ b/backend/services/tool_executor.py @@ -43,8 +43,6 @@ class ToolExecutor: @staticmethod def input_schema(tool: RuntimeTool) -> tuple[dict[str, Any], list[str]]: config = (tool.definition or {}).get("config") or {} - if tool.type == "task": - return {"instruction": {"type": "string", "description": "需要远程执行的完整任务要求"}}, ["instruction"] if tool.type == "mcp": schema = config.get("input_schema") or {} return ( diff --git a/backend/tests/test_background_tasks.py b/backend/tests/test_background_tasks.py index 530773a..5f31df8 100644 --- a/backend/tests/test_background_tasks.py +++ b/backend/tests/test_background_tasks.py @@ -6,8 +6,6 @@ from types import SimpleNamespace from services.background_tasks.service import TaskService from services.background_tasks.records import TERMINAL -from schemas import ToolUpsert -from routes.tools import _payload class MemoryRepository: @@ -58,7 +56,7 @@ class TaskServiceTest(unittest.IsolatedAsyncioTestCase): 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"}) + self.backend = SimpleNamespace(type="opencode", id="backend_test", config={"url": "https://remote.test", "directory": "/demo", "timeout_seconds": 30}, secrets={"password": "never-public"}) async def asyncTearDown(self): await self.service.close() async def wait_status(self, task_id, expected): @@ -66,7 +64,7 @@ class TaskServiceTest(unittest.IsolatedAsyncioTestCase): while (await self.service.get(task_id)).status != expected: await asyncio.sleep(.005) async def submit(self, key="call_1"): - return await self.service.submit(assistant_id="asst_1", conversation_id="conv_1", tool=self.tool, instruction="执行测试", invocation_id=key) + return await self.service.submit(assistant_id="asst_1", conversation_id="conv_1", backend=self.backend, instruction="执行测试", invocation_id=key) async def test_idempotence_public_projection_and_completion(self): task = await self.submit() again = await self.submit() @@ -152,12 +150,6 @@ class TaskServiceTest(unittest.IsolatedAsyncioTestCase): 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 @@ -174,14 +166,14 @@ class TaskApiTest(unittest.IsolatedAsyncioTestCase): 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={})]) + cfg = SimpleNamespace(type="prompt", runtimeMode="pipeline", task_backend=SimpleNamespace(type="opencode", id="backend_test", config={"url": "https://remote.test", "directory": "/demo", "timeout_seconds": 30}, secrets={}), dynamic_variables={}, dynamic_variable_definitions={}, secret_dynamic_variables={}) 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"}) + response = await client.post("/api/background-tasks", json={"assistantId": "asst_1", "instruction": "测试请求", "requestId": "request_1"}) self.assertEqual(response.status_code, 200) task_id = response.json()["id"] self.assertEqual(response.json()["assistantId"], "asst_1") diff --git a/backend/tests/test_opencode_tasks.py b/backend/tests/test_opencode_tasks.py index b2ab340..5cc89d7 100644 --- a/backend/tests/test_opencode_tasks.py +++ b/backend/tests/test_opencode_tasks.py @@ -32,7 +32,11 @@ class OpenCodeTaskTest(unittest.IsolatedAsyncioTestCase): return httpx.Response(200, text="".join("data: " + json.dumps(e) + "\n\n" for e in events)) if request.url.path.endswith("prompt_async"): self.assertFalse(submitted) - self.assertEqual(json.loads(request.content)["messageID"], "msg_test") + payload = json.loads(request.content) + self.assertEqual(payload["messageID"], "msg_test") + self.assertEqual(payload["agent"], "build") + self.assertEqual(payload["model"], {"providerID": "test-provider", "modelID": "test-model"}) + self.assertEqual(payload["system"], "通用说明\n\n项目 demo") submitted = True return httpx.Response(204) if request.url.path.endswith("/message"): @@ -42,7 +46,10 @@ class OpenCodeTaskTest(unittest.IsolatedAsyncioTestCase): if request.url.path == "/session/status": return httpx.Response(200, json={}) raise AssertionError(request.url) - executor = OpenCodeExecutor({"url": "https://remote.test", "directory": "/workspace/demo"}, {}, transport=httpx.MockTransport(respond)) + executor = OpenCodeExecutor({"url": "https://remote.test", "directory": "/workspace/demo", "agent": "build", + "model_provider_id": "test-provider", "model_id": "test-model", + "instructions": "通用说明", "_backend": {"instructions": "项目 {{project}}"}, + "_variables": {"project": "demo"}}, {}, transport=httpx.MockTransport(respond)) async def emit(event): updates.append(event) result = await executor.run(handle, "修复测试", emit) diff --git a/backend/tests/test_prompt_background_tasks.py b/backend/tests/test_prompt_background_tasks.py index a0c8d78..90b23c1 100644 --- a/backend/tests/test_prompt_background_tasks.py +++ b/backend/tests/test_prompt_background_tasks.py @@ -1,22 +1,24 @@ import unittest from types import SimpleNamespace from unittest.mock import AsyncMock, Mock -from models import AssistantConfig, RuntimeTool, RuntimeToolBinding +from models import AssistantConfig, RuntimeTaskBackend from services.brains.prompt_tasks import PromptTaskBridge from services.tool_executor import ToolExecutor from services.runtime_variables import DynamicVariableStore class PromptTaskBridgeTest(unittest.IsolatedAsyncioTestCase): + def config(self, **kwargs): + return AssistantConfig(task_backend=RuntimeTaskBackend(id="backend_1", type="opencode"), **kwargs) + async def test_receipt_and_single_terminal_notification(self): - cfg = AssistantConfig(conversation_id="conv_1", dynamic_variables={"system__agent_id": "asst_1"}) + cfg = self.config(conversation_id="conv_1", dynamic_variables={"system__agent_id": "asst_1"}) runtime = SimpleNamespace(session_id="conv_1", transport=object(), tool_executor_factory=None, call_end=SimpleNamespace(ending=False), queue_frame=AsyncMock(), llm=SimpleNamespace(queue_frame=AsyncMock())) detach = Mock() service = SimpleNamespace(subscribe=Mock(return_value=detach), submit=AsyncMock(return_value={"id": "task_1", "status": "queued"})) bridge = PromptTaskBridge(cfg, runtime, ToolExecutor(DynamicVariableStore({})), service) - tool = RuntimeTool(id="tool_task", name="coding", function_name="coding_task", type="task", definition={"type": "task", "config": {}}) - binding = RuntimeToolBinding(toolId=tool.id, argumentBindings={"instruction": {"source": "model"}}) - functions = bridge.functions(tool, binding) + functions = bridge.functions() + self.assertEqual(functions[0][0].name, "delegate_task") callback = AsyncMock() await functions[0][1](SimpleNamespace(arguments={"instruction": "修复测试"}, tool_call_id="call_1", result_callback=callback)) self.assertEqual(callback.await_args.args[0]["status"], "accepted") @@ -31,22 +33,39 @@ class PromptTaskBridgeTest(unittest.IsolatedAsyncioTestCase): async def test_test_runtime_never_submits_a_real_task(self): runtime = SimpleNamespace(session_id="conv_1", transport=None, tool_executor_factory=object()) service = SimpleNamespace(submit=AsyncMock()) - bridge = PromptTaskBridge(AssistantConfig(), runtime, ToolExecutor(DynamicVariableStore({})), service) - tool = RuntimeTool(id="tool_task", name="coding", function_name="coding_task", type="task", definition={"type": "task", "config": {}}) + bridge = PromptTaskBridge(self.config(), runtime, ToolExecutor(DynamicVariableStore({})), service) callback = AsyncMock() - await bridge.functions(tool, None)[0][1](SimpleNamespace(result_callback=callback)) + await bridge.functions()[0][1](SimpleNamespace(result_callback=callback)) service.submit.assert_not_awaited() self.assertEqual(callback.await_args.args[0]["status"], "error") async def test_test_runtime_never_controls_an_existing_real_task(self): runtime = SimpleNamespace(session_id="conv_1", transport=None, tool_executor_factory=object()) service = SimpleNamespace(get=AsyncMock(), cancel=AsyncMock(), send_input=AsyncMock()) - bridge = PromptTaskBridge(AssistantConfig(), runtime, ToolExecutor(DynamicVariableStore({})), service) - tool = RuntimeTool(id="tool_task", name="coding", function_name="coding_task", type="task", definition={"type": "task", "config": {}}) - for schema, handler in bridge.functions(tool, None)[1:]: + bridge = PromptTaskBridge(self.config(), runtime, ToolExecutor(DynamicVariableStore({})), service) + for schema, handler in bridge.functions()[1:]: callback = AsyncMock() await handler(SimpleNamespace(function_name=schema.name, arguments={"task_id": "real_task", "message": "yes"}, result_callback=callback)) self.assertEqual(callback.await_args.args[0]["status"], "error") service.get.assert_not_awaited() service.cancel.assert_not_awaited() service.send_input.assert_not_awaited() + + async def test_bound_instruction_sources_and_snapshot_ignore_model_overrides(self): + cases = [({"source": "literal", "value": "固定任务"}, "固定任务"), + ({"source": "template", "value": "检查 {{project}}"}, "检查 demo"), + ({"source": "variable", "variable_name": "project"}, "demo")] + for source, expected in cases: + with self.subTest(source=source): + cfg = self.config(conversation_id="conv_1", dynamic_variables={"system__agent_id": "asst_1"}) + cfg.task_backend.instruction_source = source + runtime = SimpleNamespace(session_id="conv_1", transport=None, tool_executor_factory=None) + service = SimpleNamespace(submit=AsyncMock(return_value={"id": "task_1"})) + store = DynamicVariableStore({"project": "demo"}) + bridge = PromptTaskBridge(cfg, runtime, ToolExecutor(store), service) + schema, handler = bridge.functions()[0] + self.assertEqual(schema.to_default_dict()["parameters"]["properties"], {}) + callback = AsyncMock() + await handler(SimpleNamespace(arguments={"instruction": "unauthorized override"}, tool_call_id="call_1", result_callback=callback)) + self.assertEqual(service.submit.await_args.kwargs["instruction"], expected) + self.assertEqual(service.submit.await_args.kwargs["variable_state"]["values"], store.values) diff --git a/backend/tests/test_reasoner_backend.py b/backend/tests/test_reasoner_backend.py new file mode 100644 index 0000000..c6d55d4 --- /dev/null +++ b/backend/tests/test_reasoner_backend.py @@ -0,0 +1,207 @@ +"""Real native backend workers, with model output explicitly simulated. + +No test in this module contacts a model provider or executes a remote task. +""" + +from __future__ import annotations + +import asyncio +from copy import deepcopy +import unittest +from unittest.mock import AsyncMock, patch + +from models import RuntimeTaskBackend +from pipecat.frames.frames import ( + FunctionCallFromLLM, + LLMContextFrame, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, + LLMTextFrame, +) +from pipecat.services.llm_service import LLMService +from pipecat.services.settings import LLMSettings +from services.background_tasks.native import NativeTaskRuntime +from services.background_tasks.reasoner import ReasonerExecutor, create_reasoner_llm + + +def backend_config(**overrides): + backend = { + "id": "backend_reasoner", "type": "reasoner", "name": "Reasoner", + "config": {"reasoning_effort": "provider_default", "skills": [{"name": "核查", "content": "先检查依据,再给结论。"}]}, + "model": {"id": "model_reasoner", "capability": "LLM", "interface_type": "openai-llm", + "values": {"modelId": "simulated", "apiUrl": "https://model.example.invalid/v1"}, + "secrets": {"apiKey": "simulated-key"}}, + **overrides, + } + return {"_backend": backend, "_task_id": "task_reasoner"} + + +class SimulatedLLM(LLMService): + """Feeds declared outputs into Pipecat's actual aggregation/tool loop.""" + + def __init__(self, steps): + super().__init__(settings=LLMSettings( + model="simulated", temperature=None, max_tokens=None, top_p=None, top_k=None, + frequency_penalty=None, presence_penalty=None, seed=None, + filter_incomplete_user_turns=False, user_turn_completion_config=None, + )) + self.steps = iter(steps) + self.contexts = [] + + async def process_frame(self, frame, direction): + await super().process_frame(frame, direction) + if not isinstance(frame, LLMContextFrame): + await self.push_frame(frame, direction) + return + self.contexts.append(deepcopy(frame.context.messages)) + step = next(self.steps, "simulated output without explicit completion") + await self.push_frame(LLMFullResponseStartFrame()) + if isinstance(step, str): + await self.push_frame(LLMTextFrame(step)) + else: + await self.run_function_calls([ + FunctionCallFromLLM(function_name=name, tool_call_id=f"call_{len(self.contexts)}_{index}", arguments=arguments, context=frame.context) + for index, (name, arguments) in enumerate(step) + ]) + await self.push_frame(LLMFullResponseEndFrame()) + + +class ReasonerSettingsTest(unittest.TestCase): + def test_nondefault_effort_requires_explicit_model_support(self): + config = backend_config() + config["_backend"]["config"]["reasoning_effort"] = "high" + with self.assertRaisesRegex(ValueError, "未声明"): + create_reasoner_llm(RuntimeTaskBackend.model_validate(config["_backend"])) + + def test_effort_is_sent_as_real_parameter_without_changing_shared_model(self): + config = backend_config() + config["_backend"]["config"]["reasoning_effort"] = "high" + config["_backend"]["model"]["values"].update( + supportsReasoningEffort=True, extraBody={"reasoning_effort": "low", "provider_option": True}, + ) + backend = RuntimeTaskBackend.model_validate(config["_backend"]) + service = create_reasoner_llm(backend) + params = service.build_chat_completion_params({"messages": []}) + self.assertEqual(params["reasoning_effort"], "high") + self.assertEqual(params["extra_body"], {"provider_option": True}) + self.assertEqual(backend.model.values["extraBody"]["reasoning_effort"], "low") + + +class ReasonerNativeTest(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.events = [] + self.results = [] + self.finished = asyncio.Event() + self.executor = None + + async def execute(message, worker): + handle = await self.executor.prepare() + result = await self.executor.run(handle, "完成本次测试目标", self.emit) + self.results.append(result) + return result + + async def cancel(message): + await self.executor.cancel(await self.executor.prepare()) + + async def on_result(message): + if message.source == self.runtime.executor.name: + self.finished.set() + + self.runtime = NativeTaskRuntime(execute, cancel, self.ignore, on_result) + await self.runtime.start() + + async def asyncTearDown(self): + await self.runtime.close() + + async def ignore(self, message): + pass + + async def emit(self, event): + self.events.append(event) + + async def start_reasoner(self, steps, config=None): + self.model = SimulatedLLM(steps) + self.executor = ReasonerExecutor(config or backend_config(), {}, runtime=self.runtime) + patcher = patch("services.background_tasks.reasoner.create_reasoner_llm", return_value=self.model) + patcher.start() + self.addCleanup(patcher.stop) + return await self.runtime.submit("task_reasoner", 10) + + async def wait_status(self, status): + async def wait(): + while not any(event.get("status") == status for event in self.events): + await asyncio.sleep(0.01) + await asyncio.wait_for(wait(), 5) + + async def test_explicit_completion_uses_actual_native_backend_and_clears_context(self): + await self.start_reasoner([[('complete_task', {"result": "显式模拟结果"})]]) + await asyncio.wait_for(self.finished.wait(), 5) + self.assertEqual(self.results[0]["status"], "succeeded") + self.assertEqual(self.results[0]["result"], "显式模拟结果") + self.assertIn("核查", self.model.contexts[0][0]["content"]) + self.assertIsNone(self.executor._worker) + self.assertTrue(self.runtime._run and not self.runtime._run.done()) + self.assertFalse(any("显式模拟结果" in event.get("summary", "") for event in self.events)) + + async def test_plain_output_idle_does_not_claim_success_or_expose_reasoning(self): + await self.start_reasoner(["simulated private reasoning text"]) + await asyncio.wait_for(self.finished.wait(), 5) + self.assertEqual(self.results[0]["status"], "failed") + self.assertIn("complete_task", self.results[0]["result"]) + self.assertNotIn("private reasoning", str(self.events)) + + async def test_question_reply_continues_same_native_context(self): + await self.start_reasoner([ + [('ask_user', {"question": "需要哪一天的数据?"})], + [('complete_task', {"result": "已按用户回答完成"})], + ]) + await self.wait_status("waiting_input") + self.assertFalse(self.finished.is_set()) + request = next(event["input_request"] for event in self.events if event.get("status") == "waiting_input") + with self.assertRaisesRegex(ValueError, "已处理"): + await self.executor.send_input({**request, "id": "other_question"}, {"answers": [["周一"]]}) + await self.executor.send_input(request, {"answers": [["周一"]]}) + await asyncio.wait_for(self.finished.wait(), 5) + self.assertEqual(self.results[0]["status"], "succeeded") + self.assertEqual(len(self.model.contexts), 2) + self.assertIn("周一", str(self.model.contexts[1])) + + async def test_cancellation_stops_only_the_task_pipeline(self): + job_id = await self.start_reasoner([[('ask_user', {"question": "等待测试回复"})]]) + await self.wait_status("waiting_input") + await self.runtime.cancel(job_id) + await asyncio.wait_for(self.finished.wait(), 5) + self.assertIsNone(self.executor._worker) + self.assertTrue(self.runtime._run and not self.runtime._run.done()) + + async def test_http_tool_uses_binding_and_task_local_variable_store(self): + config = backend_config(tools=[{ + "id": "tool_lookup", "name": "查询", "function_name": "lookup", "type": "http", + "definition": {"type": "http", "config": {"url": "https://tool.example.invalid", "parameters": [{"name": "customer", "required": True}]}}, + }], tool_bindings=[{"tool_id": "tool_lookup", "argument_bindings": {"customer": {"source": "variable", "variable_name": "customer"}}, + "result_assignments": {"answer": "answer"}}]) + variables = {"customer": "local-customer", "answer": "before"} + config["_variables"] = variables + config["_variable_definitions"] = {name: {"type": "string"} for name in variables} + + # Isolate the HTTP side effect while retaining real argument/result handling. + with patch("services.tool_executor.ToolExecutor._execute_http", AsyncMock(return_value={"status": "ok", "data": {"answer": "after"}})) as http: + await self.start_reasoner([[('lookup', {})], [('complete_task', {"result": "查询完成"})]], config) + await asyncio.wait_for(self.finished.wait(), 5) + self.assertEqual(self.results[0]["status"], "succeeded") + self.assertEqual(http.await_args.args[1], {"customer": "local-customer"}) + self.assertEqual(self.executor._tools.store.values["answer"], "after") + self.assertEqual(variables["answer"], "before") + + async def test_client_tools_and_control_name_collisions_are_rejected_before_model_start(self): + for kind, name in [("client", "browser_action"), ("http", "complete_task")]: + config = backend_config(tools=[{"id": "bad", "name": "bad", "type": kind, "function_name": name}]) + executor = ReasonerExecutor(config, {}, runtime=self.runtime) + with patch("services.background_tasks.reasoner.create_reasoner_llm") as factory: + with self.assertRaises(ValueError): + executor._build_worker() + factory.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/backend/tests/test_task_backend_spine.py b/backend/tests/test_task_backend_spine.py index 4f37982..73b2b0e 100644 --- a/backend/tests/test_task_backend_spine.py +++ b/backend/tests/test_task_backend_spine.py @@ -4,9 +4,12 @@ import json from pathlib import Path from types import SimpleNamespace import unittest +from unittest.mock import patch +from models import RuntimeTaskBackend from services.background_tasks.service import TaskService from tests.test_background_tasks import MemoryRepository, SimulatedExecutor +from tests.test_reasoner_backend import SimulatedLLM, backend_config class TaskBackendSpineTest(unittest.IsolatedAsyncioTestCase): @@ -29,3 +32,89 @@ class TaskBackendSpineTest(unittest.IsolatedAsyncioTestCase): artifact.write_text(json.dumps({"simulated_remote": True, "native_job": "completed", "backend_id": backend.id, "tool_definition_required": False}) + "\n") finally: await service.close() + + +class ReasonerServiceTest(unittest.IsolatedAsyncioTestCase): + """Full business lifecycle with a real native worker and simulated model.""" + + async def asyncSetUp(self): + self.service = TaskService(MemoryRepository()) + await self.service.start() + self.backend = RuntimeTaskBackend.model_validate(backend_config()["_backend"]) + + async def asyncTearDown(self): + await self.service.close() + + async def wait_task(self, task_id, predicate): + async with asyncio.timeout(5): + while True: + record = await self.service.get(task_id) + if predicate(record): + return record + await asyncio.sleep(.005) + + async def submit(self, *, variable_state=None): + return await self.service.submit( + assistant_id="asst_reasoner", conversation_id="conv_reasoner", backend=self.backend, + instruction="显式模拟任务", invocation_id="call_reasoner", variable_state=variable_state, + ) + + async def test_snapshot_is_frozen_and_business_projection_has_no_secrets(self): + model = SimulatedLLM([[('complete_task', {"result": "模拟任务完成"})]]) + self.backend.instructions = "为 {{customer}} 核查结果" + self.backend.config["instructions"] = "字面模板 {{customer}}" + state = {"values": {"customer": "before"}, "definitions": {"customer": {"type": "string"}}, "secrets": {"secret__token": "private-token"}} + with patch("services.background_tasks.reasoner.create_reasoner_llm", return_value=model): + task = await self.submit(variable_state=state) + self.backend.config["skills"][0]["content"] = "changed after submission" + state["values"]["customer"] = "changed after submission" + record = await self.wait_task(task["id"], lambda item: item.status == "succeeded") + self.assertEqual(record.backend_type, "reasoner") + self.assertEqual(record.connection["_variables"]["customer"], "before") + self.assertEqual(record.connection["_secret_variables"]["secret__token"], "private-token") + self.assertNotIn("changed after submission", str(model.contexts)) + self.assertIn("为 before 核查结果", model.contexts[0][0]["content"]) + self.assertIn("字面模板 {{customer}}", model.contexts[0][0]["content"]) + self.assertNotIn("private-token", str(record.public())) + self.assertNotIn("simulated-key", str(record.public())) + self.assertNotIn("_backend", record.public()) + + async def test_immediate_next_question_survives_reply_projection_update(self): + model = SimulatedLLM([ + [('ask_user', {"question": "选择日期?"})], + [('ask_user', {"question": "选择范围?"})], + [('complete_task', {"result": "两个问题均已回答"})], + ]) + with patch("services.background_tasks.reasoner.create_reasoner_llm", return_value=model): + task = await self.submit() + first = await self.wait_task(task["id"], lambda item: item.status == "waiting_input") + await self.service.send_input(task["id"], {"answers": [["周一"]]}) + second = await self.wait_task(task["id"], lambda item: item.status == "waiting_input" and item.input_request.get("id") != first.input_request["id"]) + self.assertEqual(second.input_request["questions"][0]["question"], "选择范围?") + await self.service.send_input(task["id"], {"answers": [["本部门"]]}) + finished = await self.wait_task(task["id"], lambda item: item.status == "succeeded") + self.assertEqual(finished.result, "两个问题均已回答") + self.assertFalse(finished.input_request) + + async def test_shutdown_restart_does_not_replay_reasoner(self): + model = SimulatedLLM([[('ask_user', {"question": "需要补充信息"})]]) + with patch("services.background_tasks.reasoner.create_reasoner_llm", return_value=model) as factory: + task = await self.submit() + await self.wait_task(task["id"], lambda item: item.status == "waiting_input") + await self.service.close() + await self.service.start() + record = await self.wait_task(task["id"], lambda item: item.status == "failed") + self.assertIn("未自动重放工具", record.summary) + self.assertEqual(factory.call_count, 1) + + async def test_business_cancel_cleans_only_current_reasoner(self): + model = SimulatedLLM([[('ask_user', {"question": "需要补充信息"})]]) + with patch("services.background_tasks.reasoner.create_reasoner_llm", return_value=model): + task = await self.submit() + await self.wait_task(task["id"], lambda item: item.status == "waiting_input") + executor = self.service._executors[task["id"]] + await self.service.cancel(task["id"]) + record = await self.wait_task(task["id"], lambda item: item.status == "cancelled") + self.assertIsNone(executor._worker) + self.assertIn("已发出的工具请求", record.summary) + self.assertTrue(self.service.runtime._run and not self.service.runtime._run.done()) diff --git a/implement-stage/task-backends/BUILD_NOTE.md b/implement-stage/task-backends/BUILD_NOTE.md index 371bd9f..985e445 100644 --- a/implement-stage/task-backends/BUILD_NOTE.md +++ b/implement-stage/task-backends/BUILD_NOTE.md @@ -4,13 +4,15 @@ |------|---------|--------------------|------|--------| | F0 | 新后端输入→原生Job→记录主线 | cd backend && .venv/bin/python -m unittest tests.test_task_backend_spine | MUST | passed | | F1 | 后端资源、助手配置、旧数据迁移 | cd backend && .venv/bin/python -m unittest tests.test_task_backend_resources | MUST | passed | -| F2 | 原生Reasoner、OpenCode、委派桥 | cd backend && .venv/bin/python -m unittest tests.test_reasoner_backend tests.test_prompt_background_tasks tests.test_background_tasks | MUST | pending | -| F3 | 任务后端资源页与提示词编辑器 | cd frontend && npx tsc --noEmit | MUST | pending | +| F2 | 原生Reasoner、OpenCode、委派桥 | cd backend && .venv/bin/python -m unittest tests.test_reasoner_backend tests.test_prompt_background_tasks tests.test_background_tasks | MUST | passed | +| F3 | 任务后端资源页与提示词编辑器 | cd frontend && npx tsc --noEmit | MUST | passed | | F4 | 全量回归、隔离数据库迁移与审阅 | make test | MUST | pending | ## Run record - F0:新后端资源输入 + 原生Jobs + 旧生命周期累计测试14项,exit 0。显式模拟执行器;backend-spine-smoke.json 已生成。 - F1:12项资源专项测试通过,连同原生主线、桥接、生命周期累计33项通过,exit 0;真实隔离 PostgreSQL 17 完成0026旧数据→0027迁移及CRUD/助手复制/运行时解析,见database-smoke.json。 +- F2:make test-backend,391项测试通过,exit 0;包含真实 Pipecat BackendLLMWorker 与模拟模型的13项Reasoner/服务主线检查、绑定指令来源、OpenCode固定agent/model/system负载及现有回归。 +- F3:TypeScript、lint及4项前端契约测试通过,exit 0;lint保留既有BatchTestPage警告。生产构建并入F4。 ## Deferred - 真实模型/OpenCode联调:尚无服务地址与凭据。