From c4f86a73c8c6e14109f175ad7a734206456ba893 Mon Sep 17 00:00:00 2001
From: Dallas98 <990259227@qq.com>
Date: Mon, 13 Jul 2026 10:02:40 +0800
Subject: [PATCH 1/7] feat: implement agent automation features including task
management and proposal handling
---
backend/apps/agent_automation_app.py | 185 ++++
backend/apps/runtime_app.py | 17 +
backend/consts/const.py | 20 +
backend/database/agent_automation_db.py | 400 +++++++++
backend/database/db_models.py | 127 +++
backend/prompts/agent_automation_en.yaml | 38 +
backend/prompts/agent_automation_zh.yaml | 52 ++
backend/pyproject.toml | 1 +
backend/services/agent_automation/__init__.py | 1 +
.../agent_automation/capability_resolver.py | 185 ++++
.../agent_automation/conversation_adapter.py | 82 ++
backend/services/agent_automation/errors.py | 33 +
backend/services/agent_automation/facade.py | 497 ++++++++++
.../agent_automation/intent_parser.py | 172 ++++
backend/services/agent_automation/models.py | 146 +++
.../agent_automation/prompt_generator.py | 217 +++++
backend/services/agent_automation/runner.py | 322 +++++++
.../agent_automation/schedule_engine.py | 113 +++
.../services/agent_automation/scheduler.py | 122 +++
backend/services/agent_service.py | 84 ++
.../conversation_management_service.py | 10 +
backend/utils/prompt_template_utils.py | 4 +
deploy/sql/init.sql | 142 +++
.../v_next_add_agent_automation.sql | 112 +++
.../v_next_add_agent_automation_nav.sql | 37 +
.../backend/agent-automation-task-design.md | 849 ++++++++++++++++++
.../components/AutomationProposalCard.tsx | 91 ++
frontend/app/[locale]/agent-tasks/page.tsx | 474 ++++++++++
.../[locale]/chat/components/chatHeader.tsx | 27 +-
.../chat/components/chatLeftSidebar.tsx | 377 ++++----
.../[locale]/chat/internal/chatInterface.tsx | 279 ++++--
.../chat/streaming/chatStreamFinalMessage.tsx | 75 +-
.../components/navigation/SideNavigation.tsx | 31 +-
frontend/lib/chatMessageExtractor.ts | 24 +-
frontend/public/locales/en/common.json | 2 +
frontend/public/locales/zh/common.json | 2 +
frontend/server.js | 23 +-
frontend/services/agentAutomationService.ts | 143 +++
frontend/services/api.ts | 20 +
frontend/types/agentAutomation.ts | 104 +++
frontend/types/chat.ts | 2 +
test/backend/app/test_agent_automation_app.py | 115 +++
.../database/test_agent_automation_db.py | 69 ++
...t_agent_automation_conversation_adapter.py | 69 ++
...test_agent_automation_conversation_hook.py | 30 +
.../services/test_agent_automation_facade.py | 522 +++++++++++
.../test_agent_automation_intent_parser.py | 66 ++
.../test_agent_automation_lifecycle.py | 250 ++++++
.../test_agent_automation_prompt_generator.py | 53 ++
.../services/test_agent_automation_runner.py | 498 ++++++++++
.../test_agent_automation_schedule_engine.py | 90 ++
.../test_agent_automation_scheduler.py | 60 ++
test/backend/services/test_agent_service.py | 39 +
53 files changed, 7225 insertions(+), 278 deletions(-)
create mode 100644 backend/apps/agent_automation_app.py
create mode 100644 backend/database/agent_automation_db.py
create mode 100644 backend/prompts/agent_automation_en.yaml
create mode 100644 backend/prompts/agent_automation_zh.yaml
create mode 100644 backend/services/agent_automation/__init__.py
create mode 100644 backend/services/agent_automation/capability_resolver.py
create mode 100644 backend/services/agent_automation/conversation_adapter.py
create mode 100644 backend/services/agent_automation/errors.py
create mode 100644 backend/services/agent_automation/facade.py
create mode 100644 backend/services/agent_automation/intent_parser.py
create mode 100644 backend/services/agent_automation/models.py
create mode 100644 backend/services/agent_automation/prompt_generator.py
create mode 100644 backend/services/agent_automation/runner.py
create mode 100644 backend/services/agent_automation/schedule_engine.py
create mode 100644 backend/services/agent_automation/scheduler.py
create mode 100644 deploy/sql/migrations/v_next_add_agent_automation.sql
create mode 100644 deploy/sql/migrations/v_next_add_agent_automation_nav.sql
create mode 100644 doc/docs/zh/backend/agent-automation-task-design.md
create mode 100644 frontend/app/[locale]/agent-tasks/components/AutomationProposalCard.tsx
create mode 100644 frontend/app/[locale]/agent-tasks/page.tsx
create mode 100644 frontend/services/agentAutomationService.ts
create mode 100644 frontend/types/agentAutomation.ts
create mode 100644 test/backend/app/test_agent_automation_app.py
create mode 100644 test/backend/database/test_agent_automation_db.py
create mode 100644 test/backend/services/test_agent_automation_conversation_adapter.py
create mode 100644 test/backend/services/test_agent_automation_conversation_hook.py
create mode 100644 test/backend/services/test_agent_automation_facade.py
create mode 100644 test/backend/services/test_agent_automation_intent_parser.py
create mode 100644 test/backend/services/test_agent_automation_lifecycle.py
create mode 100644 test/backend/services/test_agent_automation_prompt_generator.py
create mode 100644 test/backend/services/test_agent_automation_runner.py
create mode 100644 test/backend/services/test_agent_automation_schedule_engine.py
create mode 100644 test/backend/services/test_agent_automation_scheduler.py
diff --git a/backend/apps/agent_automation_app.py b/backend/apps/agent_automation_app.py
new file mode 100644
index 000000000..e5503d5e4
--- /dev/null
+++ b/backend/apps/agent_automation_app.py
@@ -0,0 +1,185 @@
+import logging
+from http import HTTPStatus
+from typing import Optional
+
+from fastapi import APIRouter, Header, HTTPException, Query
+
+from consts.exceptions import UnauthorizedError
+from services.agent_automation.errors import (
+ AgentAutomationError,
+ AutomationConversationAlreadyBoundError,
+ AutomationNotFoundError,
+)
+from services.agent_automation.facade import agent_automation_facade
+from services.agent_automation.models import (
+ AutomationProposalConfirmRequest,
+ AutomationProposalCreateRequest,
+ AutomationResponse,
+ AutomationTaskPatchRequest,
+)
+from utils.auth_utils import get_current_user_id
+
+logger = logging.getLogger("agent_automation_app")
+
+router = APIRouter(prefix="/agent/automations")
+conversation_automation_router = APIRouter(prefix="/conversation")
+
+
+def _get_current_user(authorization: Optional[str]) -> tuple[str, str]:
+ try:
+ return get_current_user_id(authorization)
+ except UnauthorizedError as exc:
+ raise HTTPException(status_code=HTTPStatus.UNAUTHORIZED, detail=str(exc)) from exc
+
+
+def _map_error(exc: AgentAutomationError) -> HTTPException:
+ if isinstance(exc, AutomationNotFoundError):
+ status = HTTPStatus.NOT_FOUND
+ elif isinstance(exc, AutomationConversationAlreadyBoundError):
+ status = HTTPStatus.CONFLICT
+ else:
+ status = HTTPStatus.BAD_REQUEST
+ return HTTPException(
+ status_code=status,
+ detail={
+ "code": exc.error_code,
+ "message": exc.message,
+ "details": exc.details,
+ },
+ )
+
+
+@router.post("/proposals", response_model=AutomationResponse)
+async def create_proposal(request: AutomationProposalCreateRequest, authorization: Optional[str] = Header(None)):
+ try:
+ user_id, tenant_id = _get_current_user(authorization)
+ data = await agent_automation_facade.create_proposal(request, tenant_id, user_id)
+ return AutomationResponse(data=data)
+ except AgentAutomationError as exc:
+ raise _map_error(exc)
+ except HTTPException:
+ raise
+ except Exception as exc:
+ logger.error("Failed to create automation proposal: %s", exc, exc_info=True)
+ raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(exc))
+
+
+@router.post("/proposals/{proposal_id}/confirm", response_model=AutomationResponse)
+async def confirm_proposal(
+ proposal_id: int,
+ request: AutomationProposalConfirmRequest | None = None,
+ authorization: Optional[str] = Header(None),
+):
+ try:
+ user_id, tenant_id = _get_current_user(authorization)
+ data = await agent_automation_facade.confirm_proposal(
+ proposal_id,
+ request or AutomationProposalConfirmRequest(),
+ tenant_id,
+ user_id,
+ )
+ return AutomationResponse(data=data)
+ except AgentAutomationError as exc:
+ raise _map_error(exc)
+ except HTTPException:
+ raise
+ except Exception as exc:
+ logger.error("Failed to confirm automation proposal: %s", exc, exc_info=True)
+ raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(exc))
+
+
+@router.get("", response_model=AutomationResponse)
+async def list_tasks(status: Optional[str] = Query(default=None), authorization: Optional[str] = Header(None)):
+ try:
+ user_id, tenant_id = _get_current_user(authorization)
+ return AutomationResponse(data=agent_automation_facade.list_tasks(tenant_id, user_id, status))
+ except HTTPException:
+ raise
+ except Exception as exc:
+ logger.error("Failed to list automation tasks: %s", exc, exc_info=True)
+ raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(exc))
+
+
+@router.get("/{task_id}", response_model=AutomationResponse)
+async def get_task(task_id: int, authorization: Optional[str] = Header(None)):
+ try:
+ user_id, tenant_id = _get_current_user(authorization)
+ return AutomationResponse(data=agent_automation_facade.get_task(task_id, tenant_id, user_id))
+ except AgentAutomationError as exc:
+ raise _map_error(exc)
+
+
+@router.patch("/{task_id}", response_model=AutomationResponse)
+async def patch_task(task_id: int, request: AutomationTaskPatchRequest, authorization: Optional[str] = Header(None)):
+ try:
+ user_id, tenant_id = _get_current_user(authorization)
+ data = await agent_automation_facade.patch_task(task_id, request, tenant_id, user_id)
+ return AutomationResponse(data=data)
+ except AgentAutomationError as exc:
+ raise _map_error(exc)
+
+
+@router.post("/{task_id}/pause", response_model=AutomationResponse)
+async def pause_task(task_id: int, authorization: Optional[str] = Header(None)):
+ try:
+ user_id, tenant_id = _get_current_user(authorization)
+ return AutomationResponse(data=agent_automation_facade.pause_task(task_id, tenant_id, user_id))
+ except AgentAutomationError as exc:
+ raise _map_error(exc)
+
+
+@router.post("/{task_id}/resume", response_model=AutomationResponse)
+async def resume_task(task_id: int, authorization: Optional[str] = Header(None)):
+ try:
+ user_id, tenant_id = _get_current_user(authorization)
+ return AutomationResponse(data=agent_automation_facade.resume_task(task_id, tenant_id, user_id))
+ except AgentAutomationError as exc:
+ raise _map_error(exc)
+
+
+@router.post("/{task_id}/run", response_model=AutomationResponse)
+async def run_task_now(task_id: int, authorization: Optional[str] = Header(None)):
+ try:
+ user_id, tenant_id = _get_current_user(authorization)
+ return AutomationResponse(data=await agent_automation_facade.run_task_now(task_id, tenant_id, user_id))
+ except AgentAutomationError as exc:
+ raise _map_error(exc)
+
+
+@router.delete("/{task_id}", response_model=AutomationResponse)
+async def delete_task(task_id: int, authorization: Optional[str] = Header(None)):
+ try:
+ user_id, tenant_id = _get_current_user(authorization)
+ return AutomationResponse(data=agent_automation_facade.delete_task(task_id, tenant_id, user_id))
+ except AgentAutomationError as exc:
+ raise _map_error(exc)
+
+
+@router.get("/{task_id}/runs", response_model=AutomationResponse)
+async def list_runs(task_id: int, authorization: Optional[str] = Header(None)):
+ try:
+ user_id, tenant_id = _get_current_user(authorization)
+ return AutomationResponse(data=agent_automation_facade.list_runs(task_id, tenant_id, user_id))
+ except AgentAutomationError as exc:
+ raise _map_error(exc)
+
+
+@router.post("/runs/{run_id}/cancel", response_model=AutomationResponse)
+async def cancel_run(run_id: int, authorization: Optional[str] = Header(None)):
+ try:
+ user_id, tenant_id = _get_current_user(authorization)
+ return AutomationResponse(data=agent_automation_facade.cancel_run(run_id, tenant_id, user_id))
+ except AgentAutomationError as exc:
+ raise _map_error(exc)
+
+
+@conversation_automation_router.get("/{conversation_id}/automation", response_model=AutomationResponse)
+async def get_conversation_automation(conversation_id: int, authorization: Optional[str] = Header(None)):
+ try:
+ user_id, tenant_id = _get_current_user(authorization)
+ return AutomationResponse(data=agent_automation_facade.get_task_for_conversation(conversation_id, user_id))
+ except HTTPException:
+ raise
+ except Exception as exc:
+ logger.error("Failed to get conversation automation: %s", exc, exc_info=True)
+ raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(exc))
diff --git a/backend/apps/runtime_app.py b/backend/apps/runtime_app.py
index 49fe1286b..d623b14e0 100644
--- a/backend/apps/runtime_app.py
+++ b/backend/apps/runtime_app.py
@@ -2,6 +2,7 @@
from apps.app_factory import create_app
from apps.agent_app import agent_runtime_router as agent_router
+from apps.agent_automation_app import conversation_automation_router, router as agent_automation_router
from apps.voice_app import voice_runtime_router as voice_router
from apps.conversation_management_app import router as conversation_management_router
from apps.conversation_share_app import router as conversation_share_router
@@ -20,9 +21,25 @@
app.add_middleware(ExceptionHandlerMiddleware)
app.include_router(agent_router)
+app.include_router(agent_automation_router)
+app.include_router(conversation_automation_router)
app.include_router(conversation_management_router)
app.include_router(conversation_share_router)
app.include_router(memory_config_router)
app.include_router(file_management_router)
app.include_router(voice_router)
app.include_router(skill_creator_router)
+
+
+@app.on_event("startup")
+async def start_agent_automation_scheduler():
+ from services.agent_automation.scheduler import agent_automation_scheduler
+
+ await agent_automation_scheduler.start()
+
+
+@app.on_event("shutdown")
+async def stop_agent_automation_scheduler():
+ from services.agent_automation.scheduler import agent_automation_scheduler
+
+ await agent_automation_scheduler.stop()
diff --git a/backend/consts/const.py b/backend/consts/const.py
index cf950d109..1259c3f99 100644
--- a/backend/consts/const.py
+++ b/backend/consts/const.py
@@ -46,6 +46,26 @@ class VectorDatabaseType(str, Enum):
PER_WAVE_TIMEOUT = int(os.getenv("DP_SPLIT_WAIT_TIMEOUT_PER_WAVE_S", "30"))
MAX_TIMEOUT = int(os.getenv("DP_SPLIT_WAIT_TIMEOUT_MAX_S", "1800"))
+# Agent automation runtime configuration
+AGENT_AUTOMATION_ENABLED = os.getenv(
+ "AGENT_AUTOMATION_ENABLED", "true"
+).lower() in ("true", "1", "yes", "on")
+AGENT_AUTOMATION_POLL_INTERVAL_SECONDS = int(
+ os.getenv("AGENT_AUTOMATION_POLL_INTERVAL_SECONDS", "30")
+)
+AGENT_AUTOMATION_MAX_CONCURRENT_RUNS = int(
+ os.getenv("AGENT_AUTOMATION_MAX_CONCURRENT_RUNS", "2")
+)
+AGENT_AUTOMATION_LEASE_SECONDS = int(
+ os.getenv("AGENT_AUTOMATION_LEASE_SECONDS", "120")
+)
+AGENT_AUTOMATION_DEFAULT_TIMEOUT_SECONDS = int(
+ os.getenv("AGENT_AUTOMATION_DEFAULT_TIMEOUT_SECONDS", "1800")
+)
+AGENT_AUTOMATION_MIN_INTERVAL_SECONDS = int(
+ os.getenv("AGENT_AUTOMATION_MIN_INTERVAL_SECONDS", "300")
+)
+
# Container-internal skills storage path
CONTAINER_SKILLS_PATH = os.getenv("SKILLS_PATH")
diff --git a/backend/database/agent_automation_db.py b/backend/database/agent_automation_db.py
new file mode 100644
index 000000000..8a336d3d5
--- /dev/null
+++ b/backend/database/agent_automation_db.py
@@ -0,0 +1,400 @@
+from datetime import datetime, timedelta, timezone
+from typing import Any, Dict, List, Optional
+
+from sqlalchemy import desc, insert, select, text, update
+
+from .client import as_dict, get_db_session
+from .db_models import AgentAutomationProposal, AgentAutomationRun, AgentAutomationTask
+from .utils import add_creation_tracking, add_update_tracking
+
+
+def _utcnow() -> datetime:
+ return datetime.now(timezone.utc)
+
+
+def create_task(task_data: Dict[str, Any], user_id: str) -> Dict[str, Any]:
+ data = {
+ **task_data,
+ "delete_flag": "N",
+ }
+ data = add_creation_tracking(data, user_id)
+ with get_db_session() as session:
+ stmt = insert(AgentAutomationTask).values(**data).returning(AgentAutomationTask)
+ task = session.execute(stmt).scalar_one()
+ return as_dict(task)
+
+
+def get_task(task_id: int, tenant_id: str, user_id: str) -> Optional[Dict[str, Any]]:
+ with get_db_session() as session:
+ task = session.execute(
+ select(AgentAutomationTask).where(
+ AgentAutomationTask.task_id == task_id,
+ AgentAutomationTask.tenant_id == tenant_id,
+ AgentAutomationTask.user_id == user_id,
+ AgentAutomationTask.delete_flag == "N",
+ )
+ ).scalar_one_or_none()
+ return as_dict(task) if task else None
+
+
+def get_task_by_conversation(
+ conversation_id: int,
+ user_id: str,
+ include_deleted: bool = False,
+) -> Optional[Dict[str, Any]]:
+ with get_db_session() as session:
+ conditions = [
+ AgentAutomationTask.conversation_id == conversation_id,
+ AgentAutomationTask.user_id == user_id,
+ ]
+ if not include_deleted:
+ conditions.extend([
+ AgentAutomationTask.delete_flag == "N",
+ AgentAutomationTask.status != "DELETED",
+ ])
+ task = session.execute(select(AgentAutomationTask).where(*conditions)).scalar_one_or_none()
+ return as_dict(task) if task else None
+
+
+def list_tasks(tenant_id: str, user_id: str, status: Optional[str] = None) -> List[Dict[str, Any]]:
+ with get_db_session() as session:
+ conditions = [
+ AgentAutomationTask.tenant_id == tenant_id,
+ AgentAutomationTask.user_id == user_id,
+ AgentAutomationTask.delete_flag == "N",
+ AgentAutomationTask.status != "DELETED",
+ ]
+ if status:
+ conditions.append(AgentAutomationTask.status == status)
+ rows = session.execute(
+ select(AgentAutomationTask)
+ .where(*conditions)
+ .order_by(desc(AgentAutomationTask.update_time))
+ ).scalars().all()
+ return [as_dict(row) for row in rows]
+
+
+def update_task(task_id: int, tenant_id: str, user_id: str, values: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+ data = add_update_tracking({
+ **values,
+ "update_time": _utcnow(),
+ }, user_id)
+ with get_db_session() as session:
+ task = session.execute(
+ update(AgentAutomationTask)
+ .where(
+ AgentAutomationTask.task_id == task_id,
+ AgentAutomationTask.tenant_id == tenant_id,
+ AgentAutomationTask.user_id == user_id,
+ AgentAutomationTask.delete_flag == "N",
+ )
+ .values(**data)
+ .returning(AgentAutomationTask)
+ ).scalar_one_or_none()
+ return as_dict(task) if task else None
+
+
+def soft_delete_task(task_id: int, tenant_id: str, user_id: str) -> bool:
+ result = update_task(task_id, tenant_id, user_id, {
+ "status": "DELETED",
+ "delete_flag": "Y",
+ "lock_owner": None,
+ "lock_until": None,
+ })
+ return result is not None
+
+
+def soft_delete_task_by_conversation(conversation_id: int, user_id: str) -> int:
+ with get_db_session() as session:
+ result = session.execute(
+ update(AgentAutomationTask)
+ .where(
+ AgentAutomationTask.conversation_id == conversation_id,
+ AgentAutomationTask.user_id == user_id,
+ AgentAutomationTask.delete_flag == "N",
+ )
+ .values(
+ status="DELETED",
+ delete_flag="Y",
+ lock_owner=None,
+ lock_until=None,
+ update_time=_utcnow(),
+ updated_by=user_id,
+ )
+ )
+ return result.rowcount or 0
+
+
+def create_proposal(proposal_data: Dict[str, Any], user_id: str) -> Dict[str, Any]:
+ data = add_creation_tracking({**proposal_data, "delete_flag": "N"}, user_id)
+ with get_db_session() as session:
+ stmt = insert(AgentAutomationProposal).values(**data).returning(AgentAutomationProposal)
+ proposal = session.execute(stmt).scalar_one()
+ return as_dict(proposal)
+
+
+def get_proposal(proposal_id: int, tenant_id: str, user_id: str) -> Optional[Dict[str, Any]]:
+ with get_db_session() as session:
+ proposal = session.execute(
+ select(AgentAutomationProposal).where(
+ AgentAutomationProposal.proposal_id == proposal_id,
+ AgentAutomationProposal.tenant_id == tenant_id,
+ AgentAutomationProposal.user_id == user_id,
+ AgentAutomationProposal.delete_flag == "N",
+ )
+ ).scalar_one_or_none()
+ return as_dict(proposal) if proposal else None
+
+
+def update_proposal_status(proposal_id: int, tenant_id: str, user_id: str, status: str) -> bool:
+ with get_db_session() as session:
+ result = session.execute(
+ update(AgentAutomationProposal)
+ .where(
+ AgentAutomationProposal.proposal_id == proposal_id,
+ AgentAutomationProposal.tenant_id == tenant_id,
+ AgentAutomationProposal.user_id == user_id,
+ AgentAutomationProposal.delete_flag == "N",
+ )
+ .values(status=status, update_time=_utcnow(), updated_by=user_id)
+ )
+ return bool(result.rowcount)
+
+
+def update_proposal_task(
+ proposal_id: int,
+ tenant_id: str,
+ user_id: str,
+ proposed_task: Dict[str, Any],
+) -> bool:
+ with get_db_session() as session:
+ result = session.execute(
+ update(AgentAutomationProposal)
+ .where(
+ AgentAutomationProposal.proposal_id == proposal_id,
+ AgentAutomationProposal.tenant_id == tenant_id,
+ AgentAutomationProposal.user_id == user_id,
+ AgentAutomationProposal.delete_flag == "N",
+ )
+ .values(
+ proposed_task=proposed_task,
+ update_time=_utcnow(),
+ updated_by=user_id,
+ )
+ )
+ return bool(result.rowcount)
+
+
+def create_run(run_data: Dict[str, Any], user_id: str) -> Dict[str, Any]:
+ data = add_creation_tracking({**run_data, "delete_flag": "N"}, user_id)
+ with get_db_session() as session:
+ stmt = insert(AgentAutomationRun).values(**data).returning(AgentAutomationRun)
+ run = session.execute(stmt).scalar_one()
+ return as_dict(run)
+
+
+def update_run(
+ run_id: int,
+ values: Dict[str, Any],
+ user_id: Optional[str] = None,
+ expected_statuses: Optional[List[str]] = None,
+) -> Optional[Dict[str, Any]]:
+ data = {
+ **values,
+ "update_time": _utcnow(),
+ }
+ if user_id:
+ data = add_update_tracking(data, user_id)
+ with get_db_session() as session:
+ conditions = [
+ AgentAutomationRun.run_id == run_id,
+ AgentAutomationRun.delete_flag == "N",
+ ]
+ if expected_statuses:
+ conditions.append(AgentAutomationRun.status.in_(expected_statuses))
+ run = session.execute(
+ update(AgentAutomationRun)
+ .where(*conditions)
+ .values(**data)
+ .returning(AgentAutomationRun)
+ ).scalar_one_or_none()
+ return as_dict(run) if run else None
+
+
+def get_run(run_id: int, tenant_id: str, user_id: str) -> Optional[Dict[str, Any]]:
+ with get_db_session() as session:
+ run = session.execute(
+ select(AgentAutomationRun).where(
+ AgentAutomationRun.run_id == run_id,
+ AgentAutomationRun.tenant_id == tenant_id,
+ AgentAutomationRun.user_id == user_id,
+ AgentAutomationRun.delete_flag == "N",
+ )
+ ).scalar_one_or_none()
+ return as_dict(run) if run else None
+
+
+def cancel_run(run_id: int, tenant_id: str, user_id: str, reason: str) -> Optional[Dict[str, Any]]:
+ with get_db_session() as session:
+ run = session.execute(
+ update(AgentAutomationRun)
+ .where(
+ AgentAutomationRun.run_id == run_id,
+ AgentAutomationRun.tenant_id == tenant_id,
+ AgentAutomationRun.user_id == user_id,
+ AgentAutomationRun.status.in_(["QUEUED", "RUNNING"]),
+ AgentAutomationRun.delete_flag == "N",
+ )
+ .values(
+ status="CANCELED",
+ error_code="AUTOMATION_RUN_CANCELED",
+ error_message=reason,
+ finished_at=_utcnow(),
+ update_time=_utcnow(),
+ updated_by=user_id,
+ )
+ .returning(AgentAutomationRun)
+ ).scalar_one_or_none()
+ return as_dict(run) if run else None
+
+
+def cancel_runs_by_conversation(conversation_id: int, user_id: str, reason: str) -> int:
+ with get_db_session() as session:
+ result = session.execute(
+ update(AgentAutomationRun)
+ .where(
+ AgentAutomationRun.conversation_id == conversation_id,
+ AgentAutomationRun.user_id == user_id,
+ AgentAutomationRun.status.in_(["QUEUED", "RUNNING"]),
+ AgentAutomationRun.delete_flag == "N",
+ )
+ .values(
+ status="CANCELED",
+ error_code="AUTOMATION_RUN_CANCELED",
+ error_message=reason,
+ finished_at=_utcnow(),
+ update_time=_utcnow(),
+ updated_by=user_id,
+ )
+ )
+ return result.rowcount or 0
+
+
+def list_runs(task_id: int, tenant_id: str, user_id: str, limit: int = 50) -> List[Dict[str, Any]]:
+ with get_db_session() as session:
+ rows = session.execute(
+ select(AgentAutomationRun)
+ .where(
+ AgentAutomationRun.task_id == task_id,
+ AgentAutomationRun.tenant_id == tenant_id,
+ AgentAutomationRun.user_id == user_id,
+ AgentAutomationRun.delete_flag == "N",
+ )
+ .order_by(desc(AgentAutomationRun.scheduled_fire_at))
+ .limit(limit)
+ ).scalars().all()
+ return [as_dict(row) for row in rows]
+
+
+def has_active_run_for_conversation(conversation_id: int) -> bool:
+ with get_db_session() as session:
+ run = session.execute(
+ select(AgentAutomationRun.run_id)
+ .where(
+ AgentAutomationRun.conversation_id == conversation_id,
+ AgentAutomationRun.status.in_(["QUEUED", "RUNNING"]),
+ AgentAutomationRun.delete_flag == "N",
+ )
+ .limit(1)
+ ).scalar_one_or_none()
+ return run is not None
+
+
+def claim_due_tasks(instance_id: str, batch_size: int, lease_seconds: int) -> List[Dict[str, Any]]:
+ sql = text("""
+ UPDATE nexent.agent_automation_task_t
+ SET lock_owner = :instance_id,
+ lock_until = now() + (:lease_seconds * interval '1 second'),
+ update_time = now()
+ WHERE task_id IN (
+ SELECT task_id
+ FROM nexent.agent_automation_task_t
+ WHERE delete_flag = 'N'
+ AND status = 'ACTIVE'
+ AND next_fire_at <= now()
+ AND (lock_until IS NULL OR lock_until < now())
+ ORDER BY next_fire_at ASC
+ LIMIT :batch_size
+ FOR UPDATE SKIP LOCKED
+ )
+ RETURNING *
+ """)
+ with get_db_session() as session:
+ rows = session.execute(sql, {
+ "instance_id": instance_id,
+ "batch_size": batch_size,
+ "lease_seconds": lease_seconds,
+ }).fetchall()
+ return [dict(row._mapping) for row in rows]
+
+
+def release_task_lock(task_id: int, lock_owner: Optional[str] = None) -> bool:
+ conditions = [AgentAutomationTask.task_id == task_id]
+ if lock_owner:
+ conditions.append(AgentAutomationTask.lock_owner == lock_owner)
+ with get_db_session() as session:
+ result = session.execute(
+ update(AgentAutomationTask)
+ .where(*conditions)
+ .values(lock_owner=None, lock_until=None, update_time=_utcnow())
+ )
+ return bool(result.rowcount)
+
+
+def renew_task_lock(task_id: int, lock_owner: str, lease_seconds: int) -> bool:
+ with get_db_session() as session:
+ result = session.execute(
+ update(AgentAutomationTask)
+ .where(
+ AgentAutomationTask.task_id == task_id,
+ AgentAutomationTask.lock_owner == lock_owner,
+ AgentAutomationTask.delete_flag == "N",
+ AgentAutomationTask.status == "ACTIVE",
+ )
+ .values(
+ lock_until=_utcnow() + timedelta(seconds=lease_seconds),
+ update_time=_utcnow(),
+ )
+ )
+ return bool(result.rowcount)
+
+
+def recover_stale_runs(timeout_seconds: int) -> int:
+ sql = text("""
+ UPDATE nexent.agent_automation_run_t
+ SET status = 'TIMEOUT',
+ error_code = 'AUTOMATION_RUN_TIMEOUT',
+ error_message = 'Automation run timed out during runtime recovery.',
+ finished_at = now(),
+ update_time = now()
+ WHERE delete_flag = 'N'
+ AND status = 'RUNNING'
+ AND started_at < now() - (:timeout_seconds * interval '1 second')
+ """)
+ with get_db_session() as session:
+ result = session.execute(sql, {"timeout_seconds": timeout_seconds})
+ return result.rowcount or 0
+
+
+def release_expired_locks() -> int:
+ with get_db_session() as session:
+ result = session.execute(
+ update(AgentAutomationTask)
+ .where(
+ AgentAutomationTask.delete_flag == "N",
+ AgentAutomationTask.lock_until.is_not(None),
+ AgentAutomationTask.lock_until < _utcnow(),
+ )
+ .values(lock_owner=None, lock_until=None, update_time=_utcnow())
+ )
+ return result.rowcount or 0
diff --git a/backend/database/db_models.py b/backend/database/db_models.py
index 3c078d9b4..b6bfe6c5d 100644
--- a/backend/database/db_models.py
+++ b/backend/database/db_models.py
@@ -94,6 +94,133 @@ class ConversationMessageUnit(TableBase):
doc="Lifecycle status: streaming (still aggregating) or completed (fully persisted)")
+class AgentAutomationTask(TableBase):
+ """User-managed scheduled automation task bound to one conversation."""
+
+ __tablename__ = "agent_automation_task_t"
+ __table_args__ = (
+ Index(
+ "idx_agent_automation_due",
+ "status",
+ "next_fire_at",
+ postgresql_where=text("delete_flag = 'N'"),
+ ),
+ Index(
+ "idx_agent_automation_owner",
+ "tenant_id",
+ "user_id",
+ "status",
+ postgresql_where=text("delete_flag = 'N'"),
+ ),
+ Index(
+ "uq_agent_automation_conversation_active",
+ "conversation_id",
+ unique=True,
+ postgresql_where=text("delete_flag = 'N' AND status <> 'DELETED'"),
+ ),
+ {"schema": SCHEMA},
+ )
+
+ task_id = Column(BigInteger, Sequence(
+ "agent_automation_task_t_task_id_seq", schema=SCHEMA), primary_key=True, nullable=False)
+ tenant_id = Column(String(100), nullable=False, doc="Tenant ID")
+ user_id = Column(String(100), nullable=False, doc="Owner user ID")
+ conversation_id = Column(BigInteger, nullable=False, doc="Bound conversation ID")
+ agent_id = Column(BigInteger, nullable=False, doc="Bound agent ID")
+ agent_version_no = Column(Integer, nullable=True, doc="Pinned agent version")
+ title = Column(String(255), nullable=False, doc="Task title")
+ instruction = Column(Text, nullable=False, doc="Base instruction for every automation run")
+ status = Column(String(32), nullable=False, doc="Task lifecycle status")
+ source = Column(String(32), nullable=False, doc="Creation source")
+ schedule_mode = Column(String(16), nullable=False, doc="ONCE or RECURRING")
+ schedule_rule_type = Column(String(16), nullable=False, doc="AT, INTERVAL, or CRON")
+ schedule_expr = Column(Text, nullable=True, doc="Display schedule expression")
+ schedule_config = Column(JSONB, nullable=False, doc="Normalized ScheduleTrigger payload")
+ capability_requirements = Column(JSONB, doc="Capability requirements parsed from user intent")
+ capability_bindings = Column(JSONB, doc="Confirmed matched capabilities")
+ runtime_snapshot = Column(JSONB, doc="Agent/runtime capability snapshot at creation time")
+ timezone = Column(String(64), nullable=False, doc="IANA timezone")
+ next_fire_at = Column(TIMESTAMP(timezone=True), nullable=True, doc="Next scheduled fire time")
+ last_fire_at = Column(TIMESTAMP(timezone=True), nullable=True, doc="Last scheduled fire time")
+ fire_count = Column(Integer, default=0, nullable=False, doc="Number of scheduled fires")
+ last_run_status = Column(String(32), nullable=True, doc="Latest run status")
+ last_error = Column(Text, nullable=True, doc="Latest run error")
+ consecutive_failures = Column(Integer, default=0, nullable=False, doc="Consecutive failure count")
+ timeout_seconds = Column(Integer, nullable=False, doc="Single-run timeout")
+ overlap_policy = Column(String(16), nullable=False, doc="Overlap policy")
+ misfire_policy = Column(String(16), nullable=False, doc="Misfire policy")
+ lock_owner = Column(String(128), nullable=True, doc="Scheduler lease owner")
+ lock_until = Column(TIMESTAMP(timezone=True), nullable=True, doc="Scheduler lease expiry")
+
+
+class AgentAutomationRun(TableBase):
+ """Execution history for an automation task fire."""
+
+ __tablename__ = "agent_automation_run_t"
+ __table_args__ = (
+ Index(
+ "idx_agent_automation_run_task",
+ "task_id",
+ "scheduled_fire_at",
+ postgresql_where=text("delete_flag = 'N'"),
+ ),
+ Index(
+ "idx_agent_automation_run_conversation",
+ "conversation_id",
+ "status",
+ postgresql_where=text("delete_flag = 'N'"),
+ ),
+ {"schema": SCHEMA},
+ )
+
+ run_id = Column(BigInteger, Sequence(
+ "agent_automation_run_t_run_id_seq", schema=SCHEMA), primary_key=True, nullable=False)
+ task_id = Column(BigInteger, nullable=False, doc="Automation task ID")
+ tenant_id = Column(String(100), nullable=False, doc="Tenant ID")
+ user_id = Column(String(100), nullable=False, doc="Owner user ID")
+ conversation_id = Column(BigInteger, nullable=False, doc="Bound conversation ID")
+ scheduled_fire_at = Column(TIMESTAMP(timezone=True), nullable=False, doc="Scheduled fire time")
+ actual_fire_at = Column(TIMESTAMP(timezone=True), nullable=True, doc="Actual fire time")
+ trigger_type = Column(String(32), nullable=False, doc="SCHEDULED or MANUAL")
+ status = Column(String(32), nullable=False, doc="Run lifecycle status")
+ generated_prompt = Column(Text, nullable=True, doc="Prompt appended to the conversation")
+ user_message_id = Column(BigInteger, nullable=True, doc="Automation user message ID")
+ assistant_message_id = Column(BigInteger, nullable=True, doc="Assistant message ID")
+ started_at = Column(TIMESTAMP(timezone=True), nullable=True, doc="Run start time")
+ finished_at = Column(TIMESTAMP(timezone=True), nullable=True, doc="Run finish time")
+ duration_ms = Column(BigInteger, nullable=True, doc="Run duration in milliseconds")
+ error_code = Column(String(64), nullable=True, doc="Automation error code")
+ error_message = Column(Text, nullable=True, doc="Automation error message")
+ capability_check = Column(JSONB, nullable=True, doc="Capability check result before execution")
+
+
+class AgentAutomationProposal(TableBase):
+ """Pending automation task proposal created from chat intent."""
+
+ __tablename__ = "agent_automation_proposal_t"
+ __table_args__ = (
+ Index(
+ "idx_agent_automation_proposal_owner",
+ "tenant_id",
+ "user_id",
+ "status",
+ postgresql_where=text("delete_flag = 'N'"),
+ ),
+ {"schema": SCHEMA},
+ )
+
+ proposal_id = Column(BigInteger, Sequence(
+ "agent_automation_proposal_t_proposal_id_seq", schema=SCHEMA), primary_key=True, nullable=False)
+ tenant_id = Column(String(100), nullable=False, doc="Tenant ID")
+ user_id = Column(String(100), nullable=False, doc="Owner user ID")
+ conversation_id = Column(BigInteger, nullable=False, doc="Source conversation ID")
+ agent_id = Column(BigInteger, nullable=False, doc="Bound agent ID")
+ proposed_task = Column(JSONB, nullable=False, doc="Proposed automation task payload")
+ capability_resolution = Column(JSONB, nullable=False, doc="Capability matching result")
+ status = Column(String(32), nullable=False, doc="PENDING, ACCEPTED, REJECTED, or EXPIRED")
+ expires_at = Column(TIMESTAMP(timezone=True), nullable=False, doc="Proposal expiry time")
+
+
class ConversationSourceImage(TableBase):
"""
Holds the search image source information of conversation messages
diff --git a/backend/prompts/agent_automation_en.yaml b/backend/prompts/agent_automation_en.yaml
new file mode 100644
index 000000000..bf58aa8c4
--- /dev/null
+++ b/backend/prompts/agent_automation_en.yaml
@@ -0,0 +1,38 @@
+INSTRUCTION_SYSTEM_PROMPT: |-
+ You design execution-ready prompts for scheduled Agent tasks. Rewrite the user's brief intent as a direct instruction for the selected Agent.
+ Return only the instruction. Preserve the goal, omit scheduling details, use only real listed capabilities, and keep it under 300 words.
+
+INSTRUCTION_USER_PROMPT: |-
+ Original task: {{ instruction }}
+ Agent name: {{ agent_name }}
+ Agent responsibility: {{ agent_description or "Not provided" }}
+ Bound capabilities:
+ {{ capability_summary }}
+
+FALLBACK_INSTRUCTION_PROMPT: |-
+ Act within the responsibilities of "{{ agent_name }}" ({{ agent_description or "No responsibility description provided" }}) and use the current conversation context and configured capabilities to complete this task: {{ instruction }}. Provide a clear, verifiable result and explicitly report missing information or capabilities instead of fabricating them.
+
+EXECUTION_SYSTEM_PROMPT: |-
+ Generate an execution-ready prompt for this scheduled Agent run. Return only the final prompt. Preserve the task goal, use the Agent context and real capabilities, define an appropriate output, and never invent data or tools. Keep it under 500 words.
+
+EXECUTION_USER_PROMPT: |-
+ Task title: {{ title }}
+ Base instruction: {{ instruction }}
+ Agent name: {{ agent_name }}
+ Agent responsibility: {{ agent_description or "Not provided" }}
+ Scheduled time: {{ scheduled_fire_at }}
+ Timezone: {{ timezone }}
+ Trigger type: {{ trigger_type }}
+ Bound capabilities:
+ {{ capability_summary }}
+ Recent conversation context:
+ {{ conversation_context or "No recent messages are available." }}
+
+FALLBACK_EXECUTION_PROMPT: |-
+ Execute the scheduled task "{{ title }}" in this conversation. Act within the responsibilities of "{{ agent_name }}" ({{ agent_description or "No responsibility description provided" }}) and use the current conversation context and configured capabilities to complete: {{ instruction }}
+
+ Scheduled time: {{ scheduled_fire_at }} ({{ timezone }})
+ Bound capabilities:
+ {{ capability_summary }}
+
+ Return a result appropriate to the task and identify key evidence. Explicitly report missing information or capabilities instead of fabricating them.
diff --git a/backend/prompts/agent_automation_zh.yaml b/backend/prompts/agent_automation_zh.yaml
new file mode 100644
index 000000000..0259c3561
--- /dev/null
+++ b/backend/prompts/agent_automation_zh.yaml
@@ -0,0 +1,52 @@
+INSTRUCTION_SYSTEM_PROMPT: |-
+ 你是一名 Agent 自动任务提示词设计专家。请把用户的简短任务意图改写为可直接交给指定 Agent 执行的中文指令。
+
+ 要求:
+ 1. 只输出改写后的任务指令,不要解释,不要使用 Markdown 代码块。
+ 2. 保留用户的业务目标,不重复时间、频率等调度信息。
+ 3. 结合 Agent 的名称、职责和真实可用能力,不得虚构工具、知识库或数据来源。
+ 4. 指令应说明信息依据、输出内容和质量要求;缺少必要信息时要求 Agent 明确说明。
+ 5. 内容简洁,最多 300 个汉字。
+
+INSTRUCTION_USER_PROMPT: |-
+ 用户原始任务:{{ instruction }}
+ Agent 名称:{{ agent_name }}
+ Agent 职责:{{ agent_description or "未提供" }}
+ 已绑定能力:
+ {{ capability_summary }}
+
+FALLBACK_INSTRUCTION_PROMPT: |-
+ 请以「{{ agent_name }}」的职责({{ agent_description or "未提供职责描述" }})为准,基于当前会话上下文和已配置的真实能力完成以下任务:{{ instruction }}。请给出清晰、可核验的结果;如果信息或能力不足,请明确说明,不要编造。
+
+EXECUTION_SYSTEM_PROMPT: |-
+ 你是一名 Agent 自动任务提示词优化专家。请为本次定时触发生成一条可直接发送给指定 Agent 的中文提示词。
+
+ 要求:
+ 1. 只输出最终提示词,不要解释,不要使用 Markdown 代码块。
+ 2. 保持任务原始目标,并结合 Agent 职责、已绑定能力、本次计划时间及当前会话语境。
+ 3. 不得虚构工具、知识库、事实或数据来源。
+ 4. 提示词应明确本次要完成的工作、信息范围、输出结构和异常处理方式。
+ 5. 不要要求固定输出与任务无关的“后续建议”等通用章节。
+ 6. 最多 500 个汉字。
+
+EXECUTION_USER_PROMPT: |-
+ 任务名称:{{ title }}
+ 基础任务指令:{{ instruction }}
+ Agent 名称:{{ agent_name }}
+ Agent 职责:{{ agent_description or "未提供" }}
+ 本次计划时间:{{ scheduled_fire_at }}
+ 时区:{{ timezone }}
+ 触发类型:{{ trigger_type }}
+ 已绑定能力:
+ {{ capability_summary }}
+ 最近会话语境:
+ {{ conversation_context or "当前没有可摘要的历史消息,请根据基础任务指令执行。" }}
+
+FALLBACK_EXECUTION_PROMPT: |-
+ 你正在当前会话中执行自动任务「{{ title }}」。请以「{{ agent_name }}」的职责({{ agent_description or "未提供职责描述" }})为准,基于当前会话上下文和已配置的真实能力完成以下任务:{{ instruction }}
+
+ 本次计划时间:{{ scheduled_fire_at }}({{ timezone }})
+ 已绑定能力:
+ {{ capability_summary }}
+
+ 请直接给出与任务目标匹配的结果,并说明关键信息依据;如果信息或能力不足,请明确说明,不要编造。
diff --git a/backend/pyproject.toml b/backend/pyproject.toml
index 9481cdbc7..78639250b 100644
--- a/backend/pyproject.toml
+++ b/backend/pyproject.toml
@@ -30,6 +30,7 @@ dependencies = [
"pydantic-settings>=2.0.0",
"python-docx>=1.1.0",
"xlrd>=2.0.1",
+ "croniter>=2.0.0",
]
[project.optional-dependencies]
diff --git a/backend/services/agent_automation/__init__.py b/backend/services/agent_automation/__init__.py
new file mode 100644
index 000000000..514e860e2
--- /dev/null
+++ b/backend/services/agent_automation/__init__.py
@@ -0,0 +1 @@
+"""Agent automation domain module."""
diff --git a/backend/services/agent_automation/capability_resolver.py b/backend/services/agent_automation/capability_resolver.py
new file mode 100644
index 000000000..6f99c13f7
--- /dev/null
+++ b/backend/services/agent_automation/capability_resolver.py
@@ -0,0 +1,185 @@
+import re
+from typing import Any, Dict, Iterable, List, Optional
+
+from .models import CapabilityBinding, CapabilityResolution, CapabilityType
+
+
+SEARCH_HINTS = ("联网", "网络", "搜索", "新闻", "网页", "竞品", "公开")
+KNOWLEDGE_HINTS = ("知识库", "文档", "资料", "项目", "销售线索")
+
+
+def _safe_text(*parts: Any) -> str:
+ return " ".join(str(part or "") for part in parts).lower()
+
+
+def _tool_binding(tool: Any) -> CapabilityBinding:
+ class_name = getattr(tool, "class_name", "") or ""
+ name = getattr(tool, "name", None) or class_name
+ metadata = getattr(tool, "metadata", None) or {}
+ if class_name == "KnowledgeBaseSearchTool":
+ index_names = (getattr(tool, "params", None) or {}).get("index_names", [])
+ display_map = metadata.get("index_name_to_display_map", {})
+ display = ", ".join(display_map.get(index, index) for index in index_names) if index_names else name
+ return CapabilityBinding(
+ type=CapabilityType.KNOWLEDGE_BASE,
+ name=",".join(index_names) if index_names else name,
+ display_name=display,
+ binding_ref=f"tool:KnowledgeBaseSearchTool:index:{','.join(index_names)}",
+ reason="Agent has a configured knowledge-base search tool.",
+ )
+ return CapabilityBinding(
+ type=CapabilityType.TOOL,
+ name=name,
+ display_name=name,
+ binding_ref=f"tool:{name}",
+ reason=getattr(tool, "description", None) or "Agent has this tool configured.",
+ )
+
+
+def _skill_binding(skill: Dict[str, Any]) -> CapabilityBinding:
+ name = skill.get("name") or ""
+ return CapabilityBinding(
+ type=CapabilityType.SKILL,
+ name=name,
+ display_name=name,
+ binding_ref=f"skill:{name}",
+ reason=skill.get("description") or "Agent has this skill enabled.",
+ )
+
+
+def _agent_binding(agent: Any, capability_type: CapabilityType) -> CapabilityBinding:
+ name = getattr(agent, "name", None) or getattr(agent, "agent_id", "")
+ return CapabilityBinding(
+ type=capability_type,
+ name=str(name),
+ display_name=str(name),
+ binding_ref=f"{capability_type.value.lower()}:{name}",
+ reason=getattr(agent, "description", None) or "Agent is available as a callable capability.",
+ )
+
+
+def _flatten_bindings(bindings: Iterable[CapabilityBinding]) -> Dict[str, CapabilityBinding]:
+ return {binding.binding_ref: binding for binding in bindings}
+
+
+async def resolve_agent_capabilities(
+ agent_id: int,
+ tenant_id: str,
+ user_id: str,
+ instruction: str,
+ version_no: int = 0,
+) -> CapabilityResolution:
+ """Resolve capabilities from the same assembly path used by normal agent runs."""
+ from agents.create_agent_info import create_agent_config
+ from services.skill_service import SkillService
+
+ agent_config = await create_agent_config(
+ agent_id=agent_id,
+ tenant_id=tenant_id,
+ user_id=user_id,
+ last_user_query=instruction,
+ version_no=version_no,
+ allow_memory_search=False,
+ )
+
+ tool_bindings = [_tool_binding(tool) for tool in getattr(agent_config, "tools", []) or []]
+ skill_bindings = []
+ try:
+ enabled_skills = SkillService().get_enabled_skills_for_agent(
+ agent_id=agent_id,
+ tenant_id=tenant_id,
+ version_no=version_no,
+ )
+ skill_bindings.extend(_skill_binding(skill) for skill in enabled_skills)
+ except Exception:
+ enabled_skills = []
+
+ # Skills may also be carried in context components for context-manager agents.
+ for component in getattr(agent_config, "context_components", []) or []:
+ if getattr(component, "component_type", None) == "skills":
+ for skill in getattr(component, "skills", []) or []:
+ skill_bindings.append(_skill_binding(skill))
+
+ managed_bindings = [
+ _agent_binding(agent, CapabilityType.MANAGED_AGENT)
+ for agent in getattr(agent_config, "managed_agents", []) or []
+ ]
+ a2a_bindings = [
+ _agent_binding(agent, CapabilityType.EXTERNAL_A2A_AGENT)
+ for agent in getattr(agent_config, "external_a2a_agents", []) or []
+ ]
+
+ all_bindings = tool_bindings + skill_bindings + managed_bindings + a2a_bindings
+ lower_instruction = instruction.lower()
+ missing: List[Dict[str, Any]] = []
+
+ has_search_tool = any(
+ binding.type == CapabilityType.TOOL
+ and re.search(r"search|linkup|exa|web|联网|搜索", binding.name.lower())
+ for binding in all_bindings
+ )
+ if any(hint in instruction for hint in SEARCH_HINTS) and not has_search_tool:
+ missing.append({
+ "type": CapabilityType.TOOL.value,
+ "name": "web_search",
+ "suggestion": (
+ "请先为该 Agent 启用联网搜索工具,"
+ "或删除任务中的联网/新闻/网页检索要求。"
+ ),
+ })
+
+ has_knowledge = any(binding.type == CapabilityType.KNOWLEDGE_BASE for binding in all_bindings)
+ if any(hint in instruction for hint in KNOWLEDGE_HINTS) and not has_knowledge:
+ missing.append({
+ "type": CapabilityType.KNOWLEDGE_BASE.value,
+ "name": "knowledge_base",
+ "suggestion": "请先为该 Agent 选择知识库,或修改任务为仅基于当前会话执行。",
+ })
+
+ matched = []
+ for binding in all_bindings:
+ text = _safe_text(binding.name, binding.display_name, binding.reason)
+ instruction_tokens = re.findall(r"[\w\u4e00-\u9fff]+", lower_instruction)
+ if not lower_instruction or any(token in text for token in instruction_tokens):
+ matched.append(binding)
+
+ if not matched:
+ matched = all_bindings[:10]
+
+ return CapabilityResolution(
+ matched_capabilities=matched[:20],
+ missing_capabilities=missing,
+ agent_snapshot={
+ "agent_id": agent_id,
+ "version_no": version_no,
+ "name": getattr(agent_config, "name", ""),
+ "description": getattr(agent_config, "description", ""),
+ "tools_count": len(tool_bindings),
+ "skills_count": len(skill_bindings),
+ "managed_agents_count": len(managed_bindings),
+ "external_a2a_agents_count": len(a2a_bindings),
+ },
+ executable=len(missing) == 0,
+ )
+
+
+async def validate_bindings_available(
+ agent_id: int,
+ tenant_id: str,
+ user_id: str,
+ instruction: str,
+ bindings: List[Dict[str, Any]],
+ version_no: int = 0,
+) -> Dict[str, Any]:
+ resolution = await resolve_agent_capabilities(agent_id, tenant_id, user_id, instruction, version_no)
+ available = _flatten_bindings(resolution.matched_capabilities)
+ unavailable = []
+ for binding in bindings or []:
+ ref = binding.get("binding_ref")
+ if ref and ref not in available:
+ unavailable.append(binding)
+ return {
+ "available": not unavailable and resolution.executable,
+ "unavailable_bindings": unavailable,
+ "resolution": resolution.model_dump(mode="json"),
+ }
diff --git a/backend/services/agent_automation/conversation_adapter.py b/backend/services/agent_automation/conversation_adapter.py
new file mode 100644
index 000000000..fd412baab
--- /dev/null
+++ b/backend/services/agent_automation/conversation_adapter.py
@@ -0,0 +1,82 @@
+import json
+import logging
+from typing import Any, Dict, Optional
+
+from consts.model import MessageRequest, MessageUnit
+from services.conversation_management_service import (
+ get_conversation_history_service,
+ save_message,
+ save_message_unit,
+ update_unit_content,
+)
+
+logger = logging.getLogger("agent_automation.conversation_adapter")
+
+
+class AutomationConversationAdapter:
+ """Persist automation UI events through the existing conversation service."""
+
+ def append_proposal_exchange(
+ self,
+ conversation_id: int,
+ user_instruction: str,
+ payload: Dict[str, Any],
+ user_id: str,
+ tenant_id: str,
+ ) -> Dict[str, int]:
+ history_payload = get_conversation_history_service(conversation_id, user_id)
+ messages = history_payload[0].get("message", []) if history_payload else []
+ user_request = MessageRequest(
+ conversation_id=conversation_id,
+ message_idx=len(messages),
+ role="user",
+ message=[MessageUnit(type="string", content=user_instruction)],
+ )
+ user_message_id = save_message(user_request, user_id, tenant_id)
+ user_unit_id = save_message_unit(
+ message_id=user_message_id,
+ conversation_id=conversation_id,
+ unit_index=0,
+ unit_type="string",
+ unit_content=user_instruction,
+ user_id=user_id,
+ )
+ content = json.dumps(payload, ensure_ascii=False)
+ request = MessageRequest(
+ conversation_id=conversation_id,
+ message_idx=len(messages) + 1,
+ role="assistant",
+ message=[MessageUnit(type="automation_proposal", content=content)],
+ )
+ message_id = save_message(request, user_id, tenant_id)
+ unit_id = save_message_unit(
+ message_id=message_id,
+ conversation_id=conversation_id,
+ unit_index=0,
+ unit_type="automation_proposal",
+ unit_content=content,
+ user_id=user_id,
+ )
+ return {
+ "user_message_id": user_message_id,
+ "user_unit_id": user_unit_id,
+ "message_id": message_id,
+ "unit_id": unit_id,
+ }
+
+ def update_proposal(
+ self,
+ unit_id: Optional[int],
+ payload: Dict[str, Any],
+ user_id: str,
+ ) -> None:
+ if not unit_id:
+ return
+ update_unit_content(
+ unit_id,
+ json.dumps(payload, ensure_ascii=False),
+ user_id,
+ )
+
+
+automation_conversation_adapter = AutomationConversationAdapter()
diff --git a/backend/services/agent_automation/errors.py b/backend/services/agent_automation/errors.py
new file mode 100644
index 000000000..5c9e40ca9
--- /dev/null
+++ b/backend/services/agent_automation/errors.py
@@ -0,0 +1,33 @@
+class AgentAutomationError(Exception):
+ """Base domain exception for agent automation."""
+
+ error_code = "AUTOMATION_ERROR"
+
+ def __init__(self, message: str, details: dict | None = None):
+ super().__init__(message)
+ self.message = message
+ self.details = details or {}
+
+
+class AutomationCapabilityNotReadyError(AgentAutomationError):
+ error_code = "AUTOMATION_CAPABILITY_NOT_READY"
+
+
+class AutomationCapabilityUnavailableError(AgentAutomationError):
+ error_code = "AUTOMATION_CAPABILITY_UNAVAILABLE"
+
+
+class AutomationCapabilityBindingInvalidError(AgentAutomationError):
+ error_code = "AUTOMATION_CAPABILITY_BINDING_INVALID"
+
+
+class AutomationScheduleInvalidError(AgentAutomationError):
+ error_code = "AUTOMATION_SCHEDULE_INVALID"
+
+
+class AutomationNotFoundError(AgentAutomationError):
+ error_code = "AUTOMATION_TASK_NOT_FOUND"
+
+
+class AutomationConversationAlreadyBoundError(AgentAutomationError):
+ error_code = "AUTOMATION_CONVERSATION_ALREADY_BOUND"
diff --git a/backend/services/agent_automation/facade.py b/backend/services/agent_automation/facade.py
new file mode 100644
index 000000000..f8c392433
--- /dev/null
+++ b/backend/services/agent_automation/facade.py
@@ -0,0 +1,497 @@
+import logging
+from datetime import datetime, timedelta, timezone
+from typing import Any, Dict, List, Optional
+
+from sqlalchemy.exc import IntegrityError
+
+from consts.const import AGENT_AUTOMATION_DEFAULT_TIMEOUT_SECONDS, AGENT_AUTOMATION_MIN_INTERVAL_SECONDS
+from database import agent_automation_db
+from database.conversation_db import get_conversation
+from services.conversation_management_service import (
+ create_new_conversation,
+ update_conversation_agent_id_service,
+)
+from .capability_resolver import resolve_agent_capabilities, validate_bindings_available
+from .conversation_adapter import automation_conversation_adapter
+from .errors import (
+ AutomationCapabilityBindingInvalidError,
+ AutomationCapabilityNotReadyError,
+ AutomationConversationAlreadyBoundError,
+ AutomationNotFoundError,
+ AutomationScheduleInvalidError,
+)
+from .intent_parser import parse_automation_intent
+from .models import (
+ AutomationProposalConfirmRequest,
+ AutomationProposalCreateRequest,
+ AutomationProposalStatus,
+ AutomationRunStatus,
+ AutomationSource,
+ AutomationTaskCreateRequest,
+ AutomationTaskPatchRequest,
+ AutomationTaskStatus,
+ CapabilityBinding,
+ ScheduleTrigger,
+)
+from .prompt_generator import AutomationPromptContext, automation_prompt_generator
+from .schedule_engine import compute_next_fire_at
+
+logger = logging.getLogger("agent_automation.facade")
+
+
+def _utcnow() -> datetime:
+ return datetime.now(timezone.utc)
+
+
+def _as_utc(value: datetime | str) -> datetime:
+ if isinstance(value, str):
+ value = datetime.fromisoformat(value.replace("Z", "+00:00"))
+ return value.astimezone(timezone.utc) if value.tzinfo else value.replace(tzinfo=timezone.utc)
+
+
+def _json(data: Any) -> Any:
+ if hasattr(data, "model_dump"):
+ return data.model_dump(mode="json")
+ if isinstance(data, list):
+ return [_json(item) for item in data]
+ return data
+
+
+def _parse_trigger(raw: Dict[str, Any] | ScheduleTrigger) -> ScheduleTrigger:
+ return raw if isinstance(raw, ScheduleTrigger) else ScheduleTrigger.model_validate(raw)
+
+
+def _validate_schedule_policy(trigger: ScheduleTrigger) -> None:
+ if (
+ trigger.rule_type.value == "INTERVAL"
+ and trigger.interval_seconds is not None
+ and trigger.interval_seconds < AGENT_AUTOMATION_MIN_INTERVAL_SECONDS
+ ):
+ raise AutomationScheduleInvalidError(
+ f"Automation interval must be at least {AGENT_AUTOMATION_MIN_INTERVAL_SECONDS} seconds.",
+ details={
+ "min_interval_seconds": AGENT_AUTOMATION_MIN_INTERVAL_SECONDS,
+ "interval_seconds": trigger.interval_seconds,
+ },
+ )
+
+
+class AgentAutomationFacade:
+ async def create_proposal(
+ self,
+ request: AutomationProposalCreateRequest,
+ tenant_id: str,
+ user_id: str,
+ ) -> Dict[str, Any]:
+ parsed = parse_automation_intent(request.message, request.timezone, tenant_id)
+ if not parsed.get("is_automation_intent"):
+ return {
+ "proposal_id": None,
+ "conversation_id": request.conversation_id,
+ "confidence": parsed.get("confidence", 0),
+ "executable": False,
+ "task": None,
+ "capability_resolution": None,
+ }
+
+ conversation_id = request.conversation_id
+ if conversation_id is None:
+ conversation = create_new_conversation(
+ parsed["title"],
+ user_id,
+ agent_id=request.agent_id,
+ )
+ conversation_id = conversation["conversation_id"]
+ else:
+ conversation = get_conversation(conversation_id, user_id)
+ if not conversation:
+ raise AutomationNotFoundError("Conversation does not exist or is not accessible.")
+ if conversation_id == request.conversation_id:
+ update_conversation_agent_id_service(
+ conversation_id,
+ request.agent_id,
+ user_id,
+ )
+ if agent_automation_db.get_task_by_conversation(conversation_id, user_id):
+ raise AutomationConversationAlreadyBoundError("Conversation already has an active automation task.")
+
+ resolution = await resolve_agent_capabilities(
+ agent_id=request.agent_id,
+ tenant_id=tenant_id,
+ user_id=user_id,
+ instruction=parsed["instruction"],
+ version_no=request.agent_version_no or 0,
+ )
+ optimized_instruction = await automation_prompt_generator.optimize_instruction(AutomationPromptContext(
+ tenant_id=tenant_id,
+ instruction=parsed["instruction"],
+ agent_snapshot=resolution.agent_snapshot,
+ capability_bindings=resolution.model_dump(mode="json")["matched_capabilities"],
+ title=parsed["title"],
+ timezone=request.timezone,
+ ))
+ proposed_task = {
+ "title": parsed["title"],
+ "instruction": optimized_instruction,
+ "original_instruction": parsed["instruction"],
+ "agent_id": request.agent_id,
+ "agent_version_no": request.agent_version_no,
+ "model_id": request.model_id,
+ "tool_params": request.tool_params,
+ "schedule_trigger": parsed["schedule_trigger"].model_dump(mode="json"),
+ }
+ proposal = agent_automation_db.create_proposal({
+ "tenant_id": tenant_id,
+ "user_id": user_id,
+ "conversation_id": conversation_id,
+ "agent_id": request.agent_id,
+ "proposed_task": proposed_task,
+ "capability_resolution": resolution.model_dump(mode="json"),
+ "status": AutomationProposalStatus.PENDING.value,
+ "expires_at": _utcnow() + timedelta(hours=24),
+ }, user_id)
+ response = {
+ "proposal_id": proposal["proposal_id"],
+ "conversation_id": conversation_id,
+ "confidence": parsed["confidence"],
+ "executable": resolution.executable,
+ "task": proposed_task,
+ "capability_resolution": resolution.model_dump(mode="json"),
+ }
+ try:
+ message_refs = automation_conversation_adapter.append_proposal_exchange(
+ conversation_id,
+ request.message,
+ response,
+ user_id,
+ tenant_id,
+ )
+ stored_task = {
+ **proposed_task,
+ "_conversation_message_id": message_refs["message_id"],
+ "_conversation_unit_id": message_refs["unit_id"],
+ }
+ agent_automation_db.update_proposal_task(
+ proposal["proposal_id"],
+ tenant_id,
+ user_id,
+ stored_task,
+ )
+ except Exception as exc:
+ logger.warning("Failed to persist automation proposal card: %s", exc, exc_info=True)
+ return response
+
+ async def confirm_proposal(
+ self,
+ proposal_id: int,
+ request: AutomationProposalConfirmRequest,
+ tenant_id: str,
+ user_id: str,
+ ) -> Dict[str, Any]:
+ proposal = agent_automation_db.get_proposal(proposal_id, tenant_id, user_id)
+ if not proposal or proposal["status"] != AutomationProposalStatus.PENDING.value:
+ raise AutomationNotFoundError("Automation proposal does not exist or is not pending.")
+ expires_at = proposal.get("expires_at")
+ if expires_at and _as_utc(expires_at) <= _utcnow():
+ agent_automation_db.update_proposal_status(
+ proposal_id,
+ tenant_id,
+ user_id,
+ AutomationProposalStatus.EXPIRED.value,
+ )
+ raise AutomationNotFoundError("Automation proposal has expired.")
+
+ proposed_task = proposal["proposed_task"]
+ instruction = request.instruction or proposed_task["instruction"]
+ resolution = await resolve_agent_capabilities(
+ agent_id=proposal["agent_id"],
+ tenant_id=tenant_id,
+ user_id=user_id,
+ instruction=instruction,
+ version_no=proposed_task.get("agent_version_no") or 0,
+ )
+ if not resolution.executable:
+ raise AutomationCapabilityNotReadyError(
+ "Required automation capabilities are not ready.",
+ details=resolution.model_dump(mode="json"),
+ )
+
+ create_request = AutomationTaskCreateRequest(
+ title=proposed_task["title"],
+ agent_id=proposal["agent_id"],
+ instruction=instruction,
+ original_instruction=proposed_task.get("original_instruction") or instruction,
+ schedule_trigger=_parse_trigger(proposed_task["schedule_trigger"]),
+ conversation_id=proposal["conversation_id"],
+ agent_version_no=proposed_task.get("agent_version_no"),
+ model_id=proposed_task.get("model_id"),
+ tool_params=proposed_task.get("tool_params"),
+ capability_bindings=resolution.matched_capabilities,
+ )
+ task = await self.create_task(create_request, tenant_id, user_id)
+ agent_automation_db.update_proposal_status(
+ proposal_id, tenant_id, user_id, AutomationProposalStatus.ACCEPTED.value)
+ public_task = {key: value for key, value in proposed_task.items() if not key.startswith("_")}
+ try:
+ automation_conversation_adapter.update_proposal(
+ proposed_task.get("_conversation_unit_id"),
+ {
+ "proposal_id": proposal_id,
+ "executable": True,
+ "task": public_task,
+ "capability_resolution": proposal["capability_resolution"],
+ "confirmed_task_id": task["task_id"],
+ },
+ user_id,
+ )
+ except Exception as exc:
+ logger.warning("Failed to persist confirmed automation proposal card: %s", exc, exc_info=True)
+ return task
+
+ async def create_task(
+ self,
+ request: AutomationTaskCreateRequest,
+ tenant_id: str,
+ user_id: str,
+ ) -> Dict[str, Any]:
+ conversation_id = request.conversation_id
+ if not get_conversation(conversation_id, user_id):
+ raise AutomationNotFoundError("Conversation does not exist or is not accessible.")
+
+ if agent_automation_db.get_task_by_conversation(conversation_id, user_id):
+ raise AutomationConversationAlreadyBoundError("Conversation already has an active automation task.")
+
+ resolution = await resolve_agent_capabilities(
+ agent_id=request.agent_id,
+ tenant_id=tenant_id,
+ user_id=user_id,
+ instruction=request.instruction,
+ version_no=request.agent_version_no or 0,
+ )
+ if not resolution.executable:
+ raise AutomationCapabilityNotReadyError(
+ "Required automation capabilities are not ready.",
+ details=resolution.model_dump(mode="json"),
+ )
+
+ if request.capability_bindings:
+ check = await validate_bindings_available(
+ request.agent_id,
+ tenant_id,
+ user_id,
+ request.instruction,
+ [_json(binding) for binding in request.capability_bindings],
+ request.agent_version_no or 0,
+ )
+ if check["unavailable_bindings"]:
+ raise AutomationCapabilityBindingInvalidError(
+ "Submitted capability bindings are not available for this agent.",
+ details=check,
+ )
+ bindings = [_json(binding) for binding in request.capability_bindings]
+ else:
+ bindings = resolution.model_dump(mode="json")["matched_capabilities"]
+
+ trigger = request.schedule_trigger
+ _validate_schedule_policy(trigger)
+ next_fire_at = compute_next_fire_at(trigger, _utcnow(), 0)
+ runtime_snapshot = {
+ **resolution.agent_snapshot,
+ "model_id": request.model_id,
+ "tool_params": request.tool_params,
+ "original_instruction": request.original_instruction or request.instruction,
+ }
+ try:
+ task = agent_automation_db.create_task({
+ "tenant_id": tenant_id,
+ "user_id": user_id,
+ "conversation_id": conversation_id,
+ "agent_id": request.agent_id,
+ "agent_version_no": request.agent_version_no,
+ "title": request.title,
+ "instruction": request.instruction,
+ "status": AutomationTaskStatus.ACTIVE.value,
+ "source": AutomationSource.CHAT_INTENT.value,
+ "schedule_mode": trigger.mode.value,
+ "schedule_rule_type": trigger.rule_type.value,
+ "schedule_expr": trigger.cron_expr or str(trigger.interval_seconds or trigger.start_at),
+ "schedule_config": trigger.model_dump(mode="json"),
+ "capability_requirements": resolution.model_dump(mode="json"),
+ "capability_bindings": bindings,
+ "runtime_snapshot": runtime_snapshot,
+ "timezone": trigger.timezone,
+ "next_fire_at": next_fire_at,
+ "fire_count": 0,
+ "consecutive_failures": 0,
+ "timeout_seconds": request.timeout_seconds or AGENT_AUTOMATION_DEFAULT_TIMEOUT_SECONDS,
+ "overlap_policy": "SKIP",
+ "misfire_policy": "RUN_ONCE",
+ }, user_id)
+ except IntegrityError as exc:
+ constraint_name = getattr(getattr(exc.orig, "diag", None), "constraint_name", None)
+ if constraint_name == "uq_agent_automation_conversation_active":
+ raise AutomationConversationAlreadyBoundError(
+ "Conversation already has an active automation task."
+ ) from exc
+ raise
+ return task
+
+ def list_tasks(self, tenant_id: str, user_id: str, status: Optional[str] = None) -> List[Dict[str, Any]]:
+ return agent_automation_db.list_tasks(tenant_id, user_id, status)
+
+ def get_task(self, task_id: int, tenant_id: str, user_id: str) -> Dict[str, Any]:
+ task = agent_automation_db.get_task(task_id, tenant_id, user_id)
+ if not task:
+ raise AutomationNotFoundError("Automation task not found.")
+ return task
+
+ def get_task_for_conversation(self, conversation_id: int, user_id: str) -> Optional[Dict[str, Any]]:
+ return agent_automation_db.get_task_by_conversation(conversation_id, user_id)
+
+ async def patch_task(
+ self,
+ task_id: int,
+ request: AutomationTaskPatchRequest,
+ tenant_id: str,
+ user_id: str,
+ ) -> Dict[str, Any]:
+ task = self.get_task(task_id, tenant_id, user_id)
+ values: Dict[str, Any] = {}
+ instruction = request.instruction or task["instruction"]
+ if request.title is not None:
+ values["title"] = request.title
+ if request.instruction is not None:
+ values["instruction"] = request.instruction
+ if request.timeout_seconds is not None:
+ values["timeout_seconds"] = request.timeout_seconds
+ if request.model_id is not None or request.tool_params is not None:
+ snapshot = dict(task.get("runtime_snapshot") or {})
+ if request.model_id is not None:
+ snapshot["model_id"] = request.model_id
+ if request.tool_params is not None:
+ snapshot["tool_params"] = request.tool_params
+ values["runtime_snapshot"] = snapshot
+ if request.schedule_trigger is not None:
+ trigger = request.schedule_trigger
+ _validate_schedule_policy(trigger)
+ values.update({
+ "schedule_mode": trigger.mode.value,
+ "schedule_rule_type": trigger.rule_type.value,
+ "schedule_expr": trigger.cron_expr or str(trigger.interval_seconds or trigger.start_at),
+ "schedule_config": trigger.model_dump(mode="json"),
+ "timezone": trigger.timezone,
+ "next_fire_at": compute_next_fire_at(trigger, _utcnow(), int(task.get("fire_count") or 0)),
+ })
+ if request.instruction is not None or request.capability_bindings is not None:
+ resolution = await resolve_agent_capabilities(
+ task["agent_id"], tenant_id, user_id, instruction, task.get("agent_version_no") or 0)
+ if not resolution.executable:
+ raise AutomationCapabilityNotReadyError(
+ "Required automation capabilities are not ready.",
+ details=resolution.model_dump(mode="json"),
+ )
+ values["capability_requirements"] = resolution.model_dump(mode="json")
+ values["capability_bindings"] = (
+ _json(request.capability_bindings)
+ if request.capability_bindings
+ else resolution.model_dump(mode="json")["matched_capabilities"]
+ )
+ snapshot = dict(values.get("runtime_snapshot") or task.get("runtime_snapshot") or {})
+ snapshot.update(resolution.agent_snapshot)
+ if request.instruction is not None:
+ snapshot["original_instruction"] = request.instruction
+ values["runtime_snapshot"] = snapshot
+ updated = agent_automation_db.update_task(task_id, tenant_id, user_id, values)
+ if not updated:
+ raise AutomationNotFoundError("Automation task not found.")
+ return updated
+
+ def pause_task(self, task_id: int, tenant_id: str, user_id: str) -> Dict[str, Any]:
+ task = agent_automation_db.update_task(
+ task_id,
+ tenant_id,
+ user_id,
+ {"status": AutomationTaskStatus.PAUSED.value},
+ )
+ if not task:
+ raise AutomationNotFoundError("Automation task not found.")
+ return task
+
+ def resume_task(self, task_id: int, tenant_id: str, user_id: str) -> Dict[str, Any]:
+ task = self.get_task(task_id, tenant_id, user_id)
+ trigger = _parse_trigger(task["schedule_config"])
+ next_fire_at = compute_next_fire_at(trigger, _utcnow(), int(task.get("fire_count") or 0))
+ if next_fire_at is None:
+ raise AutomationScheduleInvalidError("Automation task has no future fire time.")
+ updated = agent_automation_db.update_task(task_id, tenant_id, user_id, {
+ "status": AutomationTaskStatus.ACTIVE.value,
+ "next_fire_at": next_fire_at,
+ })
+ if not updated:
+ raise AutomationNotFoundError("Automation task not found.")
+ return updated
+
+ def delete_task(self, task_id: int, tenant_id: str, user_id: str) -> bool:
+ task = self.get_task(task_id, tenant_id, user_id)
+ self._cancel_active_runs_for_conversation(
+ task["conversation_id"],
+ user_id,
+ "Automation task was deleted.",
+ )
+ if not agent_automation_db.soft_delete_task(task_id, tenant_id, user_id):
+ raise AutomationNotFoundError("Automation task not found.")
+ return True
+
+ def list_runs(self, task_id: int, tenant_id: str, user_id: str) -> List[Dict[str, Any]]:
+ self.get_task(task_id, tenant_id, user_id)
+ return agent_automation_db.list_runs(task_id, tenant_id, user_id)
+
+ async def run_task_now(self, task_id: int, tenant_id: str, user_id: str) -> Dict[str, Any]:
+ task = self.get_task(task_id, tenant_id, user_id)
+ from .runner import agent_automation_runner
+ return await agent_automation_runner.execute_task(task, trigger_type="MANUAL")
+
+ def cancel_run(self, run_id: int, tenant_id: str, user_id: str) -> Dict[str, Any]:
+ run = agent_automation_db.get_run(run_id, tenant_id, user_id)
+ if not run:
+ raise AutomationNotFoundError("Automation run not found.")
+
+ if run["status"] not in {AutomationRunStatus.QUEUED.value, AutomationRunStatus.RUNNING.value}:
+ return run
+
+ self._request_conversation_stop(run["conversation_id"], user_id)
+ canceled = agent_automation_db.cancel_run(
+ run_id,
+ tenant_id,
+ user_id,
+ "Automation run was canceled by user.",
+ )
+ agent_automation_db.update_task(run["task_id"], tenant_id, user_id, {
+ "last_run_status": AutomationRunStatus.CANCELED.value,
+ "last_error": "Automation run was canceled by user.",
+ "lock_owner": None,
+ "lock_until": None,
+ })
+ return canceled or agent_automation_db.get_run(run_id, tenant_id, user_id) or run
+
+ def on_conversation_deleted(self, conversation_id: int, user_id: str) -> int:
+ self._cancel_active_runs_for_conversation(
+ conversation_id,
+ user_id,
+ "Conversation was deleted.",
+ )
+ return agent_automation_db.soft_delete_task_by_conversation(conversation_id, user_id)
+
+ def _cancel_active_runs_for_conversation(self, conversation_id: int, user_id: str, reason: str) -> None:
+ self._request_conversation_stop(conversation_id, user_id)
+ agent_automation_db.cancel_runs_by_conversation(conversation_id, user_id, reason)
+
+ def _request_conversation_stop(self, conversation_id: int, user_id: str) -> None:
+ try:
+ from .runner import agent_automation_runner
+ agent_automation_runner.cancel_for_conversation(conversation_id, user_id)
+ except Exception:
+ pass
+
+
+agent_automation_facade = AgentAutomationFacade()
diff --git a/backend/services/agent_automation/intent_parser.py b/backend/services/agent_automation/intent_parser.py
new file mode 100644
index 000000000..80dccbc46
--- /dev/null
+++ b/backend/services/agent_automation/intent_parser.py
@@ -0,0 +1,172 @@
+import re
+from datetime import datetime, timedelta
+from zoneinfo import ZoneInfo
+
+from .models import ScheduleMode, ScheduleRuleType, ScheduleTrigger
+
+
+RECURRING_TOKENS = ("每天", "每日", "每周", "每星期", "每礼拜", "每月", "周期")
+EXPLICIT_AUTOMATION_TOKENS = ("定时", "提醒")
+RELATIVE_DATE_TOKENS = ("明天", "今天")
+ACTION_TOKENS = (
+ "帮我",
+ "请",
+ "提醒",
+ "发送",
+ "发一个",
+ "发一份",
+ "生成",
+ "汇总",
+ "总结",
+ "执行",
+ "运行",
+ "通知",
+)
+
+
+def _has_explicit_time(message: str) -> bool:
+ return bool(
+ re.search(r"\d{1,2}(?::|:|点|时)\d{0,2}", message)
+ or any(period in message for period in ("上午", "中午", "下午", "晚上", "凌晨", "早上"))
+ )
+
+
+def _is_automation_intent(message: str) -> bool:
+ if any(token in message for token in RECURRING_TOKENS + EXPLICIT_AUTOMATION_TOKENS):
+ return True
+ return (
+ any(token in message for token in RELATIVE_DATE_TOKENS)
+ and _has_explicit_time(message)
+ and any(token in message for token in ACTION_TOKENS)
+ )
+
+
+def _parse_hour_minute(message: str, default_hour: int = 9) -> tuple[int, int]:
+ match = re.search(r"(\d{1,2})[::点时](\d{1,2})?", message)
+ if match:
+ hour = int(match.group(1))
+ minute = int(match.group(2) or 0)
+ if any(period in message for period in ("下午", "晚上")) and hour < 12:
+ hour += 12
+ elif "中午" in message and hour < 11:
+ hour += 12
+ elif "凌晨" in message and hour == 12:
+ hour = 0
+ if hour > 23 or minute > 59:
+ raise ValueError("Invalid hour or minute in automation schedule.")
+ return hour, minute
+ if "上午" in message or "早上" in message:
+ return default_hour, 0
+ if "下午" in message or "晚上" in message:
+ return 18, 0
+ return default_hour, 0
+
+
+def _clean_instruction(message: str) -> str:
+ cleaned = re.sub(r"(?:每)?(?:周|星期|礼拜)[一二三四五六日天]", "", message)
+ cleaned = re.sub(r"每月\s*\d{1,2}(?:号|日)?", "", cleaned)
+ cleaned = re.sub(
+ r"(请|帮我|定时|提醒我|每天|每日|每周|每星期|每礼拜|每月|"
+ r"明天|今天|上午|中午|下午|晚上|凌晨)",
+ "",
+ cleaned,
+ )
+ cleaned = re.sub(r"\d{1,2}[::点时]\d{0,2}", "", cleaned)
+ cleaned = cleaned.strip(" ,,。")
+ return cleaned or message
+
+
+def _parse_weekday(message: str, default: int = 1) -> int:
+ weekday_map = {
+ "一": 1,
+ "二": 2,
+ "三": 3,
+ "四": 4,
+ "五": 5,
+ "六": 6,
+ "日": 0,
+ "天": 0,
+ }
+ match = re.search(r"(?:周|星期|礼拜)([一二三四五六日天])", message)
+ return weekday_map.get(match.group(1), default) if match else default
+
+
+def _parse_month_day(message: str, default: int = 1) -> int:
+ match = re.search(r"每月\s*(\d{1,2})(?:号|日)?", message)
+ day = int(match.group(1)) if match else default
+ if day < 1 or day > 31:
+ raise ValueError("Invalid day of month in automation schedule.")
+ return day
+
+
+def parse_automation_intent(
+ message: str,
+ timezone_name: str = "Asia/Shanghai",
+ tenant_id: str | None = None,
+) -> dict:
+ """Parse a natural-language automation request into a conservative proposal."""
+ zone = ZoneInfo(timezone_name)
+ now = datetime.now(zone)
+
+ if not _is_automation_intent(message):
+ return {"is_automation_intent": False, "confidence": 0.0}
+
+ hour, minute = _parse_hour_minute(message)
+ instruction = _clean_instruction(message)
+ title = instruction[:30] if instruction else "自动任务"
+
+ if "每天" in message or "每日" in message:
+ start_at = now.replace(hour=hour, minute=minute, second=0, microsecond=0)
+ if start_at <= now:
+ start_at += timedelta(days=1)
+ trigger = ScheduleTrigger(
+ mode=ScheduleMode.RECURRING,
+ rule_type=ScheduleRuleType.CRON,
+ timezone=timezone_name,
+ start_at=start_at,
+ cron_expr=f"{minute} {hour} * * *",
+ )
+ elif any(token in message for token in ("每周", "每星期", "每礼拜")):
+ weekday = _parse_weekday(message)
+ cron_weekday = weekday
+ start_at = now.replace(hour=hour, minute=minute, second=0, microsecond=0)
+ trigger = ScheduleTrigger(
+ mode=ScheduleMode.RECURRING,
+ rule_type=ScheduleRuleType.CRON,
+ timezone=timezone_name,
+ start_at=start_at,
+ cron_expr=f"{minute} {hour} * * {cron_weekday}",
+ )
+ elif "每月" in message:
+ month_day = _parse_month_day(message)
+ start_at = now.replace(hour=hour, minute=minute, second=0, microsecond=0)
+ trigger = ScheduleTrigger(
+ mode=ScheduleMode.RECURRING,
+ rule_type=ScheduleRuleType.CRON,
+ timezone=timezone_name,
+ start_at=start_at,
+ cron_expr=f"{minute} {hour} {month_day} * *",
+ )
+ else:
+ days = 1 if "明天" in message else 0
+ start_at = (now + timedelta(days=days)).replace(hour=hour, minute=minute, second=0, microsecond=0)
+ if start_at <= now:
+ start_at += timedelta(days=1)
+ trigger = ScheduleTrigger(
+ mode=ScheduleMode.ONCE,
+ rule_type=ScheduleRuleType.AT,
+ timezone=timezone_name,
+ start_at=start_at,
+ max_fire_count=1,
+ )
+
+ base_result = {
+ "is_automation_intent": True,
+ "confidence": 0.86,
+ "title": title,
+ "instruction": instruction,
+ "schedule_trigger": trigger,
+ "capability_intents": [],
+ "output_requirements": {"language": "zh"},
+ }
+ return base_result
diff --git a/backend/services/agent_automation/models.py b/backend/services/agent_automation/models.py
new file mode 100644
index 000000000..1c4420e18
--- /dev/null
+++ b/backend/services/agent_automation/models.py
@@ -0,0 +1,146 @@
+from datetime import datetime
+from enum import Enum
+from typing import Any, Dict, List, Optional
+
+from pydantic import BaseModel, Field, model_validator
+
+
+class StrEnum(str, Enum):
+ pass
+
+
+class AutomationTaskStatus(StrEnum):
+ DRAFT = "DRAFT"
+ ACTIVE = "ACTIVE"
+ PAUSED = "PAUSED"
+ PAUSED_BY_SYSTEM = "PAUSED_BY_SYSTEM"
+ COMPLETED = "COMPLETED"
+ DELETED = "DELETED"
+
+
+class AutomationRunStatus(StrEnum):
+ QUEUED = "QUEUED"
+ RUNNING = "RUNNING"
+ SUCCEEDED = "SUCCEEDED"
+ FAILED = "FAILED"
+ SKIPPED = "SKIPPED"
+ CANCELED = "CANCELED"
+ TIMEOUT = "TIMEOUT"
+
+
+class AutomationProposalStatus(StrEnum):
+ PENDING = "PENDING"
+ ACCEPTED = "ACCEPTED"
+ REJECTED = "REJECTED"
+ EXPIRED = "EXPIRED"
+
+
+class AutomationSource(StrEnum):
+ CHAT_INTENT = "CHAT_INTENT"
+
+
+class ScheduleMode(StrEnum):
+ ONCE = "ONCE"
+ RECURRING = "RECURRING"
+
+
+class ScheduleRuleType(StrEnum):
+ AT = "AT"
+ INTERVAL = "INTERVAL"
+ CRON = "CRON"
+
+
+class CapabilityType(StrEnum):
+ TOOL = "TOOL"
+ SKILL = "SKILL"
+ KNOWLEDGE_BASE = "KNOWLEDGE_BASE"
+ MANAGED_AGENT = "MANAGED_AGENT"
+ EXTERNAL_A2A_AGENT = "EXTERNAL_A2A_AGENT"
+ MEMORY = "MEMORY"
+
+
+class ScheduleTrigger(BaseModel):
+ mode: ScheduleMode
+ rule_type: ScheduleRuleType
+ timezone: str = "Asia/Shanghai"
+ start_at: datetime
+ end_at: Optional[datetime] = None
+ cron_expr: Optional[str] = None
+ interval_seconds: Optional[int] = Field(default=None, gt=0)
+ max_fire_count: Optional[int] = Field(default=None, gt=0)
+
+ @model_validator(mode="after")
+ def validate_combination(self):
+ if self.mode == ScheduleMode.ONCE:
+ if self.rule_type != ScheduleRuleType.AT:
+ raise ValueError("ONCE schedule only supports AT rule_type")
+ self.max_fire_count = 1
+ elif self.mode == ScheduleMode.RECURRING:
+ if self.rule_type == ScheduleRuleType.AT:
+ raise ValueError("RECURRING schedule does not support AT rule_type")
+ if self.rule_type == ScheduleRuleType.CRON and not self.cron_expr:
+ raise ValueError("cron_expr is required for CRON schedules")
+ if self.rule_type == ScheduleRuleType.INTERVAL and not self.interval_seconds:
+ raise ValueError("interval_seconds is required for INTERVAL schedules")
+ return self
+
+
+class CapabilityBinding(BaseModel):
+ type: CapabilityType
+ name: str
+ display_name: Optional[str] = None
+ binding_ref: str
+ reason: Optional[str] = None
+ required: bool = True
+
+
+class CapabilityResolution(BaseModel):
+ matched_capabilities: List[CapabilityBinding] = Field(default_factory=list)
+ missing_capabilities: List[Dict[str, Any]] = Field(default_factory=list)
+ optional_capabilities: List[CapabilityBinding] = Field(default_factory=list)
+ agent_snapshot: Dict[str, Any] = Field(default_factory=dict)
+ executable: bool = True
+
+
+class AutomationTaskCreateRequest(BaseModel):
+ title: str = Field(min_length=1)
+ agent_id: int = Field(gt=0)
+ instruction: str = Field(min_length=1)
+ schedule_trigger: ScheduleTrigger
+ conversation_id: int = Field(gt=0)
+ original_instruction: Optional[str] = None
+ agent_version_no: Optional[int] = None
+ model_id: Optional[int] = None
+ tool_params: Optional[Dict[str, Any]] = None
+ capability_bindings: List[CapabilityBinding] = Field(default_factory=list)
+ timeout_seconds: Optional[int] = Field(default=None, gt=0)
+
+
+class AutomationTaskPatchRequest(BaseModel):
+ title: Optional[str] = Field(default=None, min_length=1)
+ instruction: Optional[str] = Field(default=None, min_length=1)
+ schedule_trigger: Optional[ScheduleTrigger] = None
+ capability_bindings: Optional[List[CapabilityBinding]] = None
+ model_id: Optional[int] = None
+ tool_params: Optional[Dict[str, Any]] = None
+ timeout_seconds: Optional[int] = Field(default=None, gt=0)
+
+
+class AutomationProposalCreateRequest(BaseModel):
+ conversation_id: Optional[int] = Field(default=None, gt=0)
+ agent_id: int = Field(gt=0)
+ message: str = Field(min_length=1)
+ timezone: str = "Asia/Shanghai"
+ agent_version_no: Optional[int] = None
+ model_id: Optional[int] = None
+ tool_params: Optional[Dict[str, Any]] = None
+
+
+class AutomationProposalConfirmRequest(BaseModel):
+ instruction: Optional[str] = None
+
+
+class AutomationResponse(BaseModel):
+ code: int = 0
+ message: str = "success"
+ data: Any = None
diff --git a/backend/services/agent_automation/prompt_generator.py b/backend/services/agent_automation/prompt_generator.py
new file mode 100644
index 000000000..4870d04bb
--- /dev/null
+++ b/backend/services/agent_automation/prompt_generator.py
@@ -0,0 +1,217 @@
+import asyncio
+import logging
+import re
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Any, Dict, List, Optional
+
+from jinja2 import StrictUndefined, Template
+
+from consts.const import LANGUAGE, MESSAGE_ROLE, MODEL_CONFIG_MAPPING
+from utils.prompt_template_utils import get_prompt_template
+
+logger = logging.getLogger("agent_automation.prompt_generator")
+
+
+@dataclass(frozen=True)
+class AutomationPromptContext:
+ """Data required to generate a task instruction or a single-run prompt."""
+
+ tenant_id: str
+ instruction: str
+ agent_snapshot: Dict[str, Any] = field(default_factory=dict)
+ capability_bindings: List[Dict[str, Any]] = field(default_factory=list)
+ title: str = ""
+ timezone: str = "Asia/Shanghai"
+ scheduled_fire_at: Optional[datetime] = None
+ trigger_type: str = "SCHEDULED"
+ conversation_context: str = ""
+ language: str = LANGUAGE["ZH"]
+
+
+def _capability_summary(bindings: List[Dict[str, Any]], language: str) -> str:
+ if not bindings:
+ return (
+ "当前没有绑定特定能力。"
+ if language == LANGUAGE["ZH"]
+ else "No specific capabilities are bound."
+ )
+
+ lines = []
+ for binding in bindings:
+ label = binding.get("display_name") or binding.get("name") or binding.get("binding_ref")
+ lines.append(f"- {binding.get('type', 'CAPABILITY')}: {label}")
+ return "\n".join(lines)
+
+
+def _normalize_model_output(content: str, fallback: str, max_length: int) -> str:
+ normalized = re.sub(r"
+ 任务由会话创建,并始终绑定创建它的会话 +
+