From 47206812f54605abe1df0901d193c23fbb44c99d Mon Sep 17 00:00:00 2001 From: orhuoxu <2414234821@qq.com> Date: Tue, 8 Sep 2026 21:37:40 +0800 Subject: [PATCH 1/5] feat(tools): add detached async execution --- backend/app/api/external_business_tasks.py | 121 +++++++ backend/app/api/tools.py | 1 + backend/app/config.py | 1 + backend/app/core/capability_manifest.py | 20 + backend/app/core/harness_agent.py | 20 + .../app/core/harness_capability_invoker.py | 28 ++ backend/app/db/database.py | 40 ++ backend/app/db/models.py | 57 +++ backend/app/main.py | 11 +- backend/app/tools/external_task_worker.py | 36 ++ backend/app/tools/external_tasks.py | 342 ++++++++++++++++++ backend/app/tools/tool_executor.py | 169 ++++++++- backend/app/tools/tool_schema.py | 8 + backend/tests/test_enterprise_auth_guards.py | 3 + backend/tests/test_external_business_tasks.py | 242 +++++++++++++ backend/tests/test_tools_api.py | 32 +- frontend-enterprise/src/i18n/en.json | 5 + frontend-enterprise/src/pages/ToolsPage.tsx | 65 ++++ frontend-enterprise/src/types/index.ts | 8 + 19 files changed, 1203 insertions(+), 6 deletions(-) create mode 100644 backend/app/api/external_business_tasks.py create mode 100644 backend/app/tools/external_task_worker.py create mode 100644 backend/app/tools/external_tasks.py create mode 100644 backend/tests/test_external_business_tasks.py diff --git a/backend/app/api/external_business_tasks.py b/backend/app/api/external_business_tasks.py new file mode 100644 index 000000000..910ab2a56 --- /dev/null +++ b/backend/app/api/external_business_tasks.py @@ -0,0 +1,121 @@ +from __future__ import annotations + +from typing import Any, Literal + +from fastapi import APIRouter, Depends, Header, HTTPException, Query +from pydantic import BaseModel, Field +from sqlmodel import Session, select + +from app.db import get_session +from app.db.models import ExternalBusinessTask, ExternalBusinessTaskEvent, User +from app.security.auth import ensure_current_user_tenant, get_current_user +from app.tools.external_tasks import apply_task_event, verify_callback_token + +enterprise_router = APIRouter( + prefix="/api/enterprise/external-business-tasks", + tags=["enterprise:external-business-tasks"], +) +callback_router = APIRouter( + prefix="/api/external-business-tasks", + tags=["external-business-tasks"], +) + + +class ExternalTaskCallback(BaseModel): + event_id: str = Field(min_length=1, max_length=200) + event_type: str = Field(default="status", max_length=100) + status: Literal[ + "accepted", "submitted", "queued", "pending", "working", "processing", + "completed", "succeeded", "success", "failed", "cancelled", "canceled", + ] + task_id: str | None = Field(default=None, max_length=300) + result: Any = None + error: Any = None + data: dict[str, Any] = Field(default_factory=dict) + + +def task_read(task: ExternalBusinessTask, db: Session) -> dict[str, Any]: + events = db.exec( + select(ExternalBusinessTaskEvent) + .where(ExternalBusinessTaskEvent.task_id == task.id) + .order_by(ExternalBusinessTaskEvent.created_at) + ).all() + return { + "id": task.id, + "external_task_id": task.external_task_id, + "tool_id": task.tool_id, + "agent_id": task.agent_id, + "session_id": task.session_id, + "status": task.status, + "result": task.result_json or {}, + "error": task.error_json or {}, + "poll_attempts": task.poll_attempts, + "created_at": task.created_at.isoformat(), + "accepted_at": task.accepted_at.isoformat() if task.accepted_at else None, + "finished_at": task.finished_at.isoformat() if task.finished_at else None, + "updated_at": task.updated_at.isoformat(), + "events": [ + { + "event_id": event.event_id, + "event_type": event.event_type, + "data": event.data_json, + "created_at": event.created_at.isoformat(), + } + for event in events + ], + } + + +@enterprise_router.get("/{external_task_id}") +def get_external_business_task( + external_task_id: str, + tenant_id: str = Query(...), + tool_id: str | None = Query(default=None), + db: Session = Depends(get_session), + current_user: User = Depends(get_current_user), +) -> dict[str, Any]: + ensure_current_user_tenant(tenant_id, current_user) + statement = select(ExternalBusinessTask).where( + ExternalBusinessTask.tenant_id == tenant_id, + ExternalBusinessTask.user_id == current_user.id, + ExternalBusinessTask.external_task_id == external_task_id, + ) + if tool_id: + statement = statement.where(ExternalBusinessTask.tool_id == tool_id) + rows = db.exec(statement).all() + if not rows: + raise HTTPException(status_code=404, detail="External business task not found") + if len(rows) > 1: + raise HTTPException(status_code=409, detail="tool_id is required for this external task id") + return task_read(rows[0], db) + + +@callback_router.post("/{task_id}/callback") +def external_business_task_callback( + task_id: str, + request: ExternalTaskCallback, + callback_token: str = Header(default="", alias="X-StaffDeck-Callback-Token"), + db: Session = Depends(get_session), +) -> dict[str, Any]: + task = db.get(ExternalBusinessTask, task_id) + if task is None or not verify_callback_token(task, callback_token): + raise HTTPException(status_code=401, detail="Invalid callback credential") + if request.task_id and request.task_id != task.external_task_id: + raise HTTPException( + status_code=409, + detail="Provider task id does not match callback target", + ) + data = dict(request.data) + if request.result is not None: + data["result"] = request.result + if request.error is not None: + data["error"] = request.error + created = apply_task_event( + db, + task, + event_id=request.event_id, + event_type=request.event_type, + status=request.status, + data=data, + ) + return {"accepted": True, "duplicate": not created, "status": task.status} diff --git a/backend/app/api/tools.py b/backend/app/api/tools.py index 75e7f669c..530fdcc02 100644 --- a/backend/app/api/tools.py +++ b/backend/app/api/tools.py @@ -590,6 +590,7 @@ def test_tool( request.tenant_id, ToolCall(name=row.name, arguments=request.arguments), agent_id=agent_id, + user_id=current_user.id, ) diff --git a/backend/app/config.py b/backend/app/config.py index 7b7f4d51f..47bcd56b2 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -15,6 +15,7 @@ class Settings(BaseSettings): model_thinking_mode: str = "" model_thinking_models: str = "" tool_timeout_seconds: float = 8.0 + external_task_poll_seconds: float = 2.0 a2a_task_timeout_seconds: float = 600.0 a2a_poll_interval_seconds: float = 0.5 codex_a2a_enabled: bool = False diff --git a/backend/app/core/capability_manifest.py b/backend/app/core/capability_manifest.py index cf2131c3d..54ec47b90 100644 --- a/backend/app/core/capability_manifest.py +++ b/backend/app/core/capability_manifest.py @@ -45,6 +45,7 @@ "run_skill_script", "knowledge_search", "lark_cli", + "external_task_status", } @@ -392,6 +393,25 @@ def _lark_cli_descriptor( def _internal_capability_descriptors() -> list[CapabilityDescriptor]: return [ + CapabilityDescriptor( + capability_id="builtin.external_task.status", + name="external_task_status", + kind="internal", + description=( + "Query a StaffDeck detached business task by task_id. Use this when the user asks " + "for the status of a previously submitted #taskid. Only the current user's tasks " + "are visible." + ), + input_schema={ + "type": "object", + "properties": { + "task_id": {"type": "string", "minLength": 1}, + }, + "required": ["task_id"], + "additionalProperties": False, + }, + metadata={"provider": "harness", "side_effect": "read"}, + ), CapabilityDescriptor( capability_id="builtin.deliverables.list", name="list_published_deliverables", diff --git a/backend/app/core/harness_agent.py b/backend/app/core/harness_agent.py index ae2443211..fa7040e2a 100644 --- a/backend/app/core/harness_agent.py +++ b/backend/app/core/harness_agent.py @@ -538,6 +538,26 @@ def finish(result: TaskExecutionResult) -> TaskExecutionResult: if _is_non_retryable_failure(result): non_retryable_action_signatures.add(action_signature) bounded_result = _bounded_capability_result(tool_name, result) + result_data = result.get("data") + if ( + result.get("success") is True + and isinstance(result_data, dict) + and result_data.get("detached") is True + ): + capability_results.append(bounded_result) + reply = str(result_data.get("user_reply") or "").strip() + return finish(TaskExecutionResult( + task_frame_id=requirement.task_frame_id, + status="completed", + reply_fragment=reply, + capability_results=capability_results, + action_count=iteration, + task_summary="异步业务任务已受理。", + structured_result={ + "task_id": result_data.get("task_id"), + "status": result_data.get("status"), + }, + )) if _is_loaded_general_skill_result(tool_name, result): loaded_general_skill_names.append(tool_name) transcript.extend( diff --git a/backend/app/core/harness_capability_invoker.py b/backend/app/core/harness_capability_invoker.py index 98550c8f4..bcf97e7da 100644 --- a/backend/app/core/harness_capability_invoker.py +++ b/backend/app/core/harness_capability_invoker.py @@ -42,6 +42,7 @@ GeneralSkill, HarnessInvocationRecord, ModelConfig, + ExternalBusinessTask, Skill, Tool, UIConfig, @@ -518,6 +519,8 @@ def _invoke_internal( return self._search_capabilities(arguments) if name == "capability_describe": return self._describe_capabilities(arguments) + if name == "external_task_status": + return self._external_task_status(arguments) if name == "list_published_deliverables": return self._list_published_deliverables(arguments) if name == "read_published_deliverable": @@ -540,6 +543,30 @@ def _invoke_internal( "不支持的 Harness 内部能力。", ) + def _external_task_status(self, arguments: dict[str, Any]) -> dict[str, Any]: + task_id = str(arguments.get("task_id") or "").strip().lstrip("#") + if not task_id: + return _failure("INVALID_ARGUMENTS", "task_id 不能为空。") + task = self.db.exec( + select(ExternalBusinessTask).where( + ExternalBusinessTask.id == task_id, + ExternalBusinessTask.tenant_id == self.tenant_id, + ExternalBusinessTask.user_id == self.session.user_id, + ) + ).first() + if task is None: + return _failure("EXTERNAL_TASK_NOT_FOUND", "未找到属于当前用户的该任务。") + status = str(task.status or "").lower() + return { + "success": True, + "data": { + "task_id": task.id, + "status": task.status, + "result": dict(task.result_json or {}), + "error": dict(task.error_json or {}), + }, + } + def _list_published_deliverables(self, arguments: dict[str, Any]) -> dict[str, Any]: raw_limit = arguments.get("limit", MAX_PUBLISHED_DELIVERABLES) if isinstance(raw_limit, bool) or not isinstance(raw_limit, int): @@ -1050,6 +1077,7 @@ def _invoke_external_tool( agent_id=self.agent_id, session_id=self.session.id, invocation_id=call_id, + user_id=self.session.user_id, timeout_seconds_override=self._remaining_step_seconds(), ) payload = result.model_dump(mode="json") diff --git a/backend/app/db/database.py b/backend/app/db/database.py index e5c34721d..0dc872396 100644 --- a/backend/app/db/database.py +++ b/backend/app/db/database.py @@ -346,6 +346,46 @@ def _migrate_sqlite_skill_schema() -> None: ) ) + if "external_business_tasks" in tables: + task_columns = { + column["name"] + for column in inspector.get_columns("external_business_tasks") + } + additions = { + "status_config_json": ( + "ALTER TABLE external_business_tasks ADD COLUMN status_config_json JSON" + ), + "idempotency_key": ( + "ALTER TABLE external_business_tasks ADD COLUMN idempotency_key VARCHAR" + ), + "lease_owner": ( + "ALTER TABLE external_business_tasks ADD COLUMN lease_owner VARCHAR" + ), + "lease_expires_at": ( + "ALTER TABLE external_business_tasks ADD COLUMN lease_expires_at DATETIME" + ), + "expires_at": ( + "ALTER TABLE external_business_tasks ADD COLUMN expires_at DATETIME" + ), + } + for column_name, ddl in additions.items(): + if column_name not in task_columns: + conn.execute(text(ddl)) + conn.execute( + text( + "UPDATE external_business_tasks SET status_config_json = '{}' " + "WHERE status_config_json IS NULL" + ) + ) + conn.execute( + text( + "CREATE UNIQUE INDEX IF NOT EXISTS " + "uq_external_business_tasks_idempotency_key " + "ON external_business_tasks(idempotency_key) " + "WHERE idempotency_key IS NOT NULL" + ) + ) + if "mcp_servers" in tables: mcp_server_columns = { column["name"] for column in inspector.get_columns("mcp_servers") diff --git a/backend/app/db/models.py b/backend/app/db/models.py index 78925cb40..e6d1af808 100644 --- a/backend/app/db/models.py +++ b/backend/app/db/models.py @@ -217,6 +217,63 @@ class A2ATaskEvent(SQLModel, table=True): created_at: datetime = Field(default_factory=utc_now) +class ExternalBusinessTask(SQLModel, table=True): + """Durable state for an HTTP tool whose business work finishes asynchronously.""" + + __tablename__ = "external_business_tasks" + __table_args__ = ( + UniqueConstraint( + "tenant_id", + "user_id", + "tool_id", + "external_task_id", + name="uq_external_business_task", + ), + ) + + id: str = Field(default_factory=lambda: new_id("exttask"), primary_key=True) + tenant_id: str = Field(index=True) + user_id: str = Field(index=True) + agent_id: Optional[str] = Field(default=None, index=True) + session_id: Optional[str] = Field(default=None, index=True) + invocation_id: Optional[str] = Field(default=None, index=True) + tool_id: str = Field(index=True) + external_task_id: Optional[str] = Field(default=None, index=True) + idempotency_key: Optional[str] = Field(default=None, unique=True, index=True) + status: str = Field(default="submitting", index=True) + request_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + result_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + error_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + callback_token_hash: str + status_url: Optional[str] = None + status_config_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + poll_interval_seconds: float = 5.0 + next_poll_at: Optional[datetime] = Field(default=None, index=True) + poll_attempts: int = 0 + lease_owner: Optional[str] = Field(default=None, index=True) + lease_expires_at: Optional[datetime] = Field(default=None, index=True) + expires_at: Optional[datetime] = Field(default=None, index=True) + created_at: datetime = Field(default_factory=utc_now) + accepted_at: Optional[datetime] = None + finished_at: Optional[datetime] = None + updated_at: datetime = Field(default_factory=utc_now) + + +class ExternalBusinessTaskEvent(SQLModel, table=True): + __tablename__ = "external_business_task_events" + __table_args__ = ( + UniqueConstraint("task_id", "event_id", name="uq_external_business_task_event"), + ) + + id: str = Field(default_factory=lambda: new_id("exttaskevt"), primary_key=True) + tenant_id: str = Field(index=True) + task_id: str = Field(index=True) + event_id: str = Field(index=True) + event_type: str = Field(index=True) + data_json: dict[str, Any] = Field(default_factory=dict, sa_column=Column(JSON)) + created_at: datetime = Field(default_factory=utc_now) + + class WebhookEndpoint(SQLModel, table=True): __tablename__ = "webhook_endpoints" diff --git a/backend/app/main.py b/backend/app/main.py index 22c47a86a..b078bf97e 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -4,6 +4,8 @@ from fastapi.middleware.cors import CORSMiddleware from sqlmodel import Session +from app.a2a import recover_codex_a2a_tasks, stop_codex_a2a_tasks +from app.a2a import router as a2a_router from app.api import ( agents, app_updates, @@ -11,6 +13,7 @@ channels, chat, evolution, + external_business_tasks, feedback, general_skills, knowledge, @@ -29,7 +32,6 @@ wechat_kf, ) from app.async_jobs import shutdown_async_jobs, start_async_jobs -from app.a2a import recover_codex_a2a_tasks, router as a2a_router, stop_codex_a2a_tasks from app.channels import start_channel_services, stop_channel_services from app.config import get_settings from app.core.harness_recovery import ( @@ -45,8 +47,9 @@ from app.public_api.webhooks import enqueue_due_webhook_deliveries from app.runtime_lock import acquire_runtime_instance_lock, release_runtime_instance_lock from app.scheduled_tasks.worker import start_background_worker, stop_background_worker -from app.tools.a2a_recovery import recover_a2a_client_tasks from app.teams.sweeper import start_timeout_sweeper, stop_timeout_sweeper +from app.tools.a2a_recovery import recover_a2a_client_tasks +from app.tools.external_task_worker import start_external_task_worker, stop_external_task_worker from app.version import app_version settings = get_settings() @@ -79,6 +82,7 @@ def on_startup() -> None: recover_orphan_harness_runs(db, startup=True) recover_codex_a2a_tasks() recover_a2a_client_tasks() + start_external_task_worker() start_background_worker() start_channel_services() start_timeout_sweeper() @@ -99,6 +103,7 @@ def on_startup() -> None: def on_shutdown() -> None: try: stop_codex_a2a_tasks() + stop_external_task_worker() stop_public_api_maintenance() stop_channel_services() stop_background_worker() @@ -140,6 +145,8 @@ def health() -> dict[str, str]: app.include_router(teams.threads_router) app.include_router(tools.router) app.include_router(tools.mcp_router) +app.include_router(external_business_tasks.enterprise_router) +app.include_router(external_business_tasks.callback_router) app.include_router(sessions.router) app.include_router(traces.router) app.include_router(mock.router) diff --git a/backend/app/tools/external_task_worker.py b/backend/app/tools/external_task_worker.py new file mode 100644 index 000000000..bb006a9e8 --- /dev/null +++ b/backend/app/tools/external_task_worker.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +import threading + +from sqlmodel import Session + +from app.config import get_settings +from app.db import engine +from app.tools.external_tasks import poll_due_external_tasks + +_stop_event = threading.Event() +_thread: threading.Thread | None = None + + +def run_external_task_worker() -> None: + while not _stop_event.is_set(): + with Session(engine) as db: + poll_due_external_tasks(db) + _stop_event.wait(max(0.5, get_settings().external_task_poll_seconds)) + + +def start_external_task_worker() -> None: + global _thread + if _thread and _thread.is_alive(): + return + _stop_event.clear() + _thread = threading.Thread( + target=run_external_task_worker, + name="staffdeck-external-task-worker", + daemon=True, + ) + _thread.start() + + +def stop_external_task_worker() -> None: + _stop_event.set() diff --git a/backend/app/tools/external_tasks.py b/backend/app/tools/external_tasks.py new file mode 100644 index 000000000..a1d022972 --- /dev/null +++ b/backend/app/tools/external_tasks.py @@ -0,0 +1,342 @@ +from __future__ import annotations + +import hashlib +import hmac +import secrets +from datetime import timedelta +from typing import Any + +import httpx +from sqlalchemy import or_, update +from sqlalchemy.exc import IntegrityError +from sqlmodel import Session, select + +from app.db.models import ( + ExternalBusinessTask, + ExternalBusinessTaskEvent, + Tool, + new_id, + utc_now, +) + +TERMINAL_STATUSES = {"completed", "succeeded", "success", "failed", "cancelled", "canceled"} +SUCCESS_STATUSES = {"completed", "succeeded", "success"} +PERSISTED_TERMINAL_STATUSES = {"completed", "failed", "cancelled", "expired"} + + +def new_callback_token() -> str: + return secrets.token_urlsafe(32) + + +def callback_token_hash(token: str) -> str: + return hashlib.sha256(token.encode("utf-8")).hexdigest() + + +def verify_callback_token(task: ExternalBusinessTask, token: str) -> bool: + return bool(token) and hmac.compare_digest(task.callback_token_hash, callback_token_hash(token)) + + +def normalize_status(value: object) -> str: + status = str(value or "working").strip().lower() + if status in SUCCESS_STATUSES: + return "completed" + if status in {"cancelled", "canceled"}: + return "cancelled" + if status == "failed": + return "failed" + if status == "expired": + return "expired" + if status in {"accepted", "submitted", "queued", "pending"}: + return "accepted" + return "working" + + +def apply_task_event( + db: Session, + task: ExternalBusinessTask, + *, + event_id: str, + event_type: str, + status: object, + data: dict[str, Any], +) -> bool: + existing = db.exec( + select(ExternalBusinessTaskEvent).where( + ExternalBusinessTaskEvent.task_id == task.id, + ExternalBusinessTaskEvent.event_id == event_id, + ) + ).first() + if existing is not None: + return False + event = ExternalBusinessTaskEvent( + tenant_id=task.tenant_id, + task_id=task.id, + event_id=event_id, + event_type=event_type, + data_json=data, + ) + db.add(event) + try: + db.flush() + except IntegrityError: + db.rollback() + return False + + now = utc_now() + next_status = normalize_status(status) + if ( + task.status in PERSISTED_TERMINAL_STATUSES + and next_status not in PERSISTED_TERMINAL_STATUSES + ): + db.commit() + return True + task.status = next_status + result = data.get("result") + error = data.get("error") + if isinstance(result, dict): + task.result_json = result + elif result is not None: + task.result_json = {"value": result} + if isinstance(error, dict): + task.error_json = error + elif error is not None: + task.error_json = {"message": str(error)} + if task.status in PERSISTED_TERMINAL_STATUSES: + task.finished_at = now + task.next_poll_at = None + task.updated_at = now + db.add(task) + db.commit() + return True + + +def poll_due_external_tasks(db: Session) -> int: + now = utc_now() + db.exec( + update(ExternalBusinessTask) + .where( + ExternalBusinessTask.status == "working", + ExternalBusinessTask.lease_expires_at.is_not(None), + ExternalBusinessTask.lease_expires_at <= now, + ) + .values(status="queued", lease_owner=None, lease_expires_at=None, updated_at=now) + ) + db.commit() + queued = db.exec( + select(ExternalBusinessTask) + .where( + ExternalBusinessTask.status == "queued", + ExternalBusinessTask.lease_owner.is_(None), + ) + .limit(20) + ).all() + local_count = 0 + for task in queued: + _execute_local_task(db, task) + local_count += 1 + expired = db.exec( + select(ExternalBusinessTask).where( + ExternalBusinessTask.status.in_(["accepted", "working"]), + ExternalBusinessTask.expires_at.is_not(None), + ExternalBusinessTask.expires_at <= now, + ) + ).all() + for task in expired: + apply_task_event( + db, + task, + event_id=f"expired-{int(now.timestamp())}", + event_type="expired", + status="expired", + data={ + "error": { + "code": "TASK_TRACKING_EXPIRED", + "message": "External task tracking exceeded its configured deadline.", + } + }, + ) + candidates = db.exec( + select(ExternalBusinessTask).where( + ExternalBusinessTask.status.in_(["accepted", "working"]), + ExternalBusinessTask.status_url.is_not(None), + ExternalBusinessTask.next_poll_at.is_not(None), + ExternalBusinessTask.next_poll_at <= now, + or_( + ExternalBusinessTask.lease_owner.is_(None), + ExternalBusinessTask.lease_expires_at <= now, + ), + ) + .limit(20) + ).all() + claimed = 0 + for task in candidates: + owner = new_id("exttasklease") + result = db.exec( + update(ExternalBusinessTask) + .where( + ExternalBusinessTask.id == task.id, + ExternalBusinessTask.status.in_(["accepted", "working"]), + or_( + ExternalBusinessTask.lease_owner.is_(None), + ExternalBusinessTask.lease_expires_at <= now, + ), + ) + .values( + lease_owner=owner, + lease_expires_at=now + timedelta(seconds=60), + updated_at=now, + ) + .execution_options(synchronize_session=False) + ) + if getattr(result, "rowcount", 0) != 1: + db.rollback() + continue + db.commit() + task = db.get(ExternalBusinessTask, task.id) + if task is None: + continue + _poll_task(db, task, owner=owner) + claimed += 1 + return claimed + local_count + len(expired) + + +def _execute_local_task(db: Session, task: ExternalBusinessTask) -> None: + owner = new_id("exttasklease") + now = utc_now() + claimed = db.exec( + update(ExternalBusinessTask) + .where( + ExternalBusinessTask.id == task.id, + ExternalBusinessTask.status == "queued", + ExternalBusinessTask.lease_owner.is_(None), + ) + .values( + status="working", + lease_owner=owner, + lease_expires_at=now + timedelta(hours=1), + updated_at=now, + ) + .execution_options(synchronize_session=False) + ) + if getattr(claimed, "rowcount", 0) != 1: + db.rollback() + return + db.commit() + task = db.get(ExternalBusinessTask, task.id) + tool = db.get(Tool, task.tool_id) if task else None + try: + if task is None or tool is None: + raise ValueError("Detached tool no longer exists") + from app.tools.tool_executor import ToolExecutor + + result = ToolExecutor(db).execute_sync_http(tool, task.request_json) + if result.success: + apply_task_event( + db, + task, + event_id=f"local-{task.id}", + event_type="completed", + status="completed", + data={"result": result.data}, + ) + else: + error = result.error.model_dump(mode="json") if result.error else {} + apply_task_event( + db, + task, + event_id=f"local-{task.id}", + event_type="failed", + status="failed", + data={"error": error}, + ) + except Exception as exc: + apply_task_event( + db, + task, + event_id=f"local-{task.id}", + event_type="failed", + status="failed", + data={"error": {"code": "DETACHED_EXECUTION_ERROR", "message": str(exc)}}, + ) + finally: + db.exec( + update(ExternalBusinessTask) + .where( + ExternalBusinessTask.id == task.id, + ExternalBusinessTask.lease_owner == owner, + ) + .values(lease_owner=None, lease_expires_at=None) + .execution_options(synchronize_session=False) + ) + db.commit() + + +def _poll_task(db: Session, task: ExternalBusinessTask, *, owner: str) -> None: + tool = db.get(Tool, task.tool_id) + if tool is None or not task.external_task_id or not task.status_url: + return + config = dict(task.status_config_json or {}) + status_field = str(config.get("status_field") or "status") + result_field = str(config.get("result_field") or "result") + status_mapping = dict(config.get("status_mapping") or {}) + url = task.status_url.replace("{taskId}", task.external_task_id) + task.poll_attempts += 1 + task.next_poll_at = utc_now() + timedelta(seconds=task.poll_interval_seconds) + try: + from app.tools.tool_executor import ToolExecutor + + executor = ToolExecutor(db) + if url.startswith("/"): + url = f"{executor.settings.normalized_tool_base_url}{url}" + headers = executor._request_headers( + url, + executor._resolve_headers(tool.headers_json or {}, tool.auth_json or {}), + ) + response = httpx.get( + url, + headers=headers, + timeout=executor._execution_policy(tool).timeout_seconds, + ) + response.raise_for_status() + payload = response.json() + if not isinstance(payload, dict): + raise ValueError("status response must be a JSON object") + raw_status = _json_path(payload, status_field) + mapped_status = status_mapping.get(str(raw_status), raw_status) + result_data = _json_path(payload, result_field) + event_data = dict(payload) + if result_data is not None: + event_data["result"] = result_data + apply_task_event( + db, + task, + event_id=f"poll-{task.poll_attempts}", + event_type="polled", + status=mapped_status, + data=event_data, + ) + except Exception as exc: + task.error_json = {"code": "POLL_ERROR", "message": str(exc)} + task.updated_at = utc_now() + db.add(task) + db.commit() + finally: + db.exec( + update(ExternalBusinessTask) + .where( + ExternalBusinessTask.id == task.id, + ExternalBusinessTask.lease_owner == owner, + ) + .values(lease_owner=None, lease_expires_at=None) + .execution_options(synchronize_session=False) + ) + db.commit() + + +def _json_path(value: Any, path: str) -> Any: + current = value + for part in (item for item in path.split(".") if item): + if not isinstance(current, dict): + return None + current = current.get(part) + return current diff --git a/backend/app/tools/tool_executor.py b/backend/app/tools/tool_executor.py index e5733b07e..8056a7798 100644 --- a/backend/app/tools/tool_executor.py +++ b/backend/app/tools/tool_executor.py @@ -13,17 +13,26 @@ from app.agents.branching import visible_tool_rows from app.config import get_settings -from app.db.models import MCPServer, Tool +from app.db.models import ChatSession, ExternalBusinessTask, MCPServer, Tool, utc_now from app.security.internal_service import INTERNAL_SERVICE_HEADER, internal_service_token from app.tools.a2a_client import A2AClient, A2AClientError +from app.tools.external_tasks import callback_token_hash, new_callback_token from app.tools.http_request import prepare_get_request from app.tools.mcp_client import MCPClientError, execute_mcp_tool, execute_mcp_tool_result from app.tools.tool_schema import MCPAppDescriptor, ToolCall, ToolError, ToolResult - SECRET_PATTERN = re.compile(r"\$\{secret\.([A-Z0-9_]+)\}") +def _json_path(value: Any, path: str) -> Any: + current = value + for part in (item for item in path.split(".") if item): + if not isinstance(current, dict): + return None + current = current.get(part) + return current + + @dataclass(frozen=True) class ToolExecutionPolicy: timeout_seconds: float @@ -43,6 +52,7 @@ def execute( session_id: str | None = None, invocation_id: str | None = None, timeout_seconds_override: float | None = None, + user_id: str | None = None, ) -> ToolResult: with self.db.no_autoflush: tool = self.db.exec( @@ -87,6 +97,24 @@ def execute( tool.name, "UNSUPPORTED_TOOL_TYPE", f"不支持的工具类型：{tool.tool_type}" ) + execution = ( + tool.config_json.get("execution", {}) if isinstance(tool.config_json, dict) else {} + ) + if execution.get("execution_mode") == "detached": + if not user_id and session_id: + session = self.db.get(ChatSession, session_id) + if session and session.tenant_id == tenant_id: + user_id = session.user_id + return self._execute_detached_http( + tool, + tool_call.arguments, + user_id=user_id, + agent_id=agent_id, + session_id=session_id, + invocation_id=invocation_id, + timeout_seconds_override=timeout_seconds_override, + ) + headers = self._request_headers( tool.url, self._resolve_headers(tool.headers_json or {}, tool.auth_json or {}), @@ -128,6 +156,143 @@ def execute( except Exception as exc: return self._error(tool.name, "EXECUTION_ERROR", str(exc)) + def execute_sync_http( + self, + tool: Tool, + arguments: dict[str, Any], + *, + timeout_seconds_override: float | None = None, + ) -> ToolResult: + """Execute the HTTP request directly for the detached worker.""" + headers = self._request_headers( + tool.url, + self._resolve_headers(tool.headers_json or {}, tool.auth_json or {}), + ) + policy = self._execution_policy( + tool, + timeout_seconds_override=timeout_seconds_override, + ) + try: + with httpx.Client(timeout=policy.timeout_seconds) as client: + if tool.method.upper() == "GET": + request_url, request_kwargs = prepare_get_request(tool.url, arguments) + response = client.request( + tool.method.upper(), request_url, headers=headers, **request_kwargs + ) + else: + response = client.request( + tool.method.upper(), tool.url, headers=headers, json=arguments + ) + response.raise_for_status() + return ToolResult( + tool_name=tool.name, + success=True, + data=self._response_data(response), + error=None, + ) + except httpx.TimeoutException: + return self._error( + tool.name, + "TIMEOUT", + f"工具调用超过 {policy.timeout_seconds:g} 秒未返回。", + ) + except httpx.HTTPStatusError as exc: + return self._error( + tool.name, + "HTTP_ERROR", + f"工具返回异常状态码：{exc.response.status_code}", + ) + except Exception as exc: + return self._error(tool.name, "EXECUTION_ERROR", str(exc)) + + def _execute_detached_http( + self, + tool: Tool, + arguments: dict[str, Any], + *, + user_id: str | None, + agent_id: str | None, + session_id: str | None, + invocation_id: str | None, + timeout_seconds_override: float | None, + ) -> ToolResult: + if not user_id: + return self._error( + tool.name, + "USER_CONTEXT_REQUIRED", + "Detached tools require an authenticated user or a user-owned session.", + ) + idempotency_key = ( + f"staffdeck:{tool.tenant_id}:{tool.id}:{invocation_id}" + if invocation_id + else None + ) + if idempotency_key: + existing = self.db.exec( + select(ExternalBusinessTask).where( + ExternalBusinessTask.idempotency_key == idempotency_key + ) + ).first() + if existing is not None and existing.external_task_id: + return self._detached_acceptance_result(tool, existing) + callback_token = new_callback_token() + task = ExternalBusinessTask( + tenant_id=tool.tenant_id, + user_id=user_id, + agent_id=agent_id, + session_id=session_id, + invocation_id=invocation_id, + idempotency_key=idempotency_key, + tool_id=tool.id, + request_json=arguments, + callback_token_hash=callback_token_hash(callback_token), + status="queued", + poll_interval_seconds=1, + ) + self.db.add(task) + self.db.flush() + task.external_task_id = task.id + task.accepted_at = utc_now() + task.updated_at = task.accepted_at + self.db.add(task) + self.db.commit() + self.db.refresh(task) + return self._detached_acceptance_result(tool, task) + + @staticmethod + def _detached_acceptance_result( + tool: Tool, + task: ExternalBusinessTask, + ) -> ToolResult: + return ToolResult( + tool_name=tool.name, + success=True, + data={ + "accepted": True, + "detached": True, + "status": task.status, + "task_id": task.external_task_id, + "staffdeck_task_id": task.id, + "status_query": { + "method": "GET", + "path": ( + f"/api/enterprise/external-business-tasks/{task.external_task_id}" + f"?tenant_id={tool.tenant_id}&tool_id={tool.id}" + ), + "tenant_id": tool.tenant_id, + "guidance": ( + "Use a separately configured authenticated HTTP status tool or SOP " + "to query this task by task_id." + ), + }, + "user_reply": ( + f"已经帮您提交任务，任务号 #{task.external_task_id}，正在后台处理。" + f"您随时可以对我说 “查询 #{task.external_task_id} 状态” 查看结果。" + ), + }, + error=None, + ) + def _execute_a2a_tool( self, tool: Tool, diff --git a/backend/app/tools/tool_schema.py b/backend/app/tools/tool_schema.py index 2e9030e15..6936f6baa 100644 --- a/backend/app/tools/tool_schema.py +++ b/backend/app/tools/tool_schema.py @@ -9,6 +9,14 @@ class ToolExecutionPolicy(BaseModel): timeout_seconds: float = Field(ge=1, le=3600) + execution_mode: Literal["sync", "detached"] = "sync" + status_url: Optional[str] = None + poll_interval_seconds: float = Field(default=5, ge=1, le=3600) + task_id_field: str = "taskId" + status_field: str = "status" + result_field: str = "result" + status_mapping: dict[str, str] = Field(default_factory=dict) + max_tracking_seconds: int = Field(default=86400, ge=1, le=2592000) class ToolCreateRequest(BaseModel): diff --git a/backend/tests/test_enterprise_auth_guards.py b/backend/tests/test_enterprise_auth_guards.py index ee70b33fa..5ded919d4 100644 --- a/backend/tests/test_enterprise_auth_guards.py +++ b/backend/tests/test_enterprise_auth_guards.py @@ -4,6 +4,7 @@ from app.api.agents import enterprise_router as agents_router from app.api.agents import scope_router as agent_scope_router from app.api.feedback import router as feedback_router +from app.api.external_business_tasks import enterprise_router as external_tasks_router from app.api.general_skills import router as general_skills_router from app.api.knowledge import router as knowledge_router from app.api.knowledge_bases import router as knowledge_bases_router @@ -34,6 +35,7 @@ def test_enterprise_read_endpoints_require_authentication() -> None: app.include_router(agents_router) app.include_router(agent_scope_router) app.include_router(feedback_router) + app.include_router(external_tasks_router) app.include_router(scheduled_tasks_router) app.include_router(sessions_router) client = TestClient(app) @@ -60,6 +62,7 @@ def test_enterprise_read_endpoints_require_authentication() -> None: "/api/enterprise/feedback/summary?tenant_id=tenant_demo", "/api/enterprise/scheduled-tasks?tenant_id=tenant_demo", "/api/enterprise/sessions?tenant_id=tenant_demo", + "/api/enterprise/external-business-tasks/provider-1?tenant_id=tenant_demo", ] for path in paths: diff --git a/backend/tests/test_external_business_tasks.py b/backend/tests/test_external_business_tasks.py new file mode 100644 index 000000000..9e8832802 --- /dev/null +++ b/backend/tests/test_external_business_tasks.py @@ -0,0 +1,242 @@ +from __future__ import annotations + +from types import SimpleNamespace +from typing import ClassVar + +import httpx +from fastapi import HTTPException +from sqlalchemy import inspect, text +from sqlalchemy.pool import StaticPool +from sqlmodel import Session, SQLModel, create_engine, select + +from app.api.external_business_tasks import ( + ExternalTaskCallback, + external_business_task_callback, + get_external_business_task, +) +from app.db.models import ( + ExternalBusinessTask, + ExternalBusinessTaskEvent, + Tenant, + Tool, + User, +) +from app.tools.external_tasks import callback_token_hash, poll_due_external_tasks +from app.tools.tool_executor import ToolExecutor +from app.tools.tool_schema import ToolCall, ToolResult + + +class _Client: + response_json: ClassVar[dict] = {"taskId": "provider-42", "status": "queued"} + request_headers: ClassVar[dict[str, str]] = {} + + def __init__(self, **_kwargs): + pass + + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def request(self, method, url, **kwargs): + del method, url + type(self).request_headers = kwargs.get("headers", {}) + return httpx.Response( + 202, + json=type(self).response_json, + request=httpx.Request("POST", "https://provider.test/tasks"), + ) + + +def test_detached_submission_persists_task_and_returns_query_guidance(monkeypatch) -> None: + monkeypatch.setattr( + "app.tools.tool_executor.httpx.Client", + lambda **_: (_ for _ in ()).throw(AssertionError("Provider must not run during submit")), + ) + with _session() as db: + user, tool = _seed(db) + result = ToolExecutor(db).execute( + "tenant_demo", + ToolCall(name=tool.name, arguments={"order": "A-1"}), + user_id=user.id, + ) + + task = db.exec(select(ExternalBusinessTask)).one() + assert result.success is True + assert result.data["accepted"] is True + assert result.data["task_id"] == task.id + assert f"/external-business-tasks/{task.id}?" in result.data["status_query"]["path"] + assert "tenant_id=tenant_demo&tool_id=tool_detached" in result.data["status_query"]["path"] + assert task.external_task_id == task.id + assert task.status == "queued" + assert "下单任务" not in result.data["status_query"]["guidance"] + + +def test_detached_submission_does_not_require_provider_task_id(monkeypatch) -> None: + monkeypatch.setattr( + "app.tools.tool_executor.httpx.Client", + lambda **_: (_ for _ in ()).throw(AssertionError("Provider must not run during submit")), + ) + with _session() as db: + user, tool = _seed(db) + result = ToolExecutor(db).execute( + "tenant_demo", ToolCall(name=tool.name), user_id=user.id + ) + task = db.exec(select(ExternalBusinessTask)).one() + assert result.success is True + assert result.data["task_id"] == task.id + assert task.status == "queued" + + +def test_staffdeck_worker_executes_original_sync_tool(monkeypatch) -> None: + with _session() as db: + user, tool = _seed(db) + task = ExternalBusinessTask( + tenant_id="tenant_demo", user_id=user.id, tool_id=tool.id, + external_task_id="exttask-1", status="queued", + callback_token_hash=callback_token_hash("token"), + request_json={"employee_id": "12343565"}, + ) + db.add(task) + db.commit() + monkeypatch.setattr( + "app.tools.tool_executor.ToolExecutor.execute_sync_http", + lambda *_args, **_kwargs: ToolResult( + tool_name=tool.name, + success=True, + data={"remaining": 20000}, + ), + ) + assert poll_due_external_tasks(db) == 1 + db.refresh(task) + assert task.status == "completed" + assert task.result_json == {"remaining": 20000} + + +def test_callback_auth_dedupe_and_user_owned_query() -> None: + with _session() as db: + user, tool = _seed(db) + other = User( + id="user_other", tenant_id="tenant_demo", username="other", password_hash="x" + ) + db.add(other) + token = "opaque-callback-secret" + task = ExternalBusinessTask( + id="exttask_1", + tenant_id="tenant_demo", + user_id=user.id, + tool_id=tool.id, + external_task_id="provider-42", + status="accepted", + callback_token_hash=callback_token_hash(token), + ) + db.add(task) + db.commit() + + request = ExternalTaskCallback( + event_id="event-1", status="completed", result={"receipt": "ok"} + ) + try: + external_business_task_callback(task.id, request, "wrong-token", db) + except HTTPException as exc: + assert exc.status_code == 401 + else: + raise AssertionError("callback must require its per-task credential") + first = external_business_task_callback(task.id, request, token, db) + duplicate = external_business_task_callback(task.id, request, token, db) + assert first == {"accepted": True, "duplicate": False, "status": "completed"} + assert duplicate["duplicate"] is True + assert len(db.exec(select(ExternalBusinessTaskEvent)).all()) == 1 + + result = get_external_business_task("provider-42", "tenant_demo", None, db, user) + assert result["status"] == "completed" + assert result["result"] == {"receipt": "ok"} + try: + get_external_business_task("provider-42", "tenant_demo", None, db, other) + except HTTPException as exc: + assert exc.status_code == 404 + else: + raise AssertionError("another user must not read this task") + + +def test_polling_only_updates_external_task_state(monkeypatch) -> None: + with _session() as db: + user, tool = _seed(db) + task = ExternalBusinessTask( + tenant_id="tenant_demo", + user_id=user.id, + tool_id=tool.id, + external_task_id="provider-42", + status="accepted", + callback_token_hash=callback_token_hash("token"), + status_url="https://provider.test/tasks/provider-42", + status_config_json={ + "status_field": "task.state", + "result_field": "task.output", + "status_mapping": {"done": "completed"}, + }, + next_poll_at=tool.created_at, + ) + db.add(task) + db.commit() + monkeypatch.setattr( + "app.tools.external_tasks.httpx.get", + lambda *_args, **_kwargs: SimpleNamespace( + raise_for_status=lambda: None, + json=lambda: {"task": {"state": "done", "output": {"value": 9}}}, + ), + ) + + assert poll_due_external_tasks(db) == 1 + db.refresh(task) + assert task.status == "completed" + assert task.result_json == {"value": 9} + assert task.poll_attempts == 1 + + +def test_create_all_migrates_existing_database_with_external_task_tables() -> None: + engine = create_engine("sqlite://", poolclass=StaticPool) + with engine.begin() as connection: + connection.execute(text("CREATE TABLE tenants (id VARCHAR PRIMARY KEY, name VARCHAR)")) + + SQLModel.metadata.create_all(engine) + + tables = set(inspect(engine).get_table_names()) + assert {"external_business_tasks", "external_business_task_events"} <= tables + + +def _seed(db: Session) -> tuple[User, Tool]: + db.add(Tenant(id="tenant_demo", name="Demo")) + user = User( + id="user_owner", tenant_id="tenant_demo", username="owner", password_hash="x" + ) + tool = Tool( + id="tool_detached", + tenant_id="tenant_demo", + name="orders.submit", + method="POST", + url="https://provider.test/tasks", + config_json={ + "execution": { + "execution_mode": "detached", + "timeout_seconds": 8, + "status_url": "https://provider.test/tasks/{taskId}", + "poll_interval_seconds": 5, + } + }, + ) + db.add(user) + db.add(tool) + db.commit() + return user, tool + + +def _session() -> Session: + engine = create_engine( + "sqlite://", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + SQLModel.metadata.create_all(engine) + return Session(engine) diff --git a/backend/tests/test_tools_api.py b/backend/tests/test_tools_api.py index 3b1ad1c82..de3abb00f 100644 --- a/backend/tests/test_tools_api.py +++ b/backend/tests/test_tools_api.py @@ -171,14 +171,42 @@ def test_tool_config_namespaces_execution_and_preserves_existing_policy() -> Non existing={**created, "obsolete": True}, ) - assert created == {"tool": "sum", "execution": {"timeout_seconds": 20.0}} + assert created == { + "tool": "sum", + "execution": { + "timeout_seconds": 20.0, + "execution_mode": "sync", + "status_url": None, + "poll_interval_seconds": 5.0, + "task_id_field": "taskId", + "status_field": "status", + "result_field": "result", + "status_mapping": {}, + "max_tracking_seconds": 86400, + }, + } assert updated_by_legacy_client == { "tool": "echo", "obsolete": False, - "execution": {"timeout_seconds": 20.0}, + "execution": created["execution"], } +def test_tool_config_round_trips_detached_execution_policy() -> None: + policy = ToolExecutionPolicy( + timeout_seconds=12, + execution_mode="detached", + status_url="https://provider.test/tasks/{taskId}", + poll_interval_seconds=10, + ) + + config = _tool_config({}, policy) + + assert config["execution"]["execution_mode"] == "detached" + assert config["execution"]["status_url"].endswith("/{taskId}") + assert _read_execution_policy(config) == policy + + def test_tool_config_rejects_untyped_execution_and_reads_invalid_legacy_safely() -> None: config = _tool_config( {"tool": "sum", "execution": {"timeout_seconds": 3601}}, diff --git a/frontend-enterprise/src/i18n/en.json b/frontend-enterprise/src/i18n/en.json index 8295a8c8a..4867abad9 100644 --- a/frontend-enterprise/src/i18n/en.json +++ b/frontend-enterprise/src/i18n/en.json @@ -69,6 +69,7 @@ "保存后可选择并测试": "After saving, can Select and Test", "保存技能失败": "Failed to save the skill.", "保存设置": "Save Settings", + "执行模式": "Execution mode", "API 密钥": "API Keys", "API 密钥 ·": "API Keys ·", "API 密钥已创建，请立即复制保存": "API key created. Copy and save it now.", @@ -1204,6 +1205,7 @@ "通过 Trace、反馈、记忆和定时任务沉淀失败原因、稳定口径和长期任务。": "Use Trace, feedback, memory, and scheduled tasks to retain failure reasons, stable standards, and long-running work.", "通过商品知识库、比价技能和购买工具，完成推荐、价格解释、下单确认和结果反馈。": "Combine product knowledge, price-comparison skills, and purchase tools for recommendations, price explanations, order confirmation, and result feedback.", "同步结果": "Sync Result", + "同步等待结果": "Wait for result", "同步失败": "Sync Failed", "同步完成：新增 {1}，更新 {2}": "Sync Complete: Add {1}, Update {2}", "头像读取失败": "Failed to Load Avatar", @@ -1817,6 +1819,8 @@ "转为草稿": "Convert to Draft", "转为草稿失败": "Failed to Convert to Draft", "状态": "Status", + "状态查询 URL（可选）": "Status URL (optional)", + "使用 {taskId} 作为外部任务 ID 占位符；留空时由回调更新。": "Use {taskId} as the external task ID placeholder. Leave blank to rely on callbacks.", "状态：": "Status:", "状态筛选": "Status Filter", "追问文案": "Follow-up Prompt", @@ -2525,6 +2529,7 @@ "任务创建": "Task created", "任务开始": "Task started", "提交报告": "Report submitted", + "提交后异步跟踪": "Submit and track asynchronously", "任务升级": "Task escalated", "竞标开始": "Bidding started", "竞标定标": "Bid awarded", diff --git a/frontend-enterprise/src/pages/ToolsPage.tsx b/frontend-enterprise/src/pages/ToolsPage.tsx index 1b44aeb5b..9b2943c62 100644 --- a/frontend-enterprise/src/pages/ToolsPage.tsx +++ b/frontend-enterprise/src/pages/ToolsPage.tsx @@ -102,6 +102,14 @@ const TOOL_FORM_INITIAL_VALUES = { input_schema: '{}', output_schema: '{}', timeout_seconds: 8, + execution_mode: 'sync' as 'sync' | 'detached', + status_url: '', + poll_interval_seconds: 5, + task_id_field: 'taskId', + status_field: 'status', + result_field: 'result', + status_mapping: '{\n "queued": "accepted",\n "processing": "working",\n "succeeded": "completed",\n "failed": "failed"\n}', + max_tracking_seconds: 86400, capability_scope: 'general' as CapabilityScope, }; @@ -2059,6 +2067,45 @@ function ToolFormFields({ {values.tool_type === 'a2a' && } + {values.tool_type === 'http' && ( +
+ + setField('execution_mode', value as 'sync' | 'detached')} + > + + + + + 同步等待结果 + 提交后异步跟踪 + + + + {values.execution_mode === 'detached' && ( +
+
+

StaffDeck 托管异步

+

StaffDeck 会生成任务号，后台执行当前 HTTP 请求并保存结果。Provider 不需要返回 taskId；用户可通过“查询 #任务号”获取结果。

+
+ + setField('poll_interval_seconds', Number(event.target.value) || 5)} + /> + +

提交后当前对话立即结束；任务完成只更新状态，不主动推送消息。用户后续通过任务号查询。

+
+ )} +
+ )} + ; execution_policy?: { timeout_seconds: number; + execution_mode?: 'sync' | 'detached'; + status_url?: string | null; + poll_interval_seconds?: number; + task_id_field?: string; + status_field?: string; + result_field?: string; + status_mapping?: Record; + max_tracking_seconds?: number; } | null; input_schema: Record; output_schema: Record; From cb93c6153f97f9bc971663bd1e20c7fa1317764b Mon Sep 17 00:00:00 2001 From: orhuoxu <2414234821@qq.com> Date: Tue, 8 Sep 2026 23:49:19 +0800 Subject: [PATCH 2/5] fix(tools): complete detached task workflow --- backend/app/core/harness_agent.py | 4 +- .../app/core/harness_capability_invoker.py | 1 + backend/app/core/harness_v2_engine.py | 74 +++++++++++++++++++ backend/app/core/task_frame_store.py | 14 ++-- backend/app/core/task_request_compiler.py | 1 + backend/app/db/database.py | 6 ++ backend/app/db/models.py | 2 + backend/app/session/session_schema.py | 2 + backend/app/tools/external_tasks.py | 32 ++++++++ backend/app/tools/tool_executor.py | 29 +++++++- 10 files changed, 157 insertions(+), 8 deletions(-) diff --git a/backend/app/core/harness_agent.py b/backend/app/core/harness_agent.py index fa7040e2a..6ac88bf73 100644 --- a/backend/app/core/harness_agent.py +++ b/backend/app/core/harness_agent.py @@ -548,11 +548,11 @@ def finish(result: TaskExecutionResult) -> TaskExecutionResult: reply = str(result_data.get("user_reply") or "").strip() return finish(TaskExecutionResult( task_frame_id=requirement.task_frame_id, - status="completed", + status="waiting_external_task", reply_fragment=reply, capability_results=capability_results, action_count=iteration, - task_summary="异步业务任务已受理。", + task_summary="异步业务任务已受理，等待外部任务完成后恢复 SOP。", structured_result={ "task_id": result_data.get("task_id"), "status": result_data.get("status"), diff --git a/backend/app/core/harness_capability_invoker.py b/backend/app/core/harness_capability_invoker.py index bcf97e7da..9170ed371 100644 --- a/backend/app/core/harness_capability_invoker.py +++ b/backend/app/core/harness_capability_invoker.py @@ -1077,6 +1077,7 @@ def _invoke_external_tool( agent_id=self.agent_id, session_id=self.session.id, invocation_id=call_id, + task_frame_id=self.task_frame_id, user_id=self.session.user_id, timeout_seconds_override=self._remaining_step_seconds(), ) diff --git a/backend/app/core/harness_v2_engine.py b/backend/app/core/harness_v2_engine.py index c52f4abb3..f3257c424 100644 --- a/backend/app/core/harness_v2_engine.py +++ b/backend/app/core/harness_v2_engine.py @@ -55,6 +55,7 @@ from app.core.turn_planner import TurnPlanner, turn_plan_router_decision from app.db.models import ( ChatSession, + ExternalBusinessTask, HarnessRunRecord, HarnessTaskFrameRecord, HarnessTurnRecord, @@ -355,6 +356,36 @@ def run(self, request: ChatTurnRequest) -> ChatTurnResponse: interaction_mode=request.interaction_mode, team_context=team_context, ) + ready_frame = next( + ( + item + for item in self.store.planner_state(session) + if item.get("status") == "ready_to_resume" + ), + None, + ) + if ready_frame is not None and plan.decision not in { + "complete_task", + "handoff_human", + }: + resume = planned_frame_from_record( + self.db.exec( + select(HarnessTaskFrameRecord).where( + HarnessTaskFrameRecord.session_id == session.id, + HarnessTaskFrameRecord.task_id == ready_frame["task_id"], + ) + ).one() + ) + plan = plan.model_copy( + update={ + "decision": "switch_to_pending", + "selected_task_id": resume.task_id, + "target_skill_id": resume.target_skill_id, + "target_step_id": resume.target_step_id, + "task_frames": [resume], + "task_updates": [], + } + ) self._renew_session_lease() self._raise_if_cancelled(request, session) slot_hydration = SlotHydrationPolicy.hydrate_plan( @@ -976,6 +1007,32 @@ def trace(event_type: str, payload: dict[str, Any]) -> None: # The human reply is already the handoff completion signal. Do not # re-enter the same terminal handoff node during the resume turn. result.status = "completed" + if result.status == "waiting_external_task": + next_step = ( + self.owner._default_next_step(active_skill, frame.target_step_id) + if active_skill is not None + else None + ) + resume_step_id = ( + str( + (next_step or {}).get("step_id") + or (next_step or {}).get("node_id") + or "" + ).strip() + or None + ) + external_task = self.db.exec( + select(ExternalBusinessTask).where( + ExternalBusinessTask.task_frame_id == row.task_id, + ExternalBusinessTask.session_id == session.id, + ExternalBusinessTask.status.in_(["queued", "accepted", "working"]), + ) + ).first() + if external_task is not None: + external_task.resume_step_id = resume_step_id + self.db.add(external_task) + result.next_step_id = resume_step_id + self.db.commit() deferred_continuation = False if frame.kind == "sop": deferred_result = _defer_failed_step_after_completed_checkpoint( @@ -1107,6 +1164,23 @@ def trace(event_type: str, payload: dict[str, Any]) -> None: break combined = _combine_results(row.task_id, results) + if combined.status == "waiting_external_task": + external_task = self.db.exec( + select(ExternalBusinessTask).where( + ExternalBusinessTask.task_frame_id == row.task_id, + ExternalBusinessTask.session_id == session.id, + ExternalBusinessTask.status.in_( + ["completed", "failed", "cancelled", "expired"] + ), + ) + ).first() + if external_task is not None: + combined.status = "ready_to_resume" + combined.structured_result = { + "external_task_id": external_task.external_task_id, + "external_task_status": external_task.status, + "external_task_result": dict(external_task.result_json or {}), + } if run is not None: self.store.finish_run( run, diff --git a/backend/app/core/task_frame_store.py b/backend/app/core/task_frame_store.py index e36efaf82..8c2248a38 100644 --- a/backend/app/core/task_frame_store.py +++ b/backend/app/core/task_frame_store.py @@ -113,8 +113,8 @@ def finish_agent_loop_for_frame( loop = self.db.get(HarnessAgentLoopRecord, row.agent_loop_id) if loop is None: return - if result_status == "awaiting_user": - loop_status = "suspended" + if result_status in {"awaiting_user", "waiting_external_task"}: + loop_status = "suspended" elif row.kind != "sop": loop_status = "active" elif result_status == "completed": @@ -527,7 +527,7 @@ def defer_for_action_budget( """Keep unstarted frames durable and resumable after the turn budget ends.""" for row in rows: - if row.status in TERMINAL_FRAME_STATUSES: + if row.status in TERMINAL_FRAME_STATUSES or row.status == "ready_to_resume": continue row.status = "queued" row.result_json = { @@ -733,7 +733,7 @@ def ready_dependency_frames( if ( item.task_id in excluded or not dependency_ids - or item.status not in {"queued", "blocked"} + or item.status not in {"queued", "blocked", "ready_to_resume"} ): continue if item.status == "blocked" and str( @@ -1081,6 +1081,8 @@ def planned_frame_from_record(row: HarnessTaskFrameRecord) -> PlannedTaskFrame: "handoff", "failed", "cancelled", + "waiting_external_task", + "ready_to_resume", } else "queued" ), @@ -1099,7 +1101,9 @@ def _legacy_projection(row: HarnessTaskFrameRecord) -> dict[str, Any]: "task_id": row.task_id, "status": ( "pending" - if row.status in {"queued", "blocked", "action_budget"} + if row.status in { + "queued", "blocked", "action_budget", "ready_to_resume" + } else row.status ), "skill_id": row.skill_id, diff --git a/backend/app/core/task_request_compiler.py b/backend/app/core/task_request_compiler.py index f58d9e6cd..18002f4e2 100644 --- a/backend/app/core/task_request_compiler.py +++ b/backend/app/core/task_request_compiler.py @@ -83,6 +83,7 @@ class TaskExecutionResult(BaseModel): task_frame_id: str status: Literal[ "completed", + "waiting_external_task", "awaiting_user", "handoff", "failed", diff --git a/backend/app/db/database.py b/backend/app/db/database.py index 0dc872396..3c6704f9f 100644 --- a/backend/app/db/database.py +++ b/backend/app/db/database.py @@ -367,6 +367,12 @@ def _migrate_sqlite_skill_schema() -> None: "expires_at": ( "ALTER TABLE external_business_tasks ADD COLUMN expires_at DATETIME" ), + "task_frame_id": ( + "ALTER TABLE external_business_tasks ADD COLUMN task_frame_id VARCHAR" + ), + "resume_step_id": ( + "ALTER TABLE external_business_tasks ADD COLUMN resume_step_id VARCHAR" + ), } for column_name, ddl in additions.items(): if column_name not in task_columns: diff --git a/backend/app/db/models.py b/backend/app/db/models.py index e6d1af808..4ef626367 100644 --- a/backend/app/db/models.py +++ b/backend/app/db/models.py @@ -236,6 +236,8 @@ class ExternalBusinessTask(SQLModel, table=True): user_id: str = Field(index=True) agent_id: Optional[str] = Field(default=None, index=True) session_id: Optional[str] = Field(default=None, index=True) + task_frame_id: Optional[str] = Field(default=None, index=True) + resume_step_id: Optional[str] = None invocation_id: Optional[str] = Field(default=None, index=True) tool_id: str = Field(index=True) external_task_id: Optional[str] = Field(default=None, index=True) diff --git a/backend/app/session/session_schema.py b/backend/app/session/session_schema.py index 069c76dc3..a24e1287c 100644 --- a/backend/app/session/session_schema.py +++ b/backend/app/session/session_schema.py @@ -23,6 +23,8 @@ TaskFrameRunStatus = Literal[ "queued", "running", + "waiting_external_task", + "ready_to_resume", "awaiting_user", "blocked", "completed", diff --git a/backend/app/tools/external_tasks.py b/backend/app/tools/external_tasks.py index a1d022972..6f5c468ed 100644 --- a/backend/app/tools/external_tasks.py +++ b/backend/app/tools/external_tasks.py @@ -14,6 +14,7 @@ from app.db.models import ( ExternalBusinessTask, ExternalBusinessTaskEvent, + HarnessTaskFrameRecord, Tool, new_id, utc_now, @@ -107,9 +108,40 @@ def apply_task_event( task.updated_at = now db.add(task) db.commit() + if task.status in PERSISTED_TERMINAL_STATUSES: + _prepare_sop_resume(db, task) return True +def _prepare_sop_resume(db: Session, task: ExternalBusinessTask) -> None: + if not task.task_frame_id or task.status not in PERSISTED_TERMINAL_STATUSES: + return + frame = db.get(HarnessTaskFrameRecord, task.task_frame_id) + if frame is None or frame.session_id != task.session_id: + return + if frame.status in {"completed", "cancelled", "failed"}: + return + result = dict(task.result_json or {}) + frame.result_json = { + **dict(frame.result_json or {}), + "external_task_id": task.external_task_id, + "external_task_status": task.status, + "external_task_result": result, + } + frame.slots_json = {**dict(frame.slots_json or {}), **result} + if task.error_json: + frame.error_json = dict(task.error_json) + frame.status = "ready_to_resume" + if task.resume_step_id: + frame.step_id = task.resume_step_id + frame.lease_owner = None + frame.lease_expires_at = None + frame.updated_at = utc_now() + frame.state_version += 1 + db.add(frame) + db.commit() + + def poll_due_external_tasks(db: Session) -> int: now = utc_now() db.exec( diff --git a/backend/app/tools/tool_executor.py b/backend/app/tools/tool_executor.py index 8056a7798..6cb5b7e1a 100644 --- a/backend/app/tools/tool_executor.py +++ b/backend/app/tools/tool_executor.py @@ -5,6 +5,7 @@ import os import re from dataclasses import dataclass +from datetime import timedelta from typing import Any from urllib.parse import urlsplit @@ -51,6 +52,8 @@ def execute( agent_id: str | None = None, session_id: str | None = None, invocation_id: str | None = None, + task_frame_id: str | None = None, + resume_step_id: str | None = None, timeout_seconds_override: float | None = None, user_id: str | None = None, ) -> ToolResult: @@ -112,6 +115,8 @@ def execute( agent_id=agent_id, session_id=session_id, invocation_id=invocation_id, + task_frame_id=task_frame_id, + resume_step_id=resume_step_id, timeout_seconds_override=timeout_seconds_override, ) @@ -214,6 +219,8 @@ def _execute_detached_http( agent_id: str | None, session_id: str | None, invocation_id: str | None, + task_frame_id: str | None, + resume_step_id: str | None, timeout_seconds_override: float | None, ) -> ToolResult: if not user_id: @@ -222,6 +229,9 @@ def _execute_detached_http( "USER_CONTEXT_REQUIRED", "Detached tools require an authenticated user or a user-owned session.", ) + execution = tool.config_json.get("execution", {}) + if not isinstance(execution, dict): + execution = {} idempotency_key = ( f"staffdeck:{tool.tenant_id}:{tool.id}:{invocation_id}" if invocation_id @@ -236,18 +246,35 @@ def _execute_detached_http( if existing is not None and existing.external_task_id: return self._detached_acceptance_result(tool, existing) callback_token = new_callback_token() + status_url = str(execution.get("status_url") or "").strip() or None + poll_interval_seconds = max( + 1.0, float(execution.get("poll_interval_seconds") or 5) + ) + max_tracking_seconds = max( + 1, int(execution.get("max_tracking_seconds") or 86400) + ) task = ExternalBusinessTask( tenant_id=tool.tenant_id, user_id=user_id, agent_id=agent_id, session_id=session_id, invocation_id=invocation_id, + task_frame_id=task_frame_id, + resume_step_id=resume_step_id, idempotency_key=idempotency_key, tool_id=tool.id, request_json=arguments, callback_token_hash=callback_token_hash(callback_token), status="queued", - poll_interval_seconds=1, + status_url=status_url, + status_config_json={ + "status_field": str(execution.get("status_field") or "status"), + "result_field": str(execution.get("result_field") or "result"), + "status_mapping": dict(execution.get("status_mapping") or {}), + }, + poll_interval_seconds=poll_interval_seconds, + next_poll_at=utc_now() if status_url else None, + expires_at=utc_now() + timedelta(seconds=max_tracking_seconds), ) self.db.add(task) self.db.flush() From 28075f34fff8c86aa87045e1c26f99a43fbe5af6 Mon Sep 17 00:00:00 2001 From: orhuoxu <2414234821@qq.com> Date: Wed, 9 Sep 2026 00:08:02 +0800 Subject: [PATCH 3/5] fix(tools): resume SOP after async task completion --- backend/app/core/harness_v2_engine.py | 5 + backend/app/tools/external_tasks.py | 10 +- backend/tests/test_external_business_tasks.py | 104 ++++++++++++++++++ 3 files changed, 117 insertions(+), 2 deletions(-) diff --git a/backend/app/core/harness_v2_engine.py b/backend/app/core/harness_v2_engine.py index f3257c424..bd727ca6c 100644 --- a/backend/app/core/harness_v2_engine.py +++ b/backend/app/core/harness_v2_engine.py @@ -1176,6 +1176,11 @@ def trace(event_type: str, payload: dict[str, Any]) -> None: ).first() if external_task is not None: combined.status = "ready_to_resume" + row.step_id = external_task.resume_step_id or row.step_id + row.slots_json = { + **dict(row.slots_json or {}), + **dict(external_task.result_json or {}), + } combined.structured_result = { "external_task_id": external_task.external_task_id, "external_task_status": external_task.status, diff --git a/backend/app/tools/external_tasks.py b/backend/app/tools/external_tasks.py index 6f5c468ed..d8d7a8ebd 100644 --- a/backend/app/tools/external_tasks.py +++ b/backend/app/tools/external_tasks.py @@ -116,8 +116,14 @@ def apply_task_event( def _prepare_sop_resume(db: Session, task: ExternalBusinessTask) -> None: if not task.task_frame_id or task.status not in PERSISTED_TERMINAL_STATUSES: return - frame = db.get(HarnessTaskFrameRecord, task.task_frame_id) - if frame is None or frame.session_id != task.session_id: + frame = db.exec( + select(HarnessTaskFrameRecord).where( + HarnessTaskFrameRecord.tenant_id == task.tenant_id, + HarnessTaskFrameRecord.session_id == task.session_id, + HarnessTaskFrameRecord.task_id == task.task_frame_id, + ) + ).first() + if frame is None: return if frame.status in {"completed", "cancelled", "failed"}: return diff --git a/backend/tests/test_external_business_tasks.py b/backend/tests/test_external_business_tasks.py index 9e8832802..1209583c7 100644 --- a/backend/tests/test_external_business_tasks.py +++ b/backend/tests/test_external_business_tasks.py @@ -15,12 +15,16 @@ get_external_business_task, ) from app.db.models import ( + ChatSession, ExternalBusinessTask, ExternalBusinessTaskEvent, + HarnessTaskFrameRecord, Tenant, Tool, User, ) +from app.core.task_frame_store import TaskFrameStore, planned_frame_from_record +from app.session.session_schema import TurnPlan from app.tools.external_tasks import callback_token_hash, poll_due_external_tasks from app.tools.tool_executor import ToolExecutor from app.tools.tool_schema import ToolCall, ToolResult @@ -195,6 +199,106 @@ def test_polling_only_updates_external_task_state(monkeypatch) -> None: assert task.poll_attempts == 1 +def test_completed_external_task_marks_sop_frame_ready_to_resume() -> None: + with _session() as db: + user, tool = _seed(db) + session = ChatSession( + id="session_order", + tenant_id="tenant_demo", + user_id=user.id, + agent_id="agent_order", + ) + frame = HarnessTaskFrameRecord( + tenant_id="tenant_demo", + session_id=session.id, + source_turn_id="turn_order", + task_id="sop_order", + kind="sop", + status="waiting_external_task", + skill_id="order_sop", + step_id="submit_order", + slots_json={"employee_id": "14534"}, + ) + task = ExternalBusinessTask( + tenant_id="tenant_demo", + user_id=user.id, + session_id=session.id, + task_frame_id=frame.task_id, + tool_id=tool.id, + external_task_id="provider-order-1", + status="accepted", + callback_token_hash=callback_token_hash("token"), + resume_step_id="confirm_order", + ) + db.add(session) + db.add(frame) + db.add(task) + db.commit() + + result = external_business_task_callback( + task.id, + ExternalTaskCallback( + event_id="order-complete-1", + status="completed", + result={"order_id": "ORD-001"}, + ), + "token", + db, + ) + + db.refresh(frame) + assert result["status"] == "completed" + assert frame.status == "ready_to_resume" + assert frame.step_id == "confirm_order" + assert frame.slots_json["order_id"] == "ORD-001" + assert frame.result_json["external_task_result"] == {"order_id": "ORD-001"} + + +def test_ready_sop_frame_is_queued_on_the_next_turn() -> None: + with _session() as db: + user, _tool = _seed(db) + session = ChatSession( + id="session_resume", + tenant_id="tenant_demo", + user_id=user.id, + agent_id="agent_order", + active_skill_id="order_sop", + active_step_id="confirm_order", + ) + frame = HarnessTaskFrameRecord( + tenant_id="tenant_demo", + session_id=session.id, + source_turn_id="turn_submit", + task_id="sop_resume", + kind="sop", + decision="continue_active", + status="ready_to_resume", + skill_id="order_sop", + step_id="confirm_order", + slots_json={"order_id": "ORD-001"}, + ) + db.add(session) + db.add(frame) + db.commit() + + records = TaskFrameStore(db).persist_plan( + session, + "turn_resume", + TurnPlan( + decision="switch_to_pending", + selected_task_id=frame.task_id, + task_frames=[planned_frame_from_record(frame)], + ), + ) + + assert len(records) == 1 + assert records[0].status == "queued" + assert records[0].step_id == "confirm_order" + assert records[0].slots_json["order_id"] == "ORD-001" + TaskFrameStore(db).mark_running(records[0]) + assert records[0].status == "running" + + def test_create_all_migrates_existing_database_with_external_task_tables() -> None: engine = create_engine("sqlite://", poolclass=StaticPool) with engine.begin() as connection: From 3b36fe4d72e31cfba541c1b539af93eee96d9e6a Mon Sep 17 00:00:00 2001 From: orhuoxu <2414234821@qq.com> Date: Wed, 9 Sep 2026 01:21:46 +0800 Subject: [PATCH 4/5] test(tools): cover resumable async SOP --- backend/tests/test_harness_v2.py | 107 +++++++++++++++++++++++++++++++ 1 file changed, 107 insertions(+) diff --git a/backend/tests/test_harness_v2.py b/backend/tests/test_harness_v2.py index e9a51d28d..b9a2c38d6 100644 --- a/backend/tests/test_harness_v2.py +++ b/backend/tests/test_harness_v2.py @@ -4603,6 +4603,113 @@ def generate_json( assert payloads[2]["agent_loop_memory"]["recent_task_summaries"] == ["已完成"] +def test_multi_step_detached_task_resumes_agent_after_external_result(monkeypatch) -> None: + """A detached step must pause, then allow the next step after result injection.""" + + actions = iter( + [ + { + "action": "tool", + "tool_name": "orders.submit", + "arguments": {"sku": "SKU-1"}, + }, + { + "action": "finish", + "status": "completed", + "reply_fragment": "订单已创建，订单号为 ORD-001。", + }, + ] + ) + + class FakeLLMClient: + def __init__(self, _model_config: ModelConfig): + pass + + def generate_json(self, _system_prompt: str, _payload: dict[str, object]): + return next(actions) + + monkeypatch.setattr(harness_agent_module, "LLMClient", FakeLLMClient) + requirement = TaskRequirement( + task_frame_id="order-frame", + kind="sop", + goal="提交订单并在任务完成后确认订单号", + capability_manifest=CapabilityManifest( + available=[ + CapabilityDescriptor( + capability_id="tool-order-submit", + name="orders.submit", + kind="tool", + ) + ] + ), + ) + accepted = { + "success": True, + "data": { + "detached": True, + "accepted": True, + "task_id": "exttask-order-1", + "status": "queued", + }, + } + + first = HarnessTaskAgent().run( + requirement, + _model_config(), + lambda _name, _arguments: accepted, + max_actions=3, + ) + + assert first.status == "waiting_external_task" + assert first.structured_result == { + "task_id": "exttask-order-1", + "status": "queued", + } + + # This is the durable continuation operation performed after the external + # task finishes: replace the pending receipt in the checkpoint with the + # final business result before re-entering the AgentLoop. + checkpoint = dict(first.loop_checkpoint) + checkpoint["transcript"] = [ + { + "role": "assistant", + "action": "tool", + "tool_name": "orders.submit", + "arguments": {"sku": "SKU-1"}, + }, + { + "role": "tool", + "tool_name": "orders.submit", + "result": { + "success": True, + "data": { + "task_id": "exttask-order-1", + "status": "completed", + "order_id": "ORD-001", + }, + "pending": False, + }, + }, + ] + checkpoint["external_task_result"] = { + "task_id": "exttask-order-1", + "status": "completed", + "order_id": "ORD-001", + } + + resumed = HarnessTaskAgent().run( + requirement, + _model_config(), + lambda _name, _arguments: {"success": True}, + max_actions=2, + checkpoint=checkpoint, + ) + + assert resumed.status == "completed" + assert resumed.reply_fragment == "订单已创建，订单号为 ORD-001。" + assert resumed.action_count == 1 + + def test_turn_action_budget_defers_unstarted_frames_as_queued() -> None: engine = _test_engine() with Session(engine) as db: From f12cb6e85d652380578c41f792d917fb1549ff20 Mon Sep 17 00:00:00 2001 From: hm1229 Date: Wed, 9 Sep 2026 12:01:02 +0800 Subject: [PATCH 5/5] fix(tools): complete provider async task lifecycle --- backend/.env.example | 4 + backend/app/api/external_business_tasks.py | 6 +- backend/app/config.py | 1 + backend/app/core/harness_agent.py | 21 +- .../app/core/harness_capability_invoker.py | 1 - backend/app/core/harness_v2_engine.py | 21 +- backend/app/tools/external_tasks.py | 293 +++++++++++++++--- backend/app/tools/tool_executor.py | 79 +++-- backend/app/tools/tool_schema.py | 13 +- backend/tests/test_external_business_tasks.py | 230 +++++++++++++- backend/tests/test_harness_v2.py | 32 +- backend/tests/test_tools_api.py | 11 + frontend-enterprise/src/i18n/en.json | 19 +- .../src/pages/ToolsPage.async.test.ts | 47 +++ frontend-enterprise/src/pages/ToolsPage.tsx | 131 ++++++-- frontend-enterprise/src/types/index.ts | 1 + 16 files changed, 805 insertions(+), 105 deletions(-) create mode 100644 frontend-enterprise/src/pages/ToolsPage.async.test.ts diff --git a/backend/.env.example b/backend/.env.example index 4f2c369d6..e3bb7b5e6 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -7,6 +7,10 @@ DEMO_MODEL_API_KEY="" MODEL_THINKING_MODE="" MODEL_THINKING_MODELS="" TOOL_TIMEOUT_SECONDS="8" +EXTERNAL_TASK_POLL_SECONDS="2" +# Optional externally reachable StaffDeck base URL used for provider callbacks. +# Production deployments should use HTTPS. Leave empty for polling-only tracking. +EXTERNAL_TASK_CALLBACK_BASE_URL="" A2A_TASK_TIMEOUT_SECONDS="600" A2A_POLL_INTERVAL_SECONDS="0.5" # Optional local A2A server backed by Codex CLI. Keep disabled unless the host diff --git a/backend/app/api/external_business_tasks.py b/backend/app/api/external_business_tasks.py index 910ab2a56..2f2b5d5cb 100644 --- a/backend/app/api/external_business_tasks.py +++ b/backend/app/api/external_business_tasks.py @@ -4,6 +4,7 @@ from fastapi import APIRouter, Depends, Header, HTTPException, Query from pydantic import BaseModel, Field +from sqlalchemy import or_ from sqlmodel import Session, select from app.db import get_session @@ -78,7 +79,10 @@ def get_external_business_task( statement = select(ExternalBusinessTask).where( ExternalBusinessTask.tenant_id == tenant_id, ExternalBusinessTask.user_id == current_user.id, - ExternalBusinessTask.external_task_id == external_task_id, + or_( + ExternalBusinessTask.id == external_task_id, + ExternalBusinessTask.external_task_id == external_task_id, + ), ) if tool_id: statement = statement.where(ExternalBusinessTask.tool_id == tool_id) diff --git a/backend/app/config.py b/backend/app/config.py index 47bcd56b2..8d152e57c 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -16,6 +16,7 @@ class Settings(BaseSettings): model_thinking_models: str = "" tool_timeout_seconds: float = 8.0 external_task_poll_seconds: float = 2.0 + external_task_callback_base_url: str = "" a2a_task_timeout_seconds: float = 600.0 a2a_poll_interval_seconds: float = 0.5 codex_a2a_enabled: bool = False diff --git a/backend/app/core/harness_agent.py b/backend/app/core/harness_agent.py index 6ac88bf73..bdda54ee6 100644 --- a/backend/app/core/harness_agent.py +++ b/backend/app/core/harness_agent.py @@ -545,6 +545,21 @@ def finish(result: TaskExecutionResult) -> TaskExecutionResult: and result_data.get("detached") is True ): capability_results.append(bounded_result) + transcript.extend( + [ + { + "role": "assistant", + "action": "tool", + "tool_name": tool_name, + "arguments": action.arguments, + }, + { + "role": "tool", + "tool_name": tool_name, + "result": bounded_result, + }, + ] + ) reply = str(result_data.get("user_reply") or "").strip() return finish(TaskExecutionResult( task_frame_id=requirement.task_frame_id, @@ -552,7 +567,11 @@ def finish(result: TaskExecutionResult) -> TaskExecutionResult: reply_fragment=reply, capability_results=capability_results, action_count=iteration, - task_summary="异步业务任务已受理，等待外部任务完成后恢复 SOP。", + task_summary=( + "异步业务任务已受理，等待完成后恢复 SOP。" + if requirement.kind == "sop" + else "异步业务任务已受理，可通过任务号查询进度。" + ), structured_result={ "task_id": result_data.get("task_id"), "status": result_data.get("status"), diff --git a/backend/app/core/harness_capability_invoker.py b/backend/app/core/harness_capability_invoker.py index 9170ed371..d2250a0ae 100644 --- a/backend/app/core/harness_capability_invoker.py +++ b/backend/app/core/harness_capability_invoker.py @@ -556,7 +556,6 @@ def _external_task_status(self, arguments: dict[str, Any]) -> dict[str, Any]: ).first() if task is None: return _failure("EXTERNAL_TASK_NOT_FOUND", "未找到属于当前用户的该任务。") - status = str(task.status or "").lower() return { "success": True, "data": { diff --git a/backend/app/core/harness_v2_engine.py b/backend/app/core/harness_v2_engine.py index bd727ca6c..b52c37246 100644 --- a/backend/app/core/harness_v2_engine.py +++ b/backend/app/core/harness_v2_engine.py @@ -73,6 +73,7 @@ TurnPlan, ) from app.skills.nesting import discoverable_sops, expand_visible_sops +from app.tools.external_tasks import update_external_task_checkpoint def _turn_skill_projection( @@ -361,6 +362,7 @@ def run(self, request: ChatTurnRequest) -> ChatTurnResponse: item for item in self.store.planner_state(session) if item.get("status") == "ready_to_resume" + and item.get("kind") == "sop" ), None, ) @@ -1175,12 +1177,27 @@ def trace(event_type: str, payload: dict[str, Any]) -> None: ) ).first() if external_task is not None: - combined.status = "ready_to_resume" - row.step_id = external_task.resume_step_id or row.step_id + combined.status = ( + "ready_to_resume" + if row.kind == "sop" + else ( + "completed" + if external_task.status == "completed" + else "failed" + ) + ) + if ( + row.kind == "sop" + and external_task.status == "completed" + and external_task.resume_step_id + ): + row.step_id = external_task.resume_step_id row.slots_json = { **dict(row.slots_json or {}), **dict(external_task.result_json or {}), } + update_external_task_checkpoint(self.db, row, external_task) + loop_checkpoint = dict(agent_loop.checkpoint_json or loop_checkpoint) combined.structured_result = { "external_task_id": external_task.external_task_id, "external_task_status": external_task.status, diff --git a/backend/app/tools/external_tasks.py b/backend/app/tools/external_tasks.py index d8d7a8ebd..e2e5a2fab 100644 --- a/backend/app/tools/external_tasks.py +++ b/backend/app/tools/external_tasks.py @@ -14,6 +14,7 @@ from app.db.models import ( ExternalBusinessTask, ExternalBusinessTaskEvent, + HarnessAgentLoopRecord, HarnessTaskFrameRecord, Tool, new_id, @@ -22,7 +23,17 @@ TERMINAL_STATUSES = {"completed", "succeeded", "success", "failed", "cancelled", "canceled"} SUCCESS_STATUSES = {"completed", "succeeded", "success"} -PERSISTED_TERMINAL_STATUSES = {"completed", "failed", "cancelled", "expired"} +PERSISTED_TERMINAL_STATUSES = { + "completed", + "failed", + "cancelled", + "expired", + "outcome_unknown", +} +PROVIDER_TASK_STRATEGY = "provider_task" +CALLBACK_URL_HEADER = "X-StaffDeck-Callback-URL" +CALLBACK_TOKEN_HEADER = "X-StaffDeck-Callback-Token" +IDEMPOTENCY_HEADER = "Idempotency-Key" def new_callback_token() -> str: @@ -47,6 +58,8 @@ def normalize_status(value: object) -> str: return "failed" if status == "expired": return "expired" + if status == "outcome_unknown": + return "outcome_unknown" if status in {"accepted", "submitted", "queued", "pending"}: return "accepted" return "working" @@ -87,7 +100,7 @@ def apply_task_event( next_status = normalize_status(status) if ( task.status in PERSISTED_TERMINAL_STATUSES - and next_status not in PERSISTED_TERMINAL_STATUSES + and task.status != "outcome_unknown" ): db.commit() return True @@ -102,9 +115,13 @@ def apply_task_event( task.error_json = error elif error is not None: task.error_json = {"message": str(error)} + if task.status == "completed": + task.error_json = {} if task.status in PERSISTED_TERMINAL_STATUSES: task.finished_at = now task.next_poll_at = None + else: + task.finished_at = None task.updated_at = now db.add(task) db.commit() @@ -127,6 +144,12 @@ def _prepare_sop_resume(db: Session, task: ExternalBusinessTask) -> None: return if frame.status in {"completed", "cancelled", "failed"}: return + if frame.status == "running": + # The active engine still owns the frame lease and will fold a fast + # completion into its in-memory result before releasing that lease. + return + if frame.status not in {"waiting_external_task", "ready_to_resume"}: + return result = dict(task.result_json or {}) frame.result_json = { **dict(frame.result_json or {}), @@ -137,9 +160,13 @@ def _prepare_sop_resume(db: Session, task: ExternalBusinessTask) -> None: frame.slots_json = {**dict(frame.slots_json or {}), **result} if task.error_json: frame.error_json = dict(task.error_json) - frame.status = "ready_to_resume" - if task.resume_step_id: - frame.step_id = task.resume_step_id + update_external_task_checkpoint(db, frame, task) + if frame.kind == "sop": + frame.status = "ready_to_resume" + if task.status == "completed" and task.resume_step_id: + frame.step_id = task.resume_step_id + else: + frame.status = "completed" if task.status == "completed" else "failed" frame.lease_owner = None frame.lease_expires_at = None frame.updated_at = utc_now() @@ -148,18 +175,106 @@ def _prepare_sop_resume(db: Session, task: ExternalBusinessTask) -> None: db.commit() +def update_external_task_checkpoint( + db: Session, + frame: HarnessTaskFrameRecord, + task: ExternalBusinessTask, +) -> None: + if not frame.agent_loop_id: + return + loop = db.get(HarnessAgentLoopRecord, frame.agent_loop_id) + if loop is None: + return + task_result = { + "success": task.status == "completed", + "data": { + "detached": True, + "task_id": task.id, + "provider_task_id": task.external_task_id, + "status": task.status, + "result": dict(task.result_json or {}), + "error": dict(task.error_json or {}), + }, + "error": dict(task.error_json or {}) or None, + } + checkpoint = dict(loop.checkpoint_json or {}) + transcript = [dict(item) for item in checkpoint.get("transcript") or []] + updated = False + for item in reversed(transcript): + result = item.get("result") + data = result.get("data") if isinstance(result, dict) else None + if ( + item.get("role") == "tool" + and isinstance(data, dict) + and data.get("task_id") == task.id + ): + item["result"] = task_result + updated = True + break + if not updated: + transcript.append( + { + "role": "tool", + "tool_name": "external_task_status", + "result": task_result, + } + ) + checkpoint["transcript"] = transcript + capability_results = [ + dict(item) for item in checkpoint.get("capability_results") or [] + ] + for item in reversed(capability_results): + data = item.get("data") + if isinstance(data, dict) and data.get("task_id") == task.id: + item["success"] = task_result["success"] + item["data"] = task_result["data"] + item["error"] = task_result["error"] + break + checkpoint["capability_results"] = capability_results + checkpoint["external_task_result"] = task_result["data"] + loop.checkpoint_json = checkpoint + loop.updated_at = utc_now() + loop.state_version = max(1, int(loop.state_version or 0) + 1) + db.add(loop) + + def poll_due_external_tasks(db: Session) -> int: now = utc_now() - db.exec( - update(ExternalBusinessTask) - .where( - ExternalBusinessTask.status == "working", + stale_tasks = db.exec( + select(ExternalBusinessTask).where( + ExternalBusinessTask.status.in_(["submitting", "working"]), ExternalBusinessTask.lease_expires_at.is_not(None), ExternalBusinessTask.lease_expires_at <= now, ) - .values(status="queued", lease_owner=None, lease_expires_at=None, updated_at=now) - ) - db.commit() + ).all() + recovered = 0 + for task in stale_tasks: + if ( + _task_strategy(task) == PROVIDER_TASK_STRATEGY + and task.external_task_id + and task.status_url + ): + task.status = "working" + task.next_poll_at = now + else: + task.status = "outcome_unknown" + task.error_json = { + "code": "DETACHED_OUTCOME_UNKNOWN", + "message": ( + "The worker stopped after the external request may have been sent; " + "StaffDeck will not replay a potentially non-idempotent request." + ), + } + task.finished_at = now + task.next_poll_at = None + task.lease_owner = None + task.lease_expires_at = None + task.updated_at = now + db.add(task) + db.commit() + if task.status in PERSISTED_TERMINAL_STATUSES: + _prepare_sop_resume(db, task) + recovered += 1 queued = db.exec( select(ExternalBusinessTask) .where( @@ -174,7 +289,7 @@ def poll_due_external_tasks(db: Session) -> int: local_count += 1 expired = db.exec( select(ExternalBusinessTask).where( - ExternalBusinessTask.status.in_(["accepted", "working"]), + ExternalBusinessTask.status.in_(["accepted", "working", "submitting"]), ExternalBusinessTask.expires_at.is_not(None), ExternalBusinessTask.expires_at <= now, ) @@ -235,12 +350,13 @@ def poll_due_external_tasks(db: Session) -> int: continue _poll_task(db, task, owner=owner) claimed += 1 - return claimed + local_count + len(expired) + return claimed + local_count + len(expired) + recovered def _execute_local_task(db: Session, task: ExternalBusinessTask) -> None: owner = new_id("exttasklease") now = utc_now() + strategy = _task_strategy(task) claimed = db.exec( update(ExternalBusinessTask) .where( @@ -249,7 +365,7 @@ def _execute_local_task(db: Session, task: ExternalBusinessTask) -> None: ExternalBusinessTask.lease_owner.is_(None), ) .values( - status="working", + status=("submitting" if strategy == PROVIDER_TASK_STRATEGY else "working"), lease_owner=owner, lease_expires_at=now + timedelta(hours=1), updated_at=now, @@ -267,33 +383,44 @@ def _execute_local_task(db: Session, task: ExternalBusinessTask) -> None: raise ValueError("Detached tool no longer exists") from app.tools.tool_executor import ToolExecutor - result = ToolExecutor(db).execute_sync_http(tool, task.request_json) - if result.success: - apply_task_event( - db, - task, - event_id=f"local-{task.id}", - event_type="completed", - status="completed", - data={"result": result.data}, - ) + if strategy == PROVIDER_TASK_STRATEGY: + _submit_provider_task(db, task, tool) else: - error = result.error.model_dump(mode="json") if result.error else {} - apply_task_event( - db, - task, - event_id=f"local-{task.id}", - event_type="failed", - status="failed", - data={"error": error}, - ) + result = ToolExecutor(db).execute_sync_http(tool, task.request_json) + if result.success: + apply_task_event( + db, + task, + event_id=f"local-{task.id}", + event_type="completed", + status="completed", + data={"result": result.data}, + ) + else: + error = result.error.model_dump(mode="json") if result.error else {} + apply_task_event( + db, + task, + event_id=f"local-{task.id}", + event_type="failed", + status="failed", + data={"error": error}, + ) except Exception as exc: apply_task_event( db, task, - event_id=f"local-{task.id}", - event_type="failed", - status="failed", + event_id=( + f"submission-{task.id}" + if strategy == PROVIDER_TASK_STRATEGY + else f"local-{task.id}" + ), + event_type="submission_failed", + status=( + "outcome_unknown" + if strategy == PROVIDER_TASK_STRATEGY + else "failed" + ), data={"error": {"code": "DETACHED_EXECUTION_ERROR", "message": str(exc)}}, ) finally: @@ -309,6 +436,98 @@ def _execute_local_task(db: Session, task: ExternalBusinessTask) -> None: db.commit() +def _submit_provider_task(db: Session, task: ExternalBusinessTask, tool: Tool) -> None: + from app.config import get_settings + from app.tools.tool_executor import ToolExecutor + + callback_token = new_callback_token() + task.callback_token_hash = callback_token_hash(callback_token) + task.updated_at = utc_now() + db.add(task) + db.commit() + + headers = {IDEMPOTENCY_HEADER: str(task.idempotency_key or task.id)} + callback_base_url = get_settings().external_task_callback_base_url.strip().rstrip("/") + if callback_base_url: + headers[CALLBACK_URL_HEADER] = ( + f"{callback_base_url}/api/external-business-tasks/{task.id}/callback" + ) + headers[CALLBACK_TOKEN_HEADER] = callback_token + + response = ToolExecutor(db).execute_http_with_metadata( + tool, + task.request_json, + additional_headers=headers, + ) + if not response.result.success: + error = ( + response.result.error.model_dump(mode="json") + if response.result.error + else {"code": "PROVIDER_SUBMISSION_FAILED"} + ) + uncertain = str(error.get("code") or "") in {"TIMEOUT", "EXECUTION_ERROR"} + apply_task_event( + db, + task, + event_id=f"submission-{task.id}", + event_type="submission_failed", + status="outcome_unknown" if uncertain else "failed", + data={"error": error}, + ) + return + + payload = response.result.data + data = dict(payload) if isinstance(payload, dict) else {"result": payload} + config = dict(task.status_config_json or {}) + provider_task_id = _json_path(payload, str(config.get("task_id_field") or "taskId")) + raw_status = _json_path(payload, str(config.get("status_field") or "status")) + status_mapping = dict(config.get("status_mapping") or {}) + if raw_status is None: + raw_status = "accepted" if response.status_code == 202 or provider_task_id else "completed" + mapped_status = status_mapping.get(str(raw_status), raw_status) + next_status = normalize_status(mapped_status) + + if provider_task_id is None and next_status not in PERSISTED_TERMINAL_STATUSES: + apply_task_event( + db, + task, + event_id=f"submission-{task.id}", + event_type="submission_failed", + status="failed", + data={ + "error": { + "code": "PROVIDER_TASK_ID_MISSING", + "message": "Provider accepted an async task without returning its task ID.", + } + }, + ) + return + + if provider_task_id is not None: + task.external_task_id = str(provider_task_id) + result_field = str(config.get("result_field") or "result") + result_data = _json_path(payload, result_field) + if result_data is not None: + data["result"] = result_data + if next_status not in PERSISTED_TERMINAL_STATUSES and task.status_url: + task.next_poll_at = utc_now() + timedelta(seconds=task.poll_interval_seconds) + db.add(task) + apply_task_event( + db, + task, + event_id=f"submission-{task.id}", + event_type="submitted", + status=next_status, + data=data, + ) + + +def _task_strategy(task: ExternalBusinessTask) -> str: + config = task.status_config_json if isinstance(task.status_config_json, dict) else {} + strategy = str(config.get("async_strategy") or "staffdeck_worker") + return strategy if strategy == PROVIDER_TASK_STRATEGY else "staffdeck_worker" + + def _poll_task(db: Session, task: ExternalBusinessTask, *, owner: str) -> None: tool = db.get(Tool, task.tool_id) if tool is None or not task.external_task_id or not task.status_url: diff --git a/backend/app/tools/tool_executor.py b/backend/app/tools/tool_executor.py index 6cb5b7e1a..738422c0f 100644 --- a/backend/app/tools/tool_executor.py +++ b/backend/app/tools/tool_executor.py @@ -39,6 +39,12 @@ class ToolExecutionPolicy: timeout_seconds: float +@dataclass(frozen=True) +class HttpExecutionResponse: + result: ToolResult + status_code: int | None = None + + class ToolExecutor: def __init__(self, db: Session): self.db = db @@ -169,10 +175,26 @@ def execute_sync_http( timeout_seconds_override: float | None = None, ) -> ToolResult: """Execute the HTTP request directly for the detached worker.""" + return self.execute_http_with_metadata( + tool, + arguments, + timeout_seconds_override=timeout_seconds_override, + ).result + + def execute_http_with_metadata( + self, + tool: Tool, + arguments: dict[str, Any], + *, + timeout_seconds_override: float | None = None, + additional_headers: dict[str, str] | None = None, + ) -> HttpExecutionResponse: + """Execute HTTP while retaining the response status for async protocols.""" headers = self._request_headers( tool.url, self._resolve_headers(tool.headers_json or {}, tool.auth_json or {}), ) + headers.update(additional_headers or {}) policy = self._execution_policy( tool, timeout_seconds_override=timeout_seconds_override, @@ -189,26 +211,36 @@ def execute_sync_http( tool.method.upper(), tool.url, headers=headers, json=arguments ) response.raise_for_status() - return ToolResult( - tool_name=tool.name, - success=True, - data=self._response_data(response), - error=None, + return HttpExecutionResponse( + result=ToolResult( + tool_name=tool.name, + success=True, + data=self._response_data(response), + error=None, + ), + status_code=response.status_code, ) except httpx.TimeoutException: - return self._error( - tool.name, - "TIMEOUT", - f"工具调用超过 {policy.timeout_seconds:g} 秒未返回。", + return HttpExecutionResponse( + result=self._error( + tool.name, + "TIMEOUT", + f"工具调用超过 {policy.timeout_seconds:g} 秒未返回。", + ) ) except httpx.HTTPStatusError as exc: - return self._error( - tool.name, - "HTTP_ERROR", - f"工具返回异常状态码：{exc.response.status_code}", + return HttpExecutionResponse( + result=self._error( + tool.name, + "HTTP_ERROR", + f"工具返回异常状态码：{exc.response.status_code}", + ), + status_code=exc.response.status_code, ) except Exception as exc: - return self._error(tool.name, "EXECUTION_ERROR", str(exc)) + return HttpExecutionResponse( + result=self._error(tool.name, "EXECUTION_ERROR", str(exc)) + ) def _execute_detached_http( self, @@ -232,6 +264,7 @@ def _execute_detached_http( execution = tool.config_json.get("execution", {}) if not isinstance(execution, dict): execution = {} + async_strategy = str(execution.get("async_strategy") or "staffdeck_worker") idempotency_key = ( f"staffdeck:{tool.tenant_id}:{tool.id}:{invocation_id}" if invocation_id @@ -243,7 +276,7 @@ def _execute_detached_http( ExternalBusinessTask.idempotency_key == idempotency_key ) ).first() - if existing is not None and existing.external_task_id: + if existing is not None: return self._detached_acceptance_result(tool, existing) callback_token = new_callback_token() status_url = str(execution.get("status_url") or "").strip() or None @@ -268,17 +301,20 @@ def _execute_detached_http( status="queued", status_url=status_url, status_config_json={ + "async_strategy": async_strategy, + "task_id_field": str(execution.get("task_id_field") or "taskId"), "status_field": str(execution.get("status_field") or "status"), "result_field": str(execution.get("result_field") or "result"), "status_mapping": dict(execution.get("status_mapping") or {}), }, poll_interval_seconds=poll_interval_seconds, - next_poll_at=utc_now() if status_url else None, + next_poll_at=None, expires_at=utc_now() + timedelta(seconds=max_tracking_seconds), ) self.db.add(task) self.db.flush() - task.external_task_id = task.id + if not task.idempotency_key: + task.idempotency_key = f"staffdeck:{tool.tenant_id}:{tool.id}:{task.id}" task.accepted_at = utc_now() task.updated_at = task.accepted_at self.db.add(task) @@ -298,12 +334,13 @@ def _detached_acceptance_result( "accepted": True, "detached": True, "status": task.status, - "task_id": task.external_task_id, + "task_id": task.id, "staffdeck_task_id": task.id, + "provider_task_id": task.external_task_id, "status_query": { "method": "GET", "path": ( - f"/api/enterprise/external-business-tasks/{task.external_task_id}" + f"/api/enterprise/external-business-tasks/{task.id}" f"?tenant_id={tool.tenant_id}&tool_id={tool.id}" ), "tenant_id": tool.tenant_id, @@ -313,8 +350,8 @@ def _detached_acceptance_result( ), }, "user_reply": ( - f"已经帮您提交任务，任务号 #{task.external_task_id}，正在后台处理。" - f"您随时可以对我说 “查询 #{task.external_task_id} 状态” 查看结果。" + f"已经帮您提交任务，任务号 #{task.id}，正在后台处理。" + f"您随时可以对我说 “查询 #{task.id} 状态” 查看结果。" ), }, error=None, diff --git a/backend/app/tools/tool_schema.py b/backend/app/tools/tool_schema.py index 6936f6baa..9e3f8c377 100644 --- a/backend/app/tools/tool_schema.py +++ b/backend/app/tools/tool_schema.py @@ -2,7 +2,7 @@ from typing import Any, Literal, Optional -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, model_validator from app.capability_scope import CapabilityScope @@ -10,6 +10,7 @@ class ToolExecutionPolicy(BaseModel): timeout_seconds: float = Field(ge=1, le=3600) execution_mode: Literal["sync", "detached"] = "sync" + async_strategy: Literal["staffdeck_worker", "provider_task"] = "staffdeck_worker" status_url: Optional[str] = None poll_interval_seconds: float = Field(default=5, ge=1, le=3600) task_id_field: str = "taskId" @@ -18,6 +19,16 @@ class ToolExecutionPolicy(BaseModel): status_mapping: dict[str, str] = Field(default_factory=dict) max_tracking_seconds: int = Field(default=86400, ge=1, le=2592000) + @model_validator(mode="after") + def validate_provider_tracking(self) -> "ToolExecutionPolicy": + if ( + self.execution_mode == "detached" + and self.async_strategy == "provider_task" + and not str(self.status_url or "").strip() + ): + raise ValueError("Provider-managed detached tools require status_url") + return self + class ToolCreateRequest(BaseModel): tenant_id: str diff --git a/backend/tests/test_external_business_tasks.py b/backend/tests/test_external_business_tasks.py index 1209583c7..f3fc6fcf2 100644 --- a/backend/tests/test_external_business_tasks.py +++ b/backend/tests/test_external_business_tasks.py @@ -18,6 +18,7 @@ ChatSession, ExternalBusinessTask, ExternalBusinessTaskEvent, + HarnessAgentLoopRecord, HarnessTaskFrameRecord, Tenant, Tool, @@ -72,7 +73,7 @@ def test_detached_submission_persists_task_and_returns_query_guidance(monkeypatc assert result.data["task_id"] == task.id assert f"/external-business-tasks/{task.id}?" in result.data["status_query"]["path"] assert "tenant_id=tenant_demo&tool_id=tool_detached" in result.data["status_query"]["path"] - assert task.external_task_id == task.id + assert task.external_task_id is None assert task.status == "queued" assert "下单任务" not in result.data["status_query"]["guidance"] @@ -93,6 +94,73 @@ def test_detached_submission_does_not_require_provider_task_id(monkeypatch) -> N assert task.status == "queued" +def test_provider_submission_persists_provider_task_id_and_enters_polling(monkeypatch) -> None: + _Client.response_json = {"taskId": "provider-42", "status": "queued"} + _Client.request_headers = {} + monkeypatch.setattr("app.tools.tool_executor.httpx.Client", _Client) + monkeypatch.setattr( + "app.config.get_settings", + lambda: SimpleNamespace( + external_task_callback_base_url="https://staffdeck.example" + ), + ) + with _session() as db: + user, tool = _seed(db) + tool.config_json = { + "execution": { + **dict(tool.config_json["execution"]), + "async_strategy": "provider_task", + } + } + db.add(tool) + db.commit() + + result = ToolExecutor(db).execute( + "tenant_demo", + ToolCall(name=tool.name, arguments={"order": "A-1"}), + user_id=user.id, + ) + task = db.exec(select(ExternalBusinessTask)).one() + assert result.data["task_id"] == task.id + + assert poll_due_external_tasks(db) == 1 + db.refresh(task) + assert task.external_task_id == "provider-42" + assert task.status == "accepted" + assert task.next_poll_at is not None + assert _Client.request_headers["Idempotency-Key"] == task.idempotency_key + assert _Client.request_headers["X-StaffDeck-Callback-URL"] == ( + f"https://staffdeck.example/api/external-business-tasks/{task.id}/callback" + ) + assert _Client.request_headers["X-StaffDeck-Callback-Token"] + + +def test_provider_submission_rejects_accepted_response_without_task_id(monkeypatch) -> None: + _Client.response_json = {"status": "queued"} + monkeypatch.setattr("app.tools.tool_executor.httpx.Client", _Client) + with _session() as db: + user, tool = _seed(db) + tool.config_json = { + "execution": { + **dict(tool.config_json["execution"]), + "async_strategy": "provider_task", + } + } + db.add(tool) + db.commit() + ToolExecutor(db).execute( + "tenant_demo", + ToolCall(name=tool.name, arguments={"order": "A-1"}), + user_id=user.id, + ) + task = db.exec(select(ExternalBusinessTask)).one() + + poll_due_external_tasks(db) + db.refresh(task) + assert task.status == "failed" + assert task.error_json["code"] == "PROVIDER_TASK_ID_MISSING" + + def test_staffdeck_worker_executes_original_sync_tool(monkeypatch) -> None: with _session() as db: user, tool = _seed(db) @@ -118,6 +186,36 @@ def test_staffdeck_worker_executes_original_sync_tool(monkeypatch) -> None: assert task.result_json == {"remaining": 20000} +def test_expired_submission_lease_does_not_replay_business_request(monkeypatch) -> None: + calls: list[dict] = [] + + def execute_again(_self, _tool, arguments, **_kwargs): + calls.append(dict(arguments)) + return ToolResult(tool_name="orders.submit", success=True, data={"duplicate": True}) + + monkeypatch.setattr(ToolExecutor, "execute_sync_http", execute_again) + with _session() as db: + user, tool = _seed(db) + task = ExternalBusinessTask( + tenant_id="tenant_demo", + user_id=user.id, + tool_id=tool.id, + status="working", + request_json={"sku": "SKU-1"}, + callback_token_hash=callback_token_hash("token"), + lease_owner="dead-worker", + lease_expires_at=tool.created_at, + ) + db.add(task) + db.commit() + + assert poll_due_external_tasks(db) == 1 + db.refresh(task) + assert calls == [] + assert task.status == "outcome_unknown" + assert task.error_json["code"] == "DETACHED_OUTCOME_UNKNOWN" + + def test_callback_auth_dedupe_and_user_owned_query() -> None: with _session() as db: user, tool = _seed(db) @@ -156,6 +254,7 @@ def test_callback_auth_dedupe_and_user_owned_query() -> None: result = get_external_business_task("provider-42", "tenant_demo", None, db, user) assert result["status"] == "completed" assert result["result"] == {"receipt": "ok"} + assert get_external_business_task(task.id, "tenant_demo", None, db, user)["id"] == task.id try: get_external_business_task("provider-42", "tenant_demo", None, db, other) except HTTPException as exc: @@ -219,7 +318,34 @@ def test_completed_external_task_marks_sop_frame_ready_to_resume() -> None: step_id="submit_order", slots_json={"employee_id": "14534"}, ) + loop = HarnessAgentLoopRecord( + id="hloop_order", + tenant_id="tenant_demo", + session_id=session.id, + loop_key=f"sop:{frame.id}", + kind="sop", + owner_task_frame_record_id=frame.id, + checkpoint_json={ + "task_frame_id": frame.task_id, + "transcript": [ + { + "role": "tool", + "tool_name": "orders.submit", + "result": { + "success": True, + "data": { + "detached": True, + "task_id": "exttask_order_1", + "status": "queued", + }, + }, + } + ], + }, + ) + frame.agent_loop_id = loop.id task = ExternalBusinessTask( + id="exttask_order_1", tenant_id="tenant_demo", user_id=user.id, session_id=session.id, @@ -231,6 +357,7 @@ def test_completed_external_task_marks_sop_frame_ready_to_resume() -> None: resume_step_id="confirm_order", ) db.add(session) + db.add(loop) db.add(frame) db.add(task) db.commit() @@ -247,11 +374,112 @@ def test_completed_external_task_marks_sop_frame_ready_to_resume() -> None: ) db.refresh(frame) + db.refresh(loop) assert result["status"] == "completed" assert frame.status == "ready_to_resume" assert frame.step_id == "confirm_order" assert frame.slots_json["order_id"] == "ORD-001" assert frame.result_json["external_task_result"] == {"order_id": "ORD-001"} + assert loop.checkpoint_json["external_task_result"]["status"] == "completed" + assert ( + loop.checkpoint_json["transcript"][0]["result"]["data"]["result"] + == {"order_id": "ORD-001"} + ) + + +def test_fast_completion_does_not_release_running_frame_lease() -> None: + with _session() as db: + user, tool = _seed(db) + session = ChatSession( + id="session_running", + tenant_id="tenant_demo", + user_id=user.id, + agent_id="agent_order", + ) + frame = HarnessTaskFrameRecord( + tenant_id="tenant_demo", + session_id=session.id, + source_turn_id="turn_order", + task_id="sop_running", + kind="sop", + status="running", + lease_owner="active-engine", + lease_expires_at=tool.updated_at, + ) + task = ExternalBusinessTask( + tenant_id="tenant_demo", + user_id=user.id, + session_id=session.id, + task_frame_id=frame.task_id, + tool_id=tool.id, + status="accepted", + callback_token_hash=callback_token_hash("token"), + ) + db.add(session) + db.add(frame) + db.add(task) + db.commit() + + external_business_task_callback( + task.id, + ExternalTaskCallback( + event_id="fast-complete-1", + status="completed", + result={"order_id": "ORD-001"}, + ), + "token", + db, + ) + + db.refresh(frame) + assert frame.status == "running" + assert frame.lease_owner == "active-engine" + + +def test_completed_conversation_task_does_not_become_ready_to_resume() -> None: + with _session() as db: + user, tool = _seed(db) + session = ChatSession( + id="session_conversation", + tenant_id="tenant_demo", + user_id=user.id, + agent_id="agent_order", + ) + frame = HarnessTaskFrameRecord( + tenant_id="tenant_demo", + session_id=session.id, + source_turn_id="turn_order", + task_id="conversation_order", + kind="conversation", + status="waiting_external_task", + ) + task = ExternalBusinessTask( + tenant_id="tenant_demo", + user_id=user.id, + session_id=session.id, + task_frame_id=frame.task_id, + tool_id=tool.id, + status="accepted", + callback_token_hash=callback_token_hash("token"), + ) + db.add(session) + db.add(frame) + db.add(task) + db.commit() + + external_business_task_callback( + task.id, + ExternalTaskCallback( + event_id="conversation-complete-1", + status="completed", + result={"order_id": "ORD-001"}, + ), + "token", + db, + ) + + db.refresh(frame) + assert frame.status == "completed" def test_ready_sop_frame_is_queued_on_the_next_turn() -> None: diff --git a/backend/tests/test_harness_v2.py b/backend/tests/test_harness_v2.py index b9a2c38d6..57b77a76c 100644 --- a/backend/tests/test_harness_v2.py +++ b/backend/tests/test_harness_v2.py @@ -4670,27 +4670,19 @@ def generate_json(self, _system_prompt: str, _payload: dict[str, object]): # task finishes: replace the pending receipt in the checkpoint with the # final business result before re-entering the AgentLoop. checkpoint = dict(first.loop_checkpoint) - checkpoint["transcript"] = [ - { - "role": "assistant", - "action": "tool", - "tool_name": "orders.submit", - "arguments": {"sku": "SKU-1"}, - }, - { - "role": "tool", - "tool_name": "orders.submit", - "result": { - "success": True, - "data": { - "task_id": "exttask-order-1", - "status": "completed", - "order_id": "ORD-001", - }, - "pending": False, - }, + transcript = [dict(item) for item in checkpoint["transcript"]] + assert [item["role"] for item in transcript] == ["assistant", "tool"] + assert transcript[1]["result"]["data"]["task_id"] == "exttask-order-1" + transcript[1]["result"] = { + "success": True, + "data": { + "task_id": "exttask-order-1", + "status": "completed", + "order_id": "ORD-001", }, - ] + "pending": False, + } + checkpoint["transcript"] = transcript checkpoint["external_task_result"] = { "task_id": "exttask-order-1", "status": "completed", diff --git a/backend/tests/test_tools_api.py b/backend/tests/test_tools_api.py index de3abb00f..4b8945703 100644 --- a/backend/tests/test_tools_api.py +++ b/backend/tests/test_tools_api.py @@ -176,6 +176,7 @@ def test_tool_config_namespaces_execution_and_preserves_existing_policy() -> Non "execution": { "timeout_seconds": 20.0, "execution_mode": "sync", + "async_strategy": "staffdeck_worker", "status_url": None, "poll_interval_seconds": 5.0, "task_id_field": "taskId", @@ -196,6 +197,7 @@ def test_tool_config_round_trips_detached_execution_policy() -> None: policy = ToolExecutionPolicy( timeout_seconds=12, execution_mode="detached", + async_strategy="provider_task", status_url="https://provider.test/tasks/{taskId}", poll_interval_seconds=10, ) @@ -207,6 +209,15 @@ def test_tool_config_round_trips_detached_execution_policy() -> None: assert _read_execution_policy(config) == policy +def test_provider_detached_policy_requires_status_url() -> None: + with pytest.raises(ValueError, match="status_url"): + ToolExecutionPolicy( + timeout_seconds=12, + execution_mode="detached", + async_strategy="provider_task", + ) + + def test_tool_config_rejects_untyped_execution_and_reads_invalid_legacy_safely() -> None: config = _tool_config( {"tool": "sum", "execution": {"timeout_seconds": 3601}}, diff --git a/frontend-enterprise/src/i18n/en.json b/frontend-enterprise/src/i18n/en.json index 4867abad9..294da09a3 100644 --- a/frontend-enterprise/src/i18n/en.json +++ b/frontend-enterprise/src/i18n/en.json @@ -3364,5 +3364,22 @@ "至少保留一种历史消息角色": "Keep at least one conversation-history role", "摘要前缀不能为空": "Summary prefixes cannot be empty", "历史的信息可以被总结为：": "Earlier conversation history can be summarized as:", - "近期的历史信息总结为：": "Recent conversation history can be summarized as:" + "近期的历史信息总结为：": "Recent conversation history can be summarized as:", + "异步策略": "Async Strategy", + "StaffDeck 托管后台执行": "StaffDeck-managed Background Execution", + "Provider 原生异步任务": "Provider-managed Async Task", + "Provider 原生异步任务需要状态查询 URL": "Provider-managed async tasks require a status URL.", + "StaffDeck 托管异步": "StaffDeck-managed Async Execution", + "StaffDeck 会生成任务号，在后台执行普通同步 HTTP 请求并保存最终结果；Provider 不需要返回 taskId。": "StaffDeck generates a task ID, runs the regular synchronous HTTP request in the background, and saves the final result. The provider does not need to return a taskId.", + "状态查询 URL": "Status URL", + "使用 {taskId} 作为 Provider 任务 ID 占位符。": "Use {taskId} as the provider task ID placeholder.", + "任务 ID 字段": "Task ID Field", + "状态字段": "Status Field", + "结果字段": "Result Field", + "状态映射 JSON": "Status Mapping JSON", + "最长跟踪时间（秒）": "Maximum Tracking Time (seconds)", + "Provider 返回 taskId 后由 StaffDeck 轮询状态；配置外部回调地址后也可接收带任务密钥的回调。": "After the provider returns a taskId, StaffDeck polls its status. When an external callback URL is configured, StaffDeck can also receive callbacks authenticated with a per-task token.", + "提交后当前对话立即结束；普通会话通过任务号查询，SOP 会在任务结束后从持久化检查点继续。": "The current turn ends immediately after submission. Regular conversations query by task ID, while SOPs continue from their durable checkpoint after the task finishes.", + "客服账号名称": "Customer Service Account Name", + "(?:参考来源|参考资料|引用来源|资料来源)": "(?:References|Reference Materials|Citation Sources|Sources)" } diff --git a/frontend-enterprise/src/pages/ToolsPage.async.test.ts b/frontend-enterprise/src/pages/ToolsPage.async.test.ts new file mode 100644 index 000000000..16ca89589 --- /dev/null +++ b/frontend-enterprise/src/pages/ToolsPage.async.test.ts @@ -0,0 +1,47 @@ +import { describe, expect, it } from 'vitest'; + +import { buildToolPayload, TOOL_FORM_INITIAL_VALUES } from './ToolsPage'; + +function values() { + return { + ...TOOL_FORM_INITIAL_VALUES, + name: 'orders.submit', + display_name: '提交订单', + description: '提交订单任务', + allowed_skills: '', + url: 'https://provider.test/tasks', + }; +} + +describe('detached HTTP tool configuration', () => { + it('keeps provider tracking fields for provider-managed tasks', () => { + const payload = buildToolPayload({ + ...values(), + execution_mode: 'detached', + async_strategy: 'provider_task', + status_url: 'https://provider.test/tasks/{taskId}', + }); + + expect(payload?.execution_policy).toMatchObject({ + execution_mode: 'detached', + async_strategy: 'provider_task', + status_url: 'https://provider.test/tasks/{taskId}', + task_id_field: 'taskId', + }); + }); + + it('does not persist provider status URLs for StaffDeck-managed tasks', () => { + const payload = buildToolPayload({ + ...values(), + execution_mode: 'detached', + async_strategy: 'staffdeck_worker', + status_url: 'https://stale.example/tasks/{taskId}', + }); + + expect(payload?.execution_policy).toMatchObject({ + execution_mode: 'detached', + async_strategy: 'staffdeck_worker', + status_url: null, + }); + }); +}); diff --git a/frontend-enterprise/src/pages/ToolsPage.tsx b/frontend-enterprise/src/pages/ToolsPage.tsx index 9b2943c62..6a4ae7681 100644 --- a/frontend-enterprise/src/pages/ToolsPage.tsx +++ b/frontend-enterprise/src/pages/ToolsPage.tsx @@ -91,7 +91,7 @@ type ToolPageProps = { const ENTERPRISE_AGENT_STORAGE_KEY = 'ultrarag_enterprise_agent_scope'; const TOOL_PAGE_SIZE = 10; -const TOOL_FORM_INITIAL_VALUES = { +export const TOOL_FORM_INITIAL_VALUES = { tool_type: 'http' as 'http' | 'a2a' | 'mcp', method: 'POST', enabled: true, @@ -103,6 +103,7 @@ const TOOL_FORM_INITIAL_VALUES = { output_schema: '{}', timeout_seconds: 8, execution_mode: 'sync' as 'sync' | 'detached', + async_strategy: 'staffdeck_worker' as 'staffdeck_worker' | 'provider_task', status_url: '', poll_interval_seconds: 5, task_id_field: 'taskId', @@ -1217,15 +1218,17 @@ function Field({ label, htmlFor, hint, + className, children, }: { label: string; htmlFor?: string; hint?: ReactNode; + className?: string; children: ReactNode; }) { return ( -
+
@@ -2085,22 +2088,97 @@ function ToolFormFields({ {values.execution_mode === 'detached' && (
-
-

StaffDeck 托管异步

-

StaffDeck 会生成任务号，后台执行当前 HTTP 请求并保存结果。Provider 不需要返回 taskId；用户可通过“查询 #任务号”获取结果。

-
- - setField('poll_interval_seconds', Number(event.target.value) || 5)} - /> + + setField('async_strategy', value as 'staffdeck_worker' | 'provider_task')} + > + + + + + StaffDeck 托管后台执行 + Provider 原生异步任务 + + -

提交后当前对话立即结束；任务完成只更新状态，不主动推送消息。用户后续通过任务号查询。

+ + {values.async_strategy === 'staffdeck_worker' ? ( +
+

StaffDeck 托管异步

+

StaffDeck 会生成任务号，在后台执行普通同步 HTTP 请求并保存最终结果；Provider 不需要返回 taskId。

+
+ ) : ( + <> + + setField('status_url', event.target.value)} + /> + + + setField('task_id_field', event.target.value)} + /> + + + setField('status_field', event.target.value)} + /> + + + setField('result_field', event.target.value)} + /> + + + setField('poll_interval_seconds', Number(event.target.value) || 5)} + /> + + +