"""API routes for the Automation plugin — /api/v1/automation. Endpoints: automation definitions CRUD, execute, dry-run, runs, versions. """ from __future__ import annotations import logging import uuid from datetime import UTC from typing import Any from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.ext.asyncio import AsyncSession from app.core.db import get_db from app.deps import get_current_user, require_permission from app.plugins.builtins.automation.models import ( AutomationDefinition, AutomationRun, AutomationVersion, ) from app.plugins.builtins.automation.schemas import ( AutomationDefinitionCreate, AutomationDefinitionListResponse, AutomationDefinitionResponse, AutomationDefinitionUpdate, AutomationRunListResponse, AutomationRunResponse, AutomationSettingsResponse, AutomationSettingsUpdate, AutomationVersionListResponse, AutomationVersionResponse, MiniAppCreate, MiniAppResponse, SubtaskCreate, SubtaskListResponse, SubtaskRead, SubtaskUpdate, ) from app.plugins.builtins.automation.services import ( AutomationService, RunLogService, ) logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/v1/automation", tags=["automation"]) # ─── Helper Functions ─── def _automation_to_response(a: AutomationDefinition) -> AutomationDefinitionResponse: """Convert AutomationDefinition model to response schema.""" return AutomationDefinitionResponse( id=str(a.id), name=a.name, description=a.description or "", trigger_type=a.trigger_type, trigger_config=a.trigger_config or {}, conditions=a.conditions or [], actions=a.actions or [], is_active=a.is_active, dry_run=a.dry_run, created_by=str(a.created_by) if a.created_by else None, created_at=a.created_at.isoformat() if a.created_at else None, updated_at=a.updated_at.isoformat() if a.updated_at else None, ) def _run_to_response(r: AutomationRun) -> AutomationRunResponse: """Convert AutomationRun model to response schema.""" return AutomationRunResponse( id=str(r.id), automation_id=str(r.automation_id), status=r.status, started_at=r.started_at.isoformat() if r.started_at else "", completed_at=r.completed_at.isoformat() if r.completed_at else None, duration_seconds=r.duration_seconds, result=r.result, error=r.error, trigger_type=r.trigger_type, trigger_data=r.trigger_data or {}, dry_run=r.dry_run, created_at=r.created_at.isoformat() if r.created_at else None, ) def _version_to_response(v: AutomationVersion) -> AutomationVersionResponse: """Convert AutomationVersion model to response schema.""" return AutomationVersionResponse( id=str(v.id), automation_id=str(v.automation_id), version_number=v.version_number, snapshot=v.snapshot or {}, changed_by=str(v.changed_by) if v.changed_by else None, created_at=v.created_at.isoformat() if v.created_at else None, ) # ─── CRUD Endpoints ─── @router.get( "", dependencies=[Depends(require_permission("automation:read"))], response_model=AutomationDefinitionListResponse, ) async def list_automations( trigger_type: str | None = Query(None), is_active: bool | None = Query(None), limit: int = Query(50, ge=1, le=200), offset: int = Query(0, ge=0), current_user: dict[str, Any] = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """List automation definitions with optional filters.""" tenant_id = uuid.UUID(current_user["tenant_id"]) user_id = uuid.UUID(current_user["user_id"]) is_system_admin = current_user.get("is_system_admin", False) items, total = await AutomationService.list( db, tenant_id, trigger_type=trigger_type, is_active=is_active, limit=limit, offset=offset, user_id=user_id, is_system_admin=is_system_admin, ) return AutomationDefinitionListResponse( items=[_automation_to_response(a) for a in items], total=total, ) @router.post( "/", dependencies=[Depends(require_permission("automation:write"))], response_model=AutomationDefinitionResponse, status_code=201, ) async def create_automation( data: AutomationDefinitionCreate, current_user: dict[str, Any] = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Create a new automation definition.""" tenant_id = uuid.UUID(current_user["tenant_id"]) user_id = uuid.UUID(current_user["user_id"]) automation = await AutomationService.create( db, tenant_id, data.model_dump(), user_id=user_id ) return _automation_to_response(automation) # ─── Recent Runs ─── @router.get( "/runs/recent", dependencies=[Depends(require_permission("automation:read"))], response_model=AutomationRunListResponse, ) async def list_recent_automation_runs( limit: int = Query(20, ge=1, le=100), status: str | None = Query(None), current_user: dict[str, Any] = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """List recent automation runs across all automations (for dashboard).""" tenant_id = uuid.UUID(current_user["tenant_id"]) items, total = await RunLogService.list_automation_runs( db, tenant_id, automation_id=None, status=status, limit=limit, offset=0 ) return AutomationRunListResponse( items=[_run_to_response(r) for r in items], total=total, ) # ─── MiniApps ─── @router.get( "/miniapps", dependencies=[Depends(require_permission("automation:read"))], response_model=list[MiniAppResponse], ) async def list_miniapps( current_user: dict[str, Any] = Depends(get_current_user), ): """List custom MiniApps from plugin config.""" from app.plugins.builtins.kommunikation.contracts import get_miniapp_registry registry = get_miniapp_registry() items = registry.list_apps() return items @router.post( "/miniapps", dependencies=[Depends(require_permission("automation:write"))], response_model=MiniAppResponse, status_code=201, ) async def create_miniapp( data: MiniAppCreate, current_user: dict[str, Any] = Depends(get_current_user), ): """Create a custom MiniApp definition.""" from app.plugins.builtins.kommunikation.contracts import get_miniapp_registry registry = get_miniapp_registry() registry.register( app_id=data.app_id, name=data.name, icon=data.icon, description=data.description, plugin_name="automation", render_schema=data.render_schema, ) return MiniAppResponse( app_id=data.app_id, name=data.name, icon=data.icon, description=data.description, plugin_name="automation", render_schema=data.render_schema, ) @router.delete( "/miniapps/{app_id}", dependencies=[Depends(require_permission("automation:delete"))], ) async def delete_miniapp( app_id: str, current_user: dict[str, Any] = Depends(get_current_user), ): """Delete a custom MiniApp definition.""" from app.plugins.builtins.kommunikation.contracts import get_miniapp_registry registry = get_miniapp_registry() registry.unregister(app_id) return {"status": "ok"} # ─── Settings ─── @router.get( "/settings", dependencies=[Depends(require_permission("automation:configure"))], response_model=AutomationSettingsResponse, ) async def get_automation_settings( current_user: dict[str, Any] = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Get automation settings (persisted in system_settings metadata).""" tenant_id = uuid.UUID(current_user["tenant_id"]) from sqlalchemy import select from app.models.system_settings import SystemSettings result = await db.execute( select(SystemSettings).where(SystemSettings.tenant_id == tenant_id) ) settings = result.scalar_one_or_none() defaults = AutomationSettingsResponse() if settings and hasattr(settings, "automation_config") and settings.automation_config: cfg = settings.automation_config return AutomationSettingsResponse( default_llm_model=cfg.get("default_llm_model", defaults.default_llm_model), heartbeat_default_interval=cfg.get("heartbeat_default_interval", defaults.heartbeat_default_interval), max_concurrent_agents=cfg.get("max_concurrent_agents", defaults.max_concurrent_agents), log_level=cfg.get("log_level", defaults.log_level), ) return defaults @router.patch( "/settings", dependencies=[Depends(require_permission("automation:configure"))], response_model=AutomationSettingsResponse, ) async def update_automation_settings( data: AutomationSettingsUpdate, current_user: dict[str, Any] = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Update automation settings (persisted in system_settings metadata).""" tenant_id = uuid.UUID(current_user["tenant_id"]) from sqlalchemy import select from app.models.system_settings import SystemSettings result = await db.execute( select(SystemSettings).where(SystemSettings.tenant_id == tenant_id) ) settings = result.scalar_one_or_none() # Build new config new_config = { "default_llm_model": data.default_llm_model or "ollama/deepseek-v4-flash", "heartbeat_default_interval": data.heartbeat_default_interval or 300, "max_concurrent_agents": data.max_concurrent_agents or 5, "log_level": data.log_level or "INFO", } if settings: # Update existing settings row if hasattr(settings, "automation_config"): settings.automation_config = new_config else: # Fallback: store in a generic metadata field if available await db.execute( "UPDATE system_settings SET automation_config = :cfg WHERE id = :sid", {"cfg": new_config, "sid": settings.id}, ) else: # Create new settings row with automation config settings = SystemSettings( tenant_id=tenant_id, company_name="Default", company_street="", company_city="", company_zip="", company_country="DE", ) if hasattr(settings, "automation_config"): settings.automation_config = new_config db.add(settings) await db.flush() await db.commit() return AutomationSettingsResponse(**new_config) @router.get( "/{automation_id}", dependencies=[Depends(require_permission("automation:read"))], response_model=AutomationDefinitionResponse, ) async def get_automation( automation_id: str, current_user: dict[str, Any] = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Get a single automation definition by ID.""" tenant_id = uuid.UUID(current_user["tenant_id"]) try: aid = uuid.UUID(automation_id) except (ValueError, TypeError): raise HTTPException(status_code=400, detail="Invalid automation ID") from None automation = await AutomationService.get_by_id(db, tenant_id, aid) if automation is None: raise HTTPException(status_code=404, detail="Automation not found") return _automation_to_response(automation) @router.patch( "/{automation_id}", dependencies=[Depends(require_permission("automation:write"))], response_model=AutomationDefinitionResponse, ) async def update_automation( automation_id: str, data: AutomationDefinitionUpdate, current_user: dict[str, Any] = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Update an existing automation definition.""" tenant_id = uuid.UUID(current_user["tenant_id"]) user_id = uuid.UUID(current_user["user_id"]) try: aid = uuid.UUID(automation_id) except (ValueError, TypeError): raise HTTPException(status_code=400, detail="Invalid automation ID") from None automation = await AutomationService.update( db, tenant_id, aid, data.model_dump(exclude_none=True), user_id=user_id ) if automation is None: raise HTTPException(status_code=404, detail="Automation not found") return _automation_to_response(automation) @router.delete( "/{automation_id}", dependencies=[Depends(require_permission("automation:delete"))], ) async def delete_automation( automation_id: str, current_user: dict[str, Any] = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Delete an automation definition.""" tenant_id = uuid.UUID(current_user["tenant_id"]) try: aid = uuid.UUID(automation_id) except (ValueError, TypeError): raise HTTPException(status_code=400, detail="Invalid automation ID") from None success = await AutomationService.delete(db, tenant_id, aid) if not success: raise HTTPException(status_code=404, detail="Automation not found") return {"status": "ok"} # ─── Execute / Dry-Run ─── @router.post( "/{automation_id}/execute", dependencies=[Depends(require_permission("automation:execute"))], ) async def execute_automation( automation_id: str, current_user: dict[str, Any] = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Execute an automation definition.""" tenant_id = uuid.UUID(current_user["tenant_id"]) try: aid = uuid.UUID(automation_id) except (ValueError, TypeError): raise HTTPException(status_code=400, detail="Invalid automation ID") from None automation = await AutomationService.get_by_id(db, tenant_id, aid) if automation is None: raise HTTPException(status_code=404, detail="Automation not found") if not automation.is_active: raise HTTPException(status_code=400, detail="Automation is not active") # Create run log entry run = AutomationRun( tenant_id=tenant_id, automation_id=aid, status="running", started_at=__import__("datetime").datetime.now(__import__("datetime").timezone.utc), trigger_type="manual", trigger_data={"triggered_by": str(uuid.UUID(current_user["user_id"]))}, dry_run=False, ) db.add(run) await db.flush() # Execute automation via execution engine from app.plugins.builtins.automation.execution_engine import run_automation run_id = str(run.id) await db.commit() # Run automation asynchronously result = await run_automation( ctx={"tenant_id": str(tenant_id), "user_id": current_user["user_id"]}, automation_id=str(aid), trigger_type="manual", trigger_data={"triggered_by": current_user["user_id"]}, ) # Update run with results from datetime import datetime from sqlalchemy import update as sa_update now = datetime.now(UTC) async with db.begin(): await db.execute( sa_update(AutomationRun) .where(AutomationRun.id == run.id) .values( status=result.get("status", "completed"), completed_at=now, duration_seconds=0.0, result=result, output_data=result, ) ) return {"status": result.get("status", "ok"), "run_id": run_id, "result": result} @router.post( "/{automation_id}/dry-run", dependencies=[Depends(require_permission("automation:execute"))], ) async def dry_run_automation( automation_id: str, current_user: dict[str, Any] = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Dry-run an automation definition (validate without executing).""" tenant_id = uuid.UUID(current_user["tenant_id"]) try: aid = uuid.UUID(automation_id) except (ValueError, TypeError): raise HTTPException(status_code=400, detail="Invalid automation ID") from None automation = await AutomationService.get_by_id(db, tenant_id, aid) if automation is None: raise HTTPException(status_code=404, detail="Automation not found") # Execute dry-run via execution engine from app.plugins.builtins.automation.execution_engine import ( run_automation as execute_automation_engine, ) result = await execute_automation_engine( ctx={}, automation_id=str(aid), trigger_type="manual", trigger_data={ "triggered_by": str(uuid.UUID(current_user["user_id"])), "dry_run": True, }, ) # Create dry-run log entry run = AutomationRun( tenant_id=tenant_id, automation_id=aid, status=result.get("status", "dry_run"), started_at=__import__("datetime").datetime.now(__import__("datetime").timezone.utc), completed_at=__import__("datetime").datetime.now(__import__("datetime").timezone.utc), duration_seconds=0.0, trigger_type="manual", trigger_data={"triggered_by": str(uuid.UUID(current_user["user_id"])), "dry_run": True}, dry_run=True, result=str(result), ) db.add(run) await db.flush() return { "status": "ok", "run_id": str(run.id), "automation": _automation_to_response(automation), "validation": { "valid": True, "conditions_count": len(automation.conditions or []), "actions_count": len(automation.actions or []), }, "dry_run_result": result, } # ─── Runs ─── @router.get( "/{automation_id}/runs", dependencies=[Depends(require_permission("automation:read"))], response_model=AutomationRunListResponse, ) async def list_automation_runs( automation_id: str, status: str | None = Query(None), limit: int = Query(50, ge=1, le=200), offset: int = Query(0, ge=0), current_user: dict[str, Any] = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """List run logs for an automation definition.""" tenant_id = uuid.UUID(current_user["tenant_id"]) try: aid = uuid.UUID(automation_id) except (ValueError, TypeError): raise HTTPException(status_code=400, detail="Invalid automation ID") from None items, total = await RunLogService.list_automation_runs( db, tenant_id, automation_id=aid, status=status, limit=limit, offset=offset ) return AutomationRunListResponse( items=[_run_to_response(r) for r in items], total=total, ) # ─── Versions ─── @router.get( "/{automation_id}/versions", dependencies=[Depends(require_permission("automation:read"))], response_model=AutomationVersionListResponse, ) async def list_automation_versions( automation_id: str, limit: int = Query(50, ge=1, le=200), offset: int = Query(0, ge=0), current_user: dict[str, Any] = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """List version history for an automation definition.""" tenant_id = uuid.UUID(current_user["tenant_id"]) try: aid = uuid.UUID(automation_id) except (ValueError, TypeError): raise HTTPException(status_code=400, detail="Invalid automation ID") from None items, total = await AutomationService.get_versions( db, tenant_id, aid, limit=limit, offset=offset ) return AutomationVersionListResponse( items=[_version_to_response(v) for v in items], total=total, ) @router.post( "/{automation_id}/versions/{version_id}/restore", dependencies=[Depends(require_permission("automation:write"))], response_model=AutomationDefinitionResponse, ) async def restore_automation_version( automation_id: str, version_id: str, current_user: dict[str, Any] = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Restore an automation definition from a specific version.""" tenant_id = uuid.UUID(current_user["tenant_id"]) user_id = uuid.UUID(current_user["user_id"]) try: aid = uuid.UUID(automation_id) vid = uuid.UUID(version_id) except (ValueError, TypeError): raise HTTPException(status_code=400, detail="Invalid ID") from None automation = await AutomationService.restore_version( db, tenant_id, aid, vid, user_id=user_id ) if automation is None: raise HTTPException(status_code=404, detail="Automation or version not found") return _automation_to_response(automation) # ─── Subtask Endpoints (Phase 5.8) ─── @router.get( "/subtasks", dependencies=[Depends(require_permission("agents:read"))], response_model=SubtaskListResponse, ) async def list_subtasks( parent_agent_id: str | None = Query(None), child_agent_id: str | None = Query(None), status: str | None = Query(None), limit: int = Query(50, ge=1, le=200), offset: int = Query(0, ge=0), current_user: dict[str, Any] = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """List subtasks with optional filters.""" tenant_id = uuid.UUID(current_user["tenant_id"]) try: parent_id = uuid.UUID(parent_agent_id) if parent_agent_id else None child_id = uuid.UUID(child_agent_id) if child_agent_id else None except (ValueError, TypeError): raise HTTPException(status_code=400, detail="Invalid agent ID") from None from app.plugins.builtins.automation.agent_coordinator import AgentCoordinator items, total = await AgentCoordinator.list_subtasks( db, tenant_id, parent_agent_id=parent_id, child_agent_id=child_id, status=status, limit=limit, offset=offset, ) return SubtaskListResponse( items=[_subtask_to_response(s) for s in items], total=total, ) @router.post( "/subtasks", dependencies=[Depends(require_permission("agents:execute"))], response_model=SubtaskRead, status_code=201, ) async def create_subtask( data: SubtaskCreate, current_user: dict[str, Any] = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Create a new subtask for multi-agent orchestration.""" tenant_id = uuid.UUID(current_user["tenant_id"]) try: parent_id = uuid.UUID(data.parent_agent_id) child_id = uuid.UUID(data.child_agent_id) except (ValueError, TypeError): raise HTTPException(status_code=400, detail="Invalid agent ID") from None from app.plugins.builtins.automation.agent_coordinator import AgentCoordinator subtask = await AgentCoordinator.create_subtask( db=db, tenant_id=tenant_id, parent_agent_id=parent_id, child_agent_id=child_id, task_description=data.task_description, ) await db.commit() return _subtask_to_response(subtask) @router.get( "/subtasks/{subtask_id}", dependencies=[Depends(require_permission("agents:read"))], response_model=SubtaskRead, ) async def get_subtask( subtask_id: str, current_user: dict[str, Any] = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Get a single subtask by ID.""" tenant_id = uuid.UUID(current_user["tenant_id"]) try: sid = uuid.UUID(subtask_id) except (ValueError, TypeError): raise HTTPException(status_code=400, detail="Invalid subtask ID") from None from sqlalchemy import select from app.plugins.builtins.automation.models import AgentSubtask result = await db.execute( select(AgentSubtask) .where(AgentSubtask.id == sid) .where(AgentSubtask.tenant_id == tenant_id) .limit(1) ) subtask = result.scalar_one_or_none() if subtask is None: raise HTTPException(status_code=404, detail="Subtask not found") return _subtask_to_response(subtask) @router.patch( "/subtasks/{subtask_id}", dependencies=[Depends(require_permission("agents:execute"))], response_model=SubtaskRead, ) async def update_subtask( subtask_id: str, data: SubtaskUpdate, current_user: dict[str, Any] = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Update a subtask (status, result).""" tenant_id = uuid.UUID(current_user["tenant_id"]) try: sid = uuid.UUID(subtask_id) except (ValueError, TypeError): raise HTTPException(status_code=400, detail="Invalid subtask ID") from None from sqlalchemy import select from app.plugins.builtins.automation.models import AgentSubtask result = await db.execute( select(AgentSubtask) .where(AgentSubtask.id == sid) .where(AgentSubtask.tenant_id == tenant_id) .limit(1) ) subtask = result.scalar_one_or_none() if subtask is None: raise HTTPException(status_code=404, detail="Subtask not found") update_data = data.model_dump(exclude_none=True) if "status" in update_data: subtask.status = update_data["status"] if update_data["status"] in ("completed", "failed", "cancelled"): subtask.completed_at = __import__("datetime").datetime.now( __import__("datetime").timezone.utc ) if "result" in update_data: subtask.result = update_data["result"] await db.commit() await db.refresh(subtask) return _subtask_to_response(subtask) @router.post( "/subtasks/{subtask_id}/cancel", dependencies=[Depends(require_permission("agents:execute"))], ) async def cancel_subtask( subtask_id: str, current_user: dict[str, Any] = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Cancel a pending or running subtask.""" tenant_id = uuid.UUID(current_user["tenant_id"]) try: sid = uuid.UUID(subtask_id) except (ValueError, TypeError): raise HTTPException(status_code=400, detail="Invalid subtask ID") from None from app.plugins.builtins.automation.agent_coordinator import AgentCoordinator success = await AgentCoordinator.cancel_subtask(db, tenant_id, sid) if not success: raise HTTPException( status_code=400, detail="Subtask not found or already in terminal state", ) await db.commit() return {"status": "cancelled", "subtask_id": subtask_id} @router.post( "/subtasks/{subtask_id}/wait", dependencies=[Depends(require_permission("agents:execute"))], ) async def wait_for_subtask( subtask_id: str, current_user: dict[str, Any] = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Wait for a subtask to complete (polls until terminal state).""" tenant_id = uuid.UUID(current_user["tenant_id"]) try: sid = uuid.UUID(subtask_id) except (ValueError, TypeError): raise HTTPException(status_code=400, detail="Invalid subtask ID") from None from app.plugins.builtins.automation.agent_coordinator import AgentCoordinator result = await AgentCoordinator.wait_for_subtask(db, tenant_id, sid) return result @router.post( "/subtasks/aggregate", dependencies=[Depends(require_permission("agents:read"))], ) async def aggregate_subtasks( subtask_ids: list[str], current_user: dict[str, Any] = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Aggregate results from multiple subtasks.""" tenant_id = uuid.UUID(current_user["tenant_id"]) try: ids = [uuid.UUID(sid) for sid in subtask_ids] except (ValueError, TypeError): raise HTTPException(status_code=400, detail="Invalid subtask ID in list") from None from app.plugins.builtins.automation.agent_coordinator import AgentCoordinator result = await AgentCoordinator.aggregate_results(db, tenant_id, ids) return result def _subtask_to_response(s) -> SubtaskRead: """Convert AgentSubtask model to response schema.""" return SubtaskRead( id=str(s.id), parent_agent_id=str(s.parent_agent_id), child_agent_id=str(s.child_agent_id), task_description=s.task_description, status=s.status, result=s.result or {}, created_at=s.created_at.isoformat() if s.created_at else None, completed_at=s.completed_at.isoformat() if s.completed_at else None, updated_at=s.updated_at.isoformat() if s.updated_at else None, )