# Copyright (c) 2025 Patrick Motsch # All rights reserved. """ Mandatsweite WorkflowAutomation API. System-level API for workflows, runs, tasks — scoped by mandate membership, not by FeatureInstance. Uses mandate-scoped RBAC. RBAC model: - Read: mandate membership (user sees workflows in own mandates) - Write/Execute: mandate admin or isPlatformAdmin - isPlatformAdmin bypasses all checks """ import asyncio import json import logging import math import time import uuid from typing import Optional, List, Dict, Any from fastapi import APIRouter, Body, Depends, HTTPException, Path, Query, Request from fastapi.responses import JSONResponse, Response, StreamingResponse from slowapi import Limiter from slowapi.util import get_remote_address from modules.auth.authentication import getRequestContext, RequestContext from modules.connectors.connectorDbPostgre import DatabaseConnector from modules.datamodels.datamodelPagination import PaginationParams, PaginationMetadata, normalize_pagination_dict from modules.datamodels.datamodelWorkflowAutomation import ( AutoWorkflow, AutoVersion, AutoRun, AutoStepLog, AutoTask, ) from modules.dbHelpers.paginationHelpers import applyFiltersAndSort from modules.interfaces.interfaceDbApp import getRootInterface from modules.shared.i18nRegistry import apiRouteContext, resolveText from modules.workflowAutomation.helpers import ( _getWorkflowAutomationDb, _getUserMandateIds, _isUserMandateAdmin, _validateWorkflowAccess, _scopedWorkflowFilter, _scopedRunFilter, _parsePaginationOr400, _cascadeDeleteWorkflow, ) routeApiMsg = apiRouteContext("routeWorkflowAutomation") logger = logging.getLogger(__name__) limiter = Limiter(key_func=get_remote_address) router = APIRouter(prefix="/api/workflow-automation", tags=["WorkflowAutomation"]) # --------------------------------------------------------------------------- # Workflow CRUD # --------------------------------------------------------------------------- @router.get("/workflows") async def _listWorkflows( request: RequestContext = Depends(getRequestContext), pagination: Optional[str] = Query(default=None), mandateId: Optional[str] = Query(default=None), ): db = _getWorkflowAutomationDb() try: db._ensureTableExists(AutoWorkflow) scopeFilter = _scopedWorkflowFilter(request) if mandateId and scopeFilter is not None: if mandateId not in (scopeFilter.get("mandateId") or []): return {"items": [], "total": 0} scopeFilter = {"mandateId": mandateId} elif mandateId and scopeFilter is None: scopeFilter = {"mandateId": mandateId} params = _parsePaginationOr400(pagination) records = db.getRecordset(AutoWorkflow, recordFilter=scopeFilter, pagination=params) total = db.getRecordCount(AutoWorkflow, recordFilter=scopeFilter) if params else len(records or []) return {"items": records or [], "total": total} finally: db.close() @router.get("/workflows/{workflowId}") async def _getWorkflow( workflowId: str, request: RequestContext = Depends(getRequestContext), ): db = _getWorkflowAutomationDb() try: db._ensureTableExists(AutoWorkflow) wf = db.getRecord(AutoWorkflow, workflowId) if not wf: raise HTTPException(status_code=404, detail="Workflow not found") _validateWorkflowAccess(request, wf, "read") return wf finally: db.close() @router.post("/workflows") async def _createWorkflow( request: RequestContext = Depends(getRequestContext), body: Dict[str, Any] = {}, ): mandateId = body.get("mandateId") if not mandateId: raise HTTPException(status_code=400, detail="mandateId required") _validateWorkflowAccess(request, {"mandateId": mandateId}, "write") db = _getWorkflowAutomationDb() try: db._ensureTableExists(AutoWorkflow) data = {**body, "id": str(uuid.uuid4())} if request.user: data.setdefault("runAsPrincipal", str(request.user.id)) rec = db.recordCreate(AutoWorkflow, data) return rec finally: db.close() @router.put("/workflows/{workflowId}") async def _updateWorkflow( workflowId: str, request: RequestContext = Depends(getRequestContext), body: Dict[str, Any] = {}, ): db = _getWorkflowAutomationDb() try: db._ensureTableExists(AutoWorkflow) wf = db.getRecord(AutoWorkflow, workflowId) _validateWorkflowAccess(request, wf, "write") updated = db.recordModify(AutoWorkflow, workflowId, body) return updated finally: db.close() @router.delete("/workflows/{workflowId}") async def _deleteWorkflow( workflowId: str, request: RequestContext = Depends(getRequestContext), ): db = _getWorkflowAutomationDb() try: db._ensureTableExists(AutoWorkflow) wf = db.getRecord(AutoWorkflow, workflowId) _validateWorkflowAccess(request, wf, "delete") _cascadeDeleteWorkflow(db, workflowId) return {"deleted": True, "workflowId": workflowId} finally: db.close() # --------------------------------------------------------------------------- # Runs # --------------------------------------------------------------------------- @router.get("/runs") async def _listRuns( request: RequestContext = Depends(getRequestContext), pagination: Optional[str] = Query(default=None), mandateId: Optional[str] = Query(default=None), workflowId: Optional[str] = Query(default=None), ): db = _getWorkflowAutomationDb() try: db._ensureTableExists(AutoRun) scopeFilter = _scopedRunFilter(request) if mandateId: if scopeFilter is None: scopeFilter = {"mandateId": mandateId} elif "mandateId" in scopeFilter: if mandateId not in scopeFilter["mandateId"]: return {"items": [], "total": 0} scopeFilter = {"mandateId": mandateId} if workflowId: scopeFilter = {**(scopeFilter or {}), "workflowId": workflowId} params = _parsePaginationOr400(pagination) records = db.getRecordset(AutoRun, recordFilter=scopeFilter, pagination=params) total = db.getRecordCount(AutoRun, recordFilter=scopeFilter) if params else len(records or []) return {"items": records or [], "total": total} finally: db.close() @router.get("/runs/{runId}") async def _getRun( runId: str, request: RequestContext = Depends(getRequestContext), ): db = _getWorkflowAutomationDb() try: db._ensureTableExists(AutoRun) run = db.getRecord(AutoRun, runId) if not run: raise HTTPException(status_code=404, detail="Run not found") wfId = run.get("workflowId") if wfId: wf = db.getRecord(AutoWorkflow, wfId) _validateWorkflowAccess(request, wf, "read") return run finally: db.close() # --------------------------------------------------------------------------- # Tasks # --------------------------------------------------------------------------- @router.get("/tasks") async def _listTasks( request: RequestContext = Depends(getRequestContext), pagination: Optional[str] = Query(default=None), status: Optional[str] = Query(default=None), ): db = _getWorkflowAutomationDb() try: db._ensureTableExists(AutoTask) scopeFilter: Optional[Dict[str, Any]] = None if not request.isPlatformAdmin: userId = str(request.user.id) if request.user else None if not userId: return {"items": [], "total": 0} scopeFilter = {"assigneeId": userId} if status: scopeFilter = {**(scopeFilter or {}), "status": status} params = _parsePaginationOr400(pagination) records = db.getRecordset(AutoTask, recordFilter=scopeFilter, pagination=params) total = db.getRecordCount(AutoTask, recordFilter=scopeFilter) if params else len(records or []) return {"items": records or [], "total": total} finally: db.close() # --------------------------------------------------------------------------- # Versions # --------------------------------------------------------------------------- @router.get("/workflows/{workflowId}/versions") async def _listVersions( workflowId: str, request: RequestContext = Depends(getRequestContext), ): db = _getWorkflowAutomationDb() try: db._ensureTableExists(AutoWorkflow) wf = db.getRecord(AutoWorkflow, workflowId) _validateWorkflowAccess(request, wf, "read") db._ensureTableExists(AutoVersion) versions = db.getRecordset(AutoVersion, recordFilter={"workflowId": workflowId}) return {"items": versions or []} finally: db.close() # --------------------------------------------------------------------------- # Step logs # --------------------------------------------------------------------------- @router.get("/runs/{runId}/steps") async def _listStepLogs( runId: str, request: RequestContext = Depends(getRequestContext), ): db = _getWorkflowAutomationDb() try: db._ensureTableExists(AutoRun) run = db.getRecord(AutoRun, runId) if not run: raise HTTPException(status_code=404, detail="Run not found") wfId = run.get("workflowId") if wfId: wf = db.getRecord(AutoWorkflow, wfId) _validateWorkflowAccess(request, wf, "read") db._ensureTableExists(AutoStepLog) steps = db.getRecordset(AutoStepLog, recordFilter={"runId": runId}) return {"items": steps or []} finally: db.close() # --------------------------------------------------------------------------- # Internal helpers (mandate resolution, connector adapter) # --------------------------------------------------------------------------- def _resolveInstanceIdForWorkflow(db: DatabaseConnector, workflowId: str) -> Optional[str]: """Look up the featureInstanceId stored on the workflow record.""" if not workflowId: return None wf = db.getRecord(AutoWorkflow, workflowId) if db._ensureTableExists(AutoWorkflow) else None if not wf: return None return wf.get("featureInstanceId") or wf.get("targetFeatureInstanceId") def _resolveMandateIdForWorkflow(db: DatabaseConnector, workflowId: str) -> Optional[str]: """Look up the mandateId stored on the workflow record.""" if not workflowId: return None wf = db.getRecord(AutoWorkflow, workflowId) if db._ensureTableExists(AutoWorkflow) else None if not wf: return None return wf.get("mandateId") def _buildResolverDbInterface(chatService): """Build a DB adapter that ConnectorResolver can use to load UserConnections.""" class _ResolverDbAdapter: def __init__(self, appInterface): self._app = appInterface def getUserConnection(self, connectionId: str): if hasattr(self._app, "getUserConnectionById"): return self._app.getUserConnectionById(connectionId) return None appIf = getattr(chatService, "interfaceDbApp", None) if appIf: return _ResolverDbAdapter(appIf) return getattr(chatService, "interfaceDbComponent", None) def _getWorkflowAutomationInterface(context: RequestContext, mandateId: str, instanceId: str): """Build the WorkflowAutomation interface for template / import-export operations.""" from modules.interfaces.interfaceWorkflowAutomation import _getWorkflowAutomationInterface as _ifaceFactory return _ifaceFactory(context.user, mandateId, instanceId) def _loadEnvelopeFromFile(fileId: str, context: RequestContext) -> Optional[Dict[str, Any]]: """Load and parse a ``.workflow.json`` file from the Unified-Data-Bar by file id.""" try: import modules.interfaces.interfaceDbManagement as interfaceDbManagement mgmt = interfaceDbManagement.getInterface(context.user) rawBytes = mgmt.getFileData(fileId) except Exception as exc: logger.warning("Failed to load workflow file %s: %s", fileId, exc) raise HTTPException(status_code=404, detail=routeApiMsg(f"File {fileId} not found")) if not rawBytes: raise HTTPException(status_code=404, detail=routeApiMsg(f"File {fileId} is empty")) try: if isinstance(rawBytes, bytes): text = rawBytes.decode("utf-8") else: text = str(rawBytes) return json.loads(text) except Exception as exc: raise HTTPException( status_code=400, detail=routeApiMsg(f"File {fileId} is not valid JSON: {exc}"), ) def _getUserAccessibleInstanceIds(userId: str) -> List[str]: """Return all featureInstanceIds the user has enabled FeatureAccess for.""" rootIface = getRootInterface() allAccess = rootIface.getFeatureAccessesForUser(userId) or [] return [ a.featureInstanceId for a in allAccess if a.featureInstanceId and a.enabled ] # --------------------------------------------------------------------------- # Group 4 — Templates # --------------------------------------------------------------------------- @router.get("/templates") @limiter.limit("60/minute") def _listTemplates( request: Request, scope: Optional[str] = Query(None, description="Filter by scope: user, instance, mandate, system"), pagination: Optional[str] = Query(None, description="JSON-encoded PaginationParams object"), mode: Optional[str] = Query(None, description="'filterValues' for distinct column values, 'ids' for all filtered IDs"), column: Optional[str] = Query(None, description="Column key (required when mode=filterValues)"), mandateId: Optional[str] = Query(None, description="Mandate ID to scope templates"), context: RequestContext = Depends(getRequestContext), ): """List workflow templates with optional pagination. Supports the FormGeneratorTable backend pattern: - default: paginated/filtered/sorted ``{items, pagination}`` response - ``mode=filterValues&column=X``: distinct values for column X (cross-filtered) - ``mode=ids``: all IDs matching current filters """ if not context.user: raise HTTPException(status_code=401, detail=routeApiMsg("Authentication required")) userId = str(context.user.id) userMandateIds = _getUserMandateIds(userId) if not context.isPlatformAdmin else [] effectiveMandateId = mandateId or (userMandateIds[0] if userMandateIds else None) if not effectiveMandateId and not context.isPlatformAdmin: return {"templates": []} instanceId = None if effectiveMandateId: db = _getWorkflowAutomationDb() try: if db._ensureTableExists(AutoWorkflow): wfs = db.getRecordset(AutoWorkflow, recordFilter={"mandateId": effectiveMandateId}) for w in (wfs or []): fid = w.get("featureInstanceId") if fid: instanceId = fid break finally: db.close() iface = _getWorkflowAutomationInterface(context, effectiveMandateId or "", instanceId or "") templates = iface.getTemplates(scope=scope) from modules.dbHelpers.fkLabelResolver import enrichRowsWithFkLabels from modules.interfaces.interfaceDbApp import getRootInterface as _getRootIface enrichRowsWithFkLabels(templates, AutoWorkflow, db=_getRootIface().db) if mode == "filterValues": if not column: raise HTTPException(status_code=400, detail="column parameter required for mode=filterValues") from modules.dbHelpers.paginationHelpers import handleFilterValuesInMemory return handleFilterValuesInMemory(templates, column, pagination) if mode == "ids": from modules.dbHelpers.paginationHelpers import handleIdsInMemory return handleIdsInMemory(templates, pagination) paginationParams = None if pagination: try: paginationDict = json.loads(pagination) if paginationDict: paginationDict = normalize_pagination_dict(paginationDict) paginationParams = PaginationParams(**paginationDict) except (json.JSONDecodeError, ValueError) as e: raise HTTPException(status_code=400, detail=f"Invalid pagination parameter: {str(e)}") if paginationParams: filtered = applyFiltersAndSort(templates, paginationParams) totalItems = len(filtered) totalPages = math.ceil(totalItems / paginationParams.pageSize) if totalItems > 0 else 0 startIdx = (paginationParams.page - 1) * paginationParams.pageSize endIdx = startIdx + paginationParams.pageSize return { "items": filtered[startIdx:endIdx], "pagination": PaginationMetadata( currentPage=paginationParams.page, pageSize=paginationParams.pageSize, totalItems=totalItems, totalPages=totalPages, sort=paginationParams.sort, filters=paginationParams.filters, ).model_dump(), } return {"templates": templates} @router.post("/templates/from-workflow") @limiter.limit("30/minute") def _createTemplateFromWorkflow( request: Request, body: dict = Body(..., description="{ workflowId, scope? }"), context: RequestContext = Depends(getRequestContext), ) -> dict: """Create a template from an existing workflow.""" if not context.user: raise HTTPException(status_code=401, detail=routeApiMsg("Authentication required")) workflowId = body.get("workflowId") scope = body.get("scope", "user") if not workflowId: raise HTTPException(status_code=400, detail=routeApiMsg("workflowId required")) db = _getWorkflowAutomationDb() try: wf = db.getRecord(AutoWorkflow, workflowId) if db._ensureTableExists(AutoWorkflow) else None if not wf: raise HTTPException(status_code=404, detail=routeApiMsg("Workflow not found")) _validateWorkflowAccess(context, wf, "read") mandateId = wf.get("mandateId", "") instanceId = wf.get("featureInstanceId", "") finally: db.close() iface = _getWorkflowAutomationInterface(context, mandateId, instanceId) template = iface.createTemplateFromWorkflow(workflowId, scope=scope) if not template: raise HTTPException(status_code=404, detail=routeApiMsg("Workflow not found")) return template @router.post("/templates/{templateId}/copy") @limiter.limit("30/minute") def _copyTemplate( request: Request, templateId: str = Path(..., description="Template ID"), body: dict = Body(default={}, description="{ mandateId? }"), context: RequestContext = Depends(getRequestContext), ) -> dict: """Copy a template to a new user-owned workflow.""" if not context.user: raise HTTPException(status_code=401, detail=routeApiMsg("Authentication required")) mandateId = body.get("mandateId") if isinstance(body, dict) else None userId = str(context.user.id) if not mandateId: userMandateIds = _getUserMandateIds(userId) mandateId = userMandateIds[0] if userMandateIds else "" db = _getWorkflowAutomationDb() try: instanceId = None if db._ensureTableExists(AutoWorkflow) and mandateId: wfs = db.getRecordset(AutoWorkflow, recordFilter={"mandateId": mandateId}) for w in (wfs or []): fid = w.get("featureInstanceId") if fid: instanceId = fid break finally: db.close() iface = _getWorkflowAutomationInterface(context, mandateId or "", instanceId or "") workflow = iface.copyTemplateToUser(templateId) if not workflow: raise HTTPException(status_code=404, detail=routeApiMsg("Template not found")) return workflow @router.post("/templates/{templateId}/share") @limiter.limit("30/minute") def _shareTemplate( request: Request, templateId: str = Path(..., description="Template ID"), body: dict = Body(..., description="{ scope }"), context: RequestContext = Depends(getRequestContext), ) -> dict: """Share a template by changing its scope.""" if not context.user: raise HTTPException(status_code=401, detail=routeApiMsg("Authentication required")) scope = body.get("scope") if not scope or scope not in ("user", "instance", "mandate", "system"): raise HTTPException(status_code=400, detail=routeApiMsg("scope must be user, instance, mandate, or system")) mandateId = body.get("mandateId", "") userId = str(context.user.id) if not mandateId: userMandateIds = _getUserMandateIds(userId) mandateId = userMandateIds[0] if userMandateIds else "" db = _getWorkflowAutomationDb() try: instanceId = None if db._ensureTableExists(AutoWorkflow) and mandateId: wfs = db.getRecordset(AutoWorkflow, recordFilter={"mandateId": mandateId}) for w in (wfs or []): fid = w.get("featureInstanceId") if fid: instanceId = fid break finally: db.close() iface = _getWorkflowAutomationInterface(context, mandateId or "", instanceId or "") template = iface.shareTemplate(templateId, scope=scope) if not template: raise HTTPException(status_code=404, detail=routeApiMsg("Template not found")) return template # --------------------------------------------------------------------------- # Group 5 — Connections (SharePoint etc.) # --------------------------------------------------------------------------- def _buildServiceCenterContext(context: RequestContext, mandateId: str, instanceId: str = ""): """Build a ServiceCenterContext for connector/service calls.""" from modules.serviceCenter.context import ServiceCenterContext return ServiceCenterContext( user=context.user, mandate_id=str(context.mandateId) if context.mandateId else mandateId, feature_instance_id=instanceId, ) @router.get("/connections") @limiter.limit("300/minute") def _listConnections( request: Request, context: RequestContext = Depends(getRequestContext), ) -> dict: """Return the user's active connections (UserConnections) for Email/SharePoint node config.""" if not context.user: raise HTTPException(status_code=401, detail=routeApiMsg("Authentication required")) userId = str(context.user.id) userMandateIds = _getUserMandateIds(userId) if not context.isPlatformAdmin else [] mandateId = userMandateIds[0] if userMandateIds else "" from modules.serviceCenter import getService ctx = _buildServiceCenterContext(context, mandateId) chatService = getService("chat", ctx) connections = chatService.getUserConnections() items = [] for c in connections or []: conn = c if isinstance(c, dict) else (c.model_dump() if hasattr(c, "model_dump") else {}) authority = conn.get("authority") if hasattr(authority, "value"): authority = authority.value status = conn.get("status") if hasattr(status, "value"): status = status.value items.append({ "id": conn.get("id"), "authority": authority, "externalUsername": conn.get("externalUsername"), "externalEmail": conn.get("externalEmail"), "status": status, }) return {"connections": items} @router.get("/connections/{connectionId}/services") @limiter.limit("120/minute") async def _listConnectionServices( request: Request, connectionId: str = Path(..., description="Connection ID"), context: RequestContext = Depends(getRequestContext), ) -> dict: """Return the available services for a specific UserConnection.""" if not context.user: raise HTTPException(status_code=401, detail=routeApiMsg("Authentication required")) userId = str(context.user.id) userMandateIds = _getUserMandateIds(userId) if not context.isPlatformAdmin else [] mandateId = userMandateIds[0] if userMandateIds else "" try: from modules.connectors.connectorResolver import ConnectorResolver from modules.serviceCenter import getService as getSvc ctx = _buildServiceCenterContext(context, mandateId) chatService = getSvc("chat", ctx) securityService = getSvc("security", ctx) dbInterface = _buildResolverDbInterface(chatService) resolver = ConnectorResolver(securityService, dbInterface) provider = await resolver.resolve(connectionId) services = provider.getAvailableServices() _serviceLabels = { "sharepoint": "SharePoint", "clickup": "ClickUp", "outlook": "Outlook", "teams": "Teams", "onedrive": "OneDrive", "drive": "Google Drive", "gmail": "Gmail", "files": "Files (FTP)", "kdrive": "kDrive", "calendar": "Calendar", "contact": "Contacts", } _serviceIcons = { "sharepoint": "sharepoint", "clickup": "folder", "outlook": "mail", "teams": "chat", "onedrive": "cloud", "drive": "cloud", "gmail": "mail", "files": "folder", "kdrive": "cloud", "calendar": "calendar", "contact": "contact", } items = [ {"service": s, "label": _serviceLabels.get(s, s), "icon": _serviceIcons.get(s, "folder")} for s in services ] return {"services": items} except Exception as e: logger.error(f"Error listing services for connection {connectionId}: {e}") return JSONResponse({"services": [], "error": str(e)}, status_code=400) @router.get("/connections/{connectionId}/browse") @limiter.limit("300/minute") async def _browseConnectionService( request: Request, connectionId: str = Path(..., description="Connection ID"), service: str = Query(..., description="Service name (e.g. sharepoint, onedrive, outlook)"), path: str = Query("/", description="Path within the service to browse"), context: RequestContext = Depends(getRequestContext), ) -> dict: """Browse folders/items within a connection's service at a given path.""" if not context.user: raise HTTPException(status_code=401, detail=routeApiMsg("Authentication required")) userId = str(context.user.id) userMandateIds = _getUserMandateIds(userId) if not context.isPlatformAdmin else [] mandateId = userMandateIds[0] if userMandateIds else "" try: from modules.connectors.connectorResolver import ConnectorResolver from modules.serviceCenter import getService as getSvc ctx = _buildServiceCenterContext(context, mandateId) chatService = getSvc("chat", ctx) securityService = getSvc("security", ctx) dbInterface = _buildResolverDbInterface(chatService) resolver = ConnectorResolver(securityService, dbInterface) adapter = await resolver.resolveService(connectionId, service) entries = await adapter.browse(path, filter=None) items = [] for entry in (entries or []): items.append({ "name": entry.name, "path": entry.path, "isFolder": entry.isFolder, "size": entry.size, "mimeType": entry.mimeType, "metadata": entry.metadata if hasattr(entry, "metadata") else {}, }) return {"items": items, "path": path, "service": service} except Exception as e: logger.error(f"Error browsing {service} for connection {connectionId} at '{path}': {e}") return JSONResponse({"items": [], "error": str(e)}, status_code=400) # --------------------------------------------------------------------------- # Group 6 — Import / Export # --------------------------------------------------------------------------- @router.post("/workflows/import") @limiter.limit("30/minute") def _importWorkflow( request: Request, body: dict = Body( ..., description=( "{ envelope: , existingWorkflowId?: str, " "fileId?: str, mandateId?: str } — supply EITHER the envelope " "inline OR a fileId of a previously uploaded workflow file (.workflow.json)" ), ), context: RequestContext = Depends(getRequestContext), ) -> dict: """Import a workflow from a versioned-envelope file. Two input modes: - ``envelope``: the parsed workflow-file payload - ``fileId``: the id of a previously uploaded ``.workflow.json`` """ if not context.user: raise HTTPException(status_code=401, detail=routeApiMsg("Authentication required")) from modules.nodeCatalog._workflowFileSchema import WorkflowFileSchemaError mandateId = body.get("mandateId") if isinstance(body, dict) else None userId = str(context.user.id) if not mandateId: userMandateIds = _getUserMandateIds(userId) mandateId = userMandateIds[0] if userMandateIds else "" if not mandateId and not context.isPlatformAdmin: raise HTTPException(status_code=400, detail=routeApiMsg("mandateId required")) _validateWorkflowAccess(context, {"mandateId": mandateId}, "write") db = _getWorkflowAutomationDb() try: instanceId = None if db._ensureTableExists(AutoWorkflow) and mandateId: wfs = db.getRecordset(AutoWorkflow, recordFilter={"mandateId": mandateId}) for w in (wfs or []): fid = w.get("featureInstanceId") if fid: instanceId = fid break finally: db.close() iface = _getWorkflowAutomationInterface(context, mandateId or "", instanceId or "") envelope = body.get("envelope") if isinstance(body, dict) else None fileId = body.get("fileId") if isinstance(body, dict) else None existingWorkflowId = body.get("existingWorkflowId") if isinstance(body, dict) else None if not envelope and fileId: envelope = _loadEnvelopeFromFile(str(fileId), context) if not envelope: raise HTTPException( status_code=400, detail=routeApiMsg("Body must contain 'envelope' or 'fileId'"), ) try: result = iface.importWorkflowFromDict(envelope, existingWorkflowId=existingWorkflowId) except WorkflowFileSchemaError as exc: raise HTTPException(status_code=400, detail=str(exc)) except ValueError as exc: raise HTTPException(status_code=404, detail=str(exc)) return result @router.get("/workflows/{workflowId}/export") @limiter.limit("60/minute") def _exportWorkflow( request: Request, workflowId: str = Path(..., description="Workflow ID"), download: bool = Query(False, description="If true, return as file download"), context: RequestContext = Depends(getRequestContext), ): """Export a workflow as a versioned-envelope JSON file.""" if not context.user: raise HTTPException(status_code=401, detail=routeApiMsg("Authentication required")) from modules.nodeCatalog._workflowFileSchema import buildFileName db = _getWorkflowAutomationDb() try: wf = db.getRecord(AutoWorkflow, workflowId) if db._ensureTableExists(AutoWorkflow) else None if not wf: raise HTTPException(status_code=404, detail=routeApiMsg("Workflow not found")) _validateWorkflowAccess(context, wf, "read") mandateId = wf.get("mandateId", "") instanceId = wf.get("featureInstanceId", "") finally: db.close() iface = _getWorkflowAutomationInterface(context, mandateId, instanceId) envelope = iface.exportWorkflowToDict(workflowId) if envelope is None: raise HTTPException(status_code=404, detail=routeApiMsg("Workflow not found")) if not download: return {"envelope": envelope, "fileName": buildFileName(envelope.get("label", "workflow"))} fileName = buildFileName(envelope.get("label", "workflow")) payload = json.dumps(envelope, ensure_ascii=False, indent=2).encode("utf-8") return Response( content=payload, media_type="application/json", headers={"Content-Disposition": f'attachment; filename="{fileName}"'}, ) # --------------------------------------------------------------------------- # Group 7 — Options # --------------------------------------------------------------------------- @router.get("/options/user.connection") @limiter.limit("60/minute") def _getUserConnectionOptions( request: Request, authority: Optional[str] = Query(None, description="Optional authority filter (e.g. 'msft', 'google', 'clickup', 'local')"), activeOnly: bool = Query(True, description="If true (default), only ACTIVE connections are returned"), context: RequestContext = Depends(getRequestContext), ) -> dict: """Return current user's UserConnections as { options: [{ value, label }] }. Used by node parameters with frontendType='userConnection'. """ if not context.user: raise HTTPException(status_code=401, detail=routeApiMsg("Authentication required")) rootInterface = getRootInterface() try: connections = rootInterface.getUserConnections(str(context.user.id)) or [] except Exception as e: logger.error("_getUserConnectionOptions: failed to load connections: %s", e, exc_info=True) return {"options": []} wanted = (authority or "").strip().lower() or None options: List[Dict[str, str]] = [] for conn in connections: connStatus = getattr(conn, "status", None) statusVal = connStatus.value if hasattr(connStatus, "value") else str(connStatus or "") if activeOnly and statusVal.lower() != "active": continue connAuthority = getattr(conn, "authority", None) authorityVal = (connAuthority.value if hasattr(connAuthority, "value") else str(connAuthority or "")).lower() if wanted and authorityVal != wanted: continue username = getattr(conn, "externalUsername", "") or "" email = getattr(conn, "externalEmail", "") or "" connId = str(getattr(conn, "id", "") or "") labelParts = [p for p in [username, email] if p] label = " — ".join(labelParts) if labelParts else connId if authorityVal: label = f"[{authorityVal}] {label}" value = f"connection:{authorityVal}:{username}" if authorityVal and username else connId options.append({"value": value, "label": label}) return {"options": options} @router.get("/options/feature.instance") @limiter.limit("60/minute") def _getFeatureInstanceOptions( request: Request, featureCode: str = Query(..., description="Feature code to filter by (e.g. 'trustee', 'redmine', 'clickup')"), enabledOnly: bool = Query(True, description="If true (default), only enabled feature instances are returned"), context: RequestContext = Depends(getRequestContext), ) -> dict: """Return mandate-scoped FeatureInstances for the given featureCode. Used by node parameters with frontendType='featureInstance'. """ if not context.user: raise HTTPException(status_code=401, detail=routeApiMsg("Authentication required")) code = (featureCode or "").strip().lower() if not code: raise HTTPException(status_code=400, detail=routeApiMsg("featureCode query parameter is required")) userId = str(context.user.id) userMandateIds = _getUserMandateIds(userId) if not context.isPlatformAdmin else [] if not userMandateIds and not context.isPlatformAdmin: return {"options": []} rootInterface = getRootInterface() allOptions: List[Dict[str, str]] = [] targetMandateIds = userMandateIds if not context.isPlatformAdmin else [] if context.isPlatformAdmin: try: from modules.datamodels.datamodelUam import Mandate mandates = rootInterface.db.getRecordset(Mandate) or [] targetMandateIds = [str(m.get("id") if isinstance(m, dict) else getattr(m, "id", "")) for m in mandates] except Exception: targetMandateIds = [] for mid in targetMandateIds: try: instances = rootInterface.getFeatureInstancesByMandate(mid, enabledOnly=bool(enabledOnly)) or [] except Exception as e: logger.error("_getFeatureInstanceOptions: failed to load instances mandateId=%s: %s", mid, e, exc_info=True) continue for fi in instances: fiCode = (getattr(fi, "featureCode", "") or "").strip().lower() if fiCode != code: continue fiId = str(getattr(fi, "id", "") or "") if not fiId: continue rawLabel = getattr(fi, "label", None) or getattr(fi, "name", None) or fiId allOptions.append({"value": fiId, "label": f"{rawLabel} ({fiCode})"}) return {"options": allOptions} # --------------------------------------------------------------------------- # Group 8 — Metrics # --------------------------------------------------------------------------- @router.get("/metrics") @limiter.limit("60/minute") def _getMetrics( request: Request, mandateId: Optional[str] = Query(None, description="Filter metrics by mandate"), context: RequestContext = Depends(getRequestContext), ) -> dict: """Aggregated metrics for the monitoring dashboard.""" if not context.user: raise HTTPException(status_code=401, detail=routeApiMsg("Authentication required")) userId = str(context.user.id) userMandateIds = _getUserMandateIds(userId) if not context.isPlatformAdmin else [] if mandateId: if not context.isPlatformAdmin and mandateId not in userMandateIds: raise HTTPException(status_code=403, detail=routeApiMsg("Access denied")) scopeFilter: Dict[str, Any] = {"mandateId": mandateId, "isTemplate": False} elif context.isPlatformAdmin: scopeFilter = {"isTemplate": False} elif userMandateIds: scopeFilter = {"mandateId": userMandateIds, "isTemplate": False} else: return { "workflowCount": 0, "activeWorkflows": 0, "totalRuns": 0, "runsByStatus": {}, "totalTasks": 0, "tasksByStatus": {}, "totalTokens": 0, "totalCredits": 0.0, } db = _getWorkflowAutomationDb() try: workflows = db.getRecordset(AutoWorkflow, recordFilter=scopeFilter) or [] if db._ensureTableExists(AutoWorkflow) else [] wfIds = [w.get("id") for w in workflows] runFilter = {"workflowId": {"$in": wfIds}} if wfIds else {"workflowId": "__none__"} runs = db.getRecordset(AutoRun, recordFilter=runFilter) or [] if db._ensureTableExists(AutoRun) else [] tasks = db.getRecordset(AutoTask, recordFilter=runFilter) or [] if db._ensureTableExists(AutoTask) else [] finally: db.close() runsByStatus: Dict[str, int] = {} totalTokens = 0 totalCredits = 0.0 for r in runs: s = r.get("status", "unknown") runsByStatus[s] = runsByStatus.get(s, 0) + 1 totalTokens += r.get("costTokens", 0) or 0 totalCredits += r.get("costCredits", 0.0) or 0.0 tasksByStatus: Dict[str, int] = {} for t in tasks: s = t.get("status", "unknown") tasksByStatus[s] = tasksByStatus.get(s, 0) + 1 return { "workflowCount": len(workflows), "activeWorkflows": sum(1 for w in workflows if w.get("active")), "totalRuns": len(runs), "runsByStatus": runsByStatus, "totalTasks": len(tasks), "tasksByStatus": tasksByStatus, "totalTokens": totalTokens, "totalCredits": round(totalCredits, 4), } # --------------------------------------------------------------------------- # Group 9 — SSE Stream + Stop + Run Detail # --------------------------------------------------------------------------- @router.get("/runs/{runId}/stream") async def _getRunStream( request: Request, runId: str = Path(..., description="Run ID"), context: RequestContext = Depends(getRequestContext), ): """SSE stream for live step-log updates during a workflow run.""" db = _getWorkflowAutomationDb() try: if not db._ensureTableExists(AutoRun): raise HTTPException(status_code=404, detail=routeApiMsg("Run not found")) runs = db.getRecordset(AutoRun, recordFilter={"id": runId}) if not runs: raise HTTPException(status_code=404, detail=routeApiMsg("Run not found")) run = dict(runs[0]) finally: db.close() if not context.isPlatformAdmin: userId = str(context.user.id) if context.user else None runOwner = run.get("ownerId") runMandate = run.get("mandateId") if runOwner == userId: pass elif runMandate and userId and _isUserMandateAdmin(userId, runMandate): pass else: raise HTTPException(status_code=403, detail=routeApiMsg("Access denied")) from modules.shared.eventManager import get_event_manager sseEventManager = get_event_manager() queueId = f"run-trace-{runId}" sseEventManager.create_queue(queueId) async def _sseGenerator(): queue = sseEventManager.get_queue(queueId) if not queue: return while True: try: event = await asyncio.wait_for(queue.get(), timeout=30) except asyncio.TimeoutError: yield "data: {\"type\": \"keepalive\"}\n\n" continue if event is None: break payload = event.get("data", event) if isinstance(event, dict) else event yield f"data: {json.dumps(payload, default=str)}\n\n" eventType = payload.get("type", "") if isinstance(payload, dict) else "" if eventType in ("run_complete", "run_failed"): break await sseEventManager.cleanup(queueId, delay=10) return StreamingResponse( _sseGenerator(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", }, ) @router.post("/runs/{runId}/stop") @limiter.limit("30/minute") def _stopWorkflowRun( request: Request, runId: str = Path(..., description="Run ID"), context: RequestContext = Depends(getRequestContext), ): """Stop a running workflow execution.""" db = _getWorkflowAutomationDb() try: if not db._ensureTableExists(AutoRun): raise HTTPException(status_code=404, detail=routeApiMsg("Run not found")) runs = db.getRecordset(AutoRun, recordFilter={"id": runId}) if not runs: raise HTTPException(status_code=404, detail=routeApiMsg("Run not found")) run = dict(runs[0]) if not context.isPlatformAdmin: userId = str(context.user.id) if context.user else None runOwner = run.get("ownerId") runMandate = run.get("mandateId") if runOwner == userId: pass elif runMandate and userId and _isUserMandateAdmin(userId, runMandate): pass else: raise HTTPException(status_code=403, detail=routeApiMsg("Access denied")) from modules.workflowAutomation.engine.executionEngine import requestRunStop flagged = requestRunStop(runId) if not flagged: currentStatus = run.get("status", "") if currentStatus in ("completed", "failed", "stopped"): return {"status": currentStatus, "runId": runId, "message": "Run already finished"} stopUpdates: Dict[str, Any] = {"status": "stopped"} if not run.get("completedAt"): stopUpdates["completedAt"] = time.time() db.recordModify(AutoRun, runId, stopUpdates) return {"status": "stopped", "runId": runId, "message": "Run not active in memory, marked as stopped"} return {"status": "stopping", "runId": runId, "message": "Stop signal sent"} finally: db.close() # --------------------------------------------------------------------------- # Run Detail (enriched with step logs, workflow info, files) # --------------------------------------------------------------------------- _FILE_REF_KEYS = ("fileId", "documentId", "fileIds", "documents") def _extractFileIdsFromValue(value, accumulator: set) -> None: """Recursively scan a value (dict/list/str) for file id references.""" if isinstance(value, dict): for key, sub in value.items(): if key in _FILE_REF_KEYS: _collectFileIdsFromRef(sub, accumulator) else: _extractFileIdsFromValue(sub, accumulator) elif isinstance(value, list): for item in value: _extractFileIdsFromValue(item, accumulator) def _collectFileIdsFromRef(val, accumulator: set) -> None: """Add file ids from a value located under a known file-reference key.""" if isinstance(val, str) and val: accumulator.add(val) elif isinstance(val, list): for v in val: if isinstance(v, str) and v: accumulator.add(v) elif isinstance(v, dict) and v.get("id"): accumulator.add(v["id"]) elif isinstance(val, dict) and val.get("id"): accumulator.add(val["id"]) @router.get("/runs/{runId}/detail") @limiter.limit("60/minute") def _getRunDetail( request: Request, runId: str = Path(..., description="Run ID"), context: RequestContext = Depends(getRequestContext), ) -> dict: """Get full detail for a single run: metadata, step logs, linked files.""" if not context.user: raise HTTPException(status_code=401, detail=routeApiMsg("Authentication required")) userId = str(context.user.id) db = _getWorkflowAutomationDb() try: if not db._ensureTableExists(AutoRun): raise HTTPException(status_code=404, detail=routeApiMsg("Run not found")) runs = db.getRecordset(AutoRun, recordFilter={"id": runId}) if not runs: raise HTTPException(status_code=404, detail=routeApiMsg("Run not found")) run = dict(runs[0]) wfId = run.get("workflowId") workflow: dict = {} if wfId and db._ensureTableExists(AutoWorkflow): wfs = db.getRecordset(AutoWorkflow, recordFilter={"id": wfId}) if wfs: workflow = dict(wfs[0]) tid = workflow.get("targetFeatureInstanceId") or workflow.get("featureInstanceId") accessibleIds = _getUserAccessibleInstanceIds(userId) isOwner = run.get("ownerId") == userId if not isOwner and (not tid or tid not in accessibleIds) and not context.isPlatformAdmin: raise HTTPException(status_code=403, detail=routeApiMsg("Access denied")) steps: list = [] if db._ensureTableExists(AutoStepLog): stepRecords = db.getRecordset(AutoStepLog, recordFilter={"runId": runId}) or [] steps = [dict(s) for s in stepRecords] steps.sort(key=lambda s: s.get("startedAt") or 0) allFileIds: set = set() perStepFileIds: list = [] for step in steps: inputIds: set = set() outputIds: set = set() _extractFileIdsFromValue(step.get("inputSnapshot") or {}, inputIds) _extractFileIdsFromValue(step.get("output") or {}, outputIds) perStepFileIds.append((inputIds, outputIds)) allFileIds.update(inputIds) allFileIds.update(outputIds) nodeOutputs = run.get("nodeOutputs") or {} runLevelIds: set = set() _extractFileIdsFromValue(nodeOutputs, runLevelIds) allFileIds.update(runLevelIds) fileMetaById: dict = {} try: from modules.datamodels.datamodelFiles import FileItem from modules.interfaces.interfaceDbManagement import ComponentObjects mgmtDb = ComponentObjects().db if mgmtDb._ensureTableExists(FileItem): for fid in allFileIds: try: rec = mgmtDb.getRecord(FileItem, fid) if rec: recDict = dict(rec) fileMetaById[fid] = { "id": fid, "fileName": recDict.get("fileName") or recDict.get("name"), } except Exception: pass except Exception as e: logger.warning("_getRunDetail: file lookup failed: %s", e) from modules.workflowAutomation.engine.workflowArtifactVisibility import suppress_workflow_file_in_workspace_ui def _resolveFileList(ids: set) -> list: rows = [dict(fileMetaById[fid]) for fid in ids if fid in fileMetaById] return [m for m in rows if not suppress_workflow_file_in_workspace_ui(m)] assignedFileIds: set = set() for step, (inputIds, outputIds) in zip(steps, perStepFileIds): step["inputFiles"] = _resolveFileList(inputIds) step["outputFiles"] = _resolveFileList(outputIds) assignedFileIds.update(inputIds) assignedFileIds.update(outputIds) unassignedFiles = _resolveFileList(allFileIds - assignedFileIds) allFiles = _resolveFileList(allFileIds) run["workflowLabel"] = run.get("label") or workflow.get("label") or wfId run["targetFeatureInstanceId"] = tid targetInstanceLabel = None if tid: try: from modules.dbHelpers.fkLabelResolver import resolveInstanceLabels labelMap = resolveInstanceLabels(db, [tid]) targetInstanceLabel = labelMap.get(tid) except Exception: pass run["targetInstanceLabel"] = targetInstanceLabel return { "run": run, "workflow": { "id": workflow.get("id"), "label": workflow.get("label"), "targetFeatureInstanceId": tid, "featureInstanceId": workflow.get("featureInstanceId"), "tags": workflow.get("tags", []), } if workflow else None, "steps": steps, "files": allFiles, "unassignedFiles": unassignedFiles, } finally: db.close() # --------------------------------------------------------------------------- # Execute workflow # --------------------------------------------------------------------------- def _buildExecuteRunEnvelope( body: Dict[str, Any], workflow: Optional[Dict[str, Any]], userId: Optional[str], requestLang: Optional[str] = None, ) -> Dict[str, Any]: """Build normalized run envelope from POST /execute body.""" from modules.workflowAutomation.engine.runEnvelope import ( default_run_envelope, merge_run_envelope, normalize_run_envelope, ) from modules.workflowAutomation.editor.entryPoints import find_invocation if isinstance(body.get("runEnvelope"), dict): env = normalize_run_envelope(body["runEnvelope"], user_id=userId) pl = body.get("payload") if isinstance(pl, dict): env = merge_run_envelope(env, {"payload": pl}) return env entryPointId = body.get("entryPointId") if entryPointId: if not workflow: raise HTTPException( status_code=400, detail=routeApiMsg("entryPointId requires a saved workflow (workflowId must refer to a stored workflow)"), ) inv = find_invocation(workflow, entryPointId) if not inv: raise HTTPException(status_code=400, detail=routeApiMsg("entryPointId not found on workflow")) if not inv.get("enabled", True): raise HTTPException(status_code=400, detail=routeApiMsg("entry point is disabled")) kind = inv.get("kind", "manual") trigMap = { "manual": "manual", "form": "form", "schedule": "schedule", "always_on": "event", "email": "email", "webhook": "webhook", "api": "api", "event": "event", } trig = trigMap.get(kind, "manual") title = inv.get("title") or {} label = resolveText(title) base = default_run_envelope( trig, entry_point_id=inv.get("id"), entry_point_label=label or None, ) pl = body.get("payload") if isinstance(pl, dict): base = merge_run_envelope(base, {"payload": pl}) return normalize_run_envelope(base, user_id=userId) env = normalize_run_envelope(None, user_id=userId) pl = body.get("payload") if isinstance(pl, dict): env = merge_run_envelope(env, {"payload": pl}) return env @router.post("/workflows/{workflowId}/execute") @limiter.limit("30/minute") async def _executeWorkflow( request: Request, workflowId: str = Path(..., description="Workflow ID"), body: dict = Body(..., description="{ graph?, entryPointId?, payload?, runEnvelope? }"), context: RequestContext = Depends(getRequestContext), ) -> dict: """Execute a workflow graph.""" from modules.workflowAutomation.mainWorkflowAutomation import _getWorkflowAutomationServices from modules.workflowAutomation.engine.executionEngine import executeGraph from modules.interfaces.interfaceWorkflowAutomation import _getWorkflowAutomationInterface from modules.workflows.processing.shared.methodDiscovery import discoverMethods userId = str(context.user.id) if context.user else None logger.info("workflowAutomation execute: workflowId=%s userId=%s", workflowId, userId) db = _getWorkflowAutomationDb() try: db._ensureTableExists(AutoWorkflow) wf = db.getRecord(AutoWorkflow, workflowId) finally: db.close() if not wf: raise HTTPException(status_code=404, detail=routeApiMsg("Workflow not found")) _validateWorkflowAccess(context, wf, "execute") mandateId = wf.get("mandateId") instanceId = wf.get("featureInstanceId") or "" targetFeatureInstanceId = wf.get("targetFeatureInstanceId") services = _getWorkflowAutomationServices( context.user, mandateId=mandateId, featureInstanceId=instanceId, ) discoverMethods(services) graph = body.get("graph") or body reqNodes = graph.get("nodes") or [] workflowForEnvelope: Optional[Dict[str, Any]] = wf if len(reqNodes) == 0: graph = wf.get("graph") or {} logger.info("workflowAutomation execute: loaded graph from workflow %s", workflowId) nodesCount = len(graph.get("nodes") or []) connectionsCount = len(graph.get("connections") or []) logger.info( "workflowAutomation execute: graph nodes=%d connections=%d workflowId=%s mandateId=%s", nodesCount, connectionsCount, workflowId, mandateId, ) runEnv = _buildExecuteRunEnvelope( body, workflowForEnvelope, userId, getattr(context.user, "language", None) if context.user else None, ) wfLabel = wf.get("label") if isinstance(wf, dict) else getattr(wf, "label", None) iface = _getWorkflowAutomationInterface(context.user, mandateId, instanceId) result = await executeGraph( graph=graph, services=services, workflowId=workflowId, instanceId=instanceId, userId=userId, mandateId=mandateId, automation2_interface=iface, run_envelope=runEnv, label=wfLabel, targetFeatureInstanceId=targetFeatureInstanceId, ) logger.info( "workflowAutomation execute result: success=%s error=%s paused=%s", result.get("success"), result.get("error"), result.get("paused"), ) return result # --------------------------------------------------------------------------- # Version management # --------------------------------------------------------------------------- @router.post("/workflows/{workflowId}/versions/draft") @limiter.limit("30/minute") async def _createDraftVersion( request: Request, workflowId: str = Path(..., description="Workflow ID"), context: RequestContext = Depends(getRequestContext), ) -> dict: """Create a new draft version from the workflow's current graph.""" from modules.interfaces.interfaceWorkflowAutomation import _getWorkflowAutomationInterface db = _getWorkflowAutomationDb() try: db._ensureTableExists(AutoWorkflow) wf = db.getRecord(AutoWorkflow, workflowId) finally: db.close() if not wf: raise HTTPException(status_code=404, detail=routeApiMsg("Workflow not found")) _validateWorkflowAccess(context, wf, "write") mandateId = wf.get("mandateId") instanceId = wf.get("featureInstanceId") or "" iface = _getWorkflowAutomationInterface(context.user, mandateId, instanceId) version = iface.createDraftVersion(workflowId) if not version: raise HTTPException(status_code=404, detail=routeApiMsg("Workflow not found")) return version @router.post("/versions/{versionId}/publish") @limiter.limit("30/minute") async def _publishVersion( request: Request, versionId: str = Path(..., description="Version ID"), context: RequestContext = Depends(getRequestContext), ) -> dict: """Publish a draft version. Archives the previously published version.""" from modules.interfaces.interfaceWorkflowAutomation import _getWorkflowAutomationInterface db = _getWorkflowAutomationDb() try: db._ensureTableExists(AutoVersion) ver = db.getRecord(AutoVersion, versionId) if not ver: raise HTTPException(status_code=404, detail=routeApiMsg("Version not found")) wfId = ver.get("workflowId") db._ensureTableExists(AutoWorkflow) wf = db.getRecord(AutoWorkflow, wfId) if wfId else None finally: db.close() if not wf: raise HTTPException(status_code=404, detail=routeApiMsg("Workflow not found")) _validateWorkflowAccess(context, wf, "write") mandateId = wf.get("mandateId") instanceId = wf.get("featureInstanceId") or "" userId = str(context.user.id) if context.user else None iface = _getWorkflowAutomationInterface(context.user, mandateId, instanceId) version = iface.publishVersion(versionId, userId=userId) if not version: raise HTTPException(status_code=400, detail=routeApiMsg("Version not found or not in draft status")) return version @router.post("/versions/{versionId}/unpublish") @limiter.limit("30/minute") async def _unpublishVersion( request: Request, versionId: str = Path(..., description="Version ID"), context: RequestContext = Depends(getRequestContext), ) -> dict: """Unpublish a version (revert to draft).""" from modules.interfaces.interfaceWorkflowAutomation import _getWorkflowAutomationInterface db = _getWorkflowAutomationDb() try: db._ensureTableExists(AutoVersion) ver = db.getRecord(AutoVersion, versionId) if not ver: raise HTTPException(status_code=404, detail=routeApiMsg("Version not found")) wfId = ver.get("workflowId") db._ensureTableExists(AutoWorkflow) wf = db.getRecord(AutoWorkflow, wfId) if wfId else None finally: db.close() if not wf: raise HTTPException(status_code=404, detail=routeApiMsg("Workflow not found")) _validateWorkflowAccess(context, wf, "write") mandateId = wf.get("mandateId") instanceId = wf.get("featureInstanceId") or "" iface = _getWorkflowAutomationInterface(context.user, mandateId, instanceId) version = iface.unpublishVersion(versionId) if not version: raise HTTPException(status_code=400, detail=routeApiMsg("Version not found or not published")) return version @router.post("/versions/{versionId}/archive") @limiter.limit("30/minute") async def _archiveVersion( request: Request, versionId: str = Path(..., description="Version ID"), context: RequestContext = Depends(getRequestContext), ) -> dict: """Archive a version.""" from modules.interfaces.interfaceWorkflowAutomation import _getWorkflowAutomationInterface db = _getWorkflowAutomationDb() try: db._ensureTableExists(AutoVersion) ver = db.getRecord(AutoVersion, versionId) if not ver: raise HTTPException(status_code=404, detail=routeApiMsg("Version not found")) wfId = ver.get("workflowId") db._ensureTableExists(AutoWorkflow) wf = db.getRecord(AutoWorkflow, wfId) if wfId else None finally: db.close() if not wf: raise HTTPException(status_code=404, detail=routeApiMsg("Workflow not found")) _validateWorkflowAccess(context, wf, "write") mandateId = wf.get("mandateId") instanceId = wf.get("featureInstanceId") or "" iface = _getWorkflowAutomationInterface(context.user, mandateId, instanceId) version = iface.archiveVersion(versionId) if not version: raise HTTPException(status_code=404, detail=routeApiMsg("Version not found")) return version # --------------------------------------------------------------------------- # Node types + Editor metadata # --------------------------------------------------------------------------- @router.get("/node-types") @limiter.limit("60/minute") async def _getNodeTypes( request: Request, mandateId: str = Query(..., description="Mandate ID for context"), featureInstanceId: Optional[str] = Query(default=None, description="Feature instance ID"), language: str = Query("en", description="Localization (en, de, fr)"), context: RequestContext = Depends(getRequestContext), ) -> dict: """Return node types for the flow builder: static + I/O from methodDiscovery.""" from modules.workflowAutomation.mainWorkflowAutomation import _getWorkflowAutomationServices from modules.workflowAutomation.editor.nodeRegistry import getNodeTypesForApi logger.info("workflowAutomation node-types: mandateId=%s language=%s", mandateId, language) services = _getWorkflowAutomationServices( context.user, mandateId=mandateId, featureInstanceId=featureInstanceId or "", ) result = getNodeTypesForApi(services, language=language) logger.info( "workflowAutomation node-types response: %d nodeTypes %d categories", len(result.get("nodeTypes", [])), len(result.get("categories", [])), ) return result @router.post("/upstream-paths") @limiter.limit("60/minute") async def _postUpstreamPaths( request: Request, body: Dict[str, Any] = Body(...), context: RequestContext = Depends(getRequestContext), ) -> dict: """Return pickable upstream DataRef paths for a node (draft graph in body).""" from modules.workflowAutomation.editor.upstreamPathsService import compute_upstream_paths graph = body.get("graph") nodeId = body.get("nodeId") if not isinstance(graph, dict) or not nodeId: raise HTTPException(status_code=400, detail=routeApiMsg("graph and nodeId are required")) paths = compute_upstream_paths(graph, str(nodeId)) return {"paths": paths} @router.post("/condition-meta") @limiter.limit("120/minute") async def _postConditionMeta( request: Request, body: Dict[str, Any] = Body(...), language: str = Query("de", description="Localization (en, de, fr)"), context: RequestContext = Depends(getRequestContext), ) -> dict: """Return valueKind and operators for a DataRef (backend-driven If/Else UI).""" from modules.workflowAutomation.editor.conditionOperators import resolve_condition_meta graph = body.get("graph") ref = body.get("ref") nodeId = body.get("nodeId") if not isinstance(graph, dict) or not isinstance(ref, dict): raise HTTPException(status_code=400, detail=routeApiMsg("graph and ref are required")) graphPayload = dict(graph) if nodeId: graphPayload["targetNodeId"] = str(nodeId) return resolve_condition_meta(graphPayload, ref, lang=language) @router.post("/graph-data-sources") @limiter.limit("120/minute") async def _postGraphDataSources( request: Request, body: Dict[str, Any] = Body(...), context: RequestContext = Depends(getRequestContext), ) -> dict: """Scope-aware data sources for the DataPicker.""" from modules.workflowAutomation.editor.upstreamPathsService import compute_graph_data_sources graph = body.get("graph") nodeId = body.get("nodeId") if not isinstance(graph, dict) or not nodeId: raise HTTPException(status_code=400, detail=routeApiMsg("graph and nodeId are required")) return compute_graph_data_sources(graph, str(nodeId)) @router.get("/upstream-paths/{nodeId}") @limiter.limit("60/minute") async def _getUpstreamPathsSaved( request: Request, nodeId: str = Path(..., description="Target node id"), workflowId: str = Query(..., description="Workflow id whose saved graph is used"), context: RequestContext = Depends(getRequestContext), ) -> dict: """Return upstream paths using the persisted workflow graph.""" from modules.workflowAutomation.editor.upstreamPathsService import compute_upstream_paths if not workflowId: raise HTTPException(status_code=400, detail=routeApiMsg("workflowId is required")) db = _getWorkflowAutomationDb() try: db._ensureTableExists(AutoWorkflow) wf = db.getRecord(AutoWorkflow, workflowId) finally: db.close() if not wf: raise HTTPException(status_code=404, detail=routeApiMsg("Workflow not found")) _validateWorkflowAccess(context, wf, "read") graph = wf.get("graph") or {} paths = compute_upstream_paths(graph if isinstance(graph, dict) else {}, str(nodeId)) return {"paths": paths} # --------------------------------------------------------------------------- # Tasks complete/cancel # --------------------------------------------------------------------------- @router.post("/tasks/{taskId}/complete") @limiter.limit("30/minute") async def _completeTask( request: Request, taskId: str = Path(..., description="Task ID"), body: dict = Body(..., description="{ result }"), context: RequestContext = Depends(getRequestContext), ) -> dict: """Complete a human task and resume the workflow.""" from modules.workflowAutomation.mainWorkflowAutomation import _getWorkflowAutomationServices from modules.workflowAutomation.engine.executionEngine import executeGraph from modules.interfaces.interfaceWorkflowAutomation import _getWorkflowAutomationInterface db = _getWorkflowAutomationDb() try: db._ensureTableExists(AutoTask) task = db.getRecord(AutoTask, taskId) finally: db.close() if not task: raise HTTPException(status_code=404, detail=routeApiMsg("Task not found")) wfId = task.get("workflowId") db2 = _getWorkflowAutomationDb() try: db2._ensureTableExists(AutoWorkflow) wf = db2.getRecord(AutoWorkflow, wfId) if wfId else None finally: db2.close() if not wf: raise HTTPException(status_code=404, detail=routeApiMsg("Workflow not found")) _validateWorkflowAccess(context, wf, "execute") mandateId = wf.get("mandateId") instanceId = wf.get("featureInstanceId") or "" iface = _getWorkflowAutomationInterface(context.user, mandateId, instanceId) taskRecord = iface.getTask(taskId) if not taskRecord: raise HTTPException(status_code=404, detail=routeApiMsg("Task not found")) runId = taskRecord.get("runId") result = body.get("result") if result is None: raise HTTPException(status_code=400, detail=routeApiMsg("result required")) run = iface.getRun(runId) if not run: raise HTTPException(status_code=404, detail=routeApiMsg("Run not found")) if taskRecord.get("status") != "pending": raise HTTPException(status_code=400, detail=routeApiMsg("Task already completed")) iface.updateTask(taskId, status="completed", result=result) taskNodeId = taskRecord.get("nodeId") nodeOutputs = dict(run.get("nodeOutputs") or {}) nodeOutputs[taskNodeId] = result workflowId = run.get("workflowId") wfForGraph = iface.getWorkflow(workflowId) if workflowId else None if not wfForGraph or not wfForGraph.get("graph"): raise HTTPException(status_code=400, detail=routeApiMsg("Workflow graph not found")) graph = wfForGraph["graph"] services = _getWorkflowAutomationServices(context.user, mandateId=mandateId, featureInstanceId=instanceId) return await executeGraph( graph=graph, services=services, workflowId=workflowId, instanceId=instanceId, userId=str(context.user.id) if context.user else None, mandateId=mandateId, automation2_interface=iface, initialNodeOutputs=nodeOutputs, startAfterNodeId=taskNodeId, runId=runId, ) @router.post("/tasks/{taskId}/cancel") @limiter.limit("30/minute") async def _cancelTask( request: Request, taskId: str = Path(..., description="Human task ID"), context: RequestContext = Depends(getRequestContext), ) -> dict: """Cancel a pending human task and stop the workflow run behind it.""" from modules.workflowAutomation.engine.executionEngine import requestRunStop from modules.interfaces.interfaceWorkflowAutomation import _getWorkflowAutomationInterface db = _getWorkflowAutomationDb() try: db._ensureTableExists(AutoTask) task = db.getRecord(AutoTask, taskId) finally: db.close() if not task: raise HTTPException(status_code=404, detail=routeApiMsg("Task not found")) wfId = task.get("workflowId") db2 = _getWorkflowAutomationDb() try: db2._ensureTableExists(AutoWorkflow) wf = db2.getRecord(AutoWorkflow, wfId) if wfId else None finally: db2.close() if not wf: raise HTTPException(status_code=404, detail=routeApiMsg("Task not found")) _validateWorkflowAccess(context, wf, "execute") mandateId = wf.get("mandateId") instanceId = wf.get("featureInstanceId") or "" iface = _getWorkflowAutomationInterface(context.user, mandateId, instanceId) taskRecord = iface.getTask(taskId) if not taskRecord: raise HTTPException(status_code=404, detail=routeApiMsg("Task not found")) if taskRecord.get("status") != "pending": raise HTTPException(status_code=400, detail=routeApiMsg("Task already completed")) runId = taskRecord.get("runId") if runId: requestRunStop(runId) dbRun = iface.getRun(runId) if dbRun: current = dbRun.get("status") or "" if current not in ("completed", "failed", "cancelled"): iface.updateRun(runId, status="cancelled") pending = iface.getTasks(runId=runId, status="pending") for t in pending: tid = t.get("id") if tid: iface.updateTask(tid, status="cancelled") else: iface.updateTask(taskId, status="cancelled") return {"success": True, "runId": runId, "taskId": taskId}