diff --git a/backend/app.py b/backend/app.py index 9764444..529de29 100644 --- a/backend/app.py +++ b/backend/app.py @@ -49,6 +49,7 @@ from starlette.exceptions import HTTPException as StarletteHTTPException from routes import ( assistants, + background_tasks, auth, conversations, health, @@ -80,6 +81,8 @@ async def lifespan(_app: FastAPI): await recover_interrupted_webhooks() await recover_interrupted_test_runs() await recover_interrupted_responses() + from services.background_tasks.service import task_service + await task_service.start() analysis_worker = asyncio.create_task( run_analysis_worker(), name="post-call-analysis-worker" ) @@ -89,6 +92,7 @@ async def lifespan(_app: FastAPI): try: yield finally: + await task_service.close() analysis_worker.cancel() webhook_worker.cancel() await asyncio.gather( @@ -193,6 +197,7 @@ app.include_router(test_cases.router) app.include_router(test_assets.router) app.include_router(test_runs.router) app.include_router(tools.router) +app.include_router(background_tasks.router) app.include_router(voice_webrtc.router) app.include_router(voice_ws.router) diff --git a/backend/db/models.py b/backend/db/models.py index 3a54378..326bc1f 100644 --- a/backend/db/models.py +++ b/backend/db/models.py @@ -248,7 +248,7 @@ class Tool(Base): name="uq_tools_mcp_server_remote_name", ), CheckConstraint( - "type IN ('http', 'mcp', 'client')", + "type IN ('http', 'mcp', 'client', 'task')", name="ck_tools_external_type", ), ) @@ -647,3 +647,34 @@ class TestRunCase(Base): finished_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True ) + + +class BackgroundTask(Base): + """A durable view of a Pipecat Job and its external execution handle.""" + __tablename__ = "background_tasks" + id: Mapped[str] = mapped_column(String(40), primary_key=True) + assistant_id: Mapped[str] = mapped_column(String(40), index=True) + conversation_id: Mapped[str] = mapped_column(String(80), index=True) + tool_id: Mapped[str] = mapped_column(String(40)) + instruction: Mapped[str] = mapped_column(Text) + origin_key: Mapped[str] = mapped_column(String(256), unique=True) + connection: Mapped[dict] = mapped_column(JSONB) + secrets: Mapped[dict] = mapped_column(JSONB) + status: Mapped[str] = mapped_column(String(32), index=True) + summary: Mapped[str] = mapped_column(Text) + result: Mapped[str] = mapped_column(Text) + remote_handle: Mapped[dict] = mapped_column(JSONB) + input_request: Mapped[dict] = mapped_column(JSONB) + job_id: Mapped[str] = mapped_column(String(80)) + sequence: Mapped[int] = mapped_column(Integer) + created_at: Mapped[str] = mapped_column(String(40)) + updated_at: Mapped[str] = mapped_column(String(40)) + + +class BackgroundTaskEvent(Base): + __tablename__ = "background_task_events" + __table_args__ = (UniqueConstraint("task_id", "sequence"),) + id: Mapped[str] = mapped_column(String(40), primary_key=True) + task_id: Mapped[str] = mapped_column(String(40), ForeignKey("background_tasks.id", ondelete="CASCADE"), index=True) + sequence: Mapped[int] = mapped_column(Integer) + data: Mapped[dict] = mapped_column(JSONB) diff --git a/backend/migrations/versions/20261023_0026_add_background_tasks.py b/backend/migrations/versions/20261023_0026_add_background_tasks.py new file mode 100644 index 0000000..29343bf --- /dev/null +++ b/backend/migrations/versions/20261023_0026_add_background_tasks.py @@ -0,0 +1,49 @@ +"""Add durable projections for native Pipecat background Jobs.""" +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +revision = "20261023_0026" +down_revision = "20260907_0025" +branch_labels = None +depends_on = None + + +def upgrade(): + op.drop_constraint("ck_tools_external_type", "tools", type_="check") + op.create_check_constraint("ck_tools_external_type", "tools", "type IN ('http', 'mcp', 'client', 'task')") + op.create_table( + "background_tasks", + sa.Column("id", sa.String(40), primary_key=True), + sa.Column("assistant_id", sa.String(40), nullable=False), + sa.Column("conversation_id", sa.String(80), nullable=False), + sa.Column("tool_id", sa.String(40), nullable=False), + sa.Column("instruction", sa.Text(), nullable=False), + sa.Column("origin_key", sa.String(256), nullable=False, unique=True), + *[sa.Column(name, postgresql.JSONB(), nullable=False) for name in ("connection", "secrets", "remote_handle", "input_request")], + sa.Column("status", sa.String(32), nullable=False), + sa.Column("summary", sa.Text(), nullable=False), + sa.Column("result", sa.Text(), nullable=False), + sa.Column("job_id", sa.String(80), nullable=False), + sa.Column("sequence", sa.Integer(), nullable=False), + sa.Column("created_at", sa.String(40), nullable=False), + sa.Column("updated_at", sa.String(40), nullable=False), + ) + for name in ("assistant_id", "conversation_id", "status"): + op.create_index(f"ix_background_tasks_{name}", "background_tasks", [name]) + op.create_table( + "background_task_events", + sa.Column("id", sa.String(40), primary_key=True), + sa.Column("task_id", sa.String(40), sa.ForeignKey("background_tasks.id", ondelete="CASCADE"), nullable=False), + sa.Column("sequence", sa.Integer(), nullable=False), + sa.Column("data", postgresql.JSONB(), nullable=False), + sa.UniqueConstraint("task_id", "sequence"), + ) + op.create_index("ix_background_task_events_task_id", "background_task_events", ["task_id"]) + + +def downgrade(): + op.drop_constraint("ck_tools_external_type", "tools", type_="check") + op.create_check_constraint("ck_tools_external_type", "tools", "type IN ('http', 'mcp', 'client')") + op.drop_table("background_task_events") + op.drop_table("background_tasks") diff --git a/backend/routes/assistants.py b/backend/routes/assistants.py index 0fbf8cb..9105c0e 100644 --- a/backend/routes/assistants.py +++ b/backend/routes/assistants.py @@ -242,6 +242,11 @@ async def _validate_tool_bindings( missing = sorted(tool_ids - set(tools)) if missing: raise HTTPException(400, "工具不存在或未启用: " + ", ".join(missing)) + task_tools = [tool for tool in tools.values() if tool.type == "task"] + if task_tools and (body.type != "prompt" or body.runtime_mode != "pipeline"): + raise HTTPException(400, "后台任务工具仅支持提示词 Pipeline") + if len(task_tools) > 1: + raise HTTPException(400, "每个助手最多绑定一个后台任务工具") reserved = sorted( tool.function_name for tool in tools.values() @@ -282,6 +287,8 @@ async def _validate_tool_bindings( type=tool.type, definition=tool.definition or {}, ) + if tool.type == "task" and binding.result_assignments: + raise HTTPException(400, "后台任务结果通过任务通知返回，不支持结果变量映射") properties, required = ToolExecutor.input_schema(runtime_tool) unknown = sorted(set(binding.argument_bindings) - set(properties)) if unknown: diff --git a/backend/routes/background_tasks.py b/backend/routes/background_tasks.py new file mode 100644 index 0000000..0b50a31 --- /dev/null +++ b/backend/routes/background_tasks.py @@ -0,0 +1,68 @@ +"""Administrator task projections and controls, backed by native Jobs.""" +from fastapi import APIRouter, Depends, HTTPException +from pydantic import Field +from schemas import CamelModel +from db.session import get_session +from services.auth import require_admin +from services.background_tasks.service import task_service +from services.config_resolver import resolve_runtime_config + +router = APIRouter(prefix="/api/background-tasks", tags=["background-tasks"], dependencies=[Depends(require_admin)]) + + +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) + + +class TaskInput(CamelModel): + reply: str | None = None + answers: list[list[str]] | None = None + + +@router.get("") +async def list_tasks(assistant_id: str): + return [record.public() for record in await task_service.repository.list(assistant_id)] + + +@router.post("") +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, "助手未绑定该任务工具") + 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) + except ValueError as exc: + raise HTTPException(400, str(exc)) from exc + + +@router.get("/{task_id}") +async def get_task(task_id: str): + try: + return (await task_service.get(task_id)).public() + except ValueError as exc: + raise HTTPException(404, str(exc)) from exc + + +@router.post("/{task_id}/cancel") +async def cancel_task(task_id: str): + try: + return await task_service.cancel(task_id) + except ValueError as exc: + raise HTTPException(404, str(exc)) from exc + + +@router.post("/{task_id}/input") +async def send_input(task_id: str, body: TaskInput): + try: + return await task_service.send_input(task_id, body.model_dump(exclude_none=True)) + except ValueError as exc: + raise HTTPException(400, str(exc)) from exc + except Exception as exc: + raise HTTPException(502, "远端暂未接受回复，请稍后重试") from exc diff --git a/backend/routes/tools.py b/backend/routes/tools.py index 4fce771..19cb628 100644 --- a/backend/routes/tools.py +++ b/backend/routes/tools.py @@ -100,7 +100,7 @@ def _payload( _normalize_http_template_inputs(definition) secrets = ( merge_secrets(body.secrets, stored_secrets or {}) - if definition["type"] == "http" + if definition["type"] in {"http", "task"} else {} ) secret_variables = (secrets.get("dynamic_variables") or {}) if secrets else {} @@ -154,7 +154,7 @@ async def list_tools(session: AsyncSession = Depends(get_session)): rows = ( await session.execute( select(Tool) - .where(Tool.type.in_(("http", "mcp", "client"))) + .where(Tool.type.in_(("http", "mcp", "client", "task"))) .order_by(Tool.updated_at.desc()) ) ).scalars().all() diff --git a/backend/schemas.py b/backend/schemas.py index 30b745a..62cc8f3 100644 --- a/backend/schemas.py +++ b/backend/schemas.py @@ -21,7 +21,7 @@ AssistantType = Literal["prompt", "workflow", "dify", "fastgpt"] SUPPORTED_ASSISTANT_TYPES = get_args(AssistantType) TurnEndStrategy = Literal["silence", "smart_turn"] KnowledgeRetrievalMode = Literal["automatic", "on_demand"] -ToolType = Literal["http", "mcp", "client"] +ToolType = Literal["http", "mcp", "client", "task"] ToolStatus = Literal["active", "archived", "draft"] McpServerStatus = Literal["active", "archived", "draft"] McpTransport = Literal["streamable_http", "sse"] @@ -474,11 +474,37 @@ class McpToolDefinition(CamelModel): config: McpToolConfig +class TaskToolConfig(CamelModel): + provider: Literal["opencode"] = "opencode" + url: str = Field(min_length=1, max_length=2048) + directory: str = Field(min_length=1, max_length=2048) + username: str = Field(default="opencode", max_length=128) + timeout_seconds: int = Field(default=600, ge=30, le=7200) + allow_interruptions: Literal[True] = True + execution_mode: Literal["immediate"] = "immediate" + + @field_validator("url") + @classmethod + def validate_task_url(cls, value: str) -> str: + from urllib.parse import urlsplit + parsed = urlsplit(value) + if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment: + raise ValueError("任务服务须使用不含凭证、查询参数的 HTTP(S) 地址") + return value.rstrip("/") + + +class TaskToolDefinition(CamelModel): + schema_version: int = 2 + type: Literal["task"] = "task" + config: TaskToolConfig + + ToolDefinition = Annotated[ Union[ HttpToolDefinition, ClientToolDefinition, McpToolDefinition, + TaskToolDefinition, ], Field(discriminator="type"), ] diff --git a/backend/services/background_tasks/records.py b/backend/services/background_tasks/records.py new file mode 100644 index 0000000..955cd4c --- /dev/null +++ b/backend/services/background_tasks/records.py @@ -0,0 +1,85 @@ +"""Persistent projections of native jobs; this is not another scheduler.""" + +from dataclasses import asdict, dataclass, field +from datetime import UTC, datetime +from uuid import uuid4 + +from sqlalchemy import select +from pydantic.alias_generators import to_camel +from db.models import BackgroundTask, BackgroundTaskEvent +from db.session import SessionLocal + +TERMINAL = {"succeeded", "failed", "cancelled"} +ACTIVE = {"queued", "submitted", "running", "waiting_input", "cancel_requested", "unknown"} + + +@dataclass +class TaskRecord: + id: str + assistant_id: str + conversation_id: str + tool_id: str + instruction: str + origin_key: str + connection: dict + secrets: dict + status: str = "queued" + summary: str = "任务已排队" + result: str = "" + remote_handle: dict = field(default_factory=dict) + input_request: dict = field(default_factory=dict) + job_id: str = "" + sequence: int = 0 + created_at: str = field(default_factory=lambda: datetime.now(UTC).isoformat()) + updated_at: str = field(default_factory=lambda: datetime.now(UTC).isoformat()) + + def public(self): + hidden = {"connection", "secrets", "origin_key", "remote_handle"} + return {to_camel(key): value for key, value in asdict(self).items() if key not in hidden} + + +def from_row(row) -> TaskRecord: + return TaskRecord(**{key: getattr(row, key) for key in TaskRecord.__dataclass_fields__}) + + +class TaskRepository: + async def create(self, record: TaskRecord) -> TaskRecord: + async with SessionLocal() as db: + # All calls are serialized by the application entry point; unique + # origin_key also protects against duplicate tool delivery. + existing = (await db.execute(select(BackgroundTask).where(BackgroundTask.origin_key == record.origin_key))).scalar_one_or_none() + if existing: + return from_row(existing) + db.add(BackgroundTask(**asdict(record))) + await db.commit() + return record + + async def get(self, task_id: str) -> TaskRecord | None: + async with SessionLocal() as db: + row = await db.get(BackgroundTask, task_id) + return from_row(row) if row else None + + async def list(self, assistant_id: str | None = None, *, active=False) -> list[TaskRecord]: + statement = select(BackgroundTask).order_by(BackgroundTask.created_at.desc()) + if not active: + statement = statement.limit(100) + if assistant_id: + statement = statement.where(BackgroundTask.assistant_id == assistant_id) + if active: + statement = statement.where(BackgroundTask.status.in_(ACTIVE)) + async with SessionLocal() as db: + return [from_row(row) for row in (await db.execute(statement)).scalars().all()] + + async def patch(self, task_id: str, **changes) -> TaskRecord: + async with SessionLocal() as db: + row = (await db.execute(select(BackgroundTask).where(BackgroundTask.id == task_id).with_for_update())).scalar_one() + if row.status in TERMINAL: + return from_row(row) + for key, value in changes.items(): + setattr(row, key, value) + row.sequence += 1 + row.updated_at = datetime.now(UTC).isoformat() + record = from_row(row) + db.add(BackgroundTaskEvent(id="te_" + uuid4().hex, task_id=task_id, sequence=row.sequence, data=record.public())) + await db.commit() + return record diff --git a/backend/services/background_tasks/service.py b/backend/services/background_tasks/service.py new file mode 100644 index 0000000..f212dd4 --- /dev/null +++ b/backend/services/background_tasks/service.py @@ -0,0 +1,177 @@ +"""Business entry point backed by native Pipecat Jobs.""" + +import asyncio +from collections import defaultdict +from copy import deepcopy +from uuid import uuid4 + +from loguru import logger +from services.background_tasks.native import NativeTaskRuntime +from services.background_tasks.opencode import OpenCodeExecutor +from services.background_tasks.records import TaskRecord, TaskRepository, TERMINAL + + +class TaskService: + def __init__(self, repository=None, executor_factory=OpenCodeExecutor): + self.repository = repository or TaskRepository() + self.executor_factory = executor_factory + self.runtime = None + self._listeners = defaultdict(set) + self._jobs = {} + self._submit_lock = asyncio.Lock() + self._closing = False + + async def start(self): + self._closing = False + self.runtime = NativeTaskRuntime(self._execute, self._cancel_remote, self._on_update, self._on_result) + await self.runtime.start() + for record in await self.repository.list(active=True): + if record.remote_handle: + await self._dispatch(record) + else: + await self.repository.patch(record.id, status="unknown", summary="服务曾中断，未自动重新提交；请核查远端后再创建任务") + + async def close(self): + self._closing = True + if self.runtime: + await self.runtime.close() + self.runtime = None + self._listeners.clear() + + async def submit(self, *, assistant_id, conversation_id, tool, instruction, invocation_id): + if not self.runtime: + raise ValueError("后台任务服务尚未启动") + text = instruction.strip() + if not text or len(text) > 12000: + raise ValueError("任务指令必须为 1-12000 个字符") + if tool.type != "task": + raise ValueError("请选择后台任务工具") + async with self._submit_lock: + record = await self.repository.create(TaskRecord( + id="task_" + uuid4().hex, + assistant_id=assistant_id, conversation_id=conversation_id, + tool_id=tool.id, instruction=text, + origin_key=f"{conversation_id}:{invocation_id}"[:256], + connection=deepcopy(tool.definition["config"]), secrets=deepcopy(tool.secrets), + )) + if not record.job_id and record.status == "queued": + await self._dispatch(record) + return record.public() + + async def _dispatch(self, record): + job_id = await self.runtime.submit(record.id, record.connection.get("timeout_seconds", 600)) + self._jobs[job_id] = record.id + await self.repository.patch(record.id, job_id=job_id) + + async def get(self, task_id, assistant_id=None): + record = await self.repository.get(task_id) + if not record or (assistant_id and record.assistant_id != assistant_id): + raise ValueError("任务不存在或不属于当前助手") + return record + + def subscribe(self, conversation_id, listener): + self._listeners[conversation_id].add(listener) + def detach(): + self._listeners[conversation_id].discard(listener) + if not self._listeners[conversation_id]: + self._listeners.pop(conversation_id, None) + return detach + + async def _save(self, task_id, **changes): + record = await self.repository.patch(task_id, **changes) + for listener in tuple(self._listeners.get(record.conversation_id, ())): + try: + await listener(record.public()) + except Exception: + logger.warning("任务通知未送达，结果仍可查询: {}", task_id) + return record + + async def _execute(self, message, worker): + task_id = message.payload["task_id"] + self._jobs[message.job_id] = task_id + record = await self.get(task_id) + executor = self.executor_factory(record.connection, record.secrets) + resume = bool(record.remote_handle) + try: + if not resume: + handle = await executor.prepare() + record = await self.repository.patch(task_id, remote_handle=handle) + async def emit(event): + await self.repository.patch(task_id, **event) + await worker.send_job_update(message.job_id, {"task_id": task_id}) + # Reconnect only observations; never repeat a mutating prompt. + for attempt in range(5): + try: + result = await executor.run(record.remote_handle, record.instruction, emit, resume=resume) + await self.repository.patch(task_id, **result, input_request={}) + return {"task_id": task_id} + except asyncio.CancelledError: + raise + except Exception: + resume = True + await emit({"summary": "远端连接中断，正在核查任务状态"}) + if attempt == 4: + raise + await asyncio.sleep(min(2 ** attempt, 10)) + except asyncio.CancelledError: + raise + except Exception: + await self.repository.patch(task_id, status="unknown", summary="无法确认远端执行状态，请检查远端服务；不会自动重复提交") + return {"task_id": task_id} + + async def _on_update(self, message): + task_id = (message.update or {}).get("task_id") + if task_id: + record = await self.get(task_id) + await self._save(task_id, summary=record.summary) + + async def _on_result(self, message): + if self._closing: + return + task_id = self._jobs.pop(message.job_id, None) + if not task_id: + return + record = await self.get(task_id) + if record.status not in TERMINAL and str(message.status) in {"error", "failed"}: + await self._save(task_id, status="unknown", summary="任务运行中断，请核查远端状态") + else: + await self._save(task_id, summary=record.summary) + + async def cancel(self, task_id, assistant_id=None): + record = await self.get(task_id, assistant_id) + if record.status in TERMINAL: + return record.public() + await self._save(task_id, status="cancel_requested", summary="正在请求远端停止") + accepted = bool(self.runtime and record.job_id and await self.runtime.cancel(record.job_id)) + if not accepted: + await self._stop_record(record) + return (await self.get(task_id)).public() + + async def _cancel_remote(self, message): + task_id = self._jobs.get(message.job_id) + if not task_id or self._closing: + return + await self._stop_record(await self.get(task_id)) + + async def _stop_record(self, record): + if record.status in TERMINAL: + return + if not record.remote_handle: + await self._save(record.id, status="unknown", summary="尚无远端句柄，无法确认是否已停止") + return + try: + stopped = await self.executor_factory(record.connection, record.secrets).cancel(record.remote_handle) + except Exception: + stopped = False + await self._save(record.id, status="cancelled" if stopped else "cancel_requested", summary="远端已停止" if stopped else "停止尚未确认，请检查远端服务") + + async def send_input(self, task_id, value, assistant_id=None): + record = await self.get(task_id, assistant_id) + if record.status != "waiting_input" or not record.input_request: + raise ValueError("任务当前没有待处理的确认或问题") + await self.executor_factory(record.connection, record.secrets).send_input(record.input_request, value) + await self._save(task_id, status="running", summary="已发送回复，等待远端继续", input_request={}) + return (await self.get(task_id)).public() + + +task_service = TaskService() diff --git a/backend/services/tool_executor.py b/backend/services/tool_executor.py index 02fa48a..cce80cd 100644 --- a/backend/services/tool_executor.py +++ b/backend/services/tool_executor.py @@ -43,6 +43,8 @@ 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 new file mode 100644 index 0000000..af98b1b --- /dev/null +++ b/backend/tests/test_background_tasks.py @@ -0,0 +1,90 @@ +"""Native jobs + service lifecycle, with an explicit in-memory repository.""" +import asyncio +import unittest +from copy import deepcopy +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: + def __init__(self): + self.records = {} + async def create(self, record): + old = next((r for r in self.records.values() if r.origin_key == record.origin_key), None) + if old: + return deepcopy(old) + self.records[record.id] = deepcopy(record) + return record + async def get(self, task_id): + return deepcopy(self.records.get(task_id)) + async def list(self, assistant_id=None, active=False): + return [deepcopy(r) for r in self.records.values() if (not assistant_id or r.assistant_id == assistant_id) and (not active or r.status not in TERMINAL)] + async def patch(self, task_id, **changes): + record = self.records[task_id] + if record.status not in TERMINAL: + for key, value in changes.items(): + setattr(record, key, value) + record.sequence += 1 + return deepcopy(record) + + +class SimulatedExecutor: + release = None + calls = 0 + def __init__(self, config, secrets): + pass + async def prepare(self): + return {"session_id": "ses_test", "message_id": "msg_test"} + async def run(self, handle, instruction, emit, resume=False): + type(self).calls += 1 + await emit({"status": "running", "summary": "正在执行测试"}) + await type(self).release.wait() + return {"status": "succeeded", "summary": "完成", "result": "真实Job，模拟远端结果"} + async def cancel(self, handle): + return True + + +class TaskServiceTest(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + SimulatedExecutor.release = asyncio.Event() + SimulatedExecutor.calls = 0 + 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"}) + async def asyncTearDown(self): + await self.service.close() + async def wait_status(self, task_id, expected): + async with asyncio.timeout(3): + 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) + async def test_idempotence_public_projection_and_completion(self): + task = await self.submit() + again = await self.submit() + self.assertEqual(task["id"], again["id"]) + await self.wait_status(task["id"], "running") + self.assertEqual(SimulatedExecutor.calls, 1) + self.assertNotIn("never-public", str(task)) + SimulatedExecutor.release.set() + await self.wait_status(task["id"], "succeeded") + with self.assertRaises(ValueError): + await self.service.get(task["id"], "another-assistant") + async def test_cancel_requires_remote_confirmation(self): + task = await self.submit() + await self.wait_status(task["id"], "running") + await self.service.cancel(task["id"]) + await self.wait_status(task["id"], "cancelled") + async def test_shutdown_does_not_abort_remote_work(self): + task = await self.submit() + await self.wait_status(task["id"], "running") + await self.service.close() + self.assertEqual((await self.service.get(task["id"])).status, "running") + 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") diff --git a/implement-stage/ASSUMPTIONS.md b/implement-stage/ASSUMPTIONS.md index 51e9487..5b6c142 100644 --- a/implement-stage/ASSUMPTIONS.md +++ b/implement-stage/ASSUMPTIONS.md @@ -15,3 +15,6 @@ | 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 | diff --git a/implement-stage/BUILD_NOTE.md b/implement-stage/BUILD_NOTE.md index e41f37d..2537079 100644 --- a/implement-stage/BUILD_NOTE.md +++ b/implement-stage/BUILD_NOTE.md @@ -4,14 +4,16 @@ |------|---------|------------------|------|--------| | F0 | 原生Job主线、断线独立与取消 | backend/.venv/bin/python -m unittest discover -s backend/tests -p test_background_task_native.py | MUST | passed | | F1 | OpenCode协议适配与状态关联 | cd backend && .venv/bin/python -m unittest tests.test_background_task_native tests.test_opencode_tasks | MUST | passed | -| F2 | 持久记录、管理API、应用宿主 | cd backend && .venv/bin/python -m unittest tests.test_background_tasks | MUST | pending | +| F2 | 持久记录、管理API、应用宿主 | cd backend && .venv/bin/python -m unittest tests.test_background_tasks | MUST | passed | | F3 | 提示词绑定、任务通知、网页 | cd frontend && npx tsc --noEmit | MUST | pending | | F4 | 回归、文档与假设审阅 | make test | MUST | pending | ## Run record - F0: PYTHONPATH=backend backend/.venv/bin/python -m unittest discover -s backend/tests -p test_background_task_native.py → exit 0; native-jobs-smoke.json。 - F1: 原生 Job + httpx 模拟远端协议共 3 项测试通过；未调用真实模型。 +- F2: 累计 7 项原生Job、协议、服务生命周期与密钥保留测试通过。 ## Deferred - 真实远程 OpenCode + 模型联调：未提供服务地址/凭证。 - 多进程分布式总线及通用插件发现：超出单进程 MVP。 ## Blockers +- 开发数据库连接不可用（提升权限后仍 OSError）；本次新增迁移区间 SQL 生成通过，未执行线上/本地数据库迁移。