feat(tools): add detached async execution

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

backend/app/api/external_business_tasks.py

@@ -0,0 +1,125 @@
+from __future__ import annotations
+
+from typing import Any, Literal
+
+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
+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,
+        or_(
+            ExternalBusinessTask.id == external_task_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}

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,
     )
 
 

backend/app/config.py

@@ -15,6 +15,8 @@ class Settings(BaseSettings):
     model_thinking_mode: str = ""
     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

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",

backend/app/core/harness_agent.py

@@ -538,6 +538,45 @@ 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)
+                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,
+                    status="waiting_external_task",
+                    reply_fragment=reply,
+                    capability_results=capability_results,
+                    action_count=iteration,
+                    task_summary=(
+                        "异步业务任务已受理，等待完成后恢复 SOP。"
+                        if requirement.kind == "sop"
+                        else "异步业务任务已受理，可通过任务号查询进度。"
+                    ),
+                    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(

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,29 @@ 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", "未找到属于当前用户的该任务。")
+        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 +1076,8 @@ 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(),
         )
         payload = result.model_dump(mode="json")