Enhance agent automation with task management and scheduling features

backend/apps/agent_automation_app.py

@@ -0,0 +1,225 @@
+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,
+    AutomationProposalPatchRequest,
+    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.patch("/proposals/{proposal_id}", response_model=AutomationResponse)
+async def update_proposal(
+    proposal_id: int,
+    request: AutomationProposalPatchRequest,
+    authorization: Optional[str] = Header(None),
+):
+    try:
+        user_id, tenant_id = _get_current_user(authorization)
+        data = await agent_automation_facade.update_proposal(
+            proposal_id,
+            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 update 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),
+    search: Optional[str] = Query(default=None, max_length=200),
+    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, search))
+    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)
+
+
+@router.delete("/runs/{run_id}", response_model=AutomationResponse)
+async def delete_run(run_id: int, authorization: Optional[str] = Header(None)):
+    try:
+        user_id, tenant_id = _get_current_user(authorization)
+        return AutomationResponse(data=agent_automation_facade.delete_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, tenant_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))

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

backend/consts/const.py

@@ -46,6 +46,29 @@ 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", "5")
+)
+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_SHUTDOWN_GRACE_SECONDS = int(
+    os.getenv("AGENT_AUTOMATION_SHUTDOWN_GRACE_SECONDS", "30")
+)
+AGENT_AUTOMATION_MIN_INTERVAL_SECONDS = int(
+    os.getenv("AGENT_AUTOMATION_MIN_INTERVAL_SECONDS", "5")
+)
+
 
 # Container-internal skills storage path
 CONTAINER_SKILLS_PATH = os.getenv("SKILLS_PATH")